Skip to content

Runtime

runtime

Runtime package for deckui 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/deckui/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
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)
    """

    __slots__ = ("_handlers",)

    def __init__(self) -> None:
        self._handlers: list[AsyncHandler] = []

    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.
        """
        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)

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/deckui/runtime/async_event.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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/deckui/runtime/async_event.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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/deckui/runtime/async_event.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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/deckui/runtime/async_event.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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.
    """
    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/deckui/runtime/capabilities.py 
 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
@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.
    """

    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

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

        Parameters
        ----------
        device
            An open ``StreamDeck`` device object.

        Returns
        -------
        DeviceCapabilities
            A frozen :class:`DeviceCapabilities` instance.
        """
        key_layout = device.key_layout()
        return cls(
            deck_type=device.DECK_TYPE,
            key_count=device.key_count(),
            key_cols=key_layout[0],
            key_rows=key_layout[1],
            key_pixel_width=device.KEY_PIXEL_WIDTH,
            key_pixel_height=device.KEY_PIXEL_HEIGHT,
            key_image_format=device.KEY_IMAGE_FORMAT,
            key_flip=_coerce_flip(device.KEY_FLIP),
            key_rotation=device.KEY_ROTATION,
            has_visual=device.DECK_VISUAL,
            has_touch=getattr(device, "DECK_TOUCH", False),
            dial_count=device.dial_count(),
            touchscreen_width=device.TOUCHSCREEN_PIXEL_WIDTH,
            touchscreen_height=device.TOUCHSCREEN_PIXEL_HEIGHT,
            touchscreen_image_format=getattr(
                device, "TOUCHSCREEN_IMAGE_FORMAT", ""
            ),
            touchscreen_flip=_coerce_flip(
                getattr(device, "TOUCHSCREEN_FLIP", (False, False))
            ),
            touchscreen_rotation=getattr(device, "TOUCHSCREEN_ROTATION", 0),
            has_screen=getattr(device, "SCREEN_PIXEL_WIDTH", 0) > 0,
            screen_width=getattr(device, "SCREEN_PIXEL_WIDTH", 0),
            screen_height=getattr(device, "SCREEN_PIXEL_HEIGHT", 0),
            screen_image_format=getattr(device, "SCREEN_IMAGE_FORMAT", ""),
            screen_flip=_coerce_flip(getattr(device, "SCREEN_FLIP", (False, False))),
            screen_rotation=getattr(device, "SCREEN_ROTATION", 0),
            touch_key_count=getattr(device, "TOUCH_KEY_COUNT", 0),
        )

    @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: StreamDeck) -> DeviceCapabilities

Build capabilities from a connected (opened) device.

Parameters:

Name Type Description Default
device StreamDeck

An open StreamDeck device object.

 required

Returns:

Type Description
DeviceCapabilities

A frozen :class:DeviceCapabilities instance.
Source code in src/deckui/runtime/capabilities.py 
 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
@classmethod
def from_device(cls, device: StreamDeck) -> DeviceCapabilities:
    """Build capabilities from a connected (opened) device.

    Parameters
    ----------
    device
        An open ``StreamDeck`` device object.

    Returns
    -------
    DeviceCapabilities
        A frozen :class:`DeviceCapabilities` instance.
    """
    key_layout = device.key_layout()
    return cls(
        deck_type=device.DECK_TYPE,
        key_count=device.key_count(),
        key_cols=key_layout[0],
        key_rows=key_layout[1],
        key_pixel_width=device.KEY_PIXEL_WIDTH,
        key_pixel_height=device.KEY_PIXEL_HEIGHT,
        key_image_format=device.KEY_IMAGE_FORMAT,
        key_flip=_coerce_flip(device.KEY_FLIP),
        key_rotation=device.KEY_ROTATION,
        has_visual=device.DECK_VISUAL,
        has_touch=getattr(device, "DECK_TOUCH", False),
        dial_count=device.dial_count(),
        touchscreen_width=device.TOUCHSCREEN_PIXEL_WIDTH,
        touchscreen_height=device.TOUCHSCREEN_PIXEL_HEIGHT,
        touchscreen_image_format=getattr(
            device, "TOUCHSCREEN_IMAGE_FORMAT", ""
        ),
        touchscreen_flip=_coerce_flip(
            getattr(device, "TOUCHSCREEN_FLIP", (False, False))
        ),
        touchscreen_rotation=getattr(device, "TOUCHSCREEN_ROTATION", 0),
        has_screen=getattr(device, "SCREEN_PIXEL_WIDTH", 0) > 0,
        screen_width=getattr(device, "SCREEN_PIXEL_WIDTH", 0),
        screen_height=getattr(device, "SCREEN_PIXEL_HEIGHT", 0),
        screen_image_format=getattr(device, "SCREEN_IMAGE_FORMAT", ""),
        screen_flip=_coerce_flip(getattr(device, "SCREEN_FLIP", (False, False))),
        screen_rotation=getattr(device, "SCREEN_ROTATION", 0),
        touch_key_count=getattr(device, "TOUCH_KEY_COUNT", 0),
    )

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/deckui/runtime/deck.py 
 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
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
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:
        """
        Parameters
        ----------
        serial_number
            The serial number of the target device.
        brightness
            Initial brightness (0-100).
        """
        self._serial_number = serial_number
        self._brightness = brightness
        self._device: Any = 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._closed_event = asyncio.Event()
        self._running = False
        self._executor = ThreadPoolExecutor(max_workers=2)
        self._device_lock = asyncio.Lock()

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

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

    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(self._executor, DeviceManager().enumerate)

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

        visual = [d for d in devices if d.DECK_VISUAL]
        if not visual:
            raise DeckError("No visual Stream Deck devices found")

        target = None
        for d in visual:
            try:
                await loop.run_in_executor(self._executor, d.open)
                serial = d.get_serial_number()
                if serial == self._serial_number:
                    target = d
                    break
                await loop.run_in_executor(self._executor, d.close)
            except Exception:
                continue
        if target is None:
            raise DeckError(
                f"No device with serial '{self._serial_number}' found"
            )
        self._device = target

        await loop.run_in_executor(self._executor, self._device.reset)
        await loop.run_in_executor(
            self._executor, 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.deck_type(),
            self._device.get_serial_number(),
            self._device.get_firmware_version(),
            self._caps.key_count,
            self._caps.dial_count,
        )

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

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

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

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

        self._running = False

        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:
            loop = asyncio.get_running_loop()
            try:
                await loop.run_in_executor(self._executor, self._device.reset)
                await loop.run_in_executor(self._executor, self._device.close)
            except Exception as e:
                logger.warning("Error closing device: %s", e)

        self._closed_event.set()
        self._executor.shutdown(wait=False)
        logger.info("Deck stopped")

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

    @property
    def device_path(self) -> str | None:
        """HID device path, or ``None`` if not opened."""
        if self._device is None:
            return None
        return cast("str | None", self._device.id())

    @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.get_serial_number(),
            firmware=self._device.get_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

    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.
        """
        clamped = max(0, min(100, percent))
        if clamped == self._brightness:
            return
        if self._device is not None:
            loop = asyncio.get_running_loop()
            async with self._device_lock:
                await loop.run_in_executor(
                    self._executor, self._device.set_brightness, clamped
                )
        self._brightness = clamped
        await self.on_brightness_changed.emit(clamped)

    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`.
        """
        from ..ui.screen import Screen

        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

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

        self._wire_refresh_callbacks()

        await self._render_all_keys()
        if self._active_screen.touch_strip is not None:
            await self._render_touchscreen()
        if self._active_screen.info_screen is not None:
            await self._render_info_screen()

        await self.on_screen_changed.emit(name, self._screens)

    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."""
        screen = self._current_screen()
        if not self._device or not screen:
            return

        caps = self._caps
        if caps is None:
            return

        loop = asyncio.get_running_loop()
        metrics = self._metrics

        if metrics is None:
            return

        for key_index in range(caps.key_count):
            key_slot = screen.keys.get(key_index)
            if key_slot and self._is_dui_key(key_slot):
                await self._render_dui_key(key_slot, key_index)
            else:
                image_bytes = render_blank_key(
                    key_size=metrics.key_size,
                    image_format=caps.key_image_format,
                )
                async with self._device_lock:
                    await loop.run_in_executor(
                        self._executor,
                        self._device.set_key_image,
                        key_index,
                        image_bytes,
                    )

    @staticmethod
    def _is_dui_key(key_slot: KeySlot) -> bool:
        """Check whether a key slot is a DuiKey."""
        return getattr(key_slot, "has_dui_content", False)

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

    async def _render_dui_key(self, key_slot: KeySlot, 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.  This is
            authoritative for routing (a single DuiKey may live on
            multiple screens at different slots), so the spinner
            ``push_fn`` is rewired on every render to capture the active
            slot.
        """
        if not self._device:
            return

        dui_key = cast("DuiKey", key_slot)

        # Skip rendering if a spinner animation is active
        if self._is_animating(dui_key):
            return

        # Re-wire push_fn every render so a DuiKey reused across screens
        # always animates at the slot of the currently active screen.
        async def _push_key_frame(frame_bytes: bytes) -> None:
            loop = asyncio.get_running_loop()
            async with self._device_lock:
                await loop.run_in_executor(
                    self._executor,
                    self._device.set_key_image,
                    key_index,
                    frame_bytes,
                )

        if self._caps is None:
            return

        dui_key.set_push_fn(_push_key_frame, key_size=self._caps.key_size)

        image_bytes = dui_key.render_image(
            key_size=self._caps.key_size,
            image_format=self._caps.key_image_format,
        )
        dui_key.set_rendered_image(image_bytes)

        loop = asyncio.get_running_loop()
        async with self._device_lock:
            await loop.run_in_executor(
                self._executor,
                self._device.set_key_image,
                key_index,
                image_bytes,
            )

    async def _render_touchscreen(self) -> None:
        """Render and push the full touch-strip image for the active screen."""
        screen = self._current_screen()
        if not self._device or not screen or not self._metrics:
            return

        if screen.touch_strip is None:
            return

        metrics = self._metrics

        # Re-wire push_fn for every DuiCard each render so that a card
        # reused on multiple screens always animates at the slot of the
        # currently active screen.
        from ..dui.card import DuiCard

        for card_idx, card in enumerate(screen.cards):
            if not isinstance(card, DuiCard):
                continue
            x_pos = card_idx * metrics.panel_width
            y_pos = 0

            async def _make_push(x: int, y: int, w: int, h: int) -> PushFn:
                async def _push_card_frame(frame_bytes: bytes) -> None:
                    loop = asyncio.get_running_loop()
                    async with self._device_lock:
                        await loop.run_in_executor(
                            self._executor,
                            self._device.set_touchscreen_image,
                            frame_bytes,
                            x,
                            y,
                            w,
                            h,
                        )

                return _push_card_frame

            push_fn = await _make_push(
                x_pos,
                y_pos,
                metrics.panel_width,
                metrics.panel_height,
            )
            card.set_push_fn(
                push_fn, panel_size=(metrics.panel_width, metrics.panel_height)
            )

        card_images = []
        for card in screen.cards:
            # Skip re-rendering cards with active spinner animations
            if self._is_animating(card):
                card_images.append(card.rendered)
                continue
            await card.prepare_assets()
            img = card.render()
            card.set_rendered(img)
            card_images.append(img)

        metrics = self._metrics
        touchstrip_bytes = compose_touchstrip(
            card_images,
            background=screen.touch_strip.background_color,
            touchscreen_width=metrics.touchscreen_width,
            touchscreen_height=metrics.touchscreen_height,
            panel_count=metrics.panel_count,
            panel_width=metrics.panel_width,
            image_format=self._caps.touchscreen_image_format if self._caps else "JPEG",
        )

        loop = asyncio.get_running_loop()
        async with self._device_lock:
            await loop.run_in_executor(
                self._executor,
                self._device.set_touchscreen_image,
                touchstrip_bytes,
                0,
                0,
                metrics.touchscreen_width,
                metrics.touchscreen_height,
            )

    async def _render_info_screen(self) -> None:
        """Render and push the info screen image (e.g. Neo)."""
        screen = self._current_screen()
        if not self._device or not screen:
            return

        info = screen.info_screen
        if info is None:
            return

        image_bytes = info.render_bytes()

        loop = asyncio.get_running_loop()
        async with self._device_lock:
            await loop.run_in_executor(
                self._executor,
                self._device.set_screen_image,
                image_bytes,
                0,
                0,
                info.width,
                info.height,
            )
        info.mark_clean()

    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.
        """
        screen = self._current_screen()
        if not screen:
            return

        for card in screen.cards:
            await self._drain_card_callbacks(card)

        for key_index, key_slot in screen.keys.items():
            if key_slot.is_dirty and self._is_dui_key(key_slot):
                await self._render_dui_key(key_slot, key_index)

        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()

    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()

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

        try:
            while self._running:
                try:
                    event = await asyncio.wait_for(
                        self._transport.queue.get(), timeout=0.5
                    )
                except TimeoutError:
                    await self._check_timeouts()
                    continue

                if not self._current_screen():
                    continue

                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:
            self._closed_event.set()

    async def _drain_card_callbacks(self, card: Card) -> None:
        """Drain and await all pending callbacks queued on a card."""
        for handler, args in card.drain_pending_callbacks():
            await handler(*args)

    async def _dispatch(self, event: DeckEvent) -> None:
        """Dispatch a single event to the appropriate handler on the active screen."""
        screen = self._current_screen()
        if not screen:
            return

        if isinstance(event, KeyEvent):
            key_slot = screen.keys.get(event.key)
            if key_slot:
                await key_slot.dispatch(event.pressed)
                if key_slot.is_dirty:
                    await self.refresh()

        elif isinstance(event, EncoderTurnEvent):
            encoder = screen.encoders.get(event.encoder)
            if encoder:
                await encoder.dispatch_turn(event.direction)
            if screen.touch_strip is not None:
                card = screen.touch_strip.card(event.encoder)
                await card.dispatch_encoder_turn(event.direction)
                await self._drain_card_callbacks(card)
                if card.is_dirty:
                    await self.refresh()

        elif isinstance(event, EncoderPressEvent):
            encoder = screen.encoders.get(event.encoder)
            if encoder:
                await encoder.dispatch_press(event.pressed)
            if screen.touch_strip is not None:
                card = screen.touch_strip.card(event.encoder)
                if event.pressed:
                    await card.dispatch_encoder_press()
                else:
                    await card.dispatch_encoder_release()
                await self._drain_card_callbacks(card)
                if card.is_dirty:
                    await self.refresh()

        elif isinstance(event, TouchEvent):
            if screen.touch_strip is not None and self._metrics is not None:
                zone = event.compute_zone(self._metrics)
                card = screen.touch_strip.card(zone)
                await card.dispatch_touch(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).

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

Parameters:

Name Type Description Default
serial_number str

The serial number of the target device.

 required
brightness int

Initial brightness (0-100).

 80
Source code in src/deckui/runtime/deck.py 
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
def __init__(
    self,
    serial_number: str,
    brightness: int = 80,
) -> None:
    """
    Parameters
    ----------
    serial_number
        The serial number of the target device.
    brightness
        Initial brightness (0-100).
    """
    self._serial_number = serial_number
    self._brightness = brightness
    self._device: Any = 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._closed_event = asyncio.Event()
    self._running = False
    self._executor = ThreadPoolExecutor(max_workers=2)
    self._device_lock = asyncio.Lock()

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

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

start async 

start() -> None

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

Source code in src/deckui/runtime/deck.py 
 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
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(self._executor, DeviceManager().enumerate)

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

    visual = [d for d in devices if d.DECK_VISUAL]
    if not visual:
        raise DeckError("No visual Stream Deck devices found")

    target = None
    for d in visual:
        try:
            await loop.run_in_executor(self._executor, d.open)
            serial = d.get_serial_number()
            if serial == self._serial_number:
                target = d
                break
            await loop.run_in_executor(self._executor, d.close)
        except Exception:
            continue
    if target is None:
        raise DeckError(
            f"No device with serial '{self._serial_number}' found"
        )
    self._device = target

    await loop.run_in_executor(self._executor, self._device.reset)
    await loop.run_in_executor(
        self._executor, 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.deck_type(),
        self._device.get_serial_number(),
        self._device.get_firmware_version(),
        self._caps.key_count,
        self._caps.dial_count,
    )

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

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

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

stop async 

stop() -> None

Stop the event loop and close the device.

Source code in src/deckui/runtime/deck.py 
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
async def stop(self) -> None:
    """Stop the event loop and close the device."""
    if not self._running:
        return

    self._running = False

    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:
        loop = asyncio.get_running_loop()
        try:
            await loop.run_in_executor(self._executor, self._device.reset)
            await loop.run_in_executor(self._executor, self._device.close)
        except Exception as e:
            logger.warning("Error closing device: %s", e)

    self._closed_event.set()
    self._executor.shutdown(wait=False)
    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/deckui/runtime/deck.py 
183
184
185
async def wait_closed(self) -> None:
    """Block until the deck is closed (e.g. by stop() or disconnect)."""
    await self._closed_event.wait()

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/deckui/runtime/deck.py 
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
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.
    """
    clamped = max(0, min(100, percent))
    if clamped == self._brightness:
        return
    if self._device is not None:
        loop = asyncio.get_running_loop()
        async with self._device_lock:
            await loop.run_in_executor(
                self._executor, self._device.set_brightness, clamped
            )
    self._brightness = clamped
    await self.on_brightness_changed.emit(clamped)

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/deckui/runtime/deck.py 
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
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`.
    """
    from ..ui.screen import Screen

    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/deckui/runtime/deck.py 
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
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

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

    self._wire_refresh_callbacks()

    await self._render_all_keys()
    if self._active_screen.touch_strip is not None:
        await self._render_touchscreen()
    if self._active_screen.info_screen is not None:
        await self._render_info_screen()

    await self.on_screen_changed.emit(name, self._screens)

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.

Source code in src/deckui/runtime/deck.py 
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
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.
    """
    screen = self._current_screen()
    if not screen:
        return

    for card in screen.cards:
        await self._drain_card_callbacks(card)

    for key_index, key_slot in screen.keys.items():
        if key_slot.is_dirty and self._is_dui_key(key_slot):
            await self._render_dui_key(key_slot, key_index)

    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()

DeckError

Bases: Exception

Raised for deck-level errors.

Source code in src/deckui/runtime/deck.py 
38
39
class DeckError(Exception):
    """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/deckui/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/deckui/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/deckui/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/deckui/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/deckui/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/deckui/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/deckui/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 deckui 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()

Parameters:

Name Type Description Default
poll_interval float

Seconds between device scans (default 2.0).

 2.0
brightness int

Default brightness for new Deck instances (0-100).

 80
auto_reconnect bool

If True (default), automatically reconnect devices that disconnect. The on_connect handler is called again on reconnection.

 True
Source code in src/deckui/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
class DeckManager:
    """The main entry point for the deckui 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()

    Parameters
    ----------
    poll_interval
        Seconds between device scans (default 2.0).
    brightness
        Default brightness for new Deck instances (0-100).
    auto_reconnect
        If ``True`` (default), automatically reconnect
        devices that disconnect.  The ``on_connect`` handler is
        called again on reconnection.
    """

    def __init__(
        self,
        poll_interval: float = 2.0,
        brightness: int = 80,
        auto_reconnect: bool = True,
    ) -> None:
        self._poll_interval = poll_interval
        self._brightness = brightness
        self._auto_reconnect = auto_reconnect
        self._running = False
        self._closed_event = asyncio.Event()
        self._executor = ThreadPoolExecutor(max_workers=2)
        self._scan_task: asyncio.Task[None] | None = None

        self._decks: dict[str, Deck] = {}

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

    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)
        self._decks.clear()

        self._closed_event.set()
        self._executor.shutdown(wait=False)
        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.
        """
        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.

        Examples
        --------
        ::

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

        def decorator(handler: AsyncHandler) -> AsyncHandler:
            self._disconnect_handler = 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(
                self._executor, DeviceManager().enumerate
            )
        except Exception:
            logger.debug("Device enumeration failed")
            return

        visual = [d for d in devices if d.DECK_VISUAL]

        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, Any] = {}

        for d in visual:
            dev_path = d.id()
            if dev_path in managed_paths:
                connected_serials.add(managed_paths[dev_path])
                continue
            try:
                await loop.run_in_executor(self._executor, d.open)
                serial = d.get_serial_number()
                connected_serials.add(serial)
                device_by_serial[serial] = d
                await loop.run_in_executor(self._executor, d.close)
            except Exception:
                continue

        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 Exception:
                    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_handler:
                    try:
                        await self._disconnect_handler(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.DECK_TYPE

            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.

Examples:

::

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

decks property 

decks: dict[str, Deck]

Currently managed decks, keyed by serial number.

__aenter__ async 

__aenter__() -> DeckManager

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

Source code in src/deckui/runtime/manager.py 
86
87
88
89
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/deckui/runtime/manager.py 
91
92
93
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/deckui/runtime/manager.py 
 95
 96
 97
 98
 99
100
101
102
103
104
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/deckui/runtime/manager.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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)
    self._decks.clear()

    self._closed_event.set()
    self._executor.shutdown(wait=False)
    logger.info("DeckManager stopped")

wait_closed async 

wait_closed() -> None

Block until the manager is stopped.

Source code in src/deckui/runtime/manager.py 
128
129
130
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.
Source code in src/deckui/runtime/manager.py 
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
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.
    """
    filters = {"serial": serial, "deck_type": deck_type}

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

    return decorator

AsyncTransport

Bridge sync streamdeck callbacks into an asyncio event queue.

Conditionally registers dial and touchscreen callbacks only when the device supports those features.

Parameters:

Name Type Description Default
device StreamDeck

An open Stream Deck device.

 required
loop AbstractEventLoop

The running asyncio event loop.

 required
caps DeviceCapabilities | None

Device capabilities (used to determine which callbacks to register).

 None
Source code in src/deckui/runtime/transport.py 
 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
class AsyncTransport:
    """Bridge sync streamdeck callbacks into an asyncio event queue.

    Conditionally registers dial and touchscreen callbacks only when
    the device supports those features.

    Parameters
    ----------
    device
        An open Stream Deck device.
    loop
        The running asyncio event loop.
    caps
        Device capabilities (used to determine which callbacks to register).
    """

    def __init__(
        self,
        device: StreamDeck,
        loop: asyncio.AbstractEventLoop,
        caps: DeviceCapabilities | None = None,
    ) -> None:
        self._device = device
        self._loop = loop
        self._caps = caps
        self._queue: asyncio.Queue[DeckEvent] = asyncio.Queue()
        self._running = False

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

    def start(self) -> None:
        """Register callbacks on the low-level device."""
        self._running = True
        self._device.set_key_callback(self._on_key)

        if self._caps is None or self._caps.has_encoders:
            self._device.set_dial_callback(self._on_encoder)
        if self._caps is None or self._caps.has_touchscreen:
            self._device.set_touchscreen_callback(self._on_touch)

    def stop(self) -> None:
        """Unregister callbacks."""
        self._running = False
        self._device.set_key_callback(None)

        if self._caps is None or self._caps.has_encoders:
            self._device.set_dial_callback(None)
        if self._caps is None or self._caps.has_touchscreen:
            self._device.set_touchscreen_callback(None)

    def _enqueue(self, event: DeckEvent) -> None:
        """Thread-safe: put an event onto the asyncio queue."""
        if not self._running:
            return
        self._loop.call_soon_threadsafe(self._queue.put_nowait, event)

    def _on_key(self, deck: StreamDeck, key: int, pressed: bool) -> None:
        """Handle a physical key press/release callback.

        Parameters
        ----------
        deck : StreamDeck
            The device that generated the event.
        key : int
            Zero-based key index.
        pressed : bool
            ``True`` on key-down, ``False`` on key-up.
        """
        try:
            self._enqueue(KeyEvent(key=key, pressed=pressed))
        except Exception:
            logger.exception("Error in key callback (key=%d)", key)

    def _on_encoder(
        self, deck: StreamDeck, encoder: int, event: DialEventType, value: object
    ) -> None:
        """Handle an encoder push or turn callback.

        Parameters
        ----------
        deck : StreamDeck
            The device that generated the event.
        encoder : int
            Zero-based encoder index.
        event : DialEventType
            Whether the dial was pushed or turned.
        value : object
            ``bool`` for push events, ``int`` direction for turn events.
        """
        try:
            if event == DialEventType.PUSH:
                self._enqueue(EncoderPressEvent(encoder=encoder, pressed=bool(value)))
            elif event == DialEventType.TURN:
                direction = value if isinstance(value, int) else 0
                self._enqueue(EncoderTurnEvent(encoder=encoder, direction=direction))
        except Exception:
            logger.exception("Error in encoder callback (encoder=%d)", encoder)

    def _on_touch(
        self,
        deck: StreamDeck,
        evt_type: TouchscreenEventType,
        value: Mapping[str, object],
    ) -> None:
        """Handle a touchscreen interaction callback.

        Parameters
        ----------
        deck : StreamDeck
            The device that generated the event.
        evt_type : TouchscreenEventType
            The kind of touch (short, long, or drag).
        value : Mapping[str, object]
            Touch coordinates (``x``, ``y``, and optionally ``x_out``, ``y_out``).
        """
        try:
            if evt_type == TouchscreenEventType.SHORT:
                event_type = EventType.TOUCH_SHORT
            elif evt_type == TouchscreenEventType.LONG:
                event_type = EventType.TOUCH_LONG
            elif evt_type == TouchscreenEventType.DRAG:
                event_type = EventType.TOUCH_DRAG
            else:
                return

            self._enqueue(
                TouchEvent(
                    event_type=event_type,
                    x=_coerce_int(value.get("x", 0)),
                    y=_coerce_int(value.get("y", 0)),
                    x_out=_optional_int(value.get("x_out")),
                    y_out=_optional_int(value.get("y_out")),
                )
            )
        except Exception:
            logger.exception("Error in touch callback")

queue property 

queue: Queue[DeckEvent]

The asyncio queue that receives decoded device events.

start 

start() -> None

Register callbacks on the low-level device.

Source code in src/deckui/runtime/transport.py 
 97
 98
 99
100
101
102
103
104
105
def start(self) -> None:
    """Register callbacks on the low-level device."""
    self._running = True
    self._device.set_key_callback(self._on_key)

    if self._caps is None or self._caps.has_encoders:
        self._device.set_dial_callback(self._on_encoder)
    if self._caps is None or self._caps.has_touchscreen:
        self._device.set_touchscreen_callback(self._on_touch)

stop 

stop() -> None

Unregister callbacks.

Source code in src/deckui/runtime/transport.py 
107
108
109
110
111
112
113
114
115
def stop(self) -> None:
    """Unregister callbacks."""
    self._running = False
    self._device.set_key_callback(None)

    if self._caps is None or self._caps.has_encoders:
        self._device.set_dial_callback(None)
    if self._caps is None or self._caps.has_touchscreen:
        self._device.set_touchscreen_callback(None)

list_devices async 

list_devices(*, deck_type: str | None = None, visual_only: bool = True) -> 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
visual_only bool

If True (default), exclude non-visual devices such as the Stream Deck Pedal.

 True

Returns:

Type Description
list[DeviceInfo]

A list of :class:DeviceInfo for each discovered device.
Source code in src/deckui/runtime/discovery.py 
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
async def list_devices(
    *,
    deck_type: str | None = None,
    visual_only: bool = True,
) -> 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 +"``).
    visual_only
        If ``True`` (default), exclude non-visual devices
        such as the Stream Deck Pedal.

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

    try:
        devices: list[Any] = await loop.run_in_executor(executor, DeviceManager().enumerate)

        if visual_only:
            devices = [d for d in devices if d.DECK_VISUAL]

        results: list[DeviceInfo] = []
        for d in devices:
            try:
                await loop.run_in_executor(executor, d.open)
                info = DeviceInfo(
                    deck_type=d.deck_type(),
                    serial=d.get_serial_number(),
                    firmware=d.get_firmware_version(),
                    key_count=d.key_count(),
                    key_layout=d.key_layout(),
                    encoder_count=d.dial_count(),
                    key_pixel_size=(d.KEY_PIXEL_WIDTH, d.KEY_PIXEL_HEIGHT),
                    touchscreen_size=(
                        d.TOUCHSCREEN_PIXEL_WIDTH,
                        d.TOUCHSCREEN_PIXEL_HEIGHT,
                    ),
                    key_image_format=d.KEY_IMAGE_FORMAT,
                )
                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 Exception:
                continue

        return results
    finally:
        executor.shutdown(wait=False)