Coverage for apio/utils/serial_util.py: 82%
100 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-07 05:08 +0000
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
9"""Serial devices related utilities."""
11import re
12from dataclasses import dataclass
13from serial.tools.list_ports import comports
14from serial.tools.list_ports_common import ListPortInfo
15from apio.common.debug_util import is_debug
16from apio.common.apio_console import cout
17from apio.utils import usb_util
18from apio.apio_context import ApioContext
21@dataclass()
22class SerialDevice:
23 """A data class to hold the information of a single Serial device."""
25 # pylint: disable=too-many-instance-attributes
27 port: str
28 port_name: str
29 vid: str
30 pid: str
31 manufacturer: str
32 product: str
33 serial_number: str
34 device_type: str
35 location: str
37 def __post_init__(self):
38 """Check that vid, pid, has the format %04X."""
39 usb_util.check_usb_id_format(self.vid)
40 usb_util.check_usb_id_format(self.pid)
42 def summary(self) -> str:
43 """Returns a user friendly short description of this device."""
44 return (
45 f"[{self.port}] "
46 f"[{self.vid}:{self.pid}] "
47 f"[{self.manufacturer}] [{self.product}] "
48 f"[{self.serial_number}]"
49 )
52def scan_serial_devices(_: ApioContext) -> list[SerialDevice]:
53 """Scan the connected serial devices and return their information."""
55 # -- Initial empty device list
56 devices = []
58 # -- Use the serial.tools.list_ports module for reading the
59 # -- serial ports. More info:
60 # -- https://pyserial.readthedocs.io/en/latest/tools.html
61 list_port_info: list[ListPortInfo] = list(comports())
62 assert isinstance(list_port_info, list)
63 if list_port_info: 63 ↛ 67line 63 didn't jump to line 67 because the condition on line 63 was always true
64 assert isinstance(list_port_info[0], ListPortInfo)
66 # -- Collect the items that are USB serial ports.
67 for port in list_port_info:
69 # -- Dump for debugging.
70 if is_debug(1): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 cout("Raw serial port:")
72 cout(f" Device: {port.device}")
73 cout(f" Hwid: {port.hwid}")
74 cout(f" Interface: {port.interface}")
76 # # -- Skip if not a serial port.
77 if not port.device: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true
78 continue
80 # -- Skip if not a USB device.
81 if not port.vid or not port.pid: 81 ↛ 89line 81 didn't jump to line 89 because the condition on line 81 was always true
82 continue
84 # -- If needed, append A or B to the serial numbers of a dual channel
85 # -- FTDI devices such as FT2232H.
86 # --
87 # -- Ideally these should be done in the comports() function of
88 # -- the pyserial library.
89 serial_number = port.serial_number or ""
90 if serial_number and port.device.endswith(f"-{serial_number}0"):
91 serial_number += "A"
92 elif serial_number and port.device.endswith(f"-{serial_number}1"):
93 serial_number += "B"
95 # -- Add device to list.
96 devices.append(
97 SerialDevice(
98 port=port.device,
99 port_name=port.name,
100 vid=f"{port.vid:04X}",
101 pid=f"{port.pid:04X}",
102 manufacturer=port.manufacturer or "",
103 product=port.product or "",
104 serial_number=serial_number,
105 device_type=usb_util.get_device_type(port.vid, port.pid),
106 location=port.location or "",
107 )
108 )
110 # -- Sort by port name, case insensitive.
111 devices = sorted(devices, key=lambda d: d.port.lower())
113 if is_debug(1): 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 cout(f"Found {len(devices)} serial device:")
115 for device in devices:
116 cout(str(device))
118 # -- All done.
119 return devices
122@dataclass
123class SerialDeviceFilter:
124 """A class to filter a list of serial devices by attributes. We use the
125 Fluent Interface design pattern so we can assert that the values that
126 the caller passes as filters are not unintentionally None or empty
127 unintentionally."""
129 _vid: str | None = None
130 _pid: str | None = None
131 _product_regex: str | None = None
132 _serial_port: str | None = None
133 _serial_num: str | None = None
135 def summary(self) -> str:
136 """User friendly representation of the filter"""
137 terms = []
138 if self._vid:
139 terms.append(f"VID={self._vid}")
140 if self._pid:
141 terms.append(f"PID={self._pid}")
142 if self._product_regex:
143 terms.append(f'REGEX="{self._product_regex}"')
144 if self._serial_port:
145 terms.append(f"PORT={self._serial_port}")
146 if self._serial_num:
147 terms.append(f'S/N="{self._serial_num}"')
148 if terms:
149 return "[" + ", ".join(terms) + "]"
150 return "[all]"
152 def set_vid(self, vid: str) -> "SerialDeviceFilter":
153 """Pass only devices with given vendor id."""
154 usb_util.check_usb_id_format(vid)
155 self._vid = vid
156 return self
158 def set_pid(self, pid: str) -> "SerialDeviceFilter":
159 """Pass only devices given product id."""
160 usb_util.check_usb_id_format(pid)
161 self._pid = pid
162 return self
164 def set_product_regex(self, product_regex: str) -> "SerialDeviceFilter":
165 """Pass only devices whose product string match given regex."""
166 assert product_regex
167 self._product_regex = product_regex
168 return self
170 def set_port(self, serial_port: str) -> "SerialDeviceFilter":
171 """Pass only devices given product serial port.."""
172 assert serial_port
173 self._serial_port = serial_port
174 return self
176 def set_serial_num(self, serial_num: str) -> "SerialDeviceFilter":
177 """Pass only devices given product serial number.."""
178 assert serial_num
179 self._serial_num = serial_num
180 return self
182 def _eval(self, device: SerialDevice) -> bool:
183 """Test if the devices passes this field."""
184 if (self._vid is not None) and (self._vid != device.vid):
185 return False
187 if (self._pid is not None) and (self._pid != device.pid):
188 return False
190 if (self._product_regex is not None) and not re.search(
191 self._product_regex, device.product
192 ):
193 return False
195 # -- We allow matching by full port string or by port name.
196 if (self._serial_port is not None) and (
197 (
198 self._serial_port.lower()
199 not in [device.port.lower(), device.port_name.lower()]
200 )
201 ):
202 return False
204 if (self._serial_num is not None) and (
205 self._serial_num.lower() != device.serial_number.lower()
206 ):
207 return False
209 return True
211 def filter(self, devices: list[SerialDevice]):
212 """Return a copy of the list with items that are pass this filter.
213 Items order is preserved."""
214 result = [d for d in devices if self._eval(d)]
215 return result