Coverage for kv4p/transports/__init__.py: 62%

1"""Transport implementations Kv4pRadio can be constructed with.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6 

7 

8class Kv4pTransport: 

9 """Documents the interface Kv4pRadio needs from a transport. 

10 

11 Not enforced via abc/@abstractmethod on purpose: the shape a future 

12 Bluetooth transport needs isn't known yet, and a formal contract would 

13 just be a breaking change waiting to happen. Subclass and implement. 

14 """ 

15 

16 def open( 

17 self, 

18 on_frame: Callable[[int, bytes], None], 

19 on_error: Callable[[Exception], None] | None = None, 

20 ) -> None: 

21 """Open the transport and start delivering incoming KISS frames to `on_frame`. 

22 

23 `on_error` is called (from the transport's own background thread, if it 

24 has one) when the connection is lost unexpectedly, e.g. a disconnected 

25 USB cable or a device reboot mid-read. It is not called on a normal 

26 `close()`. 

27 """ 

28 raise NotImplementedError 

29 

30 def close(self) -> None: 

31 """Close the transport.""" 

32 raise NotImplementedError 

33 

34 def write_frame(self, command: int, payload: bytes) -> None: 

35 """Write one KISS frame.""" 

36 raise NotImplementedError 

37 

38 def flush(self) -> None: 

39 """Wait until pending output is written.""" 

40 raise NotImplementedError 

41 

42 def reset(self) -> None: 

43 """Reset the radio so it re-announces itself via HELLO.""" 

44 raise NotImplementedError