diff --git a/doc/configuration.rst b/doc/configuration.rst index 57b23f13c..439598c80 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -575,6 +575,44 @@ Arguments: Used by: - `HIDRelayDriver`_ +FTDIGPIO +++++++++ +An :class:`~labgrid.resource.udev.FTDIGPIO` resource describes a single FTDI +data-bus GPIO line. +This uses FTDI bit-bang mode for the 8-bit data bus exposed by the selected +interface, not CBUS ``ftdi-cbus`` gpiochips or EEPROM-configured ACBUS +functions. + +Supported devices are FT2232-style devices (``0403:6010``, interfaces 1..2), +FT4232H (``0403:6011``, interfaces 1..4), and FT232H (``0403:6014``, +interface 1). For each interface, ``index`` selects bit ``0`` to ``7`` on +that interface's data bus. +Writes use a read-modify-write cycle on the selected 8-bit interface to avoid +changing unrelated bits managed by labgrid. + +.. code-block:: yaml + + FTDIGPIO: + index: 0 + interface: 1 + invert: false + match: + ID_PATH: 'pci-0000:00:14.0-usb-0:2' + +Arguments: + - index (int): GPIO bit number to use, from ``0`` to ``7`` + - interface (int, default=1): FTDI interface/channel number to use + - invert (bool, default=False): whether to invert the logical GPIO value + - match (dict): key and value pairs for a udev match, see `udev Matching`_ + +Used by: + - `FTDIGPIODriver`_ + +NetworkFTDIGPIO ++++++++++++++++ +A :any:`NetworkFTDIGPIO` describes an `FTDIGPIO`_ resource available on a +remote computer. + HttpDigitalOutput +++++++++++++++++ An :any:`HttpDigitalOutput` resource describes a generic digital output that @@ -2584,6 +2622,28 @@ Implements: Arguments: - None +FTDIGPIODriver +~~~~~~~~~~~~~~ +The :any:`FTDIGPIODriver` controls one `FTDIGPIO`_ or `NetworkFTDIGPIO`_ line +via FTDI bit-bang GPIO output mode. + +The FTDI serial kernel driver must not be bound to the same interface. + +Binds to: + gpio: + - `FTDIGPIO`_ + - `NetworkFTDIGPIO`_ + +Implements: + - :any:`DigitalOutputProtocol` + +.. code-block:: yaml + + FTDIGPIODriver: {} + +Arguments: + - None + SerialPortDigitalOutputDriver ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ The :any:`SerialPortDigitalOutputDriver` makes it possible to use a UART diff --git a/labgrid/driver/__init__.py b/labgrid/driver/__init__.py index d3cd6f55e..c85b854f6 100644 --- a/labgrid/driver/__init__.py +++ b/labgrid/driver/__init__.py @@ -28,6 +28,7 @@ from .usbstoragedriver import USBStorageDriver, Mode from .resetdriver import DigitalOutputResetDriver from .gpiodriver import GpioDigitalOutputDriver +from .ftdigpiodriver import FTDIGPIODriver from .filedigitaloutput import FileDigitalOutputDriver from .serialdigitaloutput import SerialPortDigitalOutputDriver from .xenadriver import XenaDriver diff --git a/labgrid/driver/ftdigpiodriver.py b/labgrid/driver/ftdigpiodriver.py new file mode 100644 index 000000000..424db4290 --- /dev/null +++ b/labgrid/driver/ftdigpiodriver.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""FTDI GPIO driver using a labgrid agent.""" + +import threading + +import attr + +from ..factory import target_factory +from ..protocol import DigitalOutputProtocol +from ..resource.remote import NetworkFTDIGPIO, RemoteUSBResource +from ..step import step +from ..util.agentwrapper import AgentWrapper +from .common import Driver + +_shared_agents = {} +_shared_lock = threading.Lock() + + +def _acquire_agent(host, busnum, devnum, interface): + key = (host, busnum, devnum, interface) + with _shared_lock: + entry = _shared_agents.get(key) + if entry is None: + wrapper = AgentWrapper(host) + proxy = wrapper.load("ftdigpio") + entry = {"wrapper": wrapper, "proxy": proxy, "refs": 0} + _shared_agents[key] = entry + entry["refs"] += 1 + return entry["proxy"] + + +def _release_agent(host, busnum, devnum, interface): + key = (host, busnum, devnum, interface) + with _shared_lock: + entry = _shared_agents.get(key) + if entry is None: + return + entry["refs"] -= 1 + if entry["refs"] <= 0: + del _shared_agents[key] + try: + entry["proxy"].close() + finally: + entry["wrapper"].close() + + +@target_factory.reg_driver +@attr.s(eq=False) +class FTDIGPIODriver(Driver, DigitalOutputProtocol): + """Control one FTDI data-bus GPIO line through a labgrid agent.""" + + bindings = { + "gpio": {"FTDIGPIO", NetworkFTDIGPIO}, + "networkservice": {"NetworkService", None}, + } + + def __attrs_post_init__(self): + super().__attrs_post_init__() + self._proxy = None + self._host = None + + def on_activate(self): + self._host = self.gpio.host if isinstance(self.gpio, RemoteUSBResource) else None + if self.networkservice and self.networkservice.address == self._host: + self._host = f"{self.networkservice.username}@{self._host}" + self._proxy = _acquire_agent(self._host, self.gpio.busnum, self.gpio.devnum, self.gpio.interface) + + def on_deactivate(self): + self._proxy = None + _release_agent(self._host, self.gpio.busnum, self.gpio.devnum, self.gpio.interface) + self._host = None + + @Driver.check_active + @step(result=True) + def get(self): + status = bool(self._proxy.get( + self.gpio.vendor_id, + self.gpio.model_id, + self.gpio.busnum, + self.gpio.devnum, + self.gpio.interface, + self.gpio.index, + )) + if self.gpio.invert: + status = not status + return status + + @Driver.check_active + @step(args=["status"]) + def set(self, status): + if self.gpio.invert: + status = not status + self._proxy.set( + self.gpio.vendor_id, + self.gpio.model_id, + self.gpio.busnum, + self.gpio.devnum, + self.gpio.interface, + self.gpio.index, + bool(status), + ) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 069240b9e..c3a6b3ec2 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -987,7 +987,13 @@ def digital_io(self): WaveshareModbusTCPCoil, Eth008DigitalOutput, ) - from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay + from ..resource.remote import ( + NetworkDeditecRelais8, + NetworkFTDIGPIO, + NetworkHIDRelay, + NetworkLXAIOBusPIO, + NetworkSysfsGPIO, + ) drv = None try: @@ -1010,6 +1016,8 @@ def digital_io(self): drv = self._get_driver_or_new(target, "DeditecRelaisDriver", name=name) elif isinstance(resource, NetworkSysfsGPIO): drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) + elif isinstance(resource, NetworkFTDIGPIO): + drv = self._get_driver_or_new(target, "FTDIGPIODriver", name=name) elif isinstance(resource, NetworkLXAIOBusPIO): drv = self._get_driver_or_new(target, "LXAIOBusPIODriver", name=name) elif isinstance(resource, NetworkHIDRelay): diff --git a/labgrid/remote/exporter.py b/labgrid/remote/exporter.py index 82f1da6b6..1772f7f02 100755 --- a/labgrid/remote/exporter.py +++ b/labgrid/remote/exporter.py @@ -549,6 +549,25 @@ def _get_params(self): } +@attr.s(eq=False) +class USBFTDIGPIOExport(USBGenericExport): + """ResourceExport for GPIOs on FTDI data-bus bit-bang interfaces""" + + def _get_params(self): + """Helper function to return parameters""" + return { + "host": self.host, + "busnum": self.local.busnum, + "devnum": self.local.devnum, + "path": self.local.path, + "vendor_id": self.local.vendor_id, + "model_id": self.local.model_id, + "index": self.local.index, + "interface": self.local.interface, + "invert": self.local.invert, + } + + @attr.s(eq=False) class USBFlashableExport(USBGenericExport): """ResourceExport for Flashable USB devices""" @@ -591,6 +610,7 @@ def __attrs_post_init__(self): exports["USBPowerPort"] = USBPowerPortExport exports["DeditecRelais8"] = USBDeditecRelaisExport exports["HIDRelay"] = USBHIDRelayExport +exports["FTDIGPIO"] = USBFTDIGPIOExport exports["USBFlashableDevice"] = USBFlashableExport exports["LXAUSBMux"] = USBGenericExport diff --git a/labgrid/resource/__init__.py b/labgrid/resource/__init__.py index 53d88f458..bb561169a 100644 --- a/labgrid/resource/__init__.py +++ b/labgrid/resource/__init__.py @@ -12,6 +12,7 @@ AndroidUSBFastboot, DFUDevice, DeditecRelais8, + FTDIGPIO, HIDRelay, IMXUSBLoader, LXAUSBMux, diff --git a/labgrid/resource/remote.py b/labgrid/resource/remote.py index bbe13aaa2..4474a39e4 100644 --- a/labgrid/resource/remote.py +++ b/labgrid/resource/remote.py @@ -345,6 +345,33 @@ def __attrs_post_init__(self): super().__attrs_post_init__() +@target_factory.reg_resource +@attr.s(eq=False) +class NetworkFTDIGPIO(RemoteUSBResource): + """The NetworkFTDIGPIO describes a remotely accessible FTDI GPIO line""" + index = attr.ib( + default=0, + converter=int, + validator=attr.validators.and_( + attr.validators.instance_of(int), + attr.validators.ge(0), + attr.validators.le(7), + ), + ) + interface = attr.ib( + default=1, + converter=int, + validator=attr.validators.and_( + attr.validators.instance_of(int), + attr.validators.ge(1), + ), + ) + invert = attr.ib(default=False, validator=attr.validators.instance_of(bool)) + def __attrs_post_init__(self): + self.timeout = 10.0 + super().__attrs_post_init__() + + @target_factory.reg_resource @attr.s(eq=False) class NetworkSysfsGPIO(NetworkResource, ManagedResource): diff --git a/labgrid/resource/udev.py b/labgrid/resource/udev.py index 7ecd4b01a..3657bae06 100644 --- a/labgrid/resource/udev.py +++ b/labgrid/resource/udev.py @@ -797,6 +797,53 @@ def filter_match(self, device): return super().filter_match(device) +@target_factory.reg_resource +@attr.s(eq=False) +class FTDIGPIO(USBResource): + """This resource describes a single GPIO line on an FTDI bit-bang interface.""" + + index = attr.ib( + default=0, + converter=int, + validator=attr.validators.and_( + attr.validators.instance_of(int), + attr.validators.ge(0), + attr.validators.le(7), + ), + ) + interface = attr.ib( + default=1, + converter=int, + validator=attr.validators.and_( + attr.validators.instance_of(int), + attr.validators.ge(1), + ), + ) + invert = attr.ib(default=False, validator=attr.validators.instance_of(bool)) + + def __attrs_post_init__(self): + self.match["DEVTYPE"] = "usb_device" + super().__attrs_post_init__() + + def filter_match(self, device): + usb_device = device + if device.device_type != "usb_device": + usb_device = device.find_parent("usb", "usb_device") + if usb_device is None: + return False + + match = ( + usb_device.properties.get("ID_VENDOR_ID"), + usb_device.properties.get("ID_MODEL_ID"), + ) + if match not in [("0403", "6010"), # FT2232C/D/H Dual UART/FIFO IC + ("0403", "6011"), # FT4232H Quad UART/MPSSE IC + ("0403", "6014"), # FT232HL/Q + ]: + return False + + return super().filter_match(device) + @target_factory.reg_resource @attr.s(eq=False) class USBFlashableDevice(USBResource): diff --git a/labgrid/util/agents/ftdigpio.py b/labgrid/util/agents/ftdigpio.py new file mode 100644 index 000000000..f40deee2b --- /dev/null +++ b/labgrid/util/agents/ftdigpio.py @@ -0,0 +1,149 @@ +# SPDX-License-Identifier: GPL-2.0-or-later +"""Agent for controlling FTDI data-bus GPIOs via bit-bang mode.""" + +import threading + +import usb.core +import usb.util + +SIO_SET_BITMODE = 11 +SIO_READ_PINS = 12 +BITMODE_ASYNC_BITBANG = 1 +OUT_REQTYPE = 0x40 +IN_REQTYPE = 0xC0 + +USB_TIMEOUT = 1000 +GPIO_MASK = 0xFF +SUPPORTED_DEVICES = { + 0x6010: 2, # FT2232C/D/H Dual UART/FIFO IC + 0x6011: 4, # FT4232H Quad UART/MPSSE IC + 0x6014: 1, # FT232HL/Q +} + + +class FTDIGPIO: + def __init__(self, vendor_id, model_id, busnum, devnum, interface): + self._validate_device(vendor_id, model_id, interface) + self._interface = interface - 1 + self._index = interface + self._lock = threading.Lock() + + self._dev = self._find_device(vendor_id, model_id, busnum, devnum) + self._detach_kernel_driver() + try: + cfg = self._dev.get_active_configuration() + except usb.core.USBError: + self._dev.set_configuration() + self._detach_kernel_driver() + cfg = self._dev.get_active_configuration() + + intf = cfg[(self._interface, 0)] + usb.util.claim_interface(self._dev, self._interface) + self._ep_out = usb.util.find_descriptor( + intf, + custom_match=lambda ep: usb.util.endpoint_direction(ep.bEndpointAddress) == usb.util.ENDPOINT_OUT, + ) + if self._ep_out is None: + raise ValueError("FTDI output endpoint not found") + + def close(self): + try: + try: + usb.util.release_interface(self._dev, self._interface) + except usb.core.USBError: + pass + finally: + usb.util.dispose_resources(self._dev) + + def _detach_kernel_driver(self): + if self._dev.is_kernel_driver_active(self._interface): + self._dev.detach_kernel_driver(self._interface) + + @staticmethod + def _validate_device(vendor_id, model_id, interface): + if vendor_id != 0x0403 or model_id not in SUPPORTED_DEVICES: + raise ValueError("Unsupported FTDI GPIO device") + if not 1 <= interface <= SUPPORTED_DEVICES[model_id]: + raise ValueError("FTDI GPIO interface is not supported by this device") + + @staticmethod + def _find_device(vendor_id, model_id, busnum, devnum): + for dev in usb.core.find(find_all=True, idVendor=vendor_id, idProduct=model_id): + if dev.bus == busnum and dev.address == devnum: + return dev + raise ValueError("FTDI device not found") + + def _ctrl_out(self, request, value): + self._dev.ctrl_transfer(OUT_REQTYPE, request, value, self._index, None, USB_TIMEOUT) + + def _ctrl_in(self, request, value, length): + return bytes(self._dev.ctrl_transfer(IN_REQTYPE, request, value, self._index, length, USB_TIMEOUT)) + + def _set_bitmode(self, mask, mode): + self._ctrl_out(SIO_SET_BITMODE, mask | (mode << 8)) + + def _write(self, data): + self._ep_out.write(bytes(data), USB_TIMEOUT) + + @staticmethod + def _validate_index(index): + if not 0 <= index <= 7: + raise ValueError("FTDI bit-bang GPIO only supports indexes 0-7") + + def _read_gpio_byte(self): + data = self._ctrl_in(SIO_READ_PINS, 0, 1) + if not data: + raise TimeoutError("FTDI GPIO read returned no data") + return data[0] + + def get(self, index): + self._validate_index(index) + with self._lock: + value = self._read_gpio_byte() + return bool(value & (1 << (index % 8))) + + def set(self, index, status): + self._validate_index(index) + mask = 1 << index + with self._lock: + output = self._read_gpio_byte() + if status: + output |= mask + else: + output &= ~mask + self._set_bitmode(GPIO_MASK, BITMODE_ASYNC_BITBANG) + self._write([output]) + + +def _run_with_device(vendor_id, model_id, busnum, devnum, interface, callback): + device = FTDIGPIO(vendor_id, model_id, busnum, devnum, interface) + try: + return callback(device) + finally: + device.close() + + +def handle_get(vendor_id, model_id, busnum, devnum, interface, index): + return _run_with_device( + vendor_id, model_id, busnum, devnum, interface, + lambda device: device.get(int(index)), + ) + + +def handle_set(vendor_id, model_id, busnum, devnum, interface, index, status): + _run_with_device( + vendor_id, model_id, busnum, devnum, interface, + lambda device: device.set(int(index), bool(status)), + ) + return True + + +def handle_close(): + return True + + +methods = { + "get": handle_get, + "set": handle_set, + "close": handle_close, +} diff --git a/tests/test_ftdigpiodriver.py b/tests/test_ftdigpiodriver.py new file mode 100644 index 000000000..0b7a24b85 --- /dev/null +++ b/tests/test_ftdigpiodriver.py @@ -0,0 +1,516 @@ +import types + +import pytest + +from labgrid.driver import ftdigpiodriver +from labgrid.driver.ftdigpiodriver import FTDIGPIODriver +from labgrid.remote.client import ClientSession +from labgrid.resource.common import ResourceManager +from labgrid.resource.remote import NetworkFTDIGPIO +from labgrid.resource.udev import FTDIGPIO + + +def test_ftdigpio_resource_create(target, monkeypatch): + monkeypatch.setattr(FTDIGPIO, "manager_cls", ResourceManager) + resource = FTDIGPIO(target, name=None, index=0, interface=1) + + assert resource.index == 0 + assert resource.interface == 1 + assert resource.match["DEVTYPE"] == "usb_device" + + +def test_ftdigpio_resource_validates_range(target, monkeypatch): + monkeypatch.setattr(FTDIGPIO, "manager_cls", ResourceManager) + + with pytest.raises(ValueError): + FTDIGPIO(target, name=None, index=8, interface=1) + + with pytest.raises(ValueError): + FTDIGPIO(target, name=None, index=0, interface=0) + + +def test_ftdigpio_resource_filters_supported_devices(target, monkeypatch): + monkeypatch.setattr(FTDIGPIO, "manager_cls", ResourceManager) + resource = FTDIGPIO(target, name=None) + + def device(vendor_id, model_id): + return types.SimpleNamespace( + device_type="usb_device", + properties={ + "ID_VENDOR_ID": vendor_id, + "ID_MODEL_ID": model_id, + }, + ) + + def child_device(parent): + return types.SimpleNamespace( + device_type="usb_interface", + find_parent=lambda subsystem, device_type: parent, + ) + + assert resource.filter_match(device("0403", "6014")) is True + assert resource.filter_match(child_device(device("0403", "6014"))) is True + assert resource.filter_match(child_device(None)) is False + assert resource.filter_match(device("0403", "6010")) is True + assert resource.filter_match(device("0403", "6011")) is True + assert resource.filter_match(device("0403", "6001")) is False + + +def test_network_ftdigpio_resource_validates_range(target, monkeypatch): + monkeypatch.setattr(NetworkFTDIGPIO, "manager_cls", ResourceManager) + kwargs = { + "host": "exporter", + "busnum": 1, + "devnum": 39, + "path": "1-12.4.2", + "vendor_id": 0x0403, + "model_id": 0x6014, + } + + with pytest.raises(ValueError): + NetworkFTDIGPIO(target, name=None, index=8, interface=1, **kwargs) + + with pytest.raises(ValueError): + NetworkFTDIGPIO(target, name=None, index=0, interface=0, **kwargs) + + +def test_ftdigpio_driver_create(target, monkeypatch): + monkeypatch.setattr(NetworkFTDIGPIO, "manager_cls", ResourceManager) + NetworkFTDIGPIO( + target, + name=None, + host="exporter", + busnum=1, + devnum=39, + path="1-12.4.2", + vendor_id=0x0403, + model_id=0x6014, + index=0, + interface=1, + ) + + driver = FTDIGPIODriver(target, name=None) + + assert isinstance(driver, FTDIGPIODriver) + + +def test_ftdigpio_driver_set_get(target, monkeypatch): + monkeypatch.setattr(NetworkFTDIGPIO, "manager_cls", ResourceManager) + proxy = types.SimpleNamespace(calls=[], value=False) + + def proxy_set(vendor_id, model_id, busnum, devnum, interface, index, status): + proxy.calls.append((vendor_id, model_id, busnum, devnum, interface, index, status)) + proxy.value = status + + def proxy_get(vendor_id, model_id, busnum, devnum, interface, index): + proxy.calls.append((vendor_id, model_id, busnum, devnum, interface, index)) + return proxy.value + + proxy.set = proxy_set + proxy.get = proxy_get + proxy.close = lambda: None + + class FakeWrapper: + def __init__(self, host): + self.host = host + + def load(self, name): + assert name == "ftdigpio" + return proxy + + def close(self): + pass + + monkeypatch.setattr(ftdigpiodriver, "AgentWrapper", FakeWrapper) + ftdigpiodriver._shared_agents.clear() + + resource = NetworkFTDIGPIO( + target, + name=None, + host="exporter", + busnum=1, + devnum=39, + path="1-12.4.2", + vendor_id=0x0403, + model_id=0x6014, + index=2, + interface=1, + invert=True, + ) + resource.avail = True + driver = FTDIGPIODriver(target, name=None) + + target.activate(driver) + driver.set(True) + assert proxy.calls[-1] == (0x0403, 0x6014, 1, 39, 1, 2, False) + assert driver.get() is True + + target.deactivate(driver) + + +def test_ftdigpio_driver_release_ignores_unknown_agent(): + ftdigpiodriver._shared_agents.clear() + + ftdigpiodriver._release_agent("exporter", 1, 39, 1) + + +def test_ftdigpio_agent(monkeypatch): + from labgrid.util.agents import ftdigpio + + class FakeEndpoint: + def __init__(self, address): + self.bEndpointAddress = address + self.writes = [] + + def write(self, data, timeout=None): + self.writes.append(bytes(data)) + + out_ep = FakeEndpoint(0x02) + in_ep = FakeEndpoint(0x81) + + class FakeConfig: + def __getitem__(self, item): + assert item == (0, 0) + return [out_ep, in_ep] + + class FakeDevice: + bus = 1 + address = 39 + + def __init__(self): + self.control = [] + self.detached = [] + self.released = [] + self.pins = b"\x00" + + def set_configuration(self): + pass + + def is_kernel_driver_active(self, interface): + assert interface == 0 + return False + + def detach_kernel_driver(self, interface): + self.detached.append(interface) + + def get_active_configuration(self): + return FakeConfig() + + def ctrl_transfer(self, request_type, request, value, index, data, timeout=None): + self.control.append((request_type, request, value, index, data)) + if request_type == ftdigpio.IN_REQTYPE and request == ftdigpio.SIO_READ_PINS: + return self.pins + return None + + device = FakeDevice() + monkeypatch.setattr(ftdigpio.usb.core, "find", lambda **kwargs: [device]) + monkeypatch.setattr(ftdigpio.usb.util, "claim_interface", lambda dev, interface: None) + monkeypatch.setattr( + ftdigpio.usb.util, + "release_interface", + lambda dev, interface: device.released.append(interface), + ) + monkeypatch.setattr(ftdigpio.usb.util, "dispose_resources", lambda dev: None) + + device.pins = b"\x02" + assert ftdigpio.handle_set(0x0403, 0x6014, 1, 39, 1, 2, True) is True + assert device.control[-1] == ( + ftdigpio.OUT_REQTYPE, + ftdigpio.SIO_SET_BITMODE, + 0x01ff, + 1, + None, + ) + assert out_ep.writes[-1] == b"\x06" + + device.pins = b"\x06" + ftdigpio.handle_set(0x0403, 0x6014, 1, 39, 1, 3, True) + assert device.control[-1] == ( + ftdigpio.OUT_REQTYPE, + ftdigpio.SIO_SET_BITMODE, + 0x01ff, + 1, + None, + ) + assert out_ep.writes[-1] == b"\x0e" + + device.pins = b"\x02" + ftdigpio.handle_set(0x0403, 0x6014, 1, 39, 1, 2, False) + assert out_ep.writes[-1] == b"\x02" + + device.pins = b"\x08" + assert ftdigpio.handle_get(0x0403, 0x6014, 1, 39, 1, 3) is True + assert ftdigpio.handle_get(0x0403, 0x6014, 1, 39, 1, 2) is False + assert device.released == [0, 0, 0, 0, 0] + assert ftdigpio.handle_close() is True + + +def test_ftdigpio_agent_rejects_empty_pin_read(monkeypatch): + from labgrid.util.agents import ftdigpio + + class FakeEndpoint: + def __init__(self, address): + self.bEndpointAddress = address + + def write(self, data, timeout=None): + pass + + out_ep = FakeEndpoint(0x02) + in_ep = FakeEndpoint(0x81) + + class FakeConfig: + def __getitem__(self, item): + assert item == (0, 0) + return [out_ep, in_ep] + + class FakeDevice: + bus = 1 + address = 39 + + def set_configuration(self): + pass + + def is_kernel_driver_active(self, interface): + return False + + def get_active_configuration(self): + return FakeConfig() + + def ctrl_transfer(self, request_type, request, value, index, data, timeout=None): + if request_type == ftdigpio.IN_REQTYPE and request == ftdigpio.SIO_READ_PINS: + return b"" + return None + + device = FakeDevice() + monkeypatch.setattr(ftdigpio.usb.core, "find", lambda **kwargs: [device]) + monkeypatch.setattr(ftdigpio.usb.util, "claim_interface", lambda dev, interface: None) + monkeypatch.setattr(ftdigpio.usb.util, "release_interface", lambda dev, interface: None) + monkeypatch.setattr(ftdigpio.usb.util, "dispose_resources", lambda dev: None) + + with pytest.raises(TimeoutError, match="no data"): + ftdigpio.handle_get(0x0403, 0x6014, 1, 39, 1, 3) + + +def test_ftdigpio_agent_configures_device_after_usb_error(monkeypatch): + from labgrid.util.agents import ftdigpio + + class FakeEndpoint: + bEndpointAddress = 0x02 + + def write(self, data, timeout=None): + pass + + class FakeConfig: + def __getitem__(self, item): + assert item == (0, 0) + return [FakeEndpoint()] + + class FakeDevice: + bus = 1 + address = 39 + + def __init__(self): + self.active_config_attempts = 0 + self.configured = False + self.detached = [] + + def set_configuration(self): + self.configured = True + + def is_kernel_driver_active(self, interface): + assert interface == 0 + return True + + def detach_kernel_driver(self, interface): + self.detached.append(interface) + + def get_active_configuration(self): + self.active_config_attempts += 1 + if self.active_config_attempts == 1: + raise ftdigpio.usb.core.USBError("not configured") + return FakeConfig() + + def ctrl_transfer(self, request_type, request, value, index, data, timeout=None): + if request_type == ftdigpio.IN_REQTYPE and request == ftdigpio.SIO_READ_PINS: + return b"\x00" + return None + + device = FakeDevice() + monkeypatch.setattr(ftdigpio.usb.core, "find", lambda **kwargs: [device]) + monkeypatch.setattr(ftdigpio.usb.util, "claim_interface", lambda dev, interface: None) + monkeypatch.setattr(ftdigpio.usb.util, "release_interface", lambda dev, interface: None) + monkeypatch.setattr(ftdigpio.usb.util, "dispose_resources", lambda dev: None) + + assert ftdigpio.handle_get(0x0403, 0x6014, 1, 39, 1, 0) is False + assert device.configured is True + assert device.detached == [0, 0] + + +def test_ftdigpio_agent_rejects_missing_output_endpoint(monkeypatch): + from labgrid.util.agents import ftdigpio + + class FakeEndpoint: + bEndpointAddress = 0x81 + + class FakeConfig: + def __getitem__(self, item): + assert item == (0, 0) + return [FakeEndpoint()] + + class FakeDevice: + bus = 1 + address = 39 + + def set_configuration(self): + pass + + def is_kernel_driver_active(self, interface): + return False + + def get_active_configuration(self): + return FakeConfig() + + device = FakeDevice() + monkeypatch.setattr(ftdigpio.usb.core, "find", lambda **kwargs: [device]) + monkeypatch.setattr(ftdigpio.usb.util, "claim_interface", lambda dev, interface: None) + + with pytest.raises(ValueError, match="output endpoint"): + ftdigpio.FTDIGPIO(0x0403, 0x6014, 1, 39, 1) + + +def test_ftdigpio_agent_close_ignores_release_error(monkeypatch): + from labgrid.util.agents import ftdigpio + + released = [] + disposed = [] + device = types.SimpleNamespace() + gpio = object.__new__(ftdigpio.FTDIGPIO) + gpio._dev = device + gpio._interface = 0 + + def release_interface(dev, interface): + released.append((dev, interface)) + raise ftdigpio.usb.core.USBError("already released") + + monkeypatch.setattr(ftdigpio.usb.util, "release_interface", release_interface) + monkeypatch.setattr(ftdigpio.usb.util, "dispose_resources", lambda dev: disposed.append(dev)) + + gpio.close() + + assert released == [(device, 0)] + assert disposed == [device] + + +def test_ftdigpio_agent_rejects_missing_device(monkeypatch): + from labgrid.util.agents import ftdigpio + + monkeypatch.setattr(ftdigpio.usb.core, "find", lambda **kwargs: []) + + with pytest.raises(ValueError, match="not found"): + ftdigpio.FTDIGPIO._find_device(0x0403, 0x6014, 1, 39) + + +def test_ftdigpio_agent_rejects_unsupported_index(): + from labgrid.util.agents import ftdigpio + + with pytest.raises(ValueError, match="0-7"): + ftdigpio.FTDIGPIO._validate_index(8) + + +def test_ftdigpio_agent_validates_supported_devices(): + from labgrid.util.agents import ftdigpio + + ftdigpio.FTDIGPIO._validate_device(0x0403, 0x6010, 2) + ftdigpio.FTDIGPIO._validate_device(0x0403, 0x6011, 4) + ftdigpio.FTDIGPIO._validate_device(0x0403, 0x6014, 1) + + with pytest.raises(ValueError, match="Unsupported"): + ftdigpio.FTDIGPIO._validate_device(0x0403, 0x6001, 1) + + with pytest.raises(ValueError, match="interface"): + ftdigpio.FTDIGPIO._validate_device(0x0403, 0x6014, 2) + + +def test_ftdigpio_export_params(): + from labgrid.remote.exporter import USBFTDIGPIOExport + + export = object.__new__(USBFTDIGPIOExport) + export.host = "exporter" + export.local = types.SimpleNamespace( + busnum=1, + devnum=39, + path="1-12.4.2", + vendor_id=0x0403, + model_id=0x6014, + index=2, + interface=1, + invert=True, + ) + + assert export._get_params() == { + "host": "exporter", + "busnum": 1, + "devnum": 39, + "path": "1-12.4.2", + "vendor_id": 0x0403, + "model_id": 0x6014, + "index": 2, + "interface": 1, + "invert": True, + } + + +def test_ftdigpio_client_digital_io(target, mocker, monkeypatch): + monkeypatch.setattr(NetworkFTDIGPIO, "manager_cls", ResourceManager) + driver = types.SimpleNamespace(set=mocker.MagicMock()) + session = object.__new__(ClientSession) + session.args = types.SimpleNamespace(action="high", name=None) + session.get_acquired_place = lambda: types.SimpleNamespace(name="test") + session._get_target = lambda place: target + session._get_driver_or_new = mocker.MagicMock(return_value=driver) + + NetworkFTDIGPIO( + target, + name=None, + host="exporter", + busnum=1, + devnum=39, + path="1-12.4.2", + vendor_id=0x0403, + model_id=0x6014, + index=2, + interface=1, + ) + + session.digital_io() + + session._get_driver_or_new.assert_called_once_with(target, "FTDIGPIODriver", name=None) + driver.set.assert_called_once_with(True) + + +def test_ftdigpio_client_digital_io_get(target, mocker, monkeypatch, capsys): + monkeypatch.setattr(NetworkFTDIGPIO, "manager_cls", ResourceManager) + driver = types.SimpleNamespace(get=mocker.MagicMock(return_value=True)) + session = object.__new__(ClientSession) + session.args = types.SimpleNamespace(action="get", name=None) + session.get_acquired_place = lambda: types.SimpleNamespace(name="test") + session._get_target = lambda place: target + session._get_driver_or_new = mocker.MagicMock(return_value=driver) + + NetworkFTDIGPIO( + target, + name=None, + host="exporter", + busnum=1, + devnum=39, + path="1-12.4.2", + vendor_id=0x0403, + model_id=0x6014, + index=2, + interface=1, + ) + + session.digital_io() + + session._get_driver_or_new.assert_called_once_with(target, "FTDIGPIODriver", name=None) + assert "digital IO for place test is high" in capsys.readouterr().out