From 75b821427d5db53515f66e13b1bc77ee5c64299f Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 11 Nov 2026 12:18:16 +0000 Subject: [PATCH 1/2] `FTDI`: FTDI's D2XX library as a second path beside libftdi - `ftdi_library="libftdi" | "d2xx" | None`; the default stays "libftdi", so no device changes; None picks D2XX on Windows when FTDI's D2XX library loads, else libftdi - D2XX is loaded with ctypes from FTDI's driver install; no new dependency, nothing bundled - channels resolve by serial number plus interface letter; device_id stays the serial without it; without interface_select a multi-interface chip opens channel A, as libftdi does; usb_address is refused under D2XX - every call mapped to its D2XX call with libftdi's values; 1.5 stop bits refused; reads return what has arrived without waiting; a short write raises - errors name the likely cause: a channel held by another program, or bound to another driver - a D2XX setup is recorded in captures; FTDIValidator.setup takes that record when present Co-Authored-By: Claude Opus 5.5 --- pylabrobot/io/executor_setup_tests.py | 2 + pylabrobot/io/ftdi.py | 532 ++++++++++++++++++++++++-- pylabrobot/io/ftdi_tests.py | 199 +++++++++- 3 files changed, 692 insertions(+), 41 deletions(-) diff --git a/pylabrobot/io/executor_setup_tests.py b/pylabrobot/io/executor_setup_tests.py index ad6a2e02697..e8219cf369d 100644 --- a/pylabrobot/io/executor_setup_tests.py +++ b/pylabrobot/io/executor_setup_tests.py @@ -59,8 +59,10 @@ async def _cancel_while_worker_is_running( async def test_cancelled_ftdi_setup_closes_device_before_executor_shutdown(self) -> None: io = ftdi_module.FTDI.__new__(ftdi_module.FTDI) io.human_readable_device_name = "mock FTDI" + io.ftdi_library = "libftdi" io._device_id = "mock" io._dev = None + io._d2xx_dev = None io._executor = None device = _ClosableDevice() started = threading.Event() diff --git a/pylabrobot/io/ftdi.py b/pylabrobot/io/ftdi.py index 17c121464ae..aa1e7b082b7 100644 --- a/pylabrobot/io/ftdi.py +++ b/pylabrobot/io/ftdi.py @@ -1,9 +1,10 @@ import asyncio import ctypes import logging +import sys from concurrent.futures import ThreadPoolExecutor from io import IOBase -from typing import Optional, Tuple, cast +from typing import Any, List, Optional, Tuple, cast from pylabrobot.events import emit_event @@ -47,13 +48,293 @@ def _open_device(self) -> int: logger = logging.getLogger(__name__) +# FTDI's D2XX library: its file per platform, the calls it is given, and its status codes. +_D2XX_LIBRARY_FILES = { + "win32": ("ftd2xx64.dll", "ftd2xx.dll"), + "darwin": ("libftd2xx.dylib",), + "linux": ("libftd2xx.so",), +} +_D2XX_DOWNLOAD = "https://ftdichip.com/drivers/d2xx-drivers/" +_FT_OPEN_BY_SERIAL_NUMBER = 1 +_FT_PURGE_RX = 1 +_FT_PURGE_TX = 2 +_FT_FLAGS_OPENED = 1 +_FT_DEVICE_NOT_OPENED = 3 +# FT_SetFlowControl's Xon and Xoff characters: the standard ones. +_XON = 0x11 +_XOFF = 0x13 +# PyLabRobot design choices: how long one read or write may wait. +_D2XX_READ_TIMEOUT_MS = 100 +_D2XX_WRITE_TIMEOUT_MS = 5000 +_D2XX_STATUS_NAMES = ( + "FT_OK", + "FT_INVALID_HANDLE", + "FT_DEVICE_NOT_FOUND", + "FT_DEVICE_NOT_OPENED", + "FT_IO_ERROR", + "FT_INSUFFICIENT_RESOURCES", + "FT_INVALID_PARAMETER", + "FT_INVALID_BAUD_RATE", + "FT_DEVICE_NOT_OPENED_FOR_ERASE", + "FT_DEVICE_NOT_OPENED_FOR_WRITE", + "FT_FAILED_TO_WRITE_DEVICE", + "FT_EEPROM_READ_FAILED", + "FT_EEPROM_WRITE_FAILED", + "FT_EEPROM_ERASE_FAILED", + "FT_EEPROM_NOT_PRESENT", + "FT_EEPROM_NOT_PROGRAMMED", + "FT_INVALID_ARGS", + "FT_NOT_SUPPORTED", + "FT_OTHER_ERROR", + "FT_DEVICE_LIST_NOT_READY", +) +_d2xx_library: Optional[Any] = None + + +class D2XXError(RuntimeError): + """FTDI's D2XX library answered a call with a status other than FT_OK.""" + + def __init__(self, message: str, *, status: int): + """ + Args: + message: what failed. + status: the D2XX status code. + """ + super().__init__(message) + self.status = status + + def is_ftdi_transport_error(error: BaseException) -> bool: - """Return whether an exception was raised by the optional pylibftdi transport.""" + """Return whether an exception was raised by the pylibftdi or D2XX transport.""" + if isinstance(error, D2XXError): + return True if not HAS_PYLIBFTDI: return False return isinstance(error, FtdiError) +def _load_d2xx_library() -> Any: + """FTDI's D2XX library, loaded once. + + Raises: + RuntimeError: if it cannot be loaded on this platform. + """ + global _d2xx_library + if _d2xx_library is not None: + return _d2xx_library + platform = "linux" if sys.platform.startswith("linux") else sys.platform + failures: List[str] = [] + for file_name in _D2XX_LIBRARY_FILES.get(platform, ()): + try: + if sys.platform == "win32": + library = ctypes.WinDLL(file_name) + else: + library = ctypes.CDLL(file_name) + except OSError as error: + failures.append(f"{file_name}: {error}") + continue + _d2xx_library = library + return library + raise RuntimeError( + "FTDI's D2XX library could not be loaded (" + + ("; ".join(failures) or f"none known for {sys.platform}") + + f"). It comes with FTDI's driver: {_D2XX_DOWNLOAD}" + ) + + +class _D2XXDeviceListInfoNode(ctypes.Structure): + """One entry of D2XX's device list (FT_DEVICE_LIST_INFO_NODE).""" + + _fields_ = [ + ("Flags", ctypes.c_uint32), + ("Type", ctypes.c_uint32), + ("ID", ctypes.c_uint32), + ("LocId", ctypes.c_uint32), + ("SerialNumber", ctypes.c_char * 16), + ("Description", ctypes.c_char * 64), + ("ftHandle", ctypes.c_void_p), + ] + + +def _check_d2xx_status(*, status: int, call: str) -> None: + """Raise ``D2XXError`` unless a D2XX call returned FT_OK.""" + if status != 0: + name = _D2XX_STATUS_NAMES[status] if status < len(_D2XX_STATUS_NAMES) else "unknown" + raise D2XXError(f"{call} returned {name} ({status})", status=status) + + +def _list_d2xx_devices() -> List[_D2XXDeviceListInfoNode]: + """Every FTDI channel D2XX sees: one entry per interface of a multi-interface chip.""" + library = _load_d2xx_library() + count = ctypes.c_uint32(0) + _check_d2xx_status( + status=library.FT_CreateDeviceInfoList(ctypes.byref(count)), call="FT_CreateDeviceInfoList" + ) + if count.value == 0: + return [] + nodes = (_D2XXDeviceListInfoNode * count.value)() + _check_d2xx_status( + status=library.FT_GetDeviceInfoList(nodes, ctypes.byref(count)), call="FT_GetDeviceInfoList" + ) + return list(nodes[: count.value]) + + +class _D2XXDevice: + """One FTDI channel opened through FTDI's D2XX library, by its D2XX serial number.""" + + def __init__(self, *, serial_number: str): + """ + Args: + serial_number: the channel's D2XX serial number, with its interface letter if it has one. + """ + self._library = _load_d2xx_library() + self._serial_number = serial_number + self._handle = ctypes.c_void_p() + self.closed = True + + def open(self) -> None: + """Open the channel, then set its timeouts; a channel whose timeouts fail is closed again.""" + _check_d2xx_status( + call="FT_OpenEx", + status=self._library.FT_OpenEx( + ctypes.c_char_p(self._serial_number.encode("ascii")), + _FT_OPEN_BY_SERIAL_NUMBER, + ctypes.byref(self._handle), + ), + ) + self.closed = False + try: + _check_d2xx_status( + call="FT_SetTimeouts", + status=self._library.FT_SetTimeouts( + self._handle, + ctypes.c_uint32(_D2XX_READ_TIMEOUT_MS), + ctypes.c_uint32(_D2XX_WRITE_TIMEOUT_MS), + ), + ) + except BaseException: + self.closed = True + self._library.FT_Close(self._handle) + raise + + def close(self) -> None: + if not self.closed: + self.closed = True + _check_d2xx_status(call="FT_Close", status=self._library.FT_Close(self._handle)) + + def set_baudrate(self, *, baudrate: int) -> None: + _check_d2xx_status( + call="FT_SetBaudRate", + status=self._library.FT_SetBaudRate(self._handle, ctypes.c_uint32(baudrate)), + ) + + def set_line_property(self, *, bits: int, stopbits: int, parity: int) -> None: + """Set data bits, stop bits and parity, given as libftdi's values. + + Raises: + ValueError: for 1.5 stop bits (libftdi's 1), which D2XX cannot set. + """ + if stopbits == 1: + raise ValueError("D2XX cannot set 1.5 stop bits (libftdi stop-bit value 1)") + _check_d2xx_status( + call="FT_SetDataCharacteristics", + status=self._library.FT_SetDataCharacteristics( + self._handle, ctypes.c_ubyte(bits), ctypes.c_ubyte(stopbits), ctypes.c_ubyte(parity) + ), + ) + + def set_flowctrl(self, *, flowctrl: int) -> None: + """Set flow control, given as libftdi's value; D2XX's values are the same.""" + _check_d2xx_status( + call="FT_SetFlowControl", + status=self._library.FT_SetFlowControl( + self._handle, + ctypes.c_uint16(flowctrl), + ctypes.c_ubyte(_XON), + ctypes.c_ubyte(_XOFF), + ), + ) + + def set_rts(self, *, level: bool) -> None: + if level: + _check_d2xx_status(call="FT_SetRts", status=self._library.FT_SetRts(self._handle)) + else: + _check_d2xx_status(call="FT_ClrRts", status=self._library.FT_ClrRts(self._handle)) + + def set_dtr(self, *, level: bool) -> None: + if level: + _check_d2xx_status(call="FT_SetDtr", status=self._library.FT_SetDtr(self._handle)) + else: + _check_d2xx_status(call="FT_ClrDtr", status=self._library.FT_ClrDtr(self._handle)) + + def set_latency_timer(self, *, latency: int) -> None: + _check_d2xx_status( + call="FT_SetLatencyTimer", + status=self._library.FT_SetLatencyTimer(self._handle, ctypes.c_ubyte(latency)), + ) + + def reset(self) -> None: + """The chip's reset request; not a reset of the USB port.""" + _check_d2xx_status(call="FT_ResetDevice", status=self._library.FT_ResetDevice(self._handle)) + + def purge(self, *, mask: int) -> None: + _check_d2xx_status( + call="FT_Purge", status=self._library.FT_Purge(self._handle, ctypes.c_uint32(mask)) + ) + + def poll_modem_status(self) -> int: + """The modem status in the low byte, its bits as libftdi's; above it the line status. + + The line status holds only the error bits (overrun, parity, framing, break), and only on Windows, + so a whole value differs from libftdi's. + """ + status = ctypes.c_uint32(0) + _check_d2xx_status( + call="FT_GetModemStatus", + status=self._library.FT_GetModemStatus(self._handle, ctypes.byref(status)), + ) + return status.value + + def write(self, data: bytes) -> int: + """Write ``data`` whole. + + Raises: + D2XXError: if the write timeout ends before every byte went out. + """ + written = ctypes.c_uint32(0) + _check_d2xx_status( + call="FT_Write", + status=self._library.FT_Write( + self._handle, ctypes.c_char_p(data), ctypes.c_uint32(len(data)), ctypes.byref(written) + ), + ) + if written.value != len(data): + raise D2XXError( + f"FT_Write wrote {written.value} of {len(data)} bytes before its timeout", status=0 + ) + return written.value + + def read(self, num_bytes: int) -> bytes: + """What has arrived, at most ``num_bytes``; empty, without waiting, when nothing has.""" + queued = ctypes.c_uint32(0) + _check_d2xx_status( + call="FT_GetQueueStatus", + status=self._library.FT_GetQueueStatus(self._handle, ctypes.byref(queued)), + ) + count = min(num_bytes, queued.value) + if count == 0: + return b"" + buffer = ctypes.create_string_buffer(count) + received = ctypes.c_uint32(0) + _check_d2xx_status( + call="FT_Read", + status=self._library.FT_Read( + self._handle, buffer, ctypes.c_uint32(count), ctypes.byref(received) + ), + ) + return buffer.raw[: received.value] + + def _parse_usb_address(address: str) -> Tuple[int, Tuple[int, ...]]: """Parse a USB topology path '-[....]' into its bus and ports.""" bus_str, sep, port_str = address.partition("-") @@ -76,7 +357,7 @@ def __init__(self, device_id: str, action: str, data: str, module: str = "ftdi") class FTDI(IOBase): - """Thin wrapper around pylibftdi with device resolution and logging. + """An FTDI chip's channel, through libftdi (over libusb) or FTDI's own D2XX library. Finds devices based on the following parameters: 1. device_id - serial number for explicit connection @@ -89,6 +370,8 @@ class FTDI(IOBase): device_id: Device identifier (serial number) vid: USB Vendor ID pid: USB Product ID + ftdi_library: "libftdi" or "d2xx"; None picks D2XX on Windows when FTDI's D2XX library loads, + else libftdi. """ def __init__( @@ -99,19 +382,35 @@ def __init__( pid: Optional[int] = None, interface_select: Optional[int] = None, usb_address: Optional[str] = None, + ftdi_library: Optional[str] = "libftdi", ): - if not HAS_PYLIBFTDI: - global _FTDI_ERROR - raise RuntimeError( - "pylibftdi is not installed. Install with: pip install pylabrobot[ftdi]. " - f"Import error: {_FTDI_ERROR}" - ) - if not HAS_PYUSB: - global _PYUSB_ERROR - raise RuntimeError( - "pyusb is not installed. Install with: pip install pylabrobot[ftdi]. " - f"Import error: {_PYUSB_ERROR}" - ) + if ftdi_library not in (None, "libftdi", "d2xx"): + raise ValueError(f"ftdi_library must be 'libftdi', 'd2xx' or None, got {ftdi_library!r}") + d2xx_failure = "" + if ftdi_library is None: + ftdi_library = "libftdi" + if sys.platform == "win32": + try: + _load_d2xx_library() + ftdi_library = "d2xx" + except RuntimeError as error: + d2xx_failure = f" D2XX was not used: {error}" + if ftdi_library == "d2xx": + _load_d2xx_library() + else: + if not HAS_PYLIBFTDI: + global _FTDI_ERROR + raise RuntimeError( + "pylibftdi is not installed. Install with: pip install pylabrobot[ftdi]. " + f"Import error: {_FTDI_ERROR}.{d2xx_failure}" + ) + if not HAS_PYUSB: + global _PYUSB_ERROR + raise RuntimeError( + "pyusb is not installed. Install with: pip install pylabrobot[ftdi]. " + f"Import error: {_PYUSB_ERROR}" + ) + self.ftdi_library = ftdi_library self.human_readable_device_name = human_readable_device_name self._device_id = device_id @@ -122,6 +421,7 @@ def __init__( # Will be resolved in setup() self._dev: Optional[Device] = None + self._d2xx_dev: Optional[_D2XXDevice] = None self._executor: Optional[ThreadPoolExecutor] = None # Bytes off the wire that no read has taken yet, and the read still in flight, if any. self._unread = bytearray() @@ -139,6 +439,81 @@ def dev(self) -> "Device": raise RuntimeError("Device not initialized. Call setup() first.") return self._dev + @property + def _d2xx(self) -> _D2XXDevice: + if self._d2xx_dev is None: + raise RuntimeError("Device not initialized. Call setup() first.") + return self._d2xx_dev + + def _resolve_d2xx_serial_number(self) -> str: + """Pick the one D2XX channel matching the parameters; return its D2XX serial number. + + A multi-interface chip lists each channel as its serial plus a letter; ``interface_select`` 1 + is "A", 2 "B". ``device_id`` is matched against the serial without that letter. + + Raises: + ValueError: if ``usb_address`` is given; D2XX selects by serial number. + RuntimeError: if no channel, or more than one, matches. + """ + nodes = _list_d2xx_devices() + listed = ", ".join( + f"{node.SerialNumber.decode('ascii', 'replace') or '(in use)'} " + f"(VID:PID {node.ID >> 16:04x}:{node.ID & 0xFFFF:04x})" + for node in nodes + ) + if self._usb_address is not None: + raise ValueError( + "usb_address is not supported through FTDI's D2XX library; pass device_id, the serial " + f"number, instead. FTDI channels found: {listed or 'none'}" + ) + serial_numbers = {node.SerialNumber.decode("ascii", "replace") for node in nodes} + letter = "ABCD"[self._interface_select - 1] if self._interface_select else None + candidates: List[Tuple[str, str]] = [] + for node in nodes: + serial_number = node.SerialNumber.decode("ascii", "replace") + if self._vid is not None and node.ID >> 16 != self._vid: + continue + if self._pid is not None and node.ID & 0xFFFF != self._pid: + continue + # A multi-interface chip's channels share a serial; without a letter, take "A" as libftdi does. + multi_interface = serial_number[:-1] + "B" in serial_numbers + if letter is not None: + if not serial_number.endswith(letter): + continue + base = serial_number[:-1] + elif multi_interface and serial_number.endswith("A"): + base = serial_number[:-1] + elif serial_number[-1:] in ("B", "C", "D") and serial_number[:-1] + "A" in serial_numbers: + continue + else: + base = serial_number + if self._device_id is not None and base != self._device_id: + continue + candidates.append((serial_number, base)) + if len(candidates) == 0: + held = sum(1 for node in nodes if node.Flags & _FT_FLAGS_OPENED and not node.SerialNumber) + raise RuntimeError( + f"No FTDI channel found through D2XX for '{self.human_readable_device_name}' " + f"(device_id {self._device_id}, interface {letter or 'any'}). FTDI channels found: " + f"{listed or 'none'}. " + + ( + f"{held} held by another program list no serial number: close the vendor software or " + "stop its service. " + if held + else "" + ) + + "A channel bound to a driver other than FTDI's (e.g. with Zadig) is not listed: restore " + "FTDI's driver in Device Manager, or pass ftdi_library='libftdi'." + ) + if len(candidates) > 1: + raise RuntimeError( + f"Multiple FTDI channels found through D2XX for '{self.human_readable_device_name}': " + f"{listed}. Please specify the device_id parameter explicitly with the serial number." + ) + serial_number, base = candidates[0] + self._device_id = base + return serial_number + def _resolve_device_serial(self) -> str: """List connected FTDI devices and resolve which one to connect to based on parameters. @@ -299,6 +674,24 @@ def _reattach_kernel_driver(self) -> None: def _setup_sync(self) -> None: """Resolve and open the device. Runs on the executor that owns all device calls.""" + if self.ftdi_library == "d2xx": + if self._d2xx_dev is not None: + self._d2xx_dev.close() + self._d2xx_dev = None + d2xx_dev = _D2XXDevice(serial_number=self._resolve_d2xx_serial_number()) + try: + d2xx_dev.open() + except D2XXError as error: + if error.status == _FT_DEVICE_NOT_OPENED: + raise D2XXError( + f"{error}: another program holds the channel (close the vendor software or stop its " + "service), or a driver other than FTDI's is bound to it (e.g. installed with Zadig; " + "restore FTDI's driver in Device Manager, or pass ftdi_library='libftdi')", + status=error.status, + ) from error + raise + self._d2xx_dev = d2xx_dev + return if self._dev is not None and not self._dev.closed: self._dev.close() self._dev = None @@ -355,16 +748,19 @@ async def setup(self): except Exception: logger.warning("Failed to close FTDI device after setup failure", exc_info=True) self._dev = None + self._d2xx_dev = None await loop.run_in_executor(self._executor, self._reattach_kernel_driver) self._shutdown_executor() - if isinstance(exc, FtdiError): + if HAS_PYLIBFTDI and isinstance(exc, FtdiError): raise RuntimeError( f"Failed to open FTDI device for '{self.human_readable_device_name}': {exc}. " "Is the device connected? Is it in use by another process? " "Try restarting the kernel." ) from exc raise - logger.info(f"Successfully opened FTDI device: {self.device_id}") + logger.info(f"Successfully opened FTDI device: {self.device_id} ({self.ftdi_library})") + if self.ftdi_library == "d2xx": + capturer.record(FTDICommand(device_id=self.device_id, action="setup", data="d2xx")) def _shutdown_executor(self) -> None: if self._executor is not None: @@ -380,7 +776,10 @@ def device_id(self) -> str: async def set_baudrate(self, baudrate: int): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: setattr(self.dev, "baudrate", baudrate)) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.set_baudrate(baudrate=baudrate)) + else: + await loop.run_in_executor(self._executor, lambda: setattr(self.dev, "baudrate", baudrate)) logger.log(LOG_LEVEL_IO, "[%s] set_baudrate %s", self._device_id, baudrate) capturer.record( FTDICommand(device_id=self.device_id, action="set_baudrate", data=str(baudrate)) @@ -388,28 +787,42 @@ async def set_baudrate(self, baudrate: int): async def set_rts(self, level: bool): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_setrts(level)) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.set_rts(level=level)) + else: + await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_setrts(level)) logger.log(LOG_LEVEL_IO, "[%s] set_rts %s", self._device_id, level) capturer.record(FTDICommand(device_id=self.device_id, action="set_rts", data=str(level))) async def set_dtr(self, level: bool): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_setdtr(level)) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.set_dtr(level=level)) + else: + await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_setdtr(level)) logger.log(LOG_LEVEL_IO, "[%s] set_dtr %s", self._device_id, level) capturer.record(FTDICommand(device_id=self.device_id, action="set_dtr", data=str(level))) async def usb_reset(self): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_reset()) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, self._d2xx.reset) + else: + await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_reset()) self._discard_reads() logger.log(LOG_LEVEL_IO, "[%s] usb_reset", self._device_id) capturer.record(FTDICommand(device_id=self.device_id, action="usb_reset", data="")) async def set_latency_timer(self, latency: int): loop = asyncio.get_running_loop() - await loop.run_in_executor( - self._executor, lambda: self.dev.ftdi_fn.ftdi_set_latency_timer(latency) - ) + if self.ftdi_library == "d2xx": + await loop.run_in_executor( + self._executor, lambda: self._d2xx.set_latency_timer(latency=latency) + ) + else: + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_set_latency_timer(latency) + ) logger.log(LOG_LEVEL_IO, "[%s] set_latency_timer %s", self._device_id, latency) capturer.record( FTDICommand(device_id=self.device_id, action="set_latency_timer", data=str(latency)) @@ -417,9 +830,15 @@ async def set_latency_timer(self, latency: int): async def set_line_property(self, bits: int, stopbits: int, parity: int): loop = asyncio.get_running_loop() - await loop.run_in_executor( - self._executor, lambda: self.dev.ftdi_fn.ftdi_set_line_property(bits, stopbits, parity) - ) + if self.ftdi_library == "d2xx": + await loop.run_in_executor( + self._executor, + lambda: self._d2xx.set_line_property(bits=bits, stopbits=stopbits, parity=parity), + ) + else: + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_set_line_property(bits, stopbits, parity) + ) logger.log( LOG_LEVEL_IO, "[%s] set_line_property %s,%s,%s", self._device_id, bits, stopbits, parity ) @@ -431,7 +850,12 @@ async def set_line_property(self, bits: int, stopbits: int, parity: int): async def set_flowctrl(self, flowctrl: int): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_setflowctrl(flowctrl)) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.set_flowctrl(flowctrl=flowctrl)) + else: + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_setflowctrl(flowctrl) + ) logger.log(LOG_LEVEL_IO, "[%s] set_flowctrl %s", self._device_id, flowctrl) capturer.record( FTDICommand(device_id=self.device_id, action="set_flowctrl", data=str(flowctrl)) @@ -439,28 +863,42 @@ async def set_flowctrl(self, flowctrl: int): async def usb_purge_rx_buffer(self): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_purge_rx_buffer()) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.purge(mask=_FT_PURGE_RX)) + else: + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_purge_rx_buffer() + ) self._discard_reads() logger.log(LOG_LEVEL_IO, "[%s] usb_purge_rx_buffer", self._device_id) capturer.record(FTDICommand(device_id=self.device_id, action="usb_purge_rx_buffer", data="")) async def usb_purge_tx_buffer(self): loop = asyncio.get_running_loop() - await loop.run_in_executor(self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_purge_tx_buffer()) + if self.ftdi_library == "d2xx": + await loop.run_in_executor(self._executor, lambda: self._d2xx.purge(mask=_FT_PURGE_TX)) + else: + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_usb_purge_tx_buffer() + ) logger.log(LOG_LEVEL_IO, "[%s] usb_purge_tx_buffer", self._device_id) capturer.record(FTDICommand(device_id=self.device_id, action="usb_purge_tx_buffer", data="")) async def poll_modem_status(self) -> int: loop = asyncio.get_running_loop() - stat = ctypes.c_ushort(0) - await loop.run_in_executor( - self._executor, lambda: self.dev.ftdi_fn.ftdi_poll_modem_status(ctypes.byref(stat)) - ) - logger.log(LOG_LEVEL_IO, "[%s] poll_modem_status %s", self._device_id, stat.value) + if self.ftdi_library == "d2xx": + value = await loop.run_in_executor(self._executor, self._d2xx.poll_modem_status) + else: + stat = ctypes.c_ushort(0) + await loop.run_in_executor( + self._executor, lambda: self.dev.ftdi_fn.ftdi_poll_modem_status(ctypes.byref(stat)) + ) + value = stat.value + logger.log(LOG_LEVEL_IO, "[%s] poll_modem_status %s", self._device_id, value) capturer.record( - FTDICommand(device_id=self.device_id, action="poll_modem_status", data=str(stat.value)) + FTDICommand(device_id=self.device_id, action="poll_modem_status", data=str(value)) ) - return stat.value + return value async def request_serial(self) -> str: return self.device_id @@ -489,6 +927,9 @@ async def attempt(operation) -> None: logger.warning("Additional FTDI shutdown operation failed", exc_info=True) try: + if self._d2xx_dev is not None: + await attempt(self._d2xx_dev.close) + self._d2xx_dev = None if self._dev is not None: await attempt(self.dev.close) self._dev = None @@ -537,8 +978,9 @@ async def write(self, data: bytes) -> int: loop = asyncio.get_running_loop() # shield: a call handed to the worker cannot be recalled, so cancelling the caller would # otherwise leave the device holding a partial command, or none, with no way to tell which. + write = self._d2xx.write if self.ftdi_library == "d2xx" else self.dev.write bytes_written = cast( - int, await asyncio.shield(loop.run_in_executor(self._executor, self.dev.write, data)) + int, await asyncio.shield(loop.run_in_executor(self._executor, write, data)) ) emit_event( "io.write", @@ -554,7 +996,8 @@ async def read(self, num_bytes: int = 1) -> bytes: pending = self._pending_read if pending is None or pending.done(): loop = asyncio.get_running_loop() - pending = loop.run_in_executor(self._executor, self.dev.read, num_bytes) + read = self._d2xx.read if self.ftdi_library == "d2xx" else self.dev.read + pending = loop.run_in_executor(self._executor, read, num_bytes) self._pending_read = pending try: # shield: the worker keeps going after the caller is gone, so its bytes are held for the @@ -623,6 +1066,7 @@ def serialize(self): "vid": self._vid, "pid": self._pid, "usb_address": self._usb_address, + "ftdi_library": self.ftdi_library, } @@ -632,7 +1076,15 @@ def __init__(self, cr: "CaptureReader", human_readable_device_name: str, device_ self.cr = cr async def setup(self): - pass + """Take the recorded setup, if there is one: only a D2XX setup is recorded.""" + commands, index = self.cr.commands, self.cr._command_idx + if ( + index < len(commands) + and commands[index].get("module") == "ftdi" + and commands[index].get("action") == "setup" + and commands[index].get("device_id") == self._device_id + ): + self.cr.next_command() async def set_baudrate(self, baudrate: int): next_command = FTDICommand(**self.cr.next_command()) diff --git a/pylabrobot/io/ftdi_tests.py b/pylabrobot/io/ftdi_tests.py index c0f962e87c2..7b8cfd6a57c 100644 --- a/pylabrobot/io/ftdi_tests.py +++ b/pylabrobot/io/ftdi_tests.py @@ -1,4 +1,5 @@ import asyncio +import ctypes import logging import tempfile import threading @@ -7,7 +8,7 @@ from concurrent.futures import ThreadPoolExecutor from pathlib import Path from types import SimpleNamespace -from typing import Any, List, Union +from typing import Any, List, Optional, Union from unittest import mock from pylabrobot.events import EventBus, use_event_bus @@ -199,9 +200,11 @@ class FTDITopologyLifecycleTests(unittest.IsolatedAsyncioTestCase): @staticmethod def _ftdi_without_optional_dependency_checks() -> FTDI: io = FTDI.__new__(FTDI) + io.ftdi_library = "libftdi" io._interface_select = None io._detached_kernel_driver = None io._dev = None + io._d2xx_dev = None io._executor = None io._unread = bytearray() io._pending_read = None @@ -375,5 +378,199 @@ async def test_empty_terminator_is_rejected(self) -> None: await io.readline(terminator=b"") +class FakeD2XXLibrary: + """FTDI's D2XX library, answering through the same ctypes arguments: lists ``channels`` as + (serial number, VID:PID ID, flags), holds bytes queued for reading, and logs each setting.""" + + def __init__(self, *, channels: List[Any]): + self.channels = channels + self.settings: List[Any] = [] + self.opened: List[str] = [] + self.queued = bytearray() + self.write_limit: Optional[int] = None + + def FT_CreateDeviceInfoList(self, count): + count._obj.value = len(self.channels) + return 0 + + def FT_GetDeviceInfoList(self, nodes, count): + for node, (serial_number, device_id, flags) in zip(nodes, self.channels): + node.SerialNumber = serial_number.encode("ascii") + node.ID = device_id + node.Flags = flags + count._obj.value = len(self.channels) + return 0 + + def FT_OpenEx(self, serial_number, flags, handle): + self.opened.append(serial_number.value.decode("ascii")) + handle._obj.value = 1 + return 0 + + def FT_Close(self, handle): + return 0 + + def FT_SetTimeouts(self, handle, read_timeout_ms, write_timeout_ms): + self.settings.append(("timeouts", read_timeout_ms.value, write_timeout_ms.value)) + return 0 + + def FT_SetLatencyTimer(self, handle, latency): + self.settings.append(("latency", latency.value)) + return 0 + + def FT_SetFlowControl(self, handle, flowctrl, xon, xoff): + self.settings.append(("flow", flowctrl.value, xon.value, xoff.value)) + return 0 + + def FT_SetRts(self, handle): + self.settings.append(("rts", True)) + return 0 + + def FT_Write(self, handle, data, count, written): + written._obj.value = min(count.value, self.write_limit or count.value) + return 0 + + def FT_GetQueueStatus(self, handle, queued): + queued._obj.value = len(self.queued) + return 0 + + def FT_Read(self, handle, buffer, count, received): + data = bytes(self.queued[: count.value]) + del self.queued[: count.value] + ctypes.memmove(buffer, data, len(data)) + received._obj.value = len(data) + return 0 + + +# One FT2232H chip's two channels as D2XX lists them: serial number plus interface letter. +FT2232H_ID = 0x0403 << 16 | 0x6010 +TWO_CHANNEL_CHIP = [("SERIALA", FT2232H_ID, 0), ("SERIALB", FT2232H_ID, 0)] + + +class FTDID2XXTests(unittest.IsolatedAsyncioTestCase): + def use_library(self, *, channels: List[Any]) -> FakeD2XXLibrary: + library = FakeD2XXLibrary(channels=channels) + patcher = mock.patch.object(ftdi_module, "_d2xx_library", library) + patcher.start() + self.addCleanup(patcher.stop) + return library + + async def set_up_interface_b(self) -> "tuple[FTDI, FakeD2XXLibrary]": + library = self.use_library(channels=TWO_CHANNEL_CHIP) + io = FTDI(human_readable_device_name="test", interface_select=2, ftdi_library="d2xx") + await io.setup() + self.addAsyncCleanup(io.stop) + return io, library + + def test_automatic_choice_is_d2xx_only_on_windows_and_only_when_asked(self): + self.use_library(channels=[]) + with ( + mock.patch.object(ftdi_module, "HAS_PYLIBFTDI", True), + mock.patch.object(ftdi_module, "HAS_PYUSB", True), + ): + for platform, chosen in (("win32", "d2xx"), ("darwin", "libftdi"), ("linux", "libftdi")): + with ( + self.subTest(platform=platform), + mock.patch.object(ftdi_module.sys, "platform", platform), + ): + automatic = FTDI(human_readable_device_name="test", ftdi_library=None) + self.assertEqual(automatic.ftdi_library, chosen) + self.assertEqual(FTDI(human_readable_device_name="test").ftdi_library, "libftdi") + + def test_unknown_library_is_refused(self): + with self.assertRaisesRegex(ValueError, "ftdi_library must be"): + FTDI(human_readable_device_name="test", ftdi_library="serial") + + async def test_interface_b_opens_its_channel_and_keeps_the_base_serial(self): + io, library = await self.set_up_interface_b() + self.assertEqual(library.opened, ["SERIALB"]) + self.assertEqual(io.device_id, "SERIAL") + + async def test_no_interface_takes_channel_a_as_libftdi_does(self): + library = self.use_library(channels=TWO_CHANNEL_CHIP) + io = FTDI(human_readable_device_name="test", device_id="SERIAL", ftdi_library="d2xx") + await io.setup() + self.addAsyncCleanup(io.stop) + self.assertEqual(library.opened, ["SERIALA"]) + + async def test_unresolvable_channels_refuse_setup(self): + for channels, interface_select, usb_address, message in ( + ([], 2, None, "restore FTDI's driver"), + ([("", FT2232H_ID, 1)], 2, None, "close the vendor software"), + (TWO_CHANNEL_CHIP + [("OTHERA", FT2232H_ID, 0)], 1, None, "Multiple FTDI channels"), + (TWO_CHANNEL_CHIP, 2, "3-2", "usb_address is not supported"), + ): + with self.subTest(message=message): + self.use_library(channels=channels) + io = FTDI( + human_readable_device_name="test", + interface_select=interface_select, + usb_address=usb_address, + ftdi_library="d2xx", + ) + with self.assertRaisesRegex((RuntimeError, ValueError), message): + await io.setup() + + async def test_settings_take_libftdi_values(self): + io, library = await self.set_up_interface_b() + await io.set_flowctrl(flowctrl=0x100) + await io.set_rts(level=True) + await io.set_latency_timer(latency=2) + self.assertEqual( + library.settings, + [ + ("timeouts", ftdi_module._D2XX_READ_TIMEOUT_MS, ftdi_module._D2XX_WRITE_TIMEOUT_MS), + ("flow", 0x100, 0x11, 0x13), + ("rts", True), + ("latency", 2), + ], + ) + with self.assertRaisesRegex(ValueError, "1.5 stop bits"): + await io.set_line_property(bits=8, stopbits=1, parity=0) + + async def test_read_returns_what_has_arrived_without_waiting(self): + io, library = await self.set_up_interface_b() + self.assertEqual(await io.read(num_bytes=16), b"") + library.queued.extend(b"\x01\x02\x03") + self.assertEqual(await io.read(num_bytes=2), b"\x01\x02") + self.assertEqual(await io.read(num_bytes=16), b"\x03") + + async def test_a_short_write_raises(self): + io, library = await self.set_up_interface_b() + self.assertEqual(await io.write(data=b"\x01\x02"), 2) + library.write_limit = 1 + with self.assertRaisesRegex(ftdi_module.D2XXError, "wrote 1 of 2 bytes"): + await io.write(data=b"\x01\x02") + + async def test_d2xx_setup_is_recorded_and_replays_with_or_without_the_record(self): + self.use_library(channels=TWO_CHANNEL_CHIP) + io = FTDI(human_readable_device_name="test", interface_select=2, ftdi_library="d2xx") + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "capture.json" + capture_module.start_capture(path) + try: + await io.setup() + await io.stop() + finally: + capture_module.stop_capture() + reader = CaptureReader(path=str(path)) + self.assertEqual([(c["action"], c["data"]) for c in reader.commands], [("setup", "d2xx")]) + recorded = list(reader.commands) + with ( + mock.patch.object(ftdi_module, "HAS_PYLIBFTDI", True), + mock.patch.object(ftdi_module, "HAS_PYUSB", True), + ): + for commands in (recorded, []): + with self.subTest(recorded=bool(commands)): + validator = FTDIValidator( + cr=reader, human_readable_device_name="test", device_id="SERIAL" + ) + reader.commands = commands + reader.start() + try: + await validator.setup() + finally: + reader.done() + + if __name__ == "__main__": unittest.main() From ef34ab447fa7536889c2f44aa98777f7e2008180 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Wed, 11 Nov 2026 12:18:16 +0000 Subject: [PATCH 2/2] v1 `Mantis`: choose the FTDI library, D2XX on Windows by default - `ftdi_library` on `Mantis()`, `MantisDevice` and `MantisDriver`, passed to both links (interfaces A and B); None picks FTDI's D2XX library on Windows when it loads, else libftdi - a simulated Mantis refuses `ftdi_library`, as it refuses any link argument - the model-file test names the meshed parts instead of probing for a mesh attribute Co-Authored-By: Claude Opus 5.5 --- pylabrobot/formulatrix/mantis/device.py | 20 +++++++++++++++---- pylabrobot/formulatrix/mantis/device_tests.py | 4 +--- .../formulatrix/mantis/driver/master.py | 19 ++++++++++++++---- .../formulatrix/mantis/driver/simulator.py | 6 ++++-- 4 files changed, 36 insertions(+), 13 deletions(-) diff --git a/pylabrobot/formulatrix/mantis/device.py b/pylabrobot/formulatrix/mantis/device.py index 7cf92f725a2..6af1ef1ff11 100644 --- a/pylabrobot/formulatrix/mantis/device.py +++ b/pylabrobot/formulatrix/mantis/device.py @@ -40,6 +40,7 @@ def __init__( size_z: float = SIZE_Z, declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, + ftdi_library: Optional[str] = None, simulation: bool = False, ): """ @@ -56,10 +57,13 @@ def __init__( ``RECORDING_MANTIS_V4``. oem_config_directory: the unit's vendor ``Data/Device/Configs`` folder; where it and the declared file both hold a unit value, they must agree. + ftdi_library: "libftdi" or "d2xx"; None picks D2XX on Windows when FTDI's D2XX library + loads, else libftdi. simulation: build a simulated Mantis, with no link. Raises: - ValueError: if ``simulation`` is given with ``device_id``, ``io`` or ``rfid_io``. + ValueError: if ``simulation`` is given with ``device_id``, ``io``, ``rfid_io`` or + ``ftdi_library``. """ super().__init__( name=name, @@ -71,8 +75,10 @@ def __init__( ) self.driver: MantisDriver if simulation: - if device_id is not None or io is not None or rfid_io is not None: - raise ValueError("a simulated Mantis has no link: no device_id, io or rfid_io") + if any(link is not None for link in (device_id, io, rfid_io, ftdi_library)): + raise ValueError( + "a simulated Mantis has no link: no device_id, io, rfid_io or ftdi_library" + ) self.driver = MantisSimulationDriver( declared_configuration_json=declared_configuration_json or RECORDING_MANTIS_V4, name=name, @@ -86,6 +92,7 @@ def __init__( name=name, declared_configuration_json=declared_configuration_json, oem_config_directory=oem_config_directory, + ftdi_library=ftdi_library, ) self.mesh = { "path": os.path.join(os.path.dirname(__file__), "resource_model", MANIFEST["base"]), @@ -146,6 +153,7 @@ def Mantis( size_z: float = SIZE_Z, declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, + ftdi_library: Optional[str] = None, simulation: bool = False, ) -> MantisDevice: """A Formulatrix Mantis, v3 or v4. @@ -162,10 +170,13 @@ def Mantis( disagrees on what is fitted. Simulated, the configuration; None uses ``RECORDING_MANTIS_V4``. oem_config_directory: the unit's vendor ``Data/Device/Configs`` folder; where it and the declared file both hold a unit value, they must agree. + ftdi_library: "libftdi" or "d2xx"; None picks D2XX on Windows when FTDI's D2XX library loads, + else libftdi. simulation: build a simulated Mantis, with no link. Raises: - ValueError: if ``simulation`` is given with ``device_id``, ``io`` or ``rfid_io``. + ValueError: if ``simulation`` is given with ``device_id``, ``io``, ``rfid_io`` or + ``ftdi_library``. """ return MantisDevice( name=name, @@ -177,5 +188,6 @@ def Mantis( size_z=size_z, declared_configuration_json=declared_configuration_json, oem_config_directory=oem_config_directory, + ftdi_library=ftdi_library, simulation=simulation, ) diff --git a/pylabrobot/formulatrix/mantis/device_tests.py b/pylabrobot/formulatrix/mantis/device_tests.py index 467e0ce3d5d..3061ddd7f48 100644 --- a/pylabrobot/formulatrix/mantis/device_tests.py +++ b/pylabrobot/formulatrix/mantis/device_tests.py @@ -16,9 +16,7 @@ def make_device(self) -> MantisDevice: def test_every_part_has_its_model_file(self): device = self.make_device() - drawn = [r for r in [device, *device.get_all_children()] if hasattr(r, "mesh")] - self.assertIn(device.left_chip_rack, drawn) - for resource in drawn: + for resource in (device, device.left_chip_rack, device.right_chip_rack): with self.subTest(resource=resource.name): self.assertTrue(os.path.isfile(os.path.join(RESOURCE_MODEL, f"{resource.model}.glb"))) diff --git a/pylabrobot/formulatrix/mantis/driver/master.py b/pylabrobot/formulatrix/mantis/driver/master.py index c0bd27bb5df..94837d4895a 100644 --- a/pylabrobot/formulatrix/mantis/driver/master.py +++ b/pylabrobot/formulatrix/mantis/driver/master.py @@ -55,7 +55,7 @@ class MantisDriver: """Talks to a Mantis: one command in flight, events logged as they come.""" - def _build_io(self, *, device_id: Optional[str]) -> FTDI: + def _build_io(self, *, device_id: Optional[str], ftdi_library: Optional[str]) -> FTDI: """Build the link to the Mantis ``device_id`` names, or to the first one found.""" return FTDI( human_readable_device_name="Formulatrix Mantis", @@ -63,9 +63,12 @@ def _build_io(self, *, device_id: Optional[str]) -> FTDI: vid=0x0403, pid=0x6010, interface_select=2, # interface B carries the command link + ftdi_library=ftdi_library, ) - def _build_rfid_io(self, *, device_id: Optional[str]) -> Optional[FTDI]: + def _build_rfid_io( + self, *, device_id: Optional[str], ftdi_library: Optional[str] + ) -> Optional[FTDI]: """Build the link to the RFID reader, interface A of the same FTDI bridge.""" return FTDI( human_readable_device_name="Formulatrix Mantis RFID reader", @@ -73,6 +76,7 @@ def _build_rfid_io(self, *, device_id: Optional[str]) -> Optional[FTDI]: vid=0x0403, pid=0x6010, interface_select=RFID_FTDI_INTERFACE, + ftdi_library=ftdi_library, ) def __init__( @@ -84,6 +88,7 @@ def __init__( name: str = "Mantis", declared_configuration_json: Optional[str] = None, oem_config_directory: Optional[str] = None, + ftdi_library: Optional[str] = None, ): """ Args: @@ -95,12 +100,16 @@ def __init__( declared_configuration_json: a file ``save_configuration`` wrote. Setup refuses a device that disagrees with its ``device`` block on what is fitted (``FITTED_FIELDS``). oem_config_directory: the unit's vendor ``Data/Device/Configs`` folder, read at every setup. + ftdi_library: "libftdi" or "d2xx", for both links; None picks D2XX on Windows when FTDI's + D2XX library loads, else libftdi. Ignored for a link given as ``io`` or ``rfid_io``. The unit's own values (``UNIT_BLOCKS``) come from ``oem_config_directory`` and the declared file; where both have a block, they must agree. A block neither has comes from the generation's recording, with a warning: another unit's values, less accurate on this one. """ - self.io: FTDI = io if io is not None else self._build_io(device_id=device_id) + self.io: FTDI = ( + io if io is not None else self._build_io(device_id=device_id, ftdi_library=ftdi_library) + ) self.name = name self.declared_configuration_json = declared_configuration_json self.oem_config_directory = oem_config_directory @@ -115,7 +124,9 @@ def __init__( self._event_waiters: List[Tuple[int, int, "asyncio.Future[bytes]"]] = [] self._reader: Optional["asyncio.Task[None]"] = None self.rfid_io: Optional[FTDI] = ( - rfid_io if rfid_io is not None or io is not None else self._build_rfid_io(device_id=device_id) + rfid_io + if rfid_io is not None or io is not None + else self._build_rfid_io(device_id=device_id, ftdi_library=ftdi_library) ) self.plate_clamp: Optional[PlateClamp] = None self.rfid: Optional[Rfid] = None diff --git a/pylabrobot/formulatrix/mantis/driver/simulator.py b/pylabrobot/formulatrix/mantis/driver/simulator.py index 4fdfca3a744..f3e0fa54057 100644 --- a/pylabrobot/formulatrix/mantis/driver/simulator.py +++ b/pylabrobot/formulatrix/mantis/driver/simulator.py @@ -86,11 +86,13 @@ def __init__( oem_config_directory=oem_config_directory, ) - def _build_io(self, *, device_id: Optional[str]) -> FTDI: + def _build_io(self, *, device_id: Optional[str], ftdi_library: Optional[str]) -> FTDI: """Build no link.""" return None # type: ignore[return-value] # no link: nothing reaches a wire - def _build_rfid_io(self, *, device_id: Optional[str]) -> Optional[FTDI]: + def _build_rfid_io( + self, *, device_id: Optional[str], ftdi_library: Optional[str] + ) -> Optional[FTDI]: """Build no link.""" return None