Skip to content

Transport

transport

Async wrappers for blocking HID I/O and input polling.

Bridges the synchronous hidapi calls into asyncio using a shared :class:~concurrent.futures.ThreadPoolExecutor. Provides an async input polling loop that yields parsed events.

HidWriteTimeout

Bases: TimeoutError

Raised when a blocking HID write exceeds the timeout threshold.

Source code in src/deux/runtime/hid/transport.py 
28
29
class HidWriteTimeout(TimeoutError):
    """Raised when a blocking HID write exceeds the timeout threshold."""

async_hid_call async 

async_hid_call(func: Callable[..., object], *args: object, timeout: float = HID_WRITE_TIMEOUT) -> object

Run a blocking HID function in the shared executor with a timeout.

Parameters:

Name Type Description Default
func Callable

The blocking function to execute (e.g. device.set_key_image).

 required
*args object

Arguments to pass to func.

 ()
timeout float

Maximum time in seconds before raising :class:HidWriteTimeout.

 HID_WRITE_TIMEOUT

Returns:

Type Description
object

The return value of func.

Raises:

Type Description
HidWriteTimeout

If the call exceeds timeout seconds.
HidApiError

If the underlying HID operation fails.
Source code in src/deux/runtime/hid/transport.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
async def async_hid_call(
    func: Callable[..., object],
    *args: object,
    timeout: float = HID_WRITE_TIMEOUT,
) -> object:
    """Run a blocking HID function in the shared executor with a timeout.

    Parameters
    ----------
    func : Callable
        The blocking function to execute (e.g. ``device.set_key_image``).
    *args : object
        Arguments to pass to *func*.
    timeout : float, default=HID_WRITE_TIMEOUT
        Maximum time in seconds before raising :class:`HidWriteTimeout`.

    Returns
    -------
    object
        The return value of *func*.

    Raises
    ------
    HidWriteTimeout
        If the call exceeds *timeout* seconds.
    HidApiError
        If the underlying HID operation fails.
    """
    loop = asyncio.get_running_loop()
    try:
        return await asyncio.wait_for(
            loop.run_in_executor(get_executor(), func, *args),
            timeout=timeout,
        )
    except TimeoutError:
        raise HidWriteTimeout(
            f"HID call {func.__name__} timed out after {timeout}s"
        ) from None

poll_input async 

poll_input(device: HidDevice, poll_interval_ms: int = DEFAULT_POLL_INTERVAL_MS) -> AsyncIterator[InputEvent]

Async generator that polls a device for input events.

Runs device.read_input() in the executor at the specified interval, yielding parsed events as they arrive. Stops when the device is closed or disconnected.

Parameters:

Name Type Description Default
device HidDevice

An open HID device.

 required
poll_interval_ms int

HID read timeout per poll cycle in milliseconds.

 50

Yields:

Type Description
InputEvent

Parsed input events (key press, touch, encoder).

Raises:

Type Description
HidApiError

If the device disconnects unexpectedly.
Source code in src/deux/runtime/hid/transport.py 
 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
async def poll_input(
    device: HidDevice,
    poll_interval_ms: int = DEFAULT_POLL_INTERVAL_MS,
) -> AsyncIterator[InputEvent]:
    """Async generator that polls a device for input events.

    Runs ``device.read_input()`` in the executor at the specified interval,
    yielding parsed events as they arrive.  Stops when the device is closed
    or disconnected.

    Parameters
    ----------
    device : HidDevice
        An open HID device.
    poll_interval_ms : int, default=50
        HID read timeout per poll cycle in milliseconds.

    Yields
    ------
    InputEvent
        Parsed input events (key press, touch, encoder).

    Raises
    ------
    HidApiError
        If the device disconnects unexpectedly.
    """
    loop = asyncio.get_running_loop()
    executor = get_executor()

    while device.is_open:
        try:
            event = await loop.run_in_executor(
                executor, device.read_input, poll_interval_ms
            )
        except HidApiError:
            logger.warning("HID device disconnected during poll")
            break

        if event is not None:
            yield event