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