Coverage for kv4p/transports/serial.py: 0%

1"""Serial transport for the KV4P-HT (ESP32) radio.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import threading 

7import time 

8from collections.abc import Callable 

9 

10from kv4p.protocol.kiss import KissParser, encode_kiss_frame 

11from kv4p.transports import Kv4pTransport 

12 

13logger = logging.getLogger(__name__) 

14 

15# Standard ESP32 auto-program circuit: RTS drives EN/CHIP_PU (reset), DTR drives 

16# GPIO0 (boot mode select). pyserial dtr/rts=True is inverted to a LOW level on 

17# the board through the auto-program transistors. Holding DTR deasserted while 

18# pulsing RTS resets the chip into its normal firmware (not the ROM bootloader). 

19_RESET_PULSE_SECONDS = 0.1 

20_RESET_SETTLE_SECONDS = 0.05 

21 

22 

23class Kv4pSerialTransport(Kv4pTransport): 

24 """Blocking serial transport with an RX thread, for use with `Kv4pRadio`.""" 

25 

26 def __init__(self, device: str, baudrate: int) -> None: 

27 self._device = device 

28 self._baudrate = baudrate 

29 self._parser: KissParser | None = None 

30 self._serial = None 

31 self._thread: threading.Thread | None = None 

32 self._stop = threading.Event() 

33 self._write_lock = threading.Lock() 

34 self._on_error: Callable[[Exception], None] | None = None 

35 

36 def open( 

37 self, 

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

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

40 ) -> None: 

41 """Open serial port and start the RX thread.""" 

42 import serial 

43 

44 self._parser = KissParser(on_frame) 

45 self._on_error = on_error 

46 self._serial = serial.Serial(self._device, self._baudrate, timeout=0.2) 

47 self._serial.rts = False 

48 self._serial.dtr = False 

49 self._stop.clear() 

50 

51 self._thread = threading.Thread( 

52 target=self._read_loop, 

53 name="kv4p-rx", 

54 daemon=True, 

55 ) 

56 

57 self._thread.start() 

58 logger.info("serial open device=%s baudrate=%d", self._device, self._baudrate) 

59 

60 def close(self) -> None: 

61 """Stop RX thread and close serial port.""" 

62 self._stop.set() 

63 serial_port = self._serial 

64 self._serial = None 

65 

66 if serial_port is not None: 

67 serial_port.close() 

68 

69 if self._thread is not None: 

70 self._thread.join(timeout=2) 

71 

72 if self._thread.is_alive(): 

73 logger.warning("serial RX thread did not stop within timeout") 

74 

75 self._thread = None 

76 

77 logger.info("serial closed") 

78 

79 def reset(self) -> None: 

80 """Hardware-reset the ESP32 into its normal firmware via an RTS pulse. 

81 

82 DTR is held deasserted throughout so GPIO0 stays in run mode (not 

83 bootloader mode); RTS is pulsed to toggle EN/CHIP_PU. 

84 """ 

85 with self._write_lock: 

86 if self._serial is None: 

87 raise RuntimeError("serial transport is not open") 

88 self._serial.dtr = False 

89 self._serial.rts = True 

90 time.sleep(_RESET_PULSE_SECONDS) 

91 self._serial.rts = False 

92 time.sleep(_RESET_SETTLE_SECONDS) 

93 logger.info("serial reset device=%s", self._device) 

94 

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

96 """Write one KISS frame.""" 

97 frame = encode_kiss_frame(command, payload) 

98 

99 with self._write_lock: 

100 if self._serial is None: 

101 raise RuntimeError("serial transport is not open") 

102 

103 self._serial.write(frame) 

104 

105 logger.debug("serial tx KISS command=0x%02x payload=%d frame=%d", command, len(payload), len(frame)) 

106 

107 def flush(self) -> None: 

108 """Wait until serial output is written.""" 

109 with self._write_lock: 

110 self._serial.flush() 

111 

112 def _read_loop(self) -> None: 

113 while not self._stop.is_set(): 

114 try: 

115 data = self._serial.read(512) 

116 except Exception as exc: 

117 if not self._stop.is_set(): 

118 logger.exception("serial read failed") 

119 if self._on_error is not None: 

120 self._on_error(exc) 

121 return 

122 

123 if data: 

124 self._parser.feed(data)