Coverage for apio/utils/serial_util.py: 82%

1"""Serial devices related utilities.""" 

2 

3import re 

4from dataclasses import dataclass 

5from serial.tools.list_ports import comports 

6from serial.tools.list_ports_common import ListPortInfo 

7from apio.common.debug_util import is_debug 

8from apio.common.apio_console import cout 

9from apio.utils import usb_util 

10from apio.apio_context import ApioContext 

11 

12 

13@dataclass() 

14class SerialDevice: 

15 """A data class to hold the information of a single Serial device.""" 

16 

17 # pylint: disable=too-many-instance-attributes 

18 

19 port: str 

20 port_name: str 

21 vid: str 

22 pid: str 

23 manufacturer: str 

24 product: str 

25 serial_number: str 

26 device_type: str 

27 location: str 

28 

29 def __post_init__(self): 

30 """Check that vid, pid, has the format %04X.""" 

31 usb_util.check_usb_id_format(self.vid) 

32 usb_util.check_usb_id_format(self.pid) 

33 

34 def summary(self) -> str: 

35 """Returns a user friendly short description of this device.""" 

36 return ( 

37 f"[{self.port}] " 

38 f"[{self.vid}:{self.pid}] " 

39 f"[{self.manufacturer}] [{self.product}] " 

40 f"[{self.serial_number}]" 

41 ) 

42 

43 

44def scan_serial_devices(_: ApioContext) -> list[SerialDevice]: 

45 """Scan the connected serial devices and return their information.""" 

46 

47 # -- Initial empty device list 

48 devices = [] 

49 

50 # -- Use the serial.tools.list_ports module for reading the 

51 # -- serial ports. More info: 

52 # -- https://pyserial.readthedocs.io/en/latest/tools.html 

53 list_port_info: list[ListPortInfo] = list(comports()) 

54 assert isinstance(list_port_info, list) 

55 if list_port_info: 55 ↛ 59line 55 didn't jump to line 59 because the condition on line 55 was always true

56 assert isinstance(list_port_info[0], ListPortInfo) 

57 

58 # -- Collect the items that are USB serial ports. 

59 for port in list_port_info: 

60 

61 # -- Dump for debugging. 

62 if is_debug(1): 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 cout("Raw serial port:") 

64 cout(f" Device: {port.device}") 

65 cout(f" Hwid: {port.hwid}") 

66 cout(f" Interface: {port.interface}") 

67 

68 # # -- Skip if not a serial port. 

69 if not port.device: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 continue 

71 

72 # -- Skip if not a USB device. 

73 if not port.vid or not port.pid: 73 ↛ 81line 73 didn't jump to line 81 because the condition on line 73 was always true

74 continue 

75 

76 # -- If needed, append A or B to the serial numbers of a dual channel 

77 # -- FTDI devices such as FT2232H. 

78 # -- 

79 # -- Ideally these should be done in the comports() function of 

80 # -- the pyserial library. 

81 serial_number = port.serial_number or "" 

82 if serial_number and port.device.endswith(f"-{serial_number}0"): 

83 serial_number += "A" 

84 elif serial_number and port.device.endswith(f"-{serial_number}1"): 

85 serial_number += "B" 

86 

87 # -- Add device to list. 

88 devices.append( 

89 SerialDevice( 

90 port=port.device, 

91 port_name=port.name, 

92 vid=f"{port.vid:04X}", 

93 pid=f"{port.pid:04X}", 

94 manufacturer=port.manufacturer or "", 

95 product=port.product or "", 

96 serial_number=serial_number, 

97 device_type=usb_util.get_device_type(port.vid, port.pid), 

98 location=port.location or "", 

99 ) 

100 ) 

101 

102 # -- Sort by port name, case insensitive. 

103 devices = sorted(devices, key=lambda d: d.port.lower()) 

104 

105 if is_debug(1): 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true

106 cout(f"Found {len(devices)} serial device:") 

107 for device in devices: 

108 cout(str(device)) 

109 

110 # -- All done. 

111 return devices 

112 

113 

114@dataclass 

115class SerialDeviceFilter: 

116 """A class to filter a list of serial devices by attributes. We use the 

117 Fluent Interface design pattern so we can assert that the values that 

118 the caller passes as filters are not unintentionally None or empty 

119 unintentionally.""" 

120 

121 _vid: str | None = None 

122 _pid: str | None = None 

123 _product_regex: str | None = None 

124 _serial_port: str | None = None 

125 _serial_num: str | None = None 

126 

127 def summary(self) -> str: 

128 """User friendly representation of the filter""" 

129 terms = [] 

130 if self._vid: 

131 terms.append(f"VID={self._vid}") 

132 if self._pid: 

133 terms.append(f"PID={self._pid}") 

134 if self._product_regex: 

135 terms.append(f'REGEX="{self._product_regex}"') 

136 if self._serial_port: 

137 terms.append(f"PORT={self._serial_port}") 

138 if self._serial_num: 

139 terms.append(f'S/N="{self._serial_num}"') 

140 if terms: 

141 return "[" + ", ".join(terms) + "]" 

142 return "[all]" 

143 

144 def set_vid(self, vid: str) -> "SerialDeviceFilter": 

145 """Pass only devices with given vendor id.""" 

146 usb_util.check_usb_id_format(vid) 

147 self._vid = vid 

148 return self 

149 

150 def set_pid(self, pid: str) -> "SerialDeviceFilter": 

151 """Pass only devices given product id.""" 

152 usb_util.check_usb_id_format(pid) 

153 self._pid = pid 

154 return self 

155 

156 def set_product_regex(self, product_regex: str) -> "SerialDeviceFilter": 

157 """Pass only devices whose product string match given regex.""" 

158 assert product_regex 

159 self._product_regex = product_regex 

160 return self 

161 

162 def set_port(self, serial_port: str) -> "SerialDeviceFilter": 

163 """Pass only devices given product serial port..""" 

164 assert serial_port 

165 self._serial_port = serial_port 

166 return self 

167 

168 def set_serial_num(self, serial_num: str) -> "SerialDeviceFilter": 

169 """Pass only devices given product serial number..""" 

170 assert serial_num 

171 self._serial_num = serial_num 

172 return self 

173 

174 def _eval(self, device: SerialDevice) -> bool: 

175 """Test if the devices passes this field.""" 

176 if (self._vid is not None) and (self._vid != device.vid): 

177 return False 

178 

179 if (self._pid is not None) and (self._pid != device.pid): 

180 return False 

181 

182 if (self._product_regex is not None) and not re.search( 

183 self._product_regex, device.product 

184 ): 

185 return False 

186 

187 # -- We allow matching by full port string or by port name. 

188 if (self._serial_port is not None) and ( 

189 ( 

190 self._serial_port.lower() 

191 not in [device.port.lower(), device.port_name.lower()] 

192 ) 

193 ): 

194 return False 

195 

196 if (self._serial_num is not None) and ( 

197 self._serial_num.lower() != device.serial_number.lower() 

198 ): 

199 return False 

200 

201 return True 

202 

203 def filter(self, devices: list[SerialDevice]): 

204 """Return a copy of the list with items that are pass this filter. 

205 Items order is preserved.""" 

206 result = [d for d in devices if self._eval(d)] 

207 return result