Skip to content

Runtime

runtime

Runtime package for deux device sessions and events.

AsyncHandler module-attribute 

AsyncHandler = Callable[..., Coroutine[Any, Any, None]]

Type alias for an async callback that accepts any arguments and returns None.

DeckEvent module-attribute 

DeckEvent = KeyEvent | EncoderTurnEvent | EncoderPressEvent | TouchEvent

Union type of all events that can be received from a Stream Deck device.

AsyncEvent

A multicast async event that can have multiple subscribers.

Subscribers are async callables registered via :meth:subscribe or by using the event itself as a decorator. :meth:emit invokes every registered subscriber sequentially, awaiting each one in registration order.

The handler list is snapshotted at the start of every :meth:emit, so subscribers may safely register or unregister during dispatch without affecting the in-flight emission.

The signature of an event is part of its documented contract, not its static type — events here are intentionally non-generic so they can carry arbitrary positional/keyword payloads.

Examples:

::

on_volume_changed = AsyncEvent()

@on_volume_changed
async def _log(value: int) -> None:
    print(f"volume = {value}")

await on_volume_changed.emit(75)
Source code in src/deux/runtime/async_event.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
class AsyncEvent:
    """A multicast async event that can have multiple subscribers.

    Subscribers are async callables registered via :meth:`subscribe`
    or by using the event itself as a decorator.  :meth:`emit` invokes
    every registered subscriber sequentially, awaiting each one in
    registration order.

    The handler list is snapshotted at the start of every :meth:`emit`,
    so subscribers may safely register or unregister during dispatch
    without affecting the in-flight emission.

    The signature of an event is part of its documented contract, not
    its static type — events here are intentionally non-generic so
    they can carry arbitrary positional/keyword payloads.

    Examples
    --------
    ::

        on_volume_changed = AsyncEvent()

        @on_volume_changed
        async def _log(value: int) -> None:
            print(f"volume = {value}")

        await on_volume_changed.emit(75)
    """

    _UNSET: object = object()

    __slots__ = ("_handlers", "_last_args", "_last_kwargs")

    def __init__(self) -> None:
        self._handlers: list[AsyncHandler] = []
        self._last_args: tuple[Any, ...] | object = AsyncEvent._UNSET
        self._last_kwargs: dict[str, Any] = {}

    @property
    def has_value(self) -> bool:
        """Whether :meth:`emit` has been called at least once.

        Returns
        -------
        bool
            ``True`` after the first :meth:`emit`, ``False`` otherwise.
        """
        return self._last_args is not AsyncEvent._UNSET

    @property
    def last_args(self) -> tuple[Any, ...]:
        """Positional arguments from the most recent :meth:`emit`.

        Returns
        -------
        tuple
            The positional args from the last emission.

        Raises
        ------
        LookupError
            If :meth:`emit` has never been called.
        """
        if self._last_args is AsyncEvent._UNSET:
            raise LookupError("No value has been emitted yet")
        return self._last_args  # type: ignore[return-value]

    @property
    def last_kwargs(self) -> dict[str, Any]:
        """Keyword arguments from the most recent :meth:`emit`.

        Returns
        -------
        dict
            The keyword args from the last emission.

        Raises
        ------
        LookupError
            If :meth:`emit` has never been called.
        """
        if self._last_args is AsyncEvent._UNSET:
            raise LookupError("No value has been emitted yet")
        return self._last_kwargs

    def subscribe(self, handler: _H) -> _H:
        """Register *handler* as a subscriber.

        Parameters
        ----------
        handler
            Async callable to invoke on every :meth:`emit`.

        Returns
        -------
        handler
            The original *handler*, unchanged, so this can be used as
            a decorator.
        """
        self._handlers.append(handler)
        return handler

    def unsubscribe(self, handler: AsyncHandler) -> None:
        """Remove *handler* from the subscriber list.

        Parameters
        ----------
        handler
            A previously-registered subscriber.

        Raises
        ------
        ValueError
            If *handler* is not currently subscribed.
        """
        self._handlers.remove(handler)

    def __call__(self, handler: _H) -> _H:
        """Decorator alias for :meth:`subscribe`.

        Parameters
        ----------
        handler
            Async callable to register.

        Returns
        -------
        handler
            The original handler.
        """
        return self.subscribe(handler)

    async def emit(self, *args: Any, **kwargs: Any) -> None:
        """Invoke every subscriber sequentially, awaiting each in turn.

        A snapshot of the handler list is taken first, so handlers may
        subscribe or unsubscribe during dispatch without affecting the
        current emission.

        Parameters
        ----------
        *args
            Positional arguments forwarded to every handler.
        **kwargs
            Keyword arguments forwarded to every handler.
        """
        self._last_args = args
        self._last_kwargs = kwargs
        for handler in list(self._handlers):
            await handler(*args, **kwargs)

    @property
    def subscriber_count(self) -> int:
        """Number of currently-registered subscribers."""
        return len(self._handlers)

has_value property 

has_value: bool

Whether :meth:emit has been called at least once.

Returns:

Type Description
bool

True after the first :meth:emit, False otherwise.

last_args property 

last_args: tuple[Any, ...]

Positional arguments from the most recent :meth:emit.

Returns:

Type Description
tuple

The positional args from the last emission.

Raises:

Type Description
LookupError

If :meth:emit has never been called.

last_kwargs property 

last_kwargs: dict[str, Any]

Keyword arguments from the most recent :meth:emit.

Returns:

Type Description
dict

The keyword args from the last emission.

Raises:

Type Description
LookupError

If :meth:emit has never been called.

subscriber_count property 

subscriber_count: int

Number of currently-registered subscribers.

subscribe 

subscribe(handler: _H) -> _H

Register handler as a subscriber.

Parameters:

Name Type Description Default
handler _H

Async callable to invoke on every :meth:emit.

 required

Returns:

Type Description
handler

The original handler, unchanged, so this can be used as a decorator.
Source code in src/deux/runtime/async_event.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def subscribe(self, handler: _H) -> _H:
    """Register *handler* as a subscriber.

    Parameters
    ----------
    handler
        Async callable to invoke on every :meth:`emit`.

    Returns
    -------
    handler
        The original *handler*, unchanged, so this can be used as
        a decorator.
    """
    self._handlers.append(handler)
    return handler

unsubscribe 

unsubscribe(handler: AsyncHandler) -> None

Remove handler from the subscriber list.

Parameters:

Name Type Description Default
handler AsyncHandler

A previously-registered subscriber.

 required

Raises:

Type Description
ValueError

If handler is not currently subscribed.
Source code in src/deux/runtime/async_event.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def unsubscribe(self, handler: AsyncHandler) -> None:
    """Remove *handler* from the subscriber list.

    Parameters
    ----------
    handler
        A previously-registered subscriber.

    Raises
    ------
    ValueError
        If *handler* is not currently subscribed.
    """
    self._handlers.remove(handler)

__call__ 

__call__(handler: _H) -> _H

Decorator alias for :meth:subscribe.

Parameters:

Name Type Description Default
handler _H

Async callable to register.

 required

Returns:

Type Description
handler

The original handler.
Source code in src/deux/runtime/async_event.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def __call__(self, handler: _H) -> _H:
    """Decorator alias for :meth:`subscribe`.

    Parameters
    ----------
    handler
        Async callable to register.

    Returns
    -------
    handler
        The original handler.
    """
    return self.subscribe(handler)

emit async 

emit(*args: Any, **kwargs: Any) -> None

Invoke every subscriber sequentially, awaiting each in turn.

A snapshot of the handler list is taken first, so handlers may subscribe or unsubscribe during dispatch without affecting the current emission.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded to every handler.

 ()
**kwargs Any

Keyword arguments forwarded to every handler.

 {}
Source code in src/deux/runtime/async_event.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def emit(self, *args: Any, **kwargs: Any) -> None:
    """Invoke every subscriber sequentially, awaiting each in turn.

    A snapshot of the handler list is taken first, so handlers may
    subscribe or unsubscribe during dispatch without affecting the
    current emission.

    Parameters
    ----------
    *args
        Positional arguments forwarded to every handler.
    **kwargs
        Keyword arguments forwarded to every handler.
    """
    self._last_args = args
    self._last_kwargs = kwargs
    for handler in list(self._handlers):
        await handler(*args, **kwargs)

DeviceCapabilities dataclass

Immutable snapshot of a Stream Deck device's hardware capabilities.

Constructed from a connected device via :meth:from_device, this dataclass captures every property needed to drive layout, rendering, and event routing without hardcoded constants.

Source code in src/deux/runtime/capabilities.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@dataclass(frozen=True, slots=True)
class DeviceCapabilities:
    """Immutable snapshot of a Stream Deck device's hardware capabilities.

    Constructed from a connected device via :meth:`from_device`, this
    dataclass captures every property needed to drive layout, rendering,
    and event routing without hardcoded constants.
    """

    vendor_id: int
    product_id: int
    deck_type: str
    key_count: int
    key_cols: int
    key_rows: int
    key_pixel_width: int
    key_pixel_height: int
    key_image_format: str
    key_flip: tuple[bool, bool]
    key_rotation: int
    has_visual: bool
    has_touch: bool
    dial_count: int
    touchscreen_width: int
    touchscreen_height: int
    touchscreen_image_format: str
    touchscreen_flip: tuple[bool, bool]
    touchscreen_rotation: int
    has_screen: bool
    screen_width: int
    screen_height: int
    screen_image_format: str
    screen_flip: tuple[bool, bool]
    screen_rotation: int
    touch_key_count: int
    lcd_width: int = 0
    lcd_height: int = 0

    @classmethod
    def from_device(cls, device: HidDevice) -> DeviceCapabilities:
        """Build capabilities from a connected (opened) HID device.

        Reads hardware information from the device's ``Get Unit Information``
        feature report and derives all capabilities from the product ID and
        self-reported geometry.

        Parameters
        ----------
        device : HidDevice
            An open :class:`~deux.runtime.hid.device.HidDevice`.

        Returns
        -------
        DeviceCapabilities
            A frozen :class:`DeviceCapabilities` instance.
        """
        key_layout = device.key_layout  # (cols, rows)
        key_w, key_h = device.key_size
        lcd_w, lcd_h = device.lcd_size
        win_w, win_h = device.window_size
        rotation = device.rotation.value

        return cls(
            vendor_id=device.vendor_id,
            product_id=device.product_id,
            deck_type=device.family,
            key_count=device.key_count,
            key_cols=key_layout[0],
            key_rows=key_layout[1],
            key_pixel_width=key_w,
            key_pixel_height=key_h,
            key_image_format="JPEG",
            key_flip=(False, False),
            key_rotation=rotation,
            has_visual=True,
            has_touch=device.has_touch,
            dial_count=device.encoder_count,
            touchscreen_width=win_w if device.has_touch else 0,
            touchscreen_height=win_h if device.has_touch else 0,
            touchscreen_image_format="JPEG" if device.has_touch else "",
            touchscreen_flip=(False, False),
            touchscreen_rotation=0,
            has_screen=device.has_window and not device.has_touch,
            screen_width=win_w if (device.has_window and not device.has_touch) else 0,
            screen_height=win_h if (device.has_window and not device.has_touch) else 0,
            screen_image_format="JPEG" if (device.has_window and not device.has_touch) else "",
            screen_flip=(False, False),
            screen_rotation=rotation if (device.has_window and not device.has_touch) else 0,
            touch_key_count=device.sensor_count,
            lcd_width=lcd_w,
            lcd_height=lcd_h,
        )

    @property
    def key_size(self) -> tuple[int, int]:
        """Key image dimensions as ``(width, height)``."""
        return (self.key_pixel_width, self.key_pixel_height)

    @property
    def has_encoders(self) -> bool:
        """Whether the device has rotary encoders (dials)."""
        return self.dial_count > 0

    @property
    def has_touchscreen(self) -> bool:
        """Whether the device has a touchscreen strip."""
        return self.touchscreen_width > 0 and self.touchscreen_height > 0

    @property
    def has_info_screen(self) -> bool:
        """Whether the device has a non-touch info screen (e.g. Neo)."""
        return self.has_screen

    @property
    def panel_count(self) -> int:
        """Number of touchscreen card zones (equals dial count)."""
        return self.dial_count

key_size property 

key_size: tuple[int, int]

Key image dimensions as (width, height).

has_encoders property 

has_encoders: bool

Whether the device has rotary encoders (dials).

has_touchscreen property 

has_touchscreen: bool

Whether the device has a touchscreen strip.

has_info_screen property 

has_info_screen: bool

Whether the device has a non-touch info screen (e.g. Neo).

panel_count property 

panel_count: int

Number of touchscreen card zones (equals dial count).

from_device classmethod 

from_device(device: HidDevice) -> DeviceCapabilities

Build capabilities from a connected (opened) HID device.

Reads hardware information from the device's Get Unit Information feature report and derives all capabilities from the product ID and self-reported geometry.

Parameters:

Name Type Description Default
device HidDevice

An open :class:~deux.runtime.hid.device.HidDevice.

 required

Returns:

Type Description
DeviceCapabilities

A frozen :class:DeviceCapabilities instance.
Source code in src/deux/runtime/capabilities.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
def from_device(cls, device: HidDevice) -> DeviceCapabilities:
    """Build capabilities from a connected (opened) HID device.

    Reads hardware information from the device's ``Get Unit Information``
    feature report and derives all capabilities from the product ID and
    self-reported geometry.

    Parameters
    ----------
    device : HidDevice
        An open :class:`~deux.runtime.hid.device.HidDevice`.

    Returns
    -------
    DeviceCapabilities
        A frozen :class:`DeviceCapabilities` instance.
    """
    key_layout = device.key_layout  # (cols, rows)
    key_w, key_h = device.key_size
    lcd_w, lcd_h = device.lcd_size
    win_w, win_h = device.window_size
    rotation = device.rotation.value

    return cls(
        vendor_id=device.vendor_id,
        product_id=device.product_id,
        deck_type=device.family,
        key_count=device.key_count,
        key_cols=key_layout[0],
        key_rows=key_layout[1],
        key_pixel_width=key_w,
        key_pixel_height=key_h,
        key_image_format="JPEG",
        key_flip=(False, False),
        key_rotation=rotation,
        has_visual=True,
        has_touch=device.has_touch,
        dial_count=device.encoder_count,
        touchscreen_width=win_w if device.has_touch else 0,
        touchscreen_height=win_h if device.has_touch else 0,
        touchscreen_image_format="JPEG" if device.has_touch else "",
        touchscreen_flip=(False, False),
        touchscreen_rotation=0,
        has_screen=device.has_window and not device.has_touch,
        screen_width=win_w if (device.has_window and not device.has_touch) else 0,
        screen_height=win_h if (device.has_window and not device.has_touch) else 0,
        screen_image_format="JPEG" if (device.has_window and not device.has_touch) else "",
        screen_flip=(False, False),
        screen_rotation=rotation if (device.has_window and not device.has_touch) else 0,
        touch_key_count=device.sensor_count,
        lcd_width=lcd_w,
        lcd_height=lcd_h,
    )

Deck

Per-device handle for an Elgato Stream Deck.

Instances are created and managed by :class:DeckManager. Do not instantiate Deck directly — use DeckManager.on_connect to receive connected Deck instances.

The Deck object provides the per-device API for screens, keys, encoders, touchscreen cards, brightness, and rendering.

Attributes:

Name Type Description
on_brightness_changed AsyncEvent

Fires after :meth:set_brightness confirms a change to a new value (the hardware push has returned). Subscribers receive the new brightness percentage as an int in [0, 100]. Idempotent calls (same value) do not emit.
on_screen_changed AsyncEvent

Fires after :meth:set_screen finishes rendering a new active screen. Subscribers receive the new screen name (str). Idempotent calls (same screen) do not emit.
Source code in src/deux/runtime/deck.py 
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78
  79
  80
  81
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94
  95
  96
  97
  98
  99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
class Deck:
    """Per-device handle for an Elgato Stream Deck.

    Instances are created and managed by :class:`DeckManager`.  Do not
    instantiate ``Deck`` directly — use ``DeckManager.on_connect`` to
    receive connected ``Deck`` instances.

    The ``Deck`` object provides the per-device API for screens, keys,
    encoders, touchscreen cards, brightness, and rendering.

    Attributes
    ----------
    on_brightness_changed : AsyncEvent
        Fires after :meth:`set_brightness` confirms a change to a *new*
        value (the hardware push has returned).  Subscribers receive
        the new brightness percentage as an ``int`` in ``[0, 100]``.
        Idempotent calls (same value) do not emit.
    on_screen_changed : AsyncEvent
        Fires after :meth:`set_screen` finishes rendering a new active
        screen.  Subscribers receive the new screen name (``str``).
        Idempotent calls (same screen) do not emit.
    """

    def __init__(
        self,
        serial_number: str,
        brightness: int = 80,
    ) -> None:
        """Construct a deck handle for the given serial number.

        Instances are normally created by :class:`DeckManager` in
        response to a device-connect event; application code receives
        them via ``on_connect`` handlers.  For unit tests that need a
        deck without HID I/O, use :meth:`Deck.for_testing`.

        Parameters
        ----------
        serial_number : str
            Serial number of the target device.  Used during
            :meth:`start` to locate the matching HID device.
        brightness : int, default=80
            Initial brightness percentage in ``[0, 100]`` applied once
            the device is opened.

        Notes
        -----
        Construction performs no I/O.  The HID device is opened and the
        event loop started by :meth:`start`.  The
        :attr:`on_brightness_changed` and :attr:`on_screen_changed`
        events are wired up here and ready for subscription before
        :meth:`start` is called.
        """
        self._serial_number = serial_number
        self._brightness = brightness
        self._device: HidDevice | None = None
        self._caps: DeviceCapabilities | None = None
        self._metrics: RenderMetrics | None = None
        self._transport: AsyncTransport | None = None
        self._event_task: asyncio.Task[None] | None = None
        self._timeout_task: asyncio.Task[None] | None = None
        self._timeout_event: asyncio.Event = asyncio.Event()
        self._closed_event = asyncio.Event()
        self._running = False
        self._device_lock = asyncio.Lock()

        self._screens: dict[str, Screen] = {}
        self._active_screen: Screen | None = None
        self._theme: Theme | None = None

        # Batched-render gate.  When ``_batch_render_depth > 0`` the deck
        # is inside a full-screen operation (initial render, screen
        # switch, theme change) and incoming ``refresh()`` calls are
        # deferred so partial per-key writes cannot land on the LCD
        # between the old frame and the new one.  ``_refresh_pending``
        # records that at least one refresh was suppressed so a single
        # drain refresh can fire when the gate reopens.  Both fields
        # are reset by :meth:`stop` so reconnects start clean.
        self._batch_render_depth: int = 0
        self._refresh_pending: bool = False

        # Splash hold deadline.  When :meth:`show_full_screen_image` is
        # called with ``min_display_ms > 0``, this records the
        # ``perf_counter()`` timestamp before which the next batched
        # render's *push* phase must not begin.  Render work runs in
        # parallel with the splash; only the final device push is
        # gated.  Cleared after the first deadline-await consumes it,
        # so the hold applies exactly once per splash.
        self._splash_push_deadline: float | None = None

        self.on_brightness_changed = AsyncEvent()
        self.on_screen_changed = AsyncEvent()

        self._renderer = DeckRenderer(self)
        self._event_router = DeckEventRouter(self)

    @classmethod
    def for_testing(
        cls,
        capabilities: DeviceCapabilities,
        *,
        serial_number: str = "TEST",
        brightness: int = 80,
    ) -> Deck:
        """Construct a :class:`Deck` pre-seeded with capabilities for tests.

        Real construction goes through :meth:`start`, which discovers a
        physical device and derives :attr:`capabilities` and
        :attr:`metrics` from it.  Tests that exercise behaviour above
        the device layer need those two attributes populated without
        any HID I/O.

        This helper provides the supported way to do so, so that tests
        do not have to assign to the private ``_caps`` / ``_metrics``
        attributes directly.  No device is opened and no transport is
        started — :attr:`is_connected` remains ``False`` until the
        normal :meth:`start` path is invoked.

        Parameters
        ----------
        capabilities : DeviceCapabilities
            The capabilities to seed onto the deck.  Drives the
            derived :class:`~deux.render.metrics.RenderMetrics` and
            anything else that reads :attr:`capabilities`.
        serial_number : str, default="TEST"
            Serial number recorded on the instance.  Does not need to
            correspond to a real device.
        brightness : int, default=80
            Initial brightness (0-100).

        Returns
        -------
        Deck
            A deck instance with :attr:`capabilities` and
            :attr:`metrics` populated.

        Notes
        -----
        This constructor is intended for unit tests.  Production code
        should use the normal :class:`Deck` constructor and
        :meth:`start`.
        """
        deck = cls(serial_number=serial_number, brightness=brightness)
        deck._caps = capabilities
        deck._metrics = RenderMetrics(capabilities)
        return deck

    async def start(self) -> None:
        """Discover the device by serial, open it, and start the event loop."""
        if self._running:
            return

        loop = asyncio.get_running_loop()

        devices = await loop.run_in_executor(get_executor(), enumerate_devices)

        if not devices:
            raise DeckError("No Stream Deck devices found")

        target: HidDevice | None = None
        for d in devices:
            try:
                await loop.run_in_executor(get_executor(), d.open)
                if d.serial_number == self._serial_number:
                    target = d
                    break
                await loop.run_in_executor(get_executor(), d.close)
            except HidApiError:
                continue
        if target is None:
            raise DeckError(
                f"No device with serial '{self._serial_number}' found"
            )
        self._device = target

        await self._exec_device_io(self._device.show_logo)
        await self._exec_device_io(self._device.set_brightness, self._brightness)

        self._caps = DeviceCapabilities.from_device(self._device)
        self._metrics = RenderMetrics(self._caps)

        logger.info(
            "Opened %s (serial: %s, firmware: %s, keys: %d, dials: %d)",
            self._device.family,
            self._device.serial_number,
            self._device.firmware_version,
            self._caps.key_count,
            self._caps.dial_count,
        )

        self._transport = AsyncTransport(self._device, self._caps)
        self._transport.start()

        self._running = True
        self._closed_event.clear()

        # Publish the initial brightness so that any later bind_range()
        # call can seed from the event's last_value without waiting for
        # an explicit set_brightness().
        await self.on_brightness_changed.emit(self._brightness)

        self._event_task = asyncio.create_task(
            self._event_loop(), name="deux-events"
        )

    async def stop(self) -> None:
        """Stop the event loop and close the device."""
        if not self._running:
            return

        self._running = False

        await self._stop_all_spinners()

        self._detach_all_cards()

        if self._transport:
            self._transport.stop()

        if self._event_task and not self._event_task.done():
            self._event_task.cancel()
            with suppress(asyncio.CancelledError):
                await self._event_task

        if self._device:
            if self._device.is_open:
                try:
                    await self._exec_device_io(self._device.show_logo)
                    await self._exec_device_io(self._device.close)
                except Exception as e:
                    # Intentional catch-all: shutdown must never raise.
                    # Covers HidWriteTimeout/HidApiError plus any transport,
                    # OS, or backend-specific errors surfaced during close.
                    logger.warning("Error closing device: %s", e)
            else:
                logger.debug("Device already closed; skipping show_logo/close")

        self._device = None
        self._closed_event.set()
        self._batch_render_depth = 0
        self._refresh_pending = False
        self._splash_push_deadline = None
        logger.info("Deck stopped")

    async def _stop_all_spinners(self) -> None:
        """Stop spinner animations on every DuiKey and DuiCard.

        Called from :meth:`stop` before the device is closed so that
        no in-flight spinner frames are pushed to a half-torn-down
        device.  Failures to stop an individual spinner are logged
        and swallowed — shutdown must never raise.
        """
        # Inline import: dui.card transitively imports runtime.events, which
        # triggers runtime package init while this module is still loading.
        from ..dui.card import DuiCard  # noqa: PLC0415

        for screen in self._screens.values():
            for key_slot in screen.keys.values():
                if isinstance(key_slot, DuiKey) and key_slot.is_animating:
                    try:
                        await key_slot.finish_busy()
                    except Exception:
                        logger.exception(
                            "Error stopping spinner on key %r", key_slot
                        )
            if screen.touch_strip is None:
                continue
            for card in screen.touch_strip.cards:
                if isinstance(card, DuiCard) and card.is_animating:
                    try:
                        await card.finish_busy()
                    except Exception:
                        logger.exception(
                            "Error stopping spinner on card %r", card
                        )

    def _detach_all_cards(self) -> None:
        """Unsubscribe deck-owned AsyncEvent handlers on every DuiCard across all screens.

        Only removes bindings to this deck's own events (e.g.
        ``on_brightness_changed``, ``on_screen_changed``), preserving
        service-owned bindings established in controller ``__init__``.
        This prevents handler accumulation across reconnect cycles
        without breaking reactive bindings to external services.
        """
        # Inline import: dui.card transitively imports runtime.events, which
        # triggers runtime package init while this module is still loading.
        from ..dui.card import DuiCard  # noqa: PLC0415

        deck_events = (self.on_brightness_changed, self.on_screen_changed)
        for screen in self._screens.values():
            if screen.touch_strip is None:
                continue
            for card in screen.touch_strip.cards:
                if isinstance(card, DuiCard):
                    card.detach_events(*deck_events)

    async def wait_closed(self) -> None:
        """Block until the deck is closed (e.g. by stop() or disconnect)."""
        await self._closed_event.wait()

    async def _exec_device_io(self, func: Any, *args: Any) -> None:
        """Run a device I/O call in the executor with a timeout.

        Parameters
        ----------
        func
            The blocking HID function to call (e.g.
            ``self._device.set_key_image``).
        *args
            Positional arguments forwarded to *func*.

        Raises
        ------
        HidWriteTimeout
            If the call does not complete within
            :data:`_HID_WRITE_TIMEOUT` seconds.
        """
        loop = asyncio.get_running_loop()
        try:
            await asyncio.wait_for(
                loop.run_in_executor(get_executor(), func, *args),
                timeout=_HID_WRITE_TIMEOUT,
            )
        except TimeoutError:
            logger.error(
                "HID write timed out after %.1fs — device may be disconnected",
                _HID_WRITE_TIMEOUT,
            )
            raise HidWriteTimeout(
                f"HID write to {func!r} timed out after {_HID_WRITE_TIMEOUT}s"
            ) from None

    @property
    def device_path(self) -> str | None:
        """HID device path, or ``None`` if not opened."""
        if self._device is None:
            return None
        return self._device.path.decode("utf-8", errors="replace")

    @property
    def is_connected(self) -> bool:
        """Whether the device is currently connected and operational."""
        return self._device is not None and self._running

    @property
    def capabilities(self) -> DeviceCapabilities:
        """The device capabilities for the connected device.

        Raises
        ------
        DeckError
            If the device is not opened.
        """
        if self._caps is None:
            raise DeckError("Device not opened")
        return self._caps

    @property
    def metrics(self) -> RenderMetrics:
        """Rendering metrics for the connected device.

        Raises
        ------
        DeckError
            If the device is not opened.
        """
        if self._metrics is None:
            raise DeckError("Device not opened")
        return self._metrics

    @property
    def info(self) -> DeviceInfo:
        """Get device information."""
        if not self._device:
            raise DeckError("Device not opened")
        caps = self.capabilities
        return DeviceInfo(
            deck_type=caps.deck_type,
            serial=self._device.serial_number,
            firmware=self._device.firmware_version,
            key_count=caps.key_count,
            key_layout=(caps.key_cols, caps.key_rows),
            encoder_count=caps.dial_count,
            key_pixel_size=caps.key_size,
            touchscreen_size=(caps.touchscreen_width, caps.touchscreen_height),
            key_image_format=caps.key_image_format,
        )

    @property
    def brightness(self) -> int:
        """Current brightness level (0-100)."""
        return self._brightness

    @property
    def theme(self) -> Theme | None:
        """Per-deck theme override, or ``None`` to inherit the system theme.

        When set, this theme is used for all screens on this deck
        unless a screen has its own :attr:`~deux.Screen.theme`
        override.  Set to ``None`` to fall back to the system-wide
        theme.
        """
        return self._theme

    @theme.setter
    def theme(self, value: Theme | None) -> None:
        self._theme = value

    async def set_theme(self, theme: Theme | None) -> None:
        """Apply a new deck-level theme and re-render the active screen.

        Sets the deck theme, applies the CSS cascade to all renderers,
        marks every control dirty, and performs a complete re-render
        (icon prefetch, render all, push all) so the display updates
        atomically.

        Parameters
        ----------
        theme : Theme or None
            The theme to apply, or ``None`` to revert to the system
            theme.
        """
        self._theme = theme
        if self._active_screen is None:
            return

        async with self._batched_render():
            self._renderer.apply_theme()
            self._active_screen.mark_all_dirty()
            await self._renderer.render_screen_complete()

    async def preload_icons(self) -> None:
        """Prefetch Iconify icons for all registered screens.

        Collects every icon identifier across all screens and fetches
        them concurrently, warming the in-memory and disk caches.
        Call this after all screens have been set up (keys and cards
        installed) to avoid network latency on first render.
        """
        # Inline import: tests patch ``deux.dui.iconify.prefetch_icons``;
        # importing it lazily keeps that patching point effective.
        from ..dui.iconify import prefetch_icons  # noqa: PLC0415

        all_icons: set[str] = set()
        for screen in self._screens.values():
            all_icons.update(screen.collect_all_icons())
        if all_icons:
            await prefetch_icons(all_icons)

    def resolve_stylesheet(self) -> str:
        """Resolve the effective CSS stylesheet for the active screen.

        The cascade is: screen theme > deck theme > system theme.

        Returns
        -------
        str
            CSS stylesheet string from the most specific theme.
        """
        screen = self._active_screen
        if screen is not None and screen.theme is not None:
            return screen.theme.css
        if self._theme is not None:
            return self._theme.css
        return get_active_theme().css

    # Backwards-compatible alias preserved for callers that used the
    # private name prior to the encapsulation refactor.  Prefer the
    # public :meth:`resolve_stylesheet` for new code.
    _resolve_stylesheet = resolve_stylesheet

    async def set_brightness(self, percent: int) -> None:
        """Set screen brightness.

        Pushes the value to the hardware (if connected) and emits
        :attr:`on_brightness_changed` with the clamped result.  If the
        clamped value equals the current brightness, no event fires —
        observers see only confirmed transitions.

        Parameters
        ----------
        percent
            Brightness level (0-100).  Values outside the range are
            clamped.
        """
        if not isinstance(percent, int):
            raise TypeError(
                f"percent must be an int, got {type(percent).__name__}"
            )
        clamped = max(0, min(100, percent))
        if clamped == self._brightness:
            return
        if self._device is not None:
            async with self._device_lock:
                await self._exec_device_io(self._device.set_brightness, clamped)
        self._brightness = clamped
        await self.on_brightness_changed.emit(clamped)

    # -- full-screen ("splash") image --------------------------------------

    async def show_full_screen_image(
        self,
        image: Image.Image | str | Path | bytes,
        *,
        fit: SplashFitMode = "cover",
        background: tuple[int, int, int] = (0, 0, 0),
        jpeg_quality: int = 90,
        min_display_ms: int = 0,
    ) -> None:
        """Upload an image covering the entire LCD (HID command ``0x08``).

        The image is loaded (PIL image, file path, raw image bytes, or
        SVG), resized to the device's logical LCD size using *fit*,
        rotated into the device's transmit orientation, JPEG-encoded,
        and pushed via
        :meth:`HidDevice.set_full_screen_image`.

        **Important behavioural contract.**  This is a one-shot,
        whole-LCD blit.  Any subsequent per-key (``set_key_image``),
        per-window (``set_partial_window_image``), or per-screen render
        will paint over the image.  In particular, calling
        :meth:`set_screen` after :meth:`show_full_screen_image` will
        immediately clobber the image as the renderer pushes the new
        screen.

        Intended use cases are startup splashes, loading screens, and
        lock screens.  For persistent backgrounds, use the per-screen
        background layer in :class:`Screen` instead.

        Parameters
        ----------
        image : Image.Image, str, Path, or bytes
            Source image.  ``.svg`` files and SVG byte streams are
            rasterised at the target LCD size; other formats are
            decoded with Pillow.
        fit : {"cover", "contain", "stretch"}, default="cover"
            Resize strategy.  ``"cover"`` scales to fill and crops the
            overflow; ``"contain"`` letterboxes with *background*;
            ``"stretch"`` ignores aspect ratio.
        background : tuple[int, int, int], default=(0, 0, 0)
            RGB background colour used to letterbox under
            ``fit="contain"``.
        jpeg_quality : int, default=90
            JPEG encoding quality (1-95).
        min_display_ms : int, default=0
            Minimum time, in milliseconds, that this image must remain
            visible on the LCD before the *push phase* of the next
            batched render (``set_screen`` / ``set_theme``) may begin.
            The render's CPU work still runs in parallel with the
            splash; only the final device push is delayed.  Effective
            total splash time is therefore
            ``max(min_display_ms, render_time)`` -- a fast render waits
            for the remainder of the requested display time, while a
            slow render is never artificially delayed.  Default of
            ``0`` disables the hold entirely.  The deadline is one-shot
            and is consumed by the first batched push that follows.

        Raises
        ------
        DeckError
            If the device is not opened, or its PID has no known
            logical LCD size.
        SplashError
            If image preparation fails.

        Examples
        --------
        Hold a splash for at least 500 ms so the user can perceive it
        even when the first ``set_screen`` is very fast::

            await deck.show_splash("boot.png", min_display_ms=500)
            await deck.set_screen("home")  # push delayed until 500ms elapsed
        """
        if self._device is None:
            raise DeckError("Device not opened")
        logical_size = self._device.logical_lcd_size
        if logical_size == (0, 0):
            raise DeckError(
                f"Device PID 0x{self._device.product_id:04X} has no "
                "known LCD size; cannot prepare full-screen image"
            )
        rotation = self._device.rotation

        # Inline import: splash transitively pulls in the resvg backend
        # for SVG inputs.  Keeping the import local avoids paying that
        # cost on deck construction for callers that never use this API.
        from .splash import prepare_full_screen_jpeg  # noqa: PLC0415

        loop = asyncio.get_running_loop()
        jpeg_bytes = await loop.run_in_executor(
            get_executor(),
            lambda: prepare_full_screen_jpeg(
                image,
                logical_size=logical_size,
                rotation=rotation,
                fit=fit,
                background=background,
                jpeg_quality=jpeg_quality,
            ),
        )

        async with self._device_lock:
            await self._exec_device_io(
                self._device.set_full_screen_image, jpeg_bytes
            )

        if min_display_ms > 0:
            self._splash_push_deadline = (
                time.perf_counter() + min_display_ms / 1000.0
            )
        else:
            self._splash_push_deadline = None

    async def show_splash(
        self,
        image: Image.Image | str | Path | bytes,
        *,
        fit: SplashFitMode = "cover",
        background: tuple[int, int, int] = (0, 0, 0),
        jpeg_quality: int = 90,
        min_display_ms: int = 0,
    ) -> None:
        """Alias for :meth:`show_full_screen_image`, intended for startup.

        Provides a semantically clearer entry point for the common case
        of displaying a boot/splash image before the first screen is
        rendered.

        Parameters
        ----------
        image : Image.Image, str, Path, or bytes
            See :meth:`show_full_screen_image`.
        fit : {"cover", "contain", "stretch"}, default="cover"
            See :meth:`show_full_screen_image`.
        background : tuple[int, int, int], default=(0, 0, 0)
            See :meth:`show_full_screen_image`.
        jpeg_quality : int, default=90
            See :meth:`show_full_screen_image`.
        min_display_ms : int, default=0
            See :meth:`show_full_screen_image`.

        See Also
        --------
        show_full_screen_image
        clear_full_screen_image
        """
        await self.show_full_screen_image(
            image,
            fit=fit,
            background=background,
            jpeg_quality=jpeg_quality,
            min_display_ms=min_display_ms,
        )

    async def clear_full_screen_image(
        self,
        color: tuple[int, int, int] = (0, 0, 0),
        *,
        jpeg_quality: int = 90,
    ) -> None:
        """Clear the LCD by uploading a solid-colour full-screen image.

        Uses the same HID command ``0x08`` path as
        :meth:`show_full_screen_image`, ensuring a deterministic
        clear across all supported deck families.  Behaviour-wise this
        is equivalent to ``show_full_screen_image`` with a solid-colour
        source; like that method, any subsequent per-key or per-window
        write will paint over the cleared LCD.

        Parameters
        ----------
        color : tuple[int, int, int], default=(0, 0, 0)
            RGB fill colour.
        jpeg_quality : int, default=90
            JPEG encoding quality (1-95).

        Raises
        ------
        DeckError
            If the device is not opened, or its PID has no known
            logical LCD size.
        """
        if self._device is None:
            raise DeckError("Device not opened")
        logical_size = self._device.logical_lcd_size
        if logical_size == (0, 0):
            raise DeckError(
                f"Device PID 0x{self._device.product_id:04X} has no "
                "known LCD size; cannot prepare full-screen image"
            )
        rotation = self._device.rotation

        from .splash import prepare_solid_color_jpeg  # noqa: PLC0415

        loop = asyncio.get_running_loop()
        jpeg_bytes = await loop.run_in_executor(
            get_executor(),
            lambda: prepare_solid_color_jpeg(
                color,
                logical_size=logical_size,
                rotation=rotation,
                jpeg_quality=jpeg_quality,
            ),
        )

        async with self._device_lock:
            await self._exec_device_io(
                self._device.set_full_screen_image, jpeg_bytes
            )

    def screen(self, name: str) -> Screen:
        """Get or create a screen by name.

        Parameters
        ----------
        name
            Screen name.

        Returns
        -------
        Screen
            The Screen instance.

        Raises
        ------
        DeckError
            If the device is not opened — capabilities are required to
            size the screen, and they are only known after :meth:`start`.
        """
        # Inline import: ui.screen transitively imports runtime.events via
        # ui.touch_strip -> ui.cards.base, creating a cycle at module load.
        from ..ui.screen import Screen  # noqa: PLC0415

        if self._caps is None:
            raise DeckError("Device not opened")

        if name not in self._screens:
            self._screens[name] = Screen(name, self._caps)
        return self._screens[name]

    async def set_screen(self, name: str) -> None:
        """Switch to a named screen, rendering all keys and cards.

        Wires up refresh callbacks on every key and card so that any
        handler or background task can call ``request_refresh()`` to
        trigger a re-render without needing a direct reference to the
        deck.  After the new screen has finished rendering, fires
        :attr:`on_screen_changed` with the new name.  No event fires
        if the requested screen is already active.

        Parameters
        ----------
        name
            Screen name (must already exist via ``deck.screen(name)``).

        Raises
        ------
        DeckError
            If *name* does not match a previously-created screen.
        """
        if name not in self._screens:
            raise DeckError(f"Screen '{name}' does not exist")

        target = self._screens[name]
        if target is self._active_screen:
            return

        async with self._batched_render():
            self._active_screen = target
            logger.info("Switching to screen: %s", name)

            # Apply the resolved theme cascade for this screen.
            self._renderer.apply_theme()

            self._wire_refresh_callbacks()

            # Emit *before* rendering so that bind() handlers listening to
            # on_screen_changed can seed card values (e.g. the nav pager)
            # prior to the first paint, avoiding a flash of defaults.
            await self.on_screen_changed.emit(name, self._screens)

            # Force a full repaint: mark every control on the incoming
            # screen dirty so the renderer does not skip cards/keys that
            # were already rendered on a previous visit or shared with
            # another screen.
            target.mark_all_dirty()

            await self._renderer.render_screen_complete()

    def _wire_refresh_callbacks(self) -> None:
        """Register ``self.refresh`` on every key and card of the active screen.

        Called by :meth:`set_screen`.  Allows handlers and background
        tasks to call :meth:`KeySlot.request_refresh` /
        :meth:`Card.request_refresh` to trigger re-renders without
        needing a direct deck reference.
        """
        screen = self._active_screen
        if screen is None:
            return
        for key_slot in screen.keys.values():
            key_slot.set_refresh_callback(self.refresh)
        if screen.touch_strip is not None:
            for card in screen.touch_strip.cards:
                card.set_refresh_callback(self.refresh)

    @property
    def active_screen(self) -> Screen | None:
        """The currently displayed screen, or ``None`` if no screen is set."""
        return self._active_screen

    def _current_screen(self) -> Screen | None:
        return self._active_screen

    async def _render_all_keys(self) -> None:
        """Render and push all key images for the active screen."""
        await self._renderer.render_all_keys()

    @staticmethod
    def _is_dui_key(key_slot: Any) -> bool:
        """Check whether a key slot is a DuiKey."""
        return DeckRenderer.is_dui_key(key_slot)

    @staticmethod
    def _is_animating(obj: Any) -> bool:
        """Check whether a card or key has an active spinner animation."""
        return DeckRenderer.is_animating(obj)

    async def _render_dui_key(self, key_slot: Any, key_index: int) -> None:
        """Render a DuiKey and push to the device.

        Parameters
        ----------
        key_slot
            The DuiKey to render.
        key_index
            The screen slot the key is currently installed at.
        """
        await self._renderer.render_dui_key(key_slot, key_index)

    async def _render_touchscreen(self) -> None:
        """Render and push each card panel individually to the device."""
        await self._renderer.render_touchscreen()

    async def _render_info_screen(self) -> None:
        """Render and push the info screen image (e.g. Neo)."""
        await self._renderer.render_info_screen()

    async def _consume_splash_push_deadline(self) -> None:
        """Block until any pending splash-hold deadline has elapsed.

        Called by :class:`DeckRenderer` at the start of each push phase
        of a batched render.  If :meth:`show_full_screen_image` was
        called with ``min_display_ms > 0`` since the last batched push,
        this awaits the remaining time before allowing the push to
        proceed.  Render-phase CPU work runs in parallel with this
        wait, so the effective splash display time is
        ``max(min_display_ms, render_time)``.

        The deadline is consumed (cleared) on the first call so that
        subsequent pushes within the same batched render -- and any
        following batched renders -- are not delayed.
        """
        deadline = self._splash_push_deadline
        if deadline is None:
            return
        self._splash_push_deadline = None
        remaining = deadline - time.perf_counter()
        if remaining > 0:
            await asyncio.sleep(remaining)

    @asynccontextmanager
    async def _batched_render(self) -> AsyncIterator[None]:
        """Suppress per-control ``refresh()`` calls inside a batched render.

        Used to wrap operations that end in a full
        :meth:`DeckRenderer.render_screen_complete` call (initial
        screen load, screen switch, theme change).  While the gate
        is closed, any handler that calls
        :meth:`Card.request_refresh` / :meth:`KeySlot.request_refresh`
        is folded into a single drain refresh fired after the
        batched render finishes.

        Re-entrant: nested batched operations are tracked by a depth
        counter and only the outermost exit fires the drain refresh.

        Yields
        ------
        None
            Inside the context, :meth:`refresh` is gated.
        """
        self._batch_render_depth += 1
        try:
            yield
        finally:
            self._batch_render_depth -= 1
            if self._batch_render_depth == 0 and self._refresh_pending:
                self._refresh_pending = False
                await self.refresh()

    async def refresh(self) -> None:
        """Re-render and push all dirty controls on the active screen.

        Call this after changing card values if you need immediate
        updates outside of ``set_screen()``.  Also drains any pending
        callbacks queued by programmatic ``set_value()`` calls on
        range controls.

        While the deck is inside a batched render (initial screen
        load, screen switch, or theme change), this call is recorded
        as pending and a single drain refresh fires once the batched
        operation completes.  This prevents partial per-key writes
        from landing on the LCD between the old frame and the new
        one.
        """
        if self._batch_render_depth > 0:
            self._refresh_pending = True
            return

        screen = self._current_screen()
        if not screen:
            return

        # Only drain cards that actually have pending callbacks.
        cards_with_pending = [
            card for card in screen.cards if card.has_pending_callbacks
        ]
        if cards_with_pending:
            await asyncio.gather(
                *(self._drain_card_callbacks(card) for card in cards_with_pending)
            )

        dirty_keys = [
            (key_index, key_slot)
            for key_index, key_slot in screen.keys.items()
            if key_slot.is_dirty and self._is_dui_key(key_slot)
        ]
        try:
            if dirty_keys:
                await asyncio.gather(
                    *(self._render_dui_key(ks, ki) for ki, ks in dirty_keys)
                )

            if screen.key_bg_dirty:
                await self._render_all_keys()
                screen.clear_key_bg_dirty()

            if screen.touch_strip is not None and screen.touch_strip.any_dirty:
                await self._render_touchscreen()

            if screen.info_screen is not None and screen.info_screen.is_dirty:
                await self._render_info_screen()
        except (HidWriteTimeout, HidApiError) as exc:
            # The device was torn down (disconnect, stop) while a
            # background task (clock tick, spinner, timer) was driving a
            # refresh.  Refresh is best-effort: swallow the error so the
            # caller's task doesn't crash with an unhandled exception.
            logger.debug("Refresh skipped — device unavailable: %s", exc)

    async def _check_timeouts(self) -> None:
        """Check all card selection timeouts on the active screen."""
        screen = self._current_screen()
        if not screen:
            return
        any_changed = False
        for card in screen.cards:
            if card.check_selection_timeout():
                any_changed = True
        if any_changed:
            await self.refresh()

    def schedule_timeout_check(self) -> None:
        """Signal that a card timeout needs checking.

        Call this method when a card registers a selection timeout so
        the deck can fire ``_check_timeouts`` without polling.  This is
        a no-op when the timeout loop is already scheduled to wake.
        """
        self._timeout_event.set()

    async def _timeout_loop(self) -> None:
        """Deadline-driven timeout checker.

        Waits until ``schedule_timeout_check`` is called, then performs
        the timeout check.  This eliminates the 500 ms polling wake.
        """
        try:
            while self._running:
                await self._timeout_event.wait()
                self._timeout_event.clear()
                if not self._running:
                    break
                await self._check_timeouts()
        except asyncio.CancelledError:
            pass

    async def _event_loop(self) -> None:
        """Dispatch transport events to the active screen handlers.

        Encoder turn events are coalesced: when a turn event is
        dequeued, any additional turn events for the same encoder
        already waiting in the queue are merged into a single
        dispatched event.  This prevents a backlog of redundant
        renders when the user turns the encoder faster than the
        render pipeline can keep up.
        """
        if not self._transport:
            return

        self._timeout_task = asyncio.create_task(
            self._timeout_loop(), name="deux-timeouts"
        )

        try:
            while self._running:
                event = await self._transport.queue.get()

                if not self._current_screen():
                    continue

                # Coalesce consecutive encoder turn events for the
                # same encoder so fast turns don't queue up behind
                # slow renders.
                if isinstance(event, EncoderTurnEvent):
                    event = self._coalesce_encoder_turns(event)

                try:
                    await self._dispatch(event)
                except Exception:
                    logger.exception("Error in event handler")

        except asyncio.CancelledError:
            pass
        except Exception:
            logger.exception("Event loop crashed")
        finally:
            if self._timeout_task and not self._timeout_task.done():
                self._timeout_task.cancel()
                with suppress(asyncio.CancelledError):
                    await self._timeout_task
            self._closed_event.set()

    async def _drain_card_callbacks(self, card: Any) -> None:
        """Drain and await all pending callbacks queued on a card."""
        await self._renderer.drain_card_callbacks(card)

    def _coalesce_encoder_turns(self, event: EncoderTurnEvent) -> EncoderTurnEvent:
        """Merge queued encoder turn events for the same encoder.

        Drains all pending :class:`EncoderTurnEvent` items from the
        transport queue that target the same encoder as *event*,
        summing their directions.  Non-matching events are re-queued
        in their original order.

        Parameters
        ----------
        event : EncoderTurnEvent
            The initial encoder turn event just dequeued.

        Returns
        -------
        EncoderTurnEvent
            A single event whose ``direction`` is the sum of all
            coalesced turns.  If no additional events were pending,
            the original event is returned unchanged.
        """
        if self._transport is None:
            return event

        queue = self._transport.queue
        direction = event.direction
        requeue: list[DeckEvent] = []

        while not queue.empty():
            try:
                pending = queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            if (
                isinstance(pending, EncoderTurnEvent)
                and pending.encoder == event.encoder
            ):
                direction += pending.direction
            else:
                requeue.append(pending)

        # Put back non-matching events in order.
        for item in requeue:
            queue.put_nowait(item)

        if direction == event.direction:
            return event
        return EncoderTurnEvent(encoder=event.encoder, direction=direction)

    async def _dispatch(self, event: DeckEvent) -> None:
        """Dispatch a single event to the appropriate handler on the active screen."""
        await self._event_router.dispatch(event)

device_path property 

device_path: str | None

HID device path, or None if not opened.

is_connected property 

is_connected: bool

Whether the device is currently connected and operational.

capabilities property 

capabilities: DeviceCapabilities

The device capabilities for the connected device.

Raises:

Type Description
DeckError

If the device is not opened.

metrics property 

metrics: RenderMetrics

Rendering metrics for the connected device.

Raises:

Type Description
DeckError

If the device is not opened.

info property 

info: DeviceInfo

Get device information.

brightness property 

brightness: int

Current brightness level (0-100).

theme property writable 

theme: Theme | None

Per-deck theme override, or None to inherit the system theme.

When set, this theme is used for all screens on this deck unless a screen has its own :attr:~deux.Screen.theme override. Set to None to fall back to the system-wide theme.

active_screen property 

active_screen: Screen | None

The currently displayed screen, or None if no screen is set.

__init__ 

__init__(serial_number: str, brightness: int = 80) -> None

Construct a deck handle for the given serial number.

Instances are normally created by :class:DeckManager in response to a device-connect event; application code receives them via on_connect handlers. For unit tests that need a deck without HID I/O, use :meth:Deck.for_testing.

Parameters:

Name Type Description Default
serial_number str

Serial number of the target device. Used during :meth:start to locate the matching HID device.

 required
brightness int

Initial brightness percentage in [0, 100] applied once the device is opened.

 80
Notes

Construction performs no I/O. The HID device is opened and the event loop started by :meth:start. The :attr:on_brightness_changed and :attr:on_screen_changed events are wired up here and ready for subscription before :meth:start is called.

Source code in src/deux/runtime/deck.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def __init__(
    self,
    serial_number: str,
    brightness: int = 80,
) -> None:
    """Construct a deck handle for the given serial number.

    Instances are normally created by :class:`DeckManager` in
    response to a device-connect event; application code receives
    them via ``on_connect`` handlers.  For unit tests that need a
    deck without HID I/O, use :meth:`Deck.for_testing`.

    Parameters
    ----------
    serial_number : str
        Serial number of the target device.  Used during
        :meth:`start` to locate the matching HID device.
    brightness : int, default=80
        Initial brightness percentage in ``[0, 100]`` applied once
        the device is opened.

    Notes
    -----
    Construction performs no I/O.  The HID device is opened and the
    event loop started by :meth:`start`.  The
    :attr:`on_brightness_changed` and :attr:`on_screen_changed`
    events are wired up here and ready for subscription before
    :meth:`start` is called.
    """
    self._serial_number = serial_number
    self._brightness = brightness
    self._device: HidDevice | None = None
    self._caps: DeviceCapabilities | None = None
    self._metrics: RenderMetrics | None = None
    self._transport: AsyncTransport | None = None
    self._event_task: asyncio.Task[None] | None = None
    self._timeout_task: asyncio.Task[None] | None = None
    self._timeout_event: asyncio.Event = asyncio.Event()
    self._closed_event = asyncio.Event()
    self._running = False
    self._device_lock = asyncio.Lock()

    self._screens: dict[str, Screen] = {}
    self._active_screen: Screen | None = None
    self._theme: Theme | None = None

    # Batched-render gate.  When ``_batch_render_depth > 0`` the deck
    # is inside a full-screen operation (initial render, screen
    # switch, theme change) and incoming ``refresh()`` calls are
    # deferred so partial per-key writes cannot land on the LCD
    # between the old frame and the new one.  ``_refresh_pending``
    # records that at least one refresh was suppressed so a single
    # drain refresh can fire when the gate reopens.  Both fields
    # are reset by :meth:`stop` so reconnects start clean.
    self._batch_render_depth: int = 0
    self._refresh_pending: bool = False

    # Splash hold deadline.  When :meth:`show_full_screen_image` is
    # called with ``min_display_ms > 0``, this records the
    # ``perf_counter()`` timestamp before which the next batched
    # render's *push* phase must not begin.  Render work runs in
    # parallel with the splash; only the final device push is
    # gated.  Cleared after the first deadline-await consumes it,
    # so the hold applies exactly once per splash.
    self._splash_push_deadline: float | None = None

    self.on_brightness_changed = AsyncEvent()
    self.on_screen_changed = AsyncEvent()

    self._renderer = DeckRenderer(self)
    self._event_router = DeckEventRouter(self)

for_testing classmethod 

for_testing(capabilities: DeviceCapabilities, *, serial_number: str = 'TEST', brightness: int = 80) -> Deck

Construct a :class:Deck pre-seeded with capabilities for tests.

Real construction goes through :meth:start, which discovers a physical device and derives :attr:capabilities and :attr:metrics from it. Tests that exercise behaviour above the device layer need those two attributes populated without any HID I/O.

This helper provides the supported way to do so, so that tests do not have to assign to the private _caps / _metrics attributes directly. No device is opened and no transport is started — :attr:is_connected remains False until the normal :meth:start path is invoked.

Parameters:

Name Type Description Default
capabilities DeviceCapabilities

The capabilities to seed onto the deck. Drives the derived :class:~deux.render.metrics.RenderMetrics and anything else that reads :attr:capabilities.

 required
serial_number str

Serial number recorded on the instance. Does not need to correspond to a real device.

 "TEST"
brightness int

Initial brightness (0-100).

 80

Returns:

Type Description
Deck

A deck instance with :attr:capabilities and :attr:metrics populated.
Notes

This constructor is intended for unit tests. Production code should use the normal :class:Deck constructor and :meth:start.

Source code in src/deux/runtime/deck.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@classmethod
def for_testing(
    cls,
    capabilities: DeviceCapabilities,
    *,
    serial_number: str = "TEST",
    brightness: int = 80,
) -> Deck:
    """Construct a :class:`Deck` pre-seeded with capabilities for tests.

    Real construction goes through :meth:`start`, which discovers a
    physical device and derives :attr:`capabilities` and
    :attr:`metrics` from it.  Tests that exercise behaviour above
    the device layer need those two attributes populated without
    any HID I/O.

    This helper provides the supported way to do so, so that tests
    do not have to assign to the private ``_caps`` / ``_metrics``
    attributes directly.  No device is opened and no transport is
    started — :attr:`is_connected` remains ``False`` until the
    normal :meth:`start` path is invoked.

    Parameters
    ----------
    capabilities : DeviceCapabilities
        The capabilities to seed onto the deck.  Drives the
        derived :class:`~deux.render.metrics.RenderMetrics` and
        anything else that reads :attr:`capabilities`.
    serial_number : str, default="TEST"
        Serial number recorded on the instance.  Does not need to
        correspond to a real device.
    brightness : int, default=80
        Initial brightness (0-100).

    Returns
    -------
    Deck
        A deck instance with :attr:`capabilities` and
        :attr:`metrics` populated.

    Notes
    -----
    This constructor is intended for unit tests.  Production code
    should use the normal :class:`Deck` constructor and
    :meth:`start`.
    """
    deck = cls(serial_number=serial_number, brightness=brightness)
    deck._caps = capabilities
    deck._metrics = RenderMetrics(capabilities)
    return deck

start async 

start() -> None

Discover the device by serial, open it, and start the event loop.

Source code in src/deux/runtime/deck.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
async def start(self) -> None:
    """Discover the device by serial, open it, and start the event loop."""
    if self._running:
        return

    loop = asyncio.get_running_loop()

    devices = await loop.run_in_executor(get_executor(), enumerate_devices)

    if not devices:
        raise DeckError("No Stream Deck devices found")

    target: HidDevice | None = None
    for d in devices:
        try:
            await loop.run_in_executor(get_executor(), d.open)
            if d.serial_number == self._serial_number:
                target = d
                break
            await loop.run_in_executor(get_executor(), d.close)
        except HidApiError:
            continue
    if target is None:
        raise DeckError(
            f"No device with serial '{self._serial_number}' found"
        )
    self._device = target

    await self._exec_device_io(self._device.show_logo)
    await self._exec_device_io(self._device.set_brightness, self._brightness)

    self._caps = DeviceCapabilities.from_device(self._device)
    self._metrics = RenderMetrics(self._caps)

    logger.info(
        "Opened %s (serial: %s, firmware: %s, keys: %d, dials: %d)",
        self._device.family,
        self._device.serial_number,
        self._device.firmware_version,
        self._caps.key_count,
        self._caps.dial_count,
    )

    self._transport = AsyncTransport(self._device, self._caps)
    self._transport.start()

    self._running = True
    self._closed_event.clear()

    # Publish the initial brightness so that any later bind_range()
    # call can seed from the event's last_value without waiting for
    # an explicit set_brightness().
    await self.on_brightness_changed.emit(self._brightness)

    self._event_task = asyncio.create_task(
        self._event_loop(), name="deux-events"
    )

stop async 

stop() -> None

Stop the event loop and close the device.

Source code in src/deux/runtime/deck.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
async def stop(self) -> None:
    """Stop the event loop and close the device."""
    if not self._running:
        return

    self._running = False

    await self._stop_all_spinners()

    self._detach_all_cards()

    if self._transport:
        self._transport.stop()

    if self._event_task and not self._event_task.done():
        self._event_task.cancel()
        with suppress(asyncio.CancelledError):
            await self._event_task

    if self._device:
        if self._device.is_open:
            try:
                await self._exec_device_io(self._device.show_logo)
                await self._exec_device_io(self._device.close)
            except Exception as e:
                # Intentional catch-all: shutdown must never raise.
                # Covers HidWriteTimeout/HidApiError plus any transport,
                # OS, or backend-specific errors surfaced during close.
                logger.warning("Error closing device: %s", e)
        else:
            logger.debug("Device already closed; skipping show_logo/close")

    self._device = None
    self._closed_event.set()
    self._batch_render_depth = 0
    self._refresh_pending = False
    self._splash_push_deadline = None
    logger.info("Deck stopped")

wait_closed async 

wait_closed() -> None

Block until the deck is closed (e.g. by stop() or disconnect).

Source code in src/deux/runtime/deck.py 
349
350
351
async def wait_closed(self) -> None:
    """Block until the deck is closed (e.g. by stop() or disconnect)."""
    await self._closed_event.wait()

set_theme async 

set_theme(theme: Theme | None) -> None

Apply a new deck-level theme and re-render the active screen.

Sets the deck theme, applies the CSS cascade to all renderers, marks every control dirty, and performs a complete re-render (icon prefetch, render all, push all) so the display updates atomically.

Parameters:

Name Type Description Default
theme Theme or None

The theme to apply, or None to revert to the system theme.

 required
Source code in src/deux/runtime/deck.py 
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
async def set_theme(self, theme: Theme | None) -> None:
    """Apply a new deck-level theme and re-render the active screen.

    Sets the deck theme, applies the CSS cascade to all renderers,
    marks every control dirty, and performs a complete re-render
    (icon prefetch, render all, push all) so the display updates
    atomically.

    Parameters
    ----------
    theme : Theme or None
        The theme to apply, or ``None`` to revert to the system
        theme.
    """
    self._theme = theme
    if self._active_screen is None:
        return

    async with self._batched_render():
        self._renderer.apply_theme()
        self._active_screen.mark_all_dirty()
        await self._renderer.render_screen_complete()

preload_icons async 

preload_icons() -> None

Prefetch Iconify icons for all registered screens.

Collects every icon identifier across all screens and fetches them concurrently, warming the in-memory and disk caches. Call this after all screens have been set up (keys and cards installed) to avoid network latency on first render.

Source code in src/deux/runtime/deck.py 
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
async def preload_icons(self) -> None:
    """Prefetch Iconify icons for all registered screens.

    Collects every icon identifier across all screens and fetches
    them concurrently, warming the in-memory and disk caches.
    Call this after all screens have been set up (keys and cards
    installed) to avoid network latency on first render.
    """
    # Inline import: tests patch ``deux.dui.iconify.prefetch_icons``;
    # importing it lazily keeps that patching point effective.
    from ..dui.iconify import prefetch_icons  # noqa: PLC0415

    all_icons: set[str] = set()
    for screen in self._screens.values():
        all_icons.update(screen.collect_all_icons())
    if all_icons:
        await prefetch_icons(all_icons)

resolve_stylesheet 

resolve_stylesheet() -> str

Resolve the effective CSS stylesheet for the active screen.

The cascade is: screen theme > deck theme > system theme.

Returns:

Type Description
str

CSS stylesheet string from the most specific theme.
Source code in src/deux/runtime/deck.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
def resolve_stylesheet(self) -> str:
    """Resolve the effective CSS stylesheet for the active screen.

    The cascade is: screen theme > deck theme > system theme.

    Returns
    -------
    str
        CSS stylesheet string from the most specific theme.
    """
    screen = self._active_screen
    if screen is not None and screen.theme is not None:
        return screen.theme.css
    if self._theme is not None:
        return self._theme.css
    return get_active_theme().css

set_brightness async 

set_brightness(percent: int) -> None

Set screen brightness.

Pushes the value to the hardware (if connected) and emits :attr:on_brightness_changed with the clamped result. If the clamped value equals the current brightness, no event fires — observers see only confirmed transitions.

Parameters:

Name Type Description Default
percent int

Brightness level (0-100). Values outside the range are clamped.

 required
Source code in src/deux/runtime/deck.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
async def set_brightness(self, percent: int) -> None:
    """Set screen brightness.

    Pushes the value to the hardware (if connected) and emits
    :attr:`on_brightness_changed` with the clamped result.  If the
    clamped value equals the current brightness, no event fires —
    observers see only confirmed transitions.

    Parameters
    ----------
    percent
        Brightness level (0-100).  Values outside the range are
        clamped.
    """
    if not isinstance(percent, int):
        raise TypeError(
            f"percent must be an int, got {type(percent).__name__}"
        )
    clamped = max(0, min(100, percent))
    if clamped == self._brightness:
        return
    if self._device is not None:
        async with self._device_lock:
            await self._exec_device_io(self._device.set_brightness, clamped)
    self._brightness = clamped
    await self.on_brightness_changed.emit(clamped)

show_full_screen_image async 

show_full_screen_image(image: Image | str | Path | bytes, *, fit: FitMode = 'cover', background: tuple[int, int, int] = (0, 0, 0), jpeg_quality: int = 90, min_display_ms: int = 0) -> None

Upload an image covering the entire LCD (HID command 0x08).

The image is loaded (PIL image, file path, raw image bytes, or SVG), resized to the device's logical LCD size using fit, rotated into the device's transmit orientation, JPEG-encoded, and pushed via :meth:HidDevice.set_full_screen_image.

Important behavioural contract. This is a one-shot, whole-LCD blit. Any subsequent per-key (set_key_image), per-window (set_partial_window_image), or per-screen render will paint over the image. In particular, calling :meth:set_screen after :meth:show_full_screen_image will immediately clobber the image as the renderer pushes the new screen.

Intended use cases are startup splashes, loading screens, and lock screens. For persistent backgrounds, use the per-screen background layer in :class:Screen instead.

Parameters:

Name Type Description Default
image Image.Image, str, Path, or bytes

Source image. .svg files and SVG byte streams are rasterised at the target LCD size; other formats are decoded with Pillow.

 required
fit ('cover', 'contain', 'stretch')

Resize strategy. "cover" scales to fill and crops the overflow; "contain" letterboxes with background; "stretch" ignores aspect ratio.

 "cover"
background tuple[int, int, int]

RGB background colour used to letterbox under fit="contain".

 (0, 0, 0)
jpeg_quality int

JPEG encoding quality (1-95).

 90
min_display_ms int

Minimum time, in milliseconds, that this image must remain visible on the LCD before the push phase of the next batched render (set_screen / set_theme) may begin. The render's CPU work still runs in parallel with the splash; only the final device push is delayed. Effective total splash time is therefore max(min_display_ms, render_time) -- a fast render waits for the remainder of the requested display time, while a slow render is never artificially delayed. Default of 0 disables the hold entirely. The deadline is one-shot and is consumed by the first batched push that follows.

 0

Raises:

Type Description
DeckError

If the device is not opened, or its PID has no known logical LCD size.
SplashError

If image preparation fails.

Examples:

Hold a splash for at least 500 ms so the user can perceive it even when the first set_screen is very fast::

await deck.show_splash("boot.png", min_display_ms=500)
await deck.set_screen("home")  # push delayed until 500ms elapsed
Source code in src/deux/runtime/deck.py 
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
async def show_full_screen_image(
    self,
    image: Image.Image | str | Path | bytes,
    *,
    fit: SplashFitMode = "cover",
    background: tuple[int, int, int] = (0, 0, 0),
    jpeg_quality: int = 90,
    min_display_ms: int = 0,
) -> None:
    """Upload an image covering the entire LCD (HID command ``0x08``).

    The image is loaded (PIL image, file path, raw image bytes, or
    SVG), resized to the device's logical LCD size using *fit*,
    rotated into the device's transmit orientation, JPEG-encoded,
    and pushed via
    :meth:`HidDevice.set_full_screen_image`.

    **Important behavioural contract.**  This is a one-shot,
    whole-LCD blit.  Any subsequent per-key (``set_key_image``),
    per-window (``set_partial_window_image``), or per-screen render
    will paint over the image.  In particular, calling
    :meth:`set_screen` after :meth:`show_full_screen_image` will
    immediately clobber the image as the renderer pushes the new
    screen.

    Intended use cases are startup splashes, loading screens, and
    lock screens.  For persistent backgrounds, use the per-screen
    background layer in :class:`Screen` instead.

    Parameters
    ----------
    image : Image.Image, str, Path, or bytes
        Source image.  ``.svg`` files and SVG byte streams are
        rasterised at the target LCD size; other formats are
        decoded with Pillow.
    fit : {"cover", "contain", "stretch"}, default="cover"
        Resize strategy.  ``"cover"`` scales to fill and crops the
        overflow; ``"contain"`` letterboxes with *background*;
        ``"stretch"`` ignores aspect ratio.
    background : tuple[int, int, int], default=(0, 0, 0)
        RGB background colour used to letterbox under
        ``fit="contain"``.
    jpeg_quality : int, default=90
        JPEG encoding quality (1-95).
    min_display_ms : int, default=0
        Minimum time, in milliseconds, that this image must remain
        visible on the LCD before the *push phase* of the next
        batched render (``set_screen`` / ``set_theme``) may begin.
        The render's CPU work still runs in parallel with the
        splash; only the final device push is delayed.  Effective
        total splash time is therefore
        ``max(min_display_ms, render_time)`` -- a fast render waits
        for the remainder of the requested display time, while a
        slow render is never artificially delayed.  Default of
        ``0`` disables the hold entirely.  The deadline is one-shot
        and is consumed by the first batched push that follows.

    Raises
    ------
    DeckError
        If the device is not opened, or its PID has no known
        logical LCD size.
    SplashError
        If image preparation fails.

    Examples
    --------
    Hold a splash for at least 500 ms so the user can perceive it
    even when the first ``set_screen`` is very fast::

        await deck.show_splash("boot.png", min_display_ms=500)
        await deck.set_screen("home")  # push delayed until 500ms elapsed
    """
    if self._device is None:
        raise DeckError("Device not opened")
    logical_size = self._device.logical_lcd_size
    if logical_size == (0, 0):
        raise DeckError(
            f"Device PID 0x{self._device.product_id:04X} has no "
            "known LCD size; cannot prepare full-screen image"
        )
    rotation = self._device.rotation

    # Inline import: splash transitively pulls in the resvg backend
    # for SVG inputs.  Keeping the import local avoids paying that
    # cost on deck construction for callers that never use this API.
    from .splash import prepare_full_screen_jpeg  # noqa: PLC0415

    loop = asyncio.get_running_loop()
    jpeg_bytes = await loop.run_in_executor(
        get_executor(),
        lambda: prepare_full_screen_jpeg(
            image,
            logical_size=logical_size,
            rotation=rotation,
            fit=fit,
            background=background,
            jpeg_quality=jpeg_quality,
        ),
    )

    async with self._device_lock:
        await self._exec_device_io(
            self._device.set_full_screen_image, jpeg_bytes
        )

    if min_display_ms > 0:
        self._splash_push_deadline = (
            time.perf_counter() + min_display_ms / 1000.0
        )
    else:
        self._splash_push_deadline = None

show_splash async 

show_splash(image: Image | str | Path | bytes, *, fit: FitMode = 'cover', background: tuple[int, int, int] = (0, 0, 0), jpeg_quality: int = 90, min_display_ms: int = 0) -> None

Alias for :meth:show_full_screen_image, intended for startup.

Provides a semantically clearer entry point for the common case of displaying a boot/splash image before the first screen is rendered.

Parameters:

Name Type Description Default
image Image.Image, str, Path, or bytes

See :meth:show_full_screen_image.

 required
fit ('cover', 'contain', 'stretch')

See :meth:show_full_screen_image.

 "cover"
background tuple[int, int, int]

See :meth:show_full_screen_image.

 (0, 0, 0)
jpeg_quality int

See :meth:show_full_screen_image.

 90
min_display_ms int

See :meth:show_full_screen_image.

 0
See Also

show_full_screen_image clear_full_screen_image

Source code in src/deux/runtime/deck.py 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
async def show_splash(
    self,
    image: Image.Image | str | Path | bytes,
    *,
    fit: SplashFitMode = "cover",
    background: tuple[int, int, int] = (0, 0, 0),
    jpeg_quality: int = 90,
    min_display_ms: int = 0,
) -> None:
    """Alias for :meth:`show_full_screen_image`, intended for startup.

    Provides a semantically clearer entry point for the common case
    of displaying a boot/splash image before the first screen is
    rendered.

    Parameters
    ----------
    image : Image.Image, str, Path, or bytes
        See :meth:`show_full_screen_image`.
    fit : {"cover", "contain", "stretch"}, default="cover"
        See :meth:`show_full_screen_image`.
    background : tuple[int, int, int], default=(0, 0, 0)
        See :meth:`show_full_screen_image`.
    jpeg_quality : int, default=90
        See :meth:`show_full_screen_image`.
    min_display_ms : int, default=0
        See :meth:`show_full_screen_image`.

    See Also
    --------
    show_full_screen_image
    clear_full_screen_image
    """
    await self.show_full_screen_image(
        image,
        fit=fit,
        background=background,
        jpeg_quality=jpeg_quality,
        min_display_ms=min_display_ms,
    )

clear_full_screen_image async 

clear_full_screen_image(color: tuple[int, int, int] = (0, 0, 0), *, jpeg_quality: int = 90) -> None

Clear the LCD by uploading a solid-colour full-screen image.

Uses the same HID command 0x08 path as :meth:show_full_screen_image, ensuring a deterministic clear across all supported deck families. Behaviour-wise this is equivalent to show_full_screen_image with a solid-colour source; like that method, any subsequent per-key or per-window write will paint over the cleared LCD.

Parameters:

Name Type Description Default
color tuple[int, int, int]

RGB fill colour.

 (0, 0, 0)
jpeg_quality int

JPEG encoding quality (1-95).

 90

Raises:

Type Description
DeckError

If the device is not opened, or its PID has no known logical LCD size.
Source code in src/deux/runtime/deck.py 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
async def clear_full_screen_image(
    self,
    color: tuple[int, int, int] = (0, 0, 0),
    *,
    jpeg_quality: int = 90,
) -> None:
    """Clear the LCD by uploading a solid-colour full-screen image.

    Uses the same HID command ``0x08`` path as
    :meth:`show_full_screen_image`, ensuring a deterministic
    clear across all supported deck families.  Behaviour-wise this
    is equivalent to ``show_full_screen_image`` with a solid-colour
    source; like that method, any subsequent per-key or per-window
    write will paint over the cleared LCD.

    Parameters
    ----------
    color : tuple[int, int, int], default=(0, 0, 0)
        RGB fill colour.
    jpeg_quality : int, default=90
        JPEG encoding quality (1-95).

    Raises
    ------
    DeckError
        If the device is not opened, or its PID has no known
        logical LCD size.
    """
    if self._device is None:
        raise DeckError("Device not opened")
    logical_size = self._device.logical_lcd_size
    if logical_size == (0, 0):
        raise DeckError(
            f"Device PID 0x{self._device.product_id:04X} has no "
            "known LCD size; cannot prepare full-screen image"
        )
    rotation = self._device.rotation

    from .splash import prepare_solid_color_jpeg  # noqa: PLC0415

    loop = asyncio.get_running_loop()
    jpeg_bytes = await loop.run_in_executor(
        get_executor(),
        lambda: prepare_solid_color_jpeg(
            color,
            logical_size=logical_size,
            rotation=rotation,
            jpeg_quality=jpeg_quality,
        ),
    )

    async with self._device_lock:
        await self._exec_device_io(
            self._device.set_full_screen_image, jpeg_bytes
        )

screen 

screen(name: str) -> Screen

Get or create a screen by name.

Parameters:

Name Type Description Default
name str

Screen name.

 required

Returns:

Type Description
Screen

The Screen instance.

Raises:

Type Description
DeckError

If the device is not opened — capabilities are required to size the screen, and they are only known after :meth:start.
Source code in src/deux/runtime/deck.py 
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
def screen(self, name: str) -> Screen:
    """Get or create a screen by name.

    Parameters
    ----------
    name
        Screen name.

    Returns
    -------
    Screen
        The Screen instance.

    Raises
    ------
    DeckError
        If the device is not opened — capabilities are required to
        size the screen, and they are only known after :meth:`start`.
    """
    # Inline import: ui.screen transitively imports runtime.events via
    # ui.touch_strip -> ui.cards.base, creating a cycle at module load.
    from ..ui.screen import Screen  # noqa: PLC0415

    if self._caps is None:
        raise DeckError("Device not opened")

    if name not in self._screens:
        self._screens[name] = Screen(name, self._caps)
    return self._screens[name]

set_screen async 

set_screen(name: str) -> None

Switch to a named screen, rendering all keys and cards.

Wires up refresh callbacks on every key and card so that any handler or background task can call request_refresh() to trigger a re-render without needing a direct reference to the deck. After the new screen has finished rendering, fires :attr:on_screen_changed with the new name. No event fires if the requested screen is already active.

Parameters:

Name Type Description Default
name str

Screen name (must already exist via deck.screen(name)).

 required

Raises:

Type Description
DeckError

If name does not match a previously-created screen.
Source code in src/deux/runtime/deck.py 
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
async def set_screen(self, name: str) -> None:
    """Switch to a named screen, rendering all keys and cards.

    Wires up refresh callbacks on every key and card so that any
    handler or background task can call ``request_refresh()`` to
    trigger a re-render without needing a direct reference to the
    deck.  After the new screen has finished rendering, fires
    :attr:`on_screen_changed` with the new name.  No event fires
    if the requested screen is already active.

    Parameters
    ----------
    name
        Screen name (must already exist via ``deck.screen(name)``).

    Raises
    ------
    DeckError
        If *name* does not match a previously-created screen.
    """
    if name not in self._screens:
        raise DeckError(f"Screen '{name}' does not exist")

    target = self._screens[name]
    if target is self._active_screen:
        return

    async with self._batched_render():
        self._active_screen = target
        logger.info("Switching to screen: %s", name)

        # Apply the resolved theme cascade for this screen.
        self._renderer.apply_theme()

        self._wire_refresh_callbacks()

        # Emit *before* rendering so that bind() handlers listening to
        # on_screen_changed can seed card values (e.g. the nav pager)
        # prior to the first paint, avoiding a flash of defaults.
        await self.on_screen_changed.emit(name, self._screens)

        # Force a full repaint: mark every control on the incoming
        # screen dirty so the renderer does not skip cards/keys that
        # were already rendered on a previous visit or shared with
        # another screen.
        target.mark_all_dirty()

        await self._renderer.render_screen_complete()

refresh async 

refresh() -> None

Re-render and push all dirty controls on the active screen.

Call this after changing card values if you need immediate updates outside of set_screen(). Also drains any pending callbacks queued by programmatic set_value() calls on range controls.

While the deck is inside a batched render (initial screen load, screen switch, or theme change), this call is recorded as pending and a single drain refresh fires once the batched operation completes. This prevents partial per-key writes from landing on the LCD between the old frame and the new one.

Source code in src/deux/runtime/deck.py 
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
async def refresh(self) -> None:
    """Re-render and push all dirty controls on the active screen.

    Call this after changing card values if you need immediate
    updates outside of ``set_screen()``.  Also drains any pending
    callbacks queued by programmatic ``set_value()`` calls on
    range controls.

    While the deck is inside a batched render (initial screen
    load, screen switch, or theme change), this call is recorded
    as pending and a single drain refresh fires once the batched
    operation completes.  This prevents partial per-key writes
    from landing on the LCD between the old frame and the new
    one.
    """
    if self._batch_render_depth > 0:
        self._refresh_pending = True
        return

    screen = self._current_screen()
    if not screen:
        return

    # Only drain cards that actually have pending callbacks.
    cards_with_pending = [
        card for card in screen.cards if card.has_pending_callbacks
    ]
    if cards_with_pending:
        await asyncio.gather(
            *(self._drain_card_callbacks(card) for card in cards_with_pending)
        )

    dirty_keys = [
        (key_index, key_slot)
        for key_index, key_slot in screen.keys.items()
        if key_slot.is_dirty and self._is_dui_key(key_slot)
    ]
    try:
        if dirty_keys:
            await asyncio.gather(
                *(self._render_dui_key(ks, ki) for ki, ks in dirty_keys)
            )

        if screen.key_bg_dirty:
            await self._render_all_keys()
            screen.clear_key_bg_dirty()

        if screen.touch_strip is not None and screen.touch_strip.any_dirty:
            await self._render_touchscreen()

        if screen.info_screen is not None and screen.info_screen.is_dirty:
            await self._render_info_screen()
    except (HidWriteTimeout, HidApiError) as exc:
        # The device was torn down (disconnect, stop) while a
        # background task (clock tick, spinner, timer) was driving a
        # refresh.  Refresh is best-effort: swallow the error so the
        # caller's task doesn't crash with an unhandled exception.
        logger.debug("Refresh skipped — device unavailable: %s", exc)

schedule_timeout_check 

schedule_timeout_check() -> None

Signal that a card timeout needs checking.

Call this method when a card registers a selection timeout so the deck can fire _check_timeouts without polling. This is a no-op when the timeout loop is already scheduled to wake.

Source code in src/deux/runtime/deck.py 
1024
1025
1026
1027
1028
1029
1030
1031
def schedule_timeout_check(self) -> None:
    """Signal that a card timeout needs checking.

    Call this method when a card registers a selection timeout so
    the deck can fire ``_check_timeouts`` without polling.  This is
    a no-op when the timeout loop is already scheduled to wake.
    """
    self._timeout_event.set()

DeckError

Bases: DeuxError

Raised for deck-level errors.

Source code in src/deux/runtime/deck.py 
49
50
class DeckError(DeuxError):
    """Raised for deck-level errors."""

DeviceInfo dataclass

Information about a connected Stream Deck device.

Attributes:

Name Type Description
deck_type str

Human-readable device model name (e.g. "Stream Deck +").
serial str

Unique serial number reported by the hardware.
firmware str

Firmware version string.
key_count int

Total number of physical keys on the device.
key_layout tuple[int, int]

Key grid dimensions as (columns, rows).
encoder_count int

Number of rotary encoders (dials) on the device.
key_pixel_size tuple[int, int]

Pixel dimensions of a single key image as (width, height).
touchscreen_size tuple[int, int]

Pixel dimensions of the touchscreen as (width, height). (0, 0) if the device has no touchscreen.
key_image_format str

Image format expected by the device (e.g. "JPEG").
Source code in src/deux/runtime/device_info.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@dataclass(frozen=True, slots=True)
class DeviceInfo:
    """Information about a connected Stream Deck device.

    Attributes
    ----------
    deck_type : str
        Human-readable device model name (e.g. ``"Stream Deck +"``).
    serial : str
        Unique serial number reported by the hardware.
    firmware : str
        Firmware version string.
    key_count : int
        Total number of physical keys on the device.
    key_layout : tuple[int, int]
        Key grid dimensions as ``(columns, rows)``.
    encoder_count : int
        Number of rotary encoders (dials) on the device.
    key_pixel_size : tuple[int, int]
        Pixel dimensions of a single key image as ``(width, height)``.
    touchscreen_size : tuple[int, int]
        Pixel dimensions of the touchscreen as ``(width, height)``.
        ``(0, 0)`` if the device has no touchscreen.
    key_image_format : str
        Image format expected by the device (e.g. ``"JPEG"``).
    """

    deck_type: str
    serial: str
    firmware: str
    key_count: int
    key_layout: tuple[int, int]
    encoder_count: int
    key_pixel_size: tuple[int, int]
    touchscreen_size: tuple[int, int]
    key_image_format: str

EncoderPressEvent dataclass

An encoder push/release event.

Source code in src/deux/runtime/events.py 
42
43
44
45
46
47
@dataclass(frozen=True, slots=True)
class EncoderPressEvent:
    """An encoder push/release event."""

    encoder: int
    pressed: bool

EncoderTurnEvent dataclass

An encoder rotation event.

Source code in src/deux/runtime/events.py 
34
35
36
37
38
39
@dataclass(frozen=True, slots=True)
class EncoderTurnEvent:
    """An encoder rotation event."""

    encoder: int
    direction: int

EventType

Bases: Enum

Internal event types for the async bridge.

Source code in src/deux/runtime/events.py 
17
18
19
20
21
22
23
class EventType(Enum):
    """Internal event types for the async bridge."""

    ENCODER_TURN = auto()
    TOUCH_SHORT = auto()
    TOUCH_LONG = auto()
    TOUCH_DRAG = auto()

KeyEvent dataclass

A physical key press/release event.

Source code in src/deux/runtime/events.py 
26
27
28
29
30
31
@dataclass(frozen=True, slots=True)
class KeyEvent:
    """A physical key press/release event."""

    key: int
    pressed: bool

TouchEvent dataclass

A touchscreen interaction event.

Source code in src/deux/runtime/events.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@dataclass(frozen=True, slots=True)
class TouchEvent:
    """A touchscreen interaction event."""

    event_type: EventType
    x: int
    y: int
    x_out: int | None = None
    y_out: int | None = None

    def compute_zone(self, metrics: RenderMetrics) -> int:
        """Compute which touch-strip zone this touch falls in.

        Parameters
        ----------
        metrics
            The render metrics for the connected device.

        Returns
        -------
        int
            Zone index (0 to panel_count-1).
        """
        if metrics.panel_count == 0 or metrics.panel_width <= 0:
            return 0

        zone = self.x // metrics.panel_width
        return max(0, min(zone, metrics.panel_count - 1))

compute_zone 

compute_zone(metrics: RenderMetrics) -> int

Compute which touch-strip zone this touch falls in.

Parameters:

Name Type Description Default
metrics RenderMetrics

The render metrics for the connected device.

 required

Returns:

Type Description
int

Zone index (0 to panel_count-1).
Source code in src/deux/runtime/events.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def compute_zone(self, metrics: RenderMetrics) -> int:
    """Compute which touch-strip zone this touch falls in.

    Parameters
    ----------
    metrics
        The render metrics for the connected device.

    Returns
    -------
    int
        Zone index (0 to panel_count-1).
    """
    if metrics.panel_count == 0 or metrics.panel_width <= 0:
        return 0

    zone = self.x // metrics.panel_width
    return max(0, min(zone, metrics.panel_count - 1))

DeckManager

The main entry point for the deux library.

Manages one or more Stream Deck devices with automatic discovery, hot-plug detection, and reconnection. Register on_connect and on_disconnect handlers, then start the manager.

Examples:

::

manager = DeckManager()

@manager.on_connect(deck_type="Stream Deck +")
async def handle(deck: Deck):
    screen = deck.screen("main")

    @screen.key(0).on_press
    async def on_home():
        print("Home pressed!")

    await deck.set_screen("main")

@manager.on_disconnect
async def lost(info: DeviceInfo):
    print(f"Lost: {info.serial}")

async with manager:
    await manager.wait_closed()
Source code in src/deux/runtime/manager.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
class DeckManager:
    """The main entry point for the deux library.

    Manages one or more Stream Deck devices with automatic discovery,
    hot-plug detection, and reconnection.  Register ``on_connect`` and
    ``on_disconnect`` handlers, then start the manager.

    Examples
    --------
    ::

        manager = DeckManager()

        @manager.on_connect(deck_type="Stream Deck +")
        async def handle(deck: Deck):
            screen = deck.screen("main")

            @screen.key(0).on_press
            async def on_home():
                print("Home pressed!")

            await deck.set_screen("main")

        @manager.on_disconnect
        async def lost(info: DeviceInfo):
            print(f"Lost: {info.serial}")

        async with manager:
            await manager.wait_closed()
    """

    def __init__(
        self,
        poll_interval: float = 2.0,
        brightness: int = 80,
        auto_reconnect: bool = True,
    ) -> None:
        """Construct a deck manager.

        Construction is side-effect free; no devices are scanned until
        :meth:`start` is called (directly or via ``async with``).

        Parameters
        ----------
        poll_interval : float, default=2.0
            Seconds between HID device scans.  Lower values detect
            hot-plug events faster at the cost of more frequent
            enumeration.
        brightness : int, default=80
            Default brightness percentage in ``[0, 100]`` applied to
            newly connected :class:`Deck` instances.
        auto_reconnect : bool, default=True
            When ``True``, devices that disconnect are automatically
            reopened on the next scan and registered ``on_connect``
            handlers fire again.  When ``False``, disconnected devices
            stay disconnected until the next manual restart.

        Notes
        -----
        The scan loop, executor lifecycle, and event subscriptions are
        created lazily by :meth:`start`; constructing a manager is
        cheap and does no I/O.
        """
        self._poll_interval = poll_interval
        self._brightness = brightness
        self._auto_reconnect = auto_reconnect
        self._running = False
        self._closed_event = asyncio.Event()
        self._scan_task: asyncio.Task[None] | None = None

        self._decks: dict[str, Deck] = {}
        self._failed_probe_paths: set[str] = set()
        self._path_serial_cache: dict[str, str] = {}

        self._connect_handlers: list[tuple[dict[str, str | None], AsyncHandler]] = []
        self._disconnect_handlers: list[AsyncHandler] = []

    async def __aenter__(self) -> DeckManager:
        """Start the manager and return it for use as an async context manager."""
        await self.start()
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Stop the manager when exiting the async context."""
        await self.stop()

    async def start(self) -> None:
        """Start the device scanning loop."""
        if self._running:
            return
        self._running = True
        self._closed_event.clear()
        self._scan_task = asyncio.create_task(
            self._scan_loop(), name="deckmanager-scan"
        )
        logger.info("DeckManager started (poll_interval=%.1fs)", self._poll_interval)

    async def stop(self) -> None:
        """Stop scanning and close all managed decks."""
        if not self._running:
            return
        self._running = False

        if self._scan_task and not self._scan_task.done():
            self._scan_task.cancel()
            with suppress(asyncio.CancelledError):
                await self._scan_task

        for serial, deck in list(self._decks.items()):
            try:
                await deck.stop()
            except Exception:
                logger.warning("Error stopping deck %s", serial, exc_info=True)
        self._decks.clear()

        self._closed_event.set()
        shutdown_executor(wait=True)
        logger.info("DeckManager stopped")

    async def wait_closed(self) -> None:
        """Block until the manager is stopped."""
        await self._closed_event.wait()

    def on_connect(
        self,
        *,
        serial: str | None = None,
        deck_type: str | None = None,
    ) -> Callable[[AsyncHandler], AsyncHandler]:
        """Register a callback for when a matching device connects.

        The handler is also called on reconnection when
        ``auto_reconnect`` is enabled.

        Examples
        --------
        ::

            @manager.on_connect(deck_type="Stream Deck +")
            async def handle(deck: Deck):
                ...

        Parameters
        ----------
        serial
            Only match this serial number.
        deck_type
            Only match this device type.

        Returns
        -------
        Callable
            Decorator that registers the handler.

        Notes
        -----
        Unlike :attr:`on_disconnect`, this is a decorator *factory* and
        must be invoked with parentheses (``@manager.on_connect()``) even
        when no filter arguments are passed.
        """
        filters = {"serial": serial, "deck_type": deck_type}

        def decorator(handler: AsyncHandler) -> AsyncHandler:
            self._connect_handlers.append((filters, handler))
            return handler

        return decorator

    @property
    def on_disconnect(self) -> Callable[[AsyncHandler], AsyncHandler]:
        """Register a callback for when a device disconnects.

        Multiple handlers may be registered; all are called in
        registration order when a device disconnects.

        Examples
        --------
        ::

            @manager.on_disconnect
            async def handle(info: DeviceInfo):
                ...

        Returns
        -------
        Callable
            Decorator that registers the handler.

        Notes
        -----
        Unlike :meth:`on_connect`, this is a property returning the
        decorator directly. Use it bare (``@manager.on_disconnect``);
        calling it with parentheses raises :class:`TypeError`.
        """

        def decorator(handler: AsyncHandler) -> AsyncHandler:
            self._disconnect_handlers.append(handler)
            return handler

        return decorator

    @property
    def decks(self) -> dict[str, Deck]:
        """Currently managed decks, keyed by serial number."""
        return dict(self._decks)

    async def _scan_loop(self) -> None:
        """Periodically enumerate devices and manage connections.

        Runs an initial scan immediately, then polls at the configured
        ``poll_interval`` until the manager is stopped or cancelled.
        Sets the closed event on exit so that ``wait_closed()`` unblocks.
        """
        await self._scan_once()

        try:
            while self._running:
                await asyncio.sleep(self._poll_interval)
                if not self._running:
                    break
                await self._scan_once()
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.exception("Scan loop crashed")
        finally:
            self._closed_event.set()

    async def _scan_once(self) -> None:
        """Execute a single device scan cycle.

        Enumerates all connected Stream Deck devices, detects new
        connections and disconnections, and invokes the appropriate
        registered handlers.  Devices that fail to open are silently
        skipped.
        """
        loop = asyncio.get_running_loop()

        try:
            devices = await loop.run_in_executor(
                get_executor(), enumerate_devices
            )
        except OSError:
            logger.warning("Device enumeration failed", exc_info=True)
            return

        managed_paths: dict[str, str] = {}
        for serial, deck in self._decks.items():
            path = deck.device_path
            if path is not None:
                managed_paths[path] = serial

        connected_serials: set[str] = set()
        device_by_serial: dict[str, HidDevice] = {}

        current_paths: set[str] = set()
        for d in devices:
            dev_path = d.path.decode("utf-8", errors="replace")
            current_paths.add(dev_path)
            if dev_path in managed_paths:
                connected_serials.add(managed_paths[dev_path])
                continue
            if dev_path in self._path_serial_cache:
                serial = self._path_serial_cache[dev_path]
                connected_serials.add(serial)
                device_by_serial[serial] = d
                continue
            try:
                await loop.run_in_executor(get_executor(), d.open)
                serial = d.serial_number
                connected_serials.add(serial)
                device_by_serial[serial] = d
                self._path_serial_cache[dev_path] = serial
                await loop.run_in_executor(get_executor(), d.close)
            except OSError:
                if dev_path not in self._failed_probe_paths:
                    self._failed_probe_paths.add(dev_path)
                    logger.info(
                        "Failed to probe new device at %s", dev_path, exc_info=True
                    )
                else:
                    logger.debug(
                        "Failed to probe device at %s", dev_path, exc_info=True
                    )
                continue

        # Invalidate cache for disconnected paths
        stale_paths = set(self._path_serial_cache) - current_paths
        for path in stale_paths:
            del self._path_serial_cache[path]
            self._failed_probe_paths.discard(path)

        for serial in list(self._decks):
            if serial not in connected_serials:
                deck = self._decks.pop(serial)
                logger.info("Device disconnected: %s", serial)
                try:
                    info = deck.info
                except (DeckError, OSError):
                    info = DeviceInfo(
                        deck_type="unknown",
                        serial=serial,
                        firmware="",
                        key_count=0,
                        key_layout=(0, 0),
                        encoder_count=0,
                        key_pixel_size=(0, 0),
                        touchscreen_size=(0, 0),
                        key_image_format="",
                    )
                with suppress(Exception):
                    await deck.stop()
                if self._disconnect_handlers:
                    for _dh in self._disconnect_handlers:
                        try:
                            await _dh(info)
                        except Exception:
                            logger.exception("Error in disconnect handler")

        for serial in connected_serials:
            if serial in self._decks:
                continue

            raw_device = device_by_serial.get(serial)
            if raw_device is None:
                continue
            device_type = raw_device.family

            for filters, handler in self._connect_handlers:
                if filters["serial"] is not None and filters["serial"] != serial:
                    continue
                if (
                    filters["deck_type"] is not None
                    and filters["deck_type"] != device_type
                ):
                    continue

                try:
                    deck = Deck(
                        serial_number=serial,
                        brightness=self._brightness,
                    )
                    await deck.start()
                    self._decks[serial] = deck
                    logger.info("Device connected: %s (%s)", serial, device_type)

                    try:
                        await handler(deck)
                    except Exception:
                        logger.exception("Error in connect handler for %s", serial)
                except DeckError:
                    logger.debug("Failed to start deck for %s", serial)
                except Exception:
                    logger.exception("Unexpected error starting deck %s", serial)
                break

on_disconnect property 

on_disconnect: Callable[[AsyncHandler], AsyncHandler]

Register a callback for when a device disconnects.

Multiple handlers may be registered; all are called in registration order when a device disconnects.

Examples:

::

@manager.on_disconnect
async def handle(info: DeviceInfo):
    ...

Returns:

Type Description
Callable

Decorator that registers the handler.
Notes

Unlike :meth:on_connect, this is a property returning the decorator directly. Use it bare (@manager.on_disconnect); calling it with parentheses raises :class:TypeError.

decks property 

decks: dict[str, Deck]

Currently managed decks, keyed by serial number.

__init__ 

__init__(poll_interval: float = 2.0, brightness: int = 80, auto_reconnect: bool = True) -> None

Construct a deck manager.

Construction is side-effect free; no devices are scanned until :meth:start is called (directly or via async with).

Parameters:

Name Type Description Default
poll_interval float

Seconds between HID device scans. Lower values detect hot-plug events faster at the cost of more frequent enumeration.

 2.0
brightness int

Default brightness percentage in [0, 100] applied to newly connected :class:Deck instances.

 80
auto_reconnect bool

When True, devices that disconnect are automatically reopened on the next scan and registered on_connect handlers fire again. When False, disconnected devices stay disconnected until the next manual restart.

 True
Notes

The scan loop, executor lifecycle, and event subscriptions are created lazily by :meth:start; constructing a manager is cheap and does no I/O.

Source code in src/deux/runtime/manager.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    poll_interval: float = 2.0,
    brightness: int = 80,
    auto_reconnect: bool = True,
) -> None:
    """Construct a deck manager.

    Construction is side-effect free; no devices are scanned until
    :meth:`start` is called (directly or via ``async with``).

    Parameters
    ----------
    poll_interval : float, default=2.0
        Seconds between HID device scans.  Lower values detect
        hot-plug events faster at the cost of more frequent
        enumeration.
    brightness : int, default=80
        Default brightness percentage in ``[0, 100]`` applied to
        newly connected :class:`Deck` instances.
    auto_reconnect : bool, default=True
        When ``True``, devices that disconnect are automatically
        reopened on the next scan and registered ``on_connect``
        handlers fire again.  When ``False``, disconnected devices
        stay disconnected until the next manual restart.

    Notes
    -----
    The scan loop, executor lifecycle, and event subscriptions are
    created lazily by :meth:`start`; constructing a manager is
    cheap and does no I/O.
    """
    self._poll_interval = poll_interval
    self._brightness = brightness
    self._auto_reconnect = auto_reconnect
    self._running = False
    self._closed_event = asyncio.Event()
    self._scan_task: asyncio.Task[None] | None = None

    self._decks: dict[str, Deck] = {}
    self._failed_probe_paths: set[str] = set()
    self._path_serial_cache: dict[str, str] = {}

    self._connect_handlers: list[tuple[dict[str, str | None], AsyncHandler]] = []
    self._disconnect_handlers: list[AsyncHandler] = []

__aenter__ async 

__aenter__() -> DeckManager

Start the manager and return it for use as an async context manager.

Source code in src/deux/runtime/manager.py 
102
103
104
105
async def __aenter__(self) -> DeckManager:
    """Start the manager and return it for use as an async context manager."""
    await self.start()
    return self

__aexit__ async 

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Stop the manager when exiting the async context.

Source code in src/deux/runtime/manager.py 
107
108
109
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Stop the manager when exiting the async context."""
    await self.stop()

start async 

start() -> None

Start the device scanning loop.

Source code in src/deux/runtime/manager.py 
111
112
113
114
115
116
117
118
119
120
async def start(self) -> None:
    """Start the device scanning loop."""
    if self._running:
        return
    self._running = True
    self._closed_event.clear()
    self._scan_task = asyncio.create_task(
        self._scan_loop(), name="deckmanager-scan"
    )
    logger.info("DeckManager started (poll_interval=%.1fs)", self._poll_interval)

stop async 

stop() -> None

Stop scanning and close all managed decks.

Source code in src/deux/runtime/manager.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def stop(self) -> None:
    """Stop scanning and close all managed decks."""
    if not self._running:
        return
    self._running = False

    if self._scan_task and not self._scan_task.done():
        self._scan_task.cancel()
        with suppress(asyncio.CancelledError):
            await self._scan_task

    for serial, deck in list(self._decks.items()):
        try:
            await deck.stop()
        except Exception:
            logger.warning("Error stopping deck %s", serial, exc_info=True)
    self._decks.clear()

    self._closed_event.set()
    shutdown_executor(wait=True)
    logger.info("DeckManager stopped")

wait_closed async 

wait_closed() -> None

Block until the manager is stopped.

Source code in src/deux/runtime/manager.py 
144
145
146
async def wait_closed(self) -> None:
    """Block until the manager is stopped."""
    await self._closed_event.wait()

on_connect 

on_connect(*, serial: str | None = None, deck_type: str | None = None) -> Callable[[AsyncHandler], AsyncHandler]

Register a callback for when a matching device connects.

The handler is also called on reconnection when auto_reconnect is enabled.

Examples:

::

@manager.on_connect(deck_type="Stream Deck +")
async def handle(deck: Deck):
    ...

Parameters:

Name Type Description Default
serial str | None

Only match this serial number.

 None
deck_type str | None

Only match this device type.

 None

Returns:

Type Description
Callable

Decorator that registers the handler.
Notes

Unlike :attr:on_disconnect, this is a decorator factory and must be invoked with parentheses (@manager.on_connect()) even when no filter arguments are passed.

Source code in src/deux/runtime/manager.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def on_connect(
    self,
    *,
    serial: str | None = None,
    deck_type: str | None = None,
) -> Callable[[AsyncHandler], AsyncHandler]:
    """Register a callback for when a matching device connects.

    The handler is also called on reconnection when
    ``auto_reconnect`` is enabled.

    Examples
    --------
    ::

        @manager.on_connect(deck_type="Stream Deck +")
        async def handle(deck: Deck):
            ...

    Parameters
    ----------
    serial
        Only match this serial number.
    deck_type
        Only match this device type.

    Returns
    -------
    Callable
        Decorator that registers the handler.

    Notes
    -----
    Unlike :attr:`on_disconnect`, this is a decorator *factory* and
    must be invoked with parentheses (``@manager.on_connect()``) even
    when no filter arguments are passed.
    """
    filters = {"serial": serial, "deck_type": deck_type}

    def decorator(handler: AsyncHandler) -> AsyncHandler:
        self._connect_handlers.append((filters, handler))
        return handler

    return decorator

AsyncTransport

Poll-based async bridge from HID input reports to deck events.

Reads input reports from the device in a background task and translates them into :class:DeckEvent objects on an asyncio queue.

Parameters:

Name Type Description Default
device HidDevice

An open HID device.

 required
caps DeviceCapabilities or None

Device capabilities (used to determine which events to process).

 None
poll_interval_ms int

HID read timeout per poll cycle in milliseconds.

 50
Source code in src/deux/runtime/transport.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class AsyncTransport:
    """Poll-based async bridge from HID input reports to deck events.

    Reads input reports from the device in a background task and
    translates them into :class:`DeckEvent` objects on an asyncio queue.

    Parameters
    ----------
    device : HidDevice
        An open HID device.
    caps : DeviceCapabilities or None
        Device capabilities (used to determine which events to process).
    poll_interval_ms : int, default=50
        HID read timeout per poll cycle in milliseconds.
    """

    def __init__(
        self,
        device: HidDevice,
        caps: DeviceCapabilities | None = None,
        poll_interval_ms: int = _DEFAULT_POLL_MS,
    ) -> None:
        self._device = device
        self._caps = caps
        self._poll_interval_ms = poll_interval_ms
        self._queue: asyncio.Queue[DeckEvent] = asyncio.Queue()
        self._running = False
        self._poll_task: asyncio.Task[None] | None = None
        key_count = caps.key_count if caps is not None else 0
        dial_count = caps.dial_count if caps is not None else 0
        self._prev_key_states: tuple[bool, ...] = (False,) * key_count
        self._prev_encoder_states: tuple[bool, ...] = (False,) * dial_count

    @property
    def queue(self) -> asyncio.Queue[DeckEvent]:
        """The asyncio queue that receives decoded device events."""
        return self._queue

    def start(self) -> None:
        """Start the input polling background task.

        Spawns an asyncio task that continuously polls the underlying
        HID device for input events and forwards decoded events to
        :attr:`queue`. Calling :meth:`start` more than once without an
        intervening :meth:`stop` will replace the previous polling task
        reference; callers should treat the method as one-shot per
        transport lifetime.

        Notes
        -----
        Must be called from within a running asyncio event loop.
        """
        self._running = True
        self._poll_task = asyncio.create_task(
            self._poll_loop(), name="deux-hid-poll"
        )

    def stop(self) -> None:
        """Stop polling.

        Signals the polling loop to exit and cancels the background
        task if it is still running. Safe to call multiple times and
        safe to call when :meth:`start` was never invoked; in both
        cases the method becomes a no-op.
        """
        self._running = False
        if self._poll_task and not self._poll_task.done():
            self._poll_task.cancel()

    async def _poll_loop(self) -> None:
        """Background task: poll HID input and enqueue events."""
        loop = asyncio.get_running_loop()
        executor = get_executor()

        try:
            while self._running and self._device.is_open:
                try:
                    event = await loop.run_in_executor(
                        executor,
                        self._device.read_input,
                        self._poll_interval_ms,
                    )
                except HidApiError:
                    logger.warning("HID device disconnected during poll")
                    # Nullify the handle immediately so no other async
                    # task can call into a stale OS handle (which causes
                    # SIGABRT / Trace/BPT trap on macOS).
                    self._device._handle = None  # noqa: SLF001
                    break

                if event is not None:
                    self._translate_event(event)
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.exception("HID poll loop crashed")

    def _translate_event(self, event: InputEvent) -> None:
        """Translate a HID input event into DeckEvent(s) on the queue.

        Parameters
        ----------
        event : InputEvent
            A parsed HID input event.
        """
        if isinstance(event, KeyStateEvent):
            self._handle_key_state(event)
        elif isinstance(event, TouchTapEvent):
            self._queue.put_nowait(
                TouchEvent(
                    event_type=EventType.TOUCH_SHORT,
                    x=event.x,
                    y=event.y,
                )
            )
        elif isinstance(event, TouchPressEvent):
            self._queue.put_nowait(
                TouchEvent(
                    event_type=EventType.TOUCH_LONG,
                    x=event.x,
                    y=event.y,
                )
            )
        elif isinstance(event, TouchFlickEvent):
            self._queue.put_nowait(
                TouchEvent(
                    event_type=EventType.TOUCH_DRAG,
                    x=event.start_x,
                    y=event.start_y,
                    x_out=event.end_x,
                    y_out=event.end_y,
                )
            )
        elif isinstance(event, EncoderButtonEvent):
            self._handle_encoder_buttons(event)
        elif isinstance(event, EncoderRotateEvent):
            self._handle_encoder_rotate(event)

    def _handle_key_state(self, event: KeyStateEvent) -> None:
        """Emit KeyEvents for keys that changed state.

        Parameters
        ----------
        event : KeyStateEvent
            The current key state snapshot.
        """
        for idx, pressed in enumerate(event.states):
            if (
                idx < len(self._prev_key_states)
                and pressed == self._prev_key_states[idx]
            ):
                continue
            self._queue.put_nowait(KeyEvent(key=idx, pressed=pressed))
        self._prev_key_states = event.states

    def _handle_encoder_buttons(self, event: EncoderButtonEvent) -> None:
        """Emit EncoderPressEvents for encoders that changed state.

        Parameters
        ----------
        event : EncoderButtonEvent
            The current encoder button state snapshot.
        """
        for idx, pressed in enumerate(event.states):
            if idx < len(self._prev_encoder_states) and pressed == self._prev_encoder_states[idx]:
                continue
            self._queue.put_nowait(
                EncoderPressEvent(encoder=idx, pressed=pressed)
            )
        self._prev_encoder_states = event.states

    def _handle_encoder_rotate(self, event: EncoderRotateEvent) -> None:
        """Emit EncoderTurnEvents for encoders that rotated.

        Parameters
        ----------
        event : EncoderRotateEvent
            The rotation ticks snapshot.
        """
        for idx, ticks in enumerate(event.ticks):
            if ticks != 0:
                self._queue.put_nowait(
                    EncoderTurnEvent(encoder=idx, direction=ticks)
                )

queue property 

queue: Queue[DeckEvent]

The asyncio queue that receives decoded device events.

start 

start() -> None

Start the input polling background task.

Spawns an asyncio task that continuously polls the underlying HID device for input events and forwards decoded events to :attr:queue. Calling :meth:start more than once without an intervening :meth:stop will replace the previous polling task reference; callers should treat the method as one-shot per transport lifetime.

Notes

Must be called from within a running asyncio event loop.

Source code in src/deux/runtime/transport.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def start(self) -> None:
    """Start the input polling background task.

    Spawns an asyncio task that continuously polls the underlying
    HID device for input events and forwards decoded events to
    :attr:`queue`. Calling :meth:`start` more than once without an
    intervening :meth:`stop` will replace the previous polling task
    reference; callers should treat the method as one-shot per
    transport lifetime.

    Notes
    -----
    Must be called from within a running asyncio event loop.
    """
    self._running = True
    self._poll_task = asyncio.create_task(
        self._poll_loop(), name="deux-hid-poll"
    )

stop 

stop() -> None

Stop polling.

Signals the polling loop to exit and cancels the background task if it is still running. Safe to call multiple times and safe to call when :meth:start was never invoked; in both cases the method becomes a no-op.

Source code in src/deux/runtime/transport.py 
101
102
103
104
105
106
107
108
109
110
111
def stop(self) -> None:
    """Stop polling.

    Signals the polling loop to exit and cancels the background
    task if it is still running. Safe to call multiple times and
    safe to call when :meth:`start` was never invoked; in both
    cases the method becomes a no-op.
    """
    self._running = False
    if self._poll_task and not self._poll_task.done():
        self._poll_task.cancel()

get_executor 

get_executor() -> ThreadPoolExecutor

Return the shared executor, creating it on first access.

The pool is lazily initialised so that import-time side-effects are avoided. All callers share the same instance.

Returns:

Type Description
ThreadPoolExecutor

The shared executor.
Source code in src/deux/runtime/_executor.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def get_executor() -> ThreadPoolExecutor:
    """Return the shared executor, creating it on first access.

    The pool is lazily initialised so that import-time side-effects are
    avoided.  All callers share the same instance.

    Returns
    -------
    ThreadPoolExecutor
        The shared executor.
    """
    global _executor  # noqa: PLW0603
    if _executor is None or _executor._shutdown:
        _executor = ThreadPoolExecutor(max_workers=_MAX_WORKERS)
    return _executor

shutdown_executor 

shutdown_executor(*, wait: bool = True) -> None

Shut down the shared executor.

Parameters:

Name Type Description Default
wait bool

If True, block until all in-flight tasks complete, ensuring no HID locks are held non-deterministically.

 True
Source code in src/deux/runtime/_executor.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def shutdown_executor(*, wait: bool = True) -> None:
    """Shut down the shared executor.

    Parameters
    ----------
    wait : bool, default=True
        If ``True``, block until all in-flight tasks complete, ensuring
        no HID locks are held non-deterministically.
    """
    global _executor  # noqa: PLW0603
    if _executor is not None:
        _executor.shutdown(wait=wait)
        _executor = None

list_devices async 

list_devices(*, deck_type: str | None = None) -> list[DeviceInfo]

Enumerate all connected Stream Deck devices.

Discovers devices via HID, opens each briefly to read serial and firmware information, then closes them. Returns a list of :class:DeviceInfo snapshots.

Parameters:

Name Type Description Default
deck_type str | None

If set, only return devices matching this type (e.g. "Stream Deck +").

 None

Returns:

Type Description
list[DeviceInfo]

A list of :class:DeviceInfo for each discovered device.
Source code in src/deux/runtime/discovery.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
async def list_devices(
    *,
    deck_type: str | None = None,
) -> list[DeviceInfo]:
    """Enumerate all connected Stream Deck devices.

    Discovers devices via HID, opens each briefly to read serial and
    firmware information, then closes them.  Returns a list of
    :class:`DeviceInfo` snapshots.

    Parameters
    ----------
    deck_type
        If set, only return devices matching this type
        (e.g. ``"Stream Deck +"``).

    Returns
    -------
    list[DeviceInfo]
        A list of :class:`DeviceInfo` for each discovered device.
    """
    loop = asyncio.get_running_loop()
    executor = get_executor()

    devices = await loop.run_in_executor(executor, enumerate_devices)

    results: list[DeviceInfo] = []
    for d in devices:
        try:
            await loop.run_in_executor(executor, d.open)
            info = DeviceInfo(
                deck_type=d.family,
                serial=d.serial_number,
                firmware=d.firmware_version,
                key_count=d.key_count,
                key_layout=d.key_layout,
                encoder_count=d.encoder_count,
                key_pixel_size=d.key_size,
                touchscreen_size=d.window_size if d.has_touch else (0, 0),
                key_image_format="JPEG",
            )
            if deck_type is not None and info.deck_type != deck_type:
                await loop.run_in_executor(executor, d.close)
                continue
            results.append(info)
            await loop.run_in_executor(executor, d.close)
        except (HidApiError, Exception):
            continue

    return results