Coverage for apio/utils/usb_util.py: 66%

1# This file is part of the Apio project 

2# (C) 2016-present FPGAwars 

3# Project authors: 

4# * Jesús Arroyo (Jesus89) (2016-2019) 

5# * Juan Gonzalez (obijuan) (2019-2024) 

6# * The Apio Team (2024-present) 

7# License: GPLv2 

8 

9"""USB devices related utilities.""" 

10 

11import re 

12from glob import glob 

13from typing import Any 

14from dataclasses import dataclass 

15import usb.core 

16import usb.backend.libusb1 

17from apio.common.debug_util import is_debug 

18from apio.common.apio_console import cout, fatal_error 

19from apio.apio_context import ApioContext 

20 

21# -- Mapping of (VID), and (VID:PID) to device type. This is presented to the 

22# -- user as an information only. Add more as you like. 

23 

24_USB_TYPES = { 

25 # -- FTDI 

26 (0x0403): "FTDI", 

27 (0x0403, 0x6001): "FT232R", 

28 (0x0403, 0x6010): "FT2232H", 

29 (0x0403, 0x6011): "FT4232H", 

30 (0x0403, 0x6014): "FT232H", 

31 (0x0403, 0x6017): "FT313H", 

32 (0x0403, 0x8372): "FT245R", 

33 (0x0403, 0x8371): "FT232BM", 

34 (0x0403, 0x8373): "FT2232C", 

35 (0x0403, 0x8374): "FT4232", 

36} 

37 

38 

39def get_device_type(vid: int, pid: int) -> str: 

40 """Determine device type string. Try to match by (vid, pid) and if 

41 not found, by (vid). Returns "" if not found.""" 

42 device_type = _USB_TYPES.get((vid, pid), "") 

43 if not device_type: 

44 device_type = _USB_TYPES.get((vid), "") 

45 return device_type 

46 

47 

48def check_usb_id_format(usb_id: str) -> None: 

49 """Check that a vid or pid is in 4 char uppercase hex.""" 

50 if not re.search(r"^[0-9A-F]{4}$", usb_id): 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 raise ValueError(f"Invalid 04X hex value: [{usb_id}]") 

52 

53 

54@dataclass() 

55class UsbDevice: 

56 """A data class to hold the information of a single USB device.""" 

57 

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

59 

60 vid: str 

61 pid: str 

62 bus: int 

63 device: int 

64 manufacturer: str 

65 product: str 

66 serial_number: str 

67 device_type: str 

68 

69 def __post_init__(self): 

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

71 check_usb_id_format(self.vid) 

72 check_usb_id_format(self.pid) 

73 

74 def summary(self) -> str: 

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

76 return ( 

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

78 f"[{self.bus}:{self.device}] " 

79 f"[{self.manufacturer}] " 

80 f"[{self.product}] [{self.serial_number}]" 

81 ) 

82 

83 

84def _get_usb_str(device: usb.core.Device, index: int, default: str) -> str: 

85 """Extract usb string by its index.""" 

86 # pylint: disable=broad-exception-caught 

87 try: 

88 s = str(usb.util.get_string(device, index)) 

89 # For Tang 9K which contains a null char as a string separator. 

90 # It's not USB standard but C tools do that implicitly. 

91 s = s.split("\x00", 1)[0] 

92 return s 

93 except Exception as e: 

94 if is_debug(1): 

95 cout(f"Error getting USB string at index {index}: {e}") 

96 return default 

97 

98 

99def scan_usb_devices(apio_ctx: ApioContext) -> list[UsbDevice]: 

100 """Query and return a list with usb device info.""" 

101 # pylint: disable=too-many-locals 

102 

103 # -- Track the names we searched for. For diagnostics. 

104 searched_names = [] 

105 

106 def find_library(name: str): 

107 """A callback for looking up the libusb backend file.""" 

108 

109 # -- Track searched names, for diagnostics 

110 searched_names.append(name) 

111 

112 # -- Try to match to a lib in oss-cad-suite/lib. 

113 oss_dir = apio_ctx.get_package_dir("oss-cad-suite") 

114 pattern = oss_dir / "lib" / f"lib{name}*" 

115 files = glob(str(pattern)) 

116 

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

118 cout("Apio find_library() call:") 

119 cout(f" {name=}") 

120 cout(f" {pattern=}") 

121 cout(f" {files=}") 

122 

123 # -- We don't expect multiple matches. 

124 if len(files) > 1: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 fatal_error(f"Found multiple backends for '{name}': {files}") 

126 

127 if files: 127 ↛ 129line 127 didn't jump to line 129 because the condition on line 127 was always true

128 return files[0] 

129 return None 

130 

131 # -- Lookup libusb backend library file in oss-cad-suite/lib. 

132 backend = usb.backend.libusb1.get_backend(find_library=find_library) 

133 

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

135 cout(f"usb get backend attempt1: {backend}") 

136 

137 # -- If it doesn't load (e.g. it needs a newer glibc than the system 

138 # -- has), try the system's libusb. 

139 if not backend: 139 ↛ 144line 139 didn't jump to line 144 because the condition on line 139 was always true

140 backend = usb.backend.libusb1.get_backend() 

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

142 cout(f"usb get backend attempt1: {backend}") 

143 

144 if not backend: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 fatal_error( 

146 "Libusb backend not found", 

147 info=f"Searched names {searched_names} in oss-cad-suite/lib " 

148 + "and the system's libusb.", 

149 ) 

150 

151 # -- Find the usb devices. 

152 raw_devices = usb.core.find(find_all=True, backend=backend) 

153 devices: list[Any] = list(raw_devices) if raw_devices else [] 

154 

155 # -- Collect the devices 

156 result: list[UsbDevice] = [] 

157 for device in devices: 157 ↛ 159line 157 didn't jump to line 159 because the loop on line 157 never started

158 # -- Print entire raw device info for debugging. 

159 if is_debug(1): 

160 cout() 

161 cout(str(device)) 

162 cout() 

163 

164 # -- Sanity check. 

165 assert isinstance(device, usb.core.Device), type(device) 

166 

167 # -- Skip hubs, they are not interesting 

168 d = device.bDeviceClass 

169 if d == 0x09: 

170 continue 

171 

172 # -- Lookup device type or "" if not found. 

173 device_type = get_device_type( 

174 device.idVendor, 

175 device.idProduct, 

176 ) 

177 

178 # -- Create the device object. 

179 unavail = "--unavail--" 

180 vid = device.idVendor 

181 pid = device.idProduct 

182 

183 d = device 

184 man = d.iManufacturer 

185 iser = d.iSerialNumber 

186 item = UsbDevice( 

187 vid=f"{vid:04X}", 

188 pid=f"{pid:04X}", 

189 bus=device.bus, 

190 device=device.address or 0, 

191 manufacturer=_get_usb_str( 

192 device, 

193 man, 

194 default=unavail, 

195 ), 

196 product=_get_usb_str( 

197 device, 

198 device.iProduct, 

199 default=unavail, 

200 ), 

201 serial_number=_get_usb_str( 

202 device, 

203 iser, 

204 default="", 

205 ), 

206 device_type=device_type, 

207 ) 

208 result.append(item) 

209 

210 # -- Sort by (vendor, product, bus, device). 

211 result = sorted( 

212 result, 

213 key=lambda d: ( 

214 d.vid.lower(), 

215 d.pid.lower(), 

216 d.bus, 

217 d.device, 

218 ), 

219 ) 

220 

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

222 cout(f"Found {len(result)} USB devices:") 

223 for device in result: 

224 cout(str(device)) 

225 

226 # -- All done. 

227 return result 

228 

229 

230@dataclass 

231class UsbDeviceFilter: 

232 """A class to filter a list of usb devices by attributes. We use the 

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

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

235 unintentionally.""" 

236 

237 _vid: str | None = None 

238 _pid: str | None = None 

239 product_regex: str | None = None 

240 _serial_num: str | None = None 

241 

242 def summary(self) -> str: 

243 """User friendly representation of the filter""" 

244 terms = [] 

245 

246 if self._vid: 

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

248 if self._pid: 

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

250 if self.product_regex: 

251 terms.append(f'REGEX="{self.product_regex}"') 

252 if self._serial_num: 

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

254 if terms: 

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

256 return "[all]" 

257 

258 def set_vid(self, vid: str) -> "UsbDeviceFilter": 

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

260 check_usb_id_format(vid) 

261 self._vid = vid 

262 return self 

263 

264 def set_pid(self, pid: str) -> "UsbDeviceFilter": 

265 """Pass only devices with given product id.""" 

266 check_usb_id_format(pid) 

267 self._pid = pid 

268 return self 

269 

270 def set_product_regex(self, product_regex: str) -> "UsbDeviceFilter": 

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

272 assert product_regex 

273 self.product_regex = product_regex 

274 return self 

275 

276 def set_serial_num(self, serial_num: str) -> "UsbDeviceFilter": 

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

278 assert serial_num 

279 self._serial_num = serial_num 

280 return self 

281 

282 def _eval(self, device: UsbDevice) -> bool: 

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

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

285 return False 

286 

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

288 return False 

289 

290 if (self.product_regex is not None) and not re.search( 

291 self.product_regex, device.product 

292 ): 

293 return False 

294 

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

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

297 ): 

298 return False 

299 

300 return True 

301 

302 def filter(self, devices: list[UsbDevice]): 

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

304 Items order is preserved.""" 

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

306 return result