Coverage for kv4p/messages/device_state.py: 100%

1"""Firmware-applied device state.""" 

2 

3from __future__ import annotations 

4 

5import struct 

6from dataclasses import dataclass 

7from enum import IntEnum 

8 

9from kv4p.constants.messages import DEVICE_STATE_SQUELCHED 

10 

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

12DEVICE_STATE_SIZE = _DEVICE_STATE.size 

13 

14 

15class RadioMode(IntEnum): 

16 """Firmware radio mode, as reported in ``DeviceState.mode``.""" 

17 

18 TX = 0 

19 RX = 1 

20 STOPPED = 2 

21 UNKNOWN = -1 

22 

23 @classmethod 

24 def _missing_(cls, value: object) -> RadioMode: 

25 return cls.UNKNOWN 

26 

27 

28@dataclass(frozen=True, slots=True) 

29class DeviceState: 

30 """Firmware-applied state.""" 

31 

32 applied_sequence: int 

33 memory_id: int 

34 flags: int 

35 bw: int 

36 freq_tx: float 

37 freq_rx: float 

38 ctcss_tx: int 

39 squelch: int 

40 ctcss_rx: int 

41 radio_module_status: str 

42 mode: RadioMode 

43 last_error: int 

44 latest_rssi: int 

45 

46 @classmethod 

47 def from_bytes(cls, payload: bytes) -> DeviceState: 

48 """Parse DeviceState.""" 

49 if len(payload) < _DEVICE_STATE.size: 

50 raise ValueError(f"DeviceState payload too short: {len(payload)}") 

51 values = _DEVICE_STATE.unpack(payload[: _DEVICE_STATE.size]) 

52 return cls( 

53 applied_sequence=values[0], 

54 memory_id=values[1], 

55 flags=values[2], 

56 bw=values[3], 

57 freq_tx=values[4], 

58 freq_rx=values[5], 

59 ctcss_tx=values[6], 

60 squelch=values[7], 

61 ctcss_rx=values[8], 

62 radio_module_status=values[9].decode("ascii", errors="replace"), 

63 mode=RadioMode(values[10]), 

64 last_error=values[11], 

65 latest_rssi=values[12], 

66 ) 

67 

68 @property 

69 def sql_open(self) -> bool: 

70 """Return true when the device squelch is open.""" 

71 return not bool(self.flags & DEVICE_STATE_SQUELCHED)