Coverage for kv4p/state_tracker.py: 85%

1"""Protocol state machine: HELLO handshake, device-state cache, HostDesiredState building.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import threading 

7from collections.abc import Callable 

8 

9from kv4p.constants.messages import ( 

10 DEVICE_STATE_PHYS_PTT_DOWN, 

11 DEVICE_STATE_SQUELCHED, 

12 DEVICE_STATE_TX_ACTIVE, 

13 HOST_STATE_PTT_REQUESTED, 

14 HOST_STATE_RADIO_CONFIG_VALID, 

15 HOST_STATE_TX_ALLOWED, 

16) 

17from kv4p.constants.vendor import COMMAND_AUDIO_ADPCM, COMMAND_AUDIO_OPUS 

18from kv4p.messages.desired_state import HostDesiredState, dra818_to_bandwidth 

19from kv4p.messages.device_state import DeviceState, RadioMode 

20from kv4p.messages.hello import Hello 

21 

22logger = logging.getLogger(__name__) 

23 

24# Firmware 17 is the last known version using Opus TX audio on 0x07. There is 

25# no published feature bit to detect ADPCM (0x0C) support, and it's unknown 

26# at which version upstream actually switches — this threshold is a guess 

27# based on "17 is confirmed 0x07" and must be revisited once a newer firmware 

28# version's actual behavior is known. 

29_OPUS_MAX_FW = 17 

30 

31 

32class DeviceStateTracker: 

33 """Tracks the firmware handshake/state and builds outgoing HostDesiredState frames. 

34 

35 Radio settings (frequency, bandwidth, squelch, CTCSS, ...) are seeded from 

36 the DeviceState carried in HELLO — the firmware always reports its actual 

37 tuned state there, right after `Kv4pRadio.open()`/`reset()` forces a 

38 reboot. There is no separate "desired settings" object with its own 

39 defaults; `set_*()` calls mutate this tracked state directly. 

40 """ 

41 

42 def __init__( 

43 self, 

44 send_desired_state: Callable[[HostDesiredState], None], 

45 on_rx_audio: Callable[[bytes], None] | None = None, 

46 on_sql: Callable[[bool], None] | None = None, 

47 on_phy_ptt: Callable[[bool], None] | None = None, 

48 on_tx_active: Callable[[bool], None] | None = None, 

49 ) -> None: 

50 self._send_desired_state = send_desired_state 

51 self._on_rx_audio = on_rx_audio 

52 self._on_sql = on_sql 

53 self._on_phy_ptt = on_phy_ptt 

54 self._on_tx_active = on_tx_active 

55 

56 self._lock = threading.RLock() 

57 self._hello_event = threading.Event() 

58 self._hello: Hello | None = None 

59 self._device_state: DeviceState | None = None 

60 self._sequence = 0 

61 self._flags = HOST_STATE_RADIO_CONFIG_VALID 

62 self._tx_audio_command = COMMAND_AUDIO_OPUS 

63 

64 # Radio settings, seeded from HELLO's DeviceState in on_hello(). 

65 self._freq_rx = 0.0 

66 self._freq_tx = 0.0 

67 self._bw = 0 

68 self._squelch = 0 

69 self._ctcss_rx = 0 

70 self._ctcss_tx = 0 

71 

72 self._last_sql_open: bool | None = None 

73 self._last_phy_ptt: bool | None = None 

74 self._last_tx_active: bool | None = None 

75 self._last_status_key: tuple[object, ...] | None = None 

76 

77 # -- handshake / incoming state ----------------------------------------- 

78 

79 def on_hello(self, hello: Hello) -> None: 

80 """Handle a HELLO frame (only ever sent once, right after the ESP32 boots).""" 

81 with self._lock: 

82 self._hello = hello 

83 self._device_state = hello.device_state 

84 self._sequence = hello.device_state.applied_sequence 

85 self._flags = (hello.device_state.flags 

86 & ~(DEVICE_STATE_PHYS_PTT_DOWN | DEVICE_STATE_TX_ACTIVE | DEVICE_STATE_SQUELCHED )) \ 

87 | HOST_STATE_RADIO_CONFIG_VALID 

88 self._seed_settings_locked(hello.device_state) 

89 self._tx_audio_command = ( 

90 COMMAND_AUDIO_ADPCM if hello.version.ver > _OPUS_MAX_FW else COMMAND_AUDIO_OPUS 

91 ) 

92 logger.info( 

93 "HELLO firmware=%d window=%d radio=%s range=%.3f-%.3f features=0x%02x", 

94 hello.version.ver, 

95 hello.version.window_size, 

96 hello.version.radio_module_status, 

97 hello.version.min_radio_freq, 

98 hello.version.max_radio_freq, 

99 hello.version.features, 

100 ) 

101 self._hello_event.set() 

102 self.on_device_state(hello.device_state) 

103 

104 def wait_for_hello(self, timeout: float | None = None) -> bool: 

105 """Block until HELLO has been received.""" 

106 return self._hello_event.wait(timeout=timeout) 

107 

108 def on_device_state(self, state: DeviceState) -> None: 

109 """Handle a DEVICE_STATE frame.""" 

110 with self._lock: 

111 self._device_state = state 

112 sql_open = state.sql_open 

113 if state.applied_sequence > self._sequence: 

114 self._sequence = state.applied_sequence 

115 

116 self._log_device_status(state) 

117 

118 if sql_open != self._last_sql_open: 

119 self._last_sql_open = sql_open 

120 logger.info("sql %s", "open" if sql_open else "closed") 

121 if self._on_sql is not None: 

122 self._on_sql(sql_open) 

123 

124 phy_ptt = bool(state.flags & DEVICE_STATE_PHYS_PTT_DOWN) 

125 if phy_ptt != self._last_phy_ptt: 

126 self._last_phy_ptt = phy_ptt 

127 logger.info("physical ptt %s", "down" if phy_ptt else "up") 

128 if self._on_phy_ptt is not None: 

129 self._on_phy_ptt(phy_ptt) 

130 

131 tx_active = bool(state.flags & DEVICE_STATE_TX_ACTIVE) 

132 if tx_active != self._last_tx_active: 

133 self._last_tx_active = tx_active 

134 logger.info("tx active %s", tx_active) 

135 if self._on_tx_active is not None: 

136 self._on_tx_active(tx_active) 

137 

138 def on_rx_audio(self, payload: bytes) -> None: 

139 """Handle an RX audio payload.""" 

140 if self._on_rx_audio is not None: 

141 self._on_rx_audio(payload) 

142 

143 # -- setters ---------------------------------------------------------------- 

144 

145 def request_ptt(self, enabled: bool) -> bool: 

146 """Set/clear the PTT-requested bit. Returns True if the flags actually changed.""" 

147 with self._lock: 

148 if enabled and not (self._flags & HOST_STATE_TX_ALLOWED): 

149 logger.warning("PTT requested while TX is not allowed") 

150 old_flags = self._flags 

151 if enabled: 

152 self._flags |= HOST_STATE_PTT_REQUESTED 

153 else: 

154 self._flags &= ~HOST_STATE_PTT_REQUESTED 

155 if self._flags == old_flags: 

156 return False 

157 logger.info("ptt %s", "on" if enabled else "off") 

158 self._send_desired_state_locked() 

159 return True 

160 

161 def set_frequency(self, *, rx: float | None = None, tx: float | None = None) -> None: 

162 """Update RX/TX frequency and send the new desired state.""" 

163 with self._lock: 

164 if rx is not None: 

165 self._freq_rx = rx 

166 if tx is not None: 

167 self._freq_tx = tx 

168 self._send_desired_state_locked() 

169 

170 def set_bandwidth(self, bw: int) -> None: 

171 """Update bandwidth (a DRA818_* constant) and send the new desired state.""" 

172 with self._lock: 

173 self._bw = bw 

174 self._send_desired_state_locked() 

175 

176 def set_squelch(self, squelch: int) -> None: 

177 """Update squelch level and send the new desired state.""" 

178 with self._lock: 

179 self._squelch = squelch 

180 self._send_desired_state_locked() 

181 

182 def set_ctcss(self, *, rx: int | None = None, tx: int | None = None) -> None: 

183 """Update RX/TX CTCSS tone and send the new desired state.""" 

184 with self._lock: 

185 if rx is not None: 

186 self._ctcss_rx = rx 

187 if tx is not None: 

188 self._ctcss_tx = tx 

189 self._send_desired_state_locked() 

190 

191 def set_flag(self, flag: int, enabled: bool) -> None: 

192 """Set/clear one of the HOST_STATE_* option bits and send the new desired state.""" 

193 with self._lock: 

194 if enabled: 

195 self._flags |= flag 

196 else: 

197 self._flags &= ~flag 

198 self._send_desired_state_locked() 

199 

200 def _send_desired_state_locked(self) -> None: 

201 applied_sequence = self._device_state.applied_sequence if self._device_state is not None else 0 

202 self._sequence = max(self._sequence, applied_sequence) + 1 

203 state = self._build_desired_state_locked() 

204 # Release the lock before handing off to I/O: the caller's send callback 

205 # may block on flow control / serial writes and must not hold up readers 

206 # of phy_ptt/mode from other threads. 

207 self._lock.release() 

208 try: 

209 self._send_desired_state(state) 

210 finally: 

211 self._lock.acquire() 

212 logger.info( 

213 ( 

214 "desired state sequence=%d flags=0x%04x rx=%.5f tx=%.5f " 

215 "bw=%s squelch=%d ctcss_rx=%d ctcss_tx=%d" 

216 ), 

217 state.sequence, 

218 state.flags, 

219 state.freq_rx, 

220 state.freq_tx, 

221 dra818_to_bandwidth(state.bw), 

222 state.squelch, 

223 state.ctcss_rx, 

224 state.ctcss_tx, 

225 ) 

226 

227 def _build_desired_state_locked(self) -> HostDesiredState: 

228 return HostDesiredState( 

229 sequence=self._sequence, 

230 memory_id=-1, 

231 flags=self._flags, 

232 bw=self._bw, 

233 freq_tx=self._freq_tx, 

234 freq_rx=self._freq_rx, 

235 ctcss_tx=self._ctcss_tx, 

236 squelch=self._squelch, 

237 ctcss_rx=self._ctcss_rx, 

238 ) 

239 

240 # -- derived properties ----------------------------------------------- 

241 

242 @property 

243 def hello(self) -> Hello | None: 

244 with self._lock: 

245 return self._hello 

246 

247 @property 

248 def device_state(self) -> DeviceState | None: 

249 with self._lock: 

250 return self._device_state 

251 

252 @property 

253 def flags(self) -> int: 

254 with self._lock: 

255 return self._flags 

256 

257 @property 

258 def freq_rx(self) -> float: 

259 with self._lock: 

260 return self._freq_rx 

261 

262 @property 

263 def freq_tx(self) -> float: 

264 with self._lock: 

265 return self._freq_tx 

266 

267 @property 

268 def bandwidth(self) -> str: 

269 with self._lock: 

270 return dra818_to_bandwidth(self._bw) 

271 

272 @property 

273 def squelch(self) -> int: 

274 with self._lock: 

275 return self._squelch 

276 

277 @property 

278 def ctcss_rx(self) -> int: 

279 with self._lock: 

280 return self._ctcss_rx 

281 

282 @property 

283 def ctcss_tx(self) -> int: 

284 with self._lock: 

285 return self._ctcss_tx 

286 

287 @property 

288 def tx_audio_command(self) -> int: 

289 """TX audio vendor command, guessed once in on_hello() from the firmware version. 

290 

291 See `_OPUS_MAX_FW` above — there is no feature bit to detect this properly yet. 

292 """ 

293 with self._lock: 

294 return self._tx_audio_command 

295 

296 @property 

297 def phy_ptt(self) -> bool: 

298 with self._lock: 

299 if self._device_state is None: 

300 return False 

301 return bool(self._device_state.flags & DEVICE_STATE_PHYS_PTT_DOWN) 

302 

303 @property 

304 def tx_active(self) -> bool: 

305 with self._lock: 

306 if self._device_state is None: 

307 return False 

308 return bool(self._device_state.flags & DEVICE_STATE_TX_ACTIVE) 

309 

310 @property 

311 def sql_open(self) -> bool: 

312 with self._lock: 

313 if self._device_state is None: 

314 return False 

315 return self._device_state.sql_open 

316 

317 @property 

318 def mode(self) -> RadioMode | None: 

319 with self._lock: 

320 if self._device_state is None: 

321 return None 

322 return self._device_state.mode 

323 

324 # -- internals ----------------------------------------------------------- 

325 

326 def _seed_settings_locked(self, state: DeviceState) -> None: 

327 """Seed radio settings from the firmware's actual tuned state at boot.""" 

328 self._freq_rx = state.freq_rx 

329 self._freq_tx = state.freq_tx 

330 self._bw = state.bw 

331 self._squelch = state.squelch 

332 self._ctcss_rx = state.ctcss_rx 

333 self._ctcss_tx = state.ctcss_tx 

334 

335 def _log_device_status(self, state: DeviceState) -> None: 

336 key = ( 

337 state.applied_sequence, 

338 state.flags, 

339 state.mode, 

340 state.last_error, 

341 round(state.freq_rx, 5), 

342 round(state.freq_tx, 5), 

343 state.bw, 

344 state.squelch, 

345 state.ctcss_rx, 

346 state.ctcss_tx, 

347 state.radio_module_status, 

348 state.latest_rssi if state.mode == RadioMode.TX else None, 

349 ) 

350 if key == self._last_status_key: 

351 return 

352 self._last_status_key = key 

353 logger.info( 

354 ( 

355 "radio status mode=%s sql=%s rx=%.5f tx=%.5f bw=%s " 

356 "squelch=%d ctcss_rx=%d ctcss_tx=%d flags=0x%04x " 

357 "applied_sequence=%d error=%d rssi=%d module=%s" 

358 ), 

359 state.mode.name, 

360 "open" if state.sql_open else "closed", 

361 state.freq_rx, 

362 state.freq_tx, 

363 dra818_to_bandwidth(state.bw), 

364 state.squelch, 

365 state.ctcss_rx, 

366 state.ctcss_tx, 

367 state.flags, 

368 state.applied_sequence, 

369 state.last_error, 

370 state.latest_rssi, 

371 state.radio_module_status, 

372 )