"""Transport implementations Kv4pRadio can be constructed with."""
from __future__ import annotations
from collections.abc import Callable
[docs]
class Kv4pTransport:
"""Documents the interface Kv4pRadio needs from a transport.
Not enforced via abc/@abstractmethod on purpose: the shape a future
Bluetooth transport needs isn't known yet, and a formal contract would
just be a breaking change waiting to happen. Subclass and implement.
"""
[docs]
def open(
self,
on_frame: Callable[[int, bytes], None],
on_error: Callable[[Exception], None] | None = None,
) -> None:
"""Open the transport and start delivering incoming KISS frames to `on_frame`.
`on_error` is called (from the transport's own background thread, if it
has one) when the connection is lost unexpectedly, e.g. a disconnected
USB cable or a device reboot mid-read. It is not called on a normal
`close()`.
"""
raise NotImplementedError
[docs]
def close(self) -> None:
"""Close the transport."""
raise NotImplementedError
[docs]
def write_frame(self, command: int, payload: bytes) -> None:
"""Write one KISS frame."""
raise NotImplementedError
[docs]
def flush(self) -> None:
"""Wait until pending output is written."""
raise NotImplementedError
[docs]
def reset(self) -> None:
"""Reset the radio so it re-announces itself via HELLO."""
raise NotImplementedError