Coverage for kv4p/transports/dummy.py: 98%

1"""In-memory transport that fakes a firmware HELLO reply, for tests/smoke runs without hardware.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import struct 

7import threading 

8from collections.abc import Callable 

9 

10from kv4p.constants.kiss import KISS_CMD_SETHARDWARE 

11from kv4p.constants.vendor import COMMAND_HELLO, KV4P_PROTOCOL_VERSION, KV4P_VENDOR_PREFIX 

12from kv4p.transports import Kv4pTransport 

13 

14logger = logging.getLogger(__name__) 

15 

16# Same layouts as messages/version.py and messages/device_state.py from_bytes(). 

17_VERSION = struct.Struct("<HcIBffB") 

18_DEVICE_STATE = struct.Struct("<IiHBffBBBcBBB") 

19 

20 

21class DummyTransport(Kv4pTransport): 

22 """Transport with no real radio: answers reset() with a synthetic HELLO.""" 

23 

24 def __init__( 

25 self, 

26 *, 

27 firmware_version: int = 17, 

28 window_size: int = 2048, 

29 min_radio_freq: float = 134.0, 

30 max_radio_freq: float = 174.0, 

31 features: int = 0, 

32 device_state_flags: int = 0, 

33 ) -> None: 

34 self._firmware_version = firmware_version 

35 self._window_size = window_size 

36 self._min_radio_freq = min_radio_freq 

37 self._max_radio_freq = max_radio_freq 

38 self._features = features 

39 self._device_state_flags = device_state_flags 

40 self._on_frame: Callable[[int, bytes], None] | None = None 

41 self.written: list[tuple[int, bytes]] = [] 

42 

43 def open( 

44 self, 

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

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

47 ) -> None: 

48 self._on_frame = on_frame 

49 logger.info("dummy transport open") 

50 

51 def close(self) -> None: 

52 self._on_frame = None 

53 logger.info("dummy transport close") 

54 

55 def reset(self) -> None: 

56 logger.info("dummy transport reset -> synthetic HELLO") 

57 payload = _encode_vendor_payload(COMMAND_HELLO, self._build_hello_bytes()) 

58 threading.Thread(target=self._deliver, args=(payload,), daemon=True).start() 

59 

60 def _build_hello_bytes(self) -> bytes: 

61 version = _VERSION.pack( 

62 self._firmware_version, 

63 b"D", 

64 self._window_size, 

65 0, 

66 self._min_radio_freq, 

67 self._max_radio_freq, 

68 self._features, 

69 ) 

70 device_state = _DEVICE_STATE.pack( 

71 0, # applied_sequence 

72 -1, # memory_id 

73 self._device_state_flags, # flags 

74 0, # bw 

75 self._min_radio_freq, # freq_tx 

76 self._min_radio_freq, # freq_rx 

77 0, # ctcss_tx 

78 1, # squelch 

79 0, # ctcss_rx 

80 b"D", # radio_module_status 

81 1, # mode (RX) 

82 0, # last_error 

83 0, # latest_rssi 

84 ) 

85 return version + device_state 

86 

87 def _deliver(self, payload: bytes) -> None: 

88 if self._on_frame is not None: 

89 self._on_frame(KISS_CMD_SETHARDWARE, payload) 

90 

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

92 self.written.append((command, payload)) 

93 logger.debug("dummy transport write command=0x%02x bytes=%d", command, len(payload)) 

94 

95 def flush(self) -> None: 

96 pass 

97 

98 

99def _encode_vendor_payload(command: int, payload: bytes = b"") -> bytes: 

100 return KV4P_VENDOR_PREFIX + bytes([KV4P_PROTOCOL_VERSION, command]) + payload