Add FTDI bit-bang GPIO output support

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

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

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),
+        )

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):

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
 

labgrid/resource/__init__.py

@@ -12,6 +12,7 @@
     AndroidUSBFastboot,
     DFUDevice,
     DeditecRelais8,
+    FTDIGPIO,
     HIDRelay,
     IMXUSBLoader,
     LXAUSBMux,

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):

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):