Coverage for apio/utils/usb_util.py: 67%
129 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 03:53 +0000
1"""USB devices related utilities."""
3import re
4from glob import glob
5from typing import Any
6from dataclasses import dataclass
7import usb.core
8import usb.backend.libusb1
9from apio.common.debug_util import is_debug
10from apio.common.apio_console import cout, fatal_error
11from apio.apio_context import ApioContext
13# -- Mapping of (VID), and (VID:PID) to device type. This is presented to the
14# -- user as an information only. Add more as you like.
16_USB_TYPES = {
17 # -- FTDI
18 (0x0403): "FTDI",
19 (0x0403, 0x6001): "FT232R",
20 (0x0403, 0x6010): "FT2232H",
21 (0x0403, 0x6011): "FT4232H",
22 (0x0403, 0x6014): "FT232H",
23 (0x0403, 0x6017): "FT313H",
24 (0x0403, 0x8372): "FT245R",
25 (0x0403, 0x8371): "FT232BM",
26 (0x0403, 0x8373): "FT2232C",
27 (0x0403, 0x8374): "FT4232",
28}
31def get_device_type(vid: int, pid: int) -> str:
32 """Determine device type string. Try to match by (vid, pid) and if
33 not found, by (vid). Returns "" if not found."""
34 device_type = _USB_TYPES.get((vid, pid), "")
35 if not device_type:
36 device_type = _USB_TYPES.get((vid), "")
37 return device_type
40def check_usb_id_format(usb_id: str) -> None:
41 """Check that a vid or pid is in 4 char uppercase hex."""
42 if not re.search(r"^[0-9A-F]{4}$", usb_id): 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
43 raise ValueError(f"Invalid 04X hex value: [{usb_id}]")
46@dataclass()
47class UsbDevice:
48 """A data class to hold the information of a single USB device."""
50 # pylint: disable=too-many-instance-attributes
52 vid: str
53 pid: str
54 bus: int
55 device: int
56 manufacturer: str
57 product: str
58 serial_number: str
59 device_type: str
61 def __post_init__(self):
62 """Check that vid, pid, has the format %04X."""
63 check_usb_id_format(self.vid)
64 check_usb_id_format(self.pid)
66 def summary(self) -> str:
67 """Returns a user friendly short description of this device."""
68 return (
69 f"[{self.vid}:{self.pid}] "
70 f"[{self.bus}:{self.device}] "
71 f"[{self.manufacturer}] "
72 f"[{self.product}] [{self.serial_number}]"
73 )
76def _get_usb_str(device: usb.core.Device, index: int, default: str) -> str:
77 """Extract usb string by its index."""
78 # pylint: disable=broad-exception-caught
79 try:
80 s = str(usb.util.get_string(device, index))
81 # For Tang 9K which contains a null char as a string separator.
82 # It's not USB standard but C tools do that implicitly.
83 s = s.split("\x00", 1)[0]
84 return s
85 except Exception as e:
86 if is_debug(1):
87 cout(f"Error getting USB string at index {index}: {e}")
88 return default
91def scan_usb_devices(apio_ctx: ApioContext) -> list[UsbDevice]:
92 """Query and return a list with usb device info."""
93 # pylint: disable=too-many-locals
95 # -- Track the names we searched for. For diagnostics.
96 searched_names = []
98 def find_library(name: str):
99 """A callback for looking up the libusb backend file."""
101 # -- Track searched names, for diagnostics
102 searched_names.append(name)
104 # -- Try to match to a lib in oss-cad-suite/lib.
105 oss_dir = apio_ctx.get_package_dir("oss-cad-suite")
106 pattern = oss_dir / "lib" / f"lib{name}*"
107 files = glob(str(pattern))
109 if is_debug(1): 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 cout("Apio find_library() call:")
111 cout(f" {name=}")
112 cout(f" {pattern=}")
113 cout(f" {files=}")
115 # -- We don't expect multiple matches.
116 if len(files) > 1: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 fatal_error(f"Found multiple backends for '{name}': {files}")
119 if files: 119 ↛ 121line 119 didn't jump to line 121 because the condition on line 119 was always true
120 return files[0]
121 return None
123 # -- Lookup libusb backend library file in oss-cad-suite/lib.
124 backend = usb.backend.libusb1.get_backend(find_library=find_library)
126 if not backend: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 fatal_error(
128 "Libusb backend not found",
129 info=f"Searched names: {searched_names}",
130 )
132 # -- Find the usb devices.
133 raw_devices = usb.core.find(find_all=True, backend=backend)
134 devices: list[Any] = list(raw_devices) if raw_devices else []
136 # -- Collect the devices
137 result: list[UsbDevice] = []
138 for device in devices: 138 ↛ 140line 138 didn't jump to line 140 because the loop on line 138 never started
139 # -- Print entire raw device info for debugging.
140 if is_debug(1):
141 cout()
142 cout(str(device))
143 cout()
145 # -- Sanity check.
146 assert isinstance(device, usb.core.Device), type(device)
148 # -- Skip hubs, they are not interesting
149 d = device.bDeviceClass
150 if d == 0x09:
151 continue
153 # -- Lookup device type or "" if not found.
154 device_type = get_device_type(
155 device.idVendor,
156 device.idProduct,
157 )
159 # -- Create the device object.
160 unavail = "--unavail--"
161 vid = device.idVendor
162 pid = device.idProduct
164 d = device
165 man = d.iManufacturer
166 iser = d.iSerialNumber
167 item = UsbDevice(
168 vid=f"{vid:04X}",
169 pid=f"{pid:04X}",
170 bus=device.bus,
171 device=device.address or 0,
172 manufacturer=_get_usb_str(
173 device,
174 man,
175 default=unavail,
176 ),
177 product=_get_usb_str(
178 device,
179 device.iProduct,
180 default=unavail,
181 ),
182 serial_number=_get_usb_str(
183 device,
184 iser,
185 default="",
186 ),
187 device_type=device_type,
188 )
189 result.append(item)
191 # -- Sort by (vendor, product, bus, device).
192 result = sorted(
193 result,
194 key=lambda d: (
195 d.vid.lower(),
196 d.pid.lower(),
197 d.bus,
198 d.device,
199 ),
200 )
202 if is_debug(1): 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 cout(f"Found {len(result)} USB devices:")
204 for device in result:
205 cout(str(device))
207 # -- All done.
208 return result
211@dataclass
212class UsbDeviceFilter:
213 """A class to filter a list of usb devices by attributes. We use the
214 Fluent Interface design pattern so we can assert that the values that
215 the caller passes as filters are not unintentionally None or empty
216 unintentionally."""
218 _vid: str | None = None
219 _pid: str | None = None
220 product_regex: str | None = None
221 _serial_num: str | None = None
223 def summary(self) -> str:
224 """User friendly representation of the filter"""
225 terms = []
227 if self._vid:
228 terms.append(f"VID={self._vid}")
229 if self._pid:
230 terms.append(f"PID={self._pid}")
231 if self.product_regex:
232 terms.append(f'REGEX="{self.product_regex}"')
233 if self._serial_num:
234 terms.append(f'S/N="{self._serial_num}"')
235 if terms:
236 return "[" + ", ".join(terms) + "]"
237 return "[all]"
239 def set_vid(self, vid: str) -> "UsbDeviceFilter":
240 """Pass only devices with given vendor id."""
241 check_usb_id_format(vid)
242 self._vid = vid
243 return self
245 def set_pid(self, pid: str) -> "UsbDeviceFilter":
246 """Pass only devices with given product id."""
247 check_usb_id_format(pid)
248 self._pid = pid
249 return self
251 def set_product_regex(self, product_regex: str) -> "UsbDeviceFilter":
252 """Pass only devices whose product string match given regex."""
253 assert product_regex
254 self.product_regex = product_regex
255 return self
257 def set_serial_num(self, serial_num: str) -> "UsbDeviceFilter":
258 """Pass only devices given product serial number.."""
259 assert serial_num
260 self._serial_num = serial_num
261 return self
263 def _eval(self, device: UsbDevice) -> bool:
264 """Test if the devices passes this field."""
265 if (self._vid is not None) and (self._vid != device.vid):
266 return False
268 if (self._pid is not None) and (self._pid != device.pid):
269 return False
271 if (self.product_regex is not None) and not re.search(
272 self.product_regex, device.product
273 ):
274 return False
276 if (self._serial_num is not None) and (
277 self._serial_num.lower() != device.serial_number.lower()
278 ):
279 return False
281 return True
283 def filter(self, devices: list[UsbDevice]):
284 """Return a copy of the list with items that are pass this filter.
285 Items order is preserved."""
286 result = [d for d in devices if self._eval(d)]
287 return result