Add input path for sysfs GPIOs

doc/configuration.rst

@@ -651,6 +651,7 @@ Arguments:
   - invert (bool, default=False): optional, whether the logic level is inverted (active-low)
 
 Used by:
+  - `GpioDigitalInputDriver`_
   - `GpioDigitalOutputDriver`_
 
 NetworkSysfsGPIO
@@ -684,6 +685,7 @@ Arguments:
   - invert (bool, default=False): optional, whether the logic level is inverted (active-low)
 
 Used by:
+  - `GpioDigitalInputDriver`_
   - `GpioDigitalOutputDriver`_
 
 NetworkService
@@ -2559,6 +2561,30 @@ Implements:
 Arguments:
   - delay (float, default=2.0): delay in seconds between off and on
 
+GpioDigitalInputDriver
+~~~~~~~~~~~~~~~~~~~~~~
+The :any:`GpioDigitalInputDriver` reads a digital signal from a GPIO line.
+
+This driver configures GPIO lines via
+`the sysfs kernel interface <https://www.kernel.org/doc/html/latest/gpio/sysfs.html>`__
+as an input.
+
+Binds to:
+  gpio:
+    - `SysfsGPIO`_
+    - `MatchedSysfsGPIO`_
+    - `NetworkSysfsGPIO`_
+
+Implements:
+  - :any:`DigitalInputProtocol`
+
+.. code-block:: yaml
+
+   GpioDigitalInputDriver: {}
+
+Arguments:
+  - None
+
 GpioDigitalOutputDriver
 ~~~~~~~~~~~~~~~~~~~~~~~
 The :any:`GpioDigitalOutputDriver` writes a digital signal to a GPIO line.

labgrid/driver/__init__.py

@@ -27,7 +27,7 @@
 from .sigrokdriver import SigrokDriver, SigrokPowerDriver, SigrokDmmDriver
 from .usbstoragedriver import USBStorageDriver, Mode
 from .resetdriver import DigitalOutputResetDriver
-from .gpiodriver import GpioDigitalOutputDriver
+from .gpiodriver import GpioDigitalInputDriver, GpioDigitalOutputDriver
 from .filedigitaloutput import FileDigitalOutputDriver
 from .serialdigitaloutput import SerialPortDigitalOutputDriver
 from .xenadriver import XenaDriver

labgrid/driver/gpiodriver.py

@@ -2,13 +2,44 @@
 import attr
 
 from ..factory import target_factory
-from ..protocol import DigitalOutputProtocol
+from ..protocol import DigitalInputProtocol, DigitalOutputProtocol
 from ..resource.remote import NetworkSysfsGPIO
 from ..step import step
 from .common import Driver
 from ..util.agentwrapper import AgentWrapper
 
 
+@target_factory.reg_driver
+@attr.s(eq=False)
+class GpioDigitalInputDriver(Driver, DigitalInputProtocol):
+
+    bindings = {
+        "gpio": {"SysfsGPIO", "MatchedSysfsGPIO", "NetworkSysfsGPIO"},
+    }
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        self.wrapper = None
+
+    def on_activate(self):
+        if isinstance(self.gpio, NetworkSysfsGPIO):
+            host = self.gpio.host
+        else:
+            host = None
+        self.wrapper = AgentWrapper(host)
+        self.proxy = self.wrapper.load('sysfsgpio')
+
+    def on_deactivate(self):
+        self.wrapper.close()
+        self.wrapper = None
+        self.proxy = None
+
+    @Driver.check_active
+    @step(result=True)
+    def get(self):
+        return self.proxy.get(self.gpio.index, self.gpio.invert, 'in')
+
+
 @target_factory.reg_driver
 @attr.s(eq=False)
 class GpioDigitalOutputDriver(Driver, DigitalOutputProtocol):

labgrid/protocol/__init__.py

@@ -5,6 +5,7 @@
 from .powerprotocol import PowerProtocol
 from .filetransferprotocol import FileTransferProtocol
 from .infoprotocol import InfoProtocol
+from .digitalinputprotocol import DigitalInputProtocol
 from .digitaloutputprotocol import DigitalOutputProtocol
 from .mmioprotocol import MMIOProtocol
 from .filesystemprotocol import FileSystemProtocol

labgrid/protocol/digitalinputprotocol.py

@@ -0,0 +1,10 @@
+import abc
+
+
+class DigitalInputProtocol(abc.ABC):
+    """Abstract class providing the DigitalInputProtocol interface"""
+
+    @abc.abstractmethod
+    def get(self):
+        """Implementations should return the status of the digital input."""
+        raise NotImplementedError

labgrid/protocol/digitaloutputprotocol.py

@@ -1,13 +1,10 @@
 import abc
 
+from .digitalinputprotocol import DigitalInputProtocol
 
-class DigitalOutputProtocol(abc.ABC):
-    """Abstract class providing the DigitalOutputProtocol interface"""
 
-    @abc.abstractmethod
-    def get(self):
-        """Implementations should return the status of the digital output."""
-        raise NotImplementedError
+class DigitalOutputProtocol(DigitalInputProtocol):
+    """Abstract class providing the DigitalOutputProtocol interface"""
 
     @abc.abstractmethod
     def set(self, status):

labgrid/remote/client.py

@@ -991,7 +991,10 @@ def digital_io(self):
 
         drv = None
         try:
-            drv = target.get_driver("DigitalOutputProtocol", name=name)
+            if action == "get":
+                drv = target.get_driver("DigitalInputProtocol", name=name)
+            else:
+                drv = target.get_driver("DigitalOutputProtocol", name=name)
         except NoDriverFoundError:
             for resource in target.resources:
                 if name and resource.name != name:

labgrid/util/agents/sysfsgpio.py

@@ -1,5 +1,5 @@
 """
-This module implements switching GPIOs via sysfs GPIO kernel interface.
+This module implements accessing GPIOs via sysfs GPIO kernel interface.
 
 Takes an integer property 'index' which refers to the already exported GPIO device.
 Takes a boolean property 'invert' which inverts logical values if set to True (active-low)
@@ -8,8 +8,13 @@
 import logging
 import os
 
+
 class GpioDigitalOutput:
     _gpio_sysfs_path_prefix = '/sys/class/gpio'
+    _directions = {
+        'in': b'in',
+        'out': b'out',
+    }
 
     @staticmethod
     def _assert_gpio_line_is_exported(index):
@@ -24,30 +29,41 @@ def _assert_gpio_line_is_exported(index):
         if not os.path.exists(gpio_sysfs_path):
             raise ValueError("Device not found")
 
-    def __init__(self, index, invert):
+    def __init__(self, index, invert, direction='out'):
         self.gpio_sysfs_value_fd = None
+        if direction not in self._directions:
+            raise ValueError("GPIO direction is out of range.")
+
         self._logger = logging.getLogger("Device: ")
         GpioDigitalOutput._assert_gpio_line_is_exported(index)
         gpio_sysfs_path = os.path.join(GpioDigitalOutput._gpio_sysfs_path_prefix,
                                        f'gpio{index}')
 
-        gpio_sysfs_direction_path = os.path.join(gpio_sysfs_path, 'direction')
-        with open(gpio_sysfs_direction_path, 'rb') as direction_fd:
-            literal_value = direction_fd.read(3)
-        if literal_value != b"out":
-            self._logger.debug("Configuring GPIO %d as output.", index)
-            with open(gpio_sysfs_direction_path, 'wb') as direction_fd:
-                direction_fd.write(b'out')
+        self.gpio_sysfs_direction_path = os.path.join(gpio_sysfs_path, 'direction')
+        self.gpio_sysfs_active_low_path = os.path.join(gpio_sysfs_path, 'active_low')
+        self.configure(index, invert, direction)
 
         gpio_sysfs_value_path = os.path.join(gpio_sysfs_path, 'value')
-        self.gpio_sysfs_value_fd = os.open(gpio_sysfs_value_path, flags=(os.O_RDWR | os.O_SYNC))
-
-        gpio_sysfs_active_low_path = os.path.join(gpio_sysfs_path, 'active_low')
-        with open(gpio_sysfs_active_low_path, 'w') as active_low_fd:
+        flags = os.O_SYNC
+        if direction == 'out':
+            flags |= os.O_RDWR
+        else:
+            flags |= os.O_RDONLY
+        self.gpio_sysfs_value_fd = os.open(gpio_sysfs_value_path, flags=flags)
+
+    def configure(self, index, invert, direction):
+        with open(self.gpio_sysfs_direction_path, 'rb') as direction_fd:
+            literal_value = direction_fd.read(3).strip()
+        if literal_value != self._directions[direction]:
+            self._logger.debug("Configuring GPIO %d as %s.", index, direction)
+            with open(self.gpio_sysfs_direction_path, 'wb') as direction_fd:
+                direction_fd.write(self._directions[direction])
+
+        with open(self.gpio_sysfs_active_low_path, 'w') as active_low_fd:
             active_low_fd.write(str(int(invert)))
 
     def __del__(self):
-        if self.gpio_sysfs_value_fd:
+        if self.gpio_sysfs_value_fd is not None:
             os.close(self.gpio_sysfs_value_fd)
         self.gpio_sysfs_value_fd = None
 
@@ -75,18 +91,21 @@ def set(self, status):
 
 _gpios = {}
 
-def _get_gpio_line(index, invert):
-    if index not in _gpios:
-        _gpios[index] = GpioDigitalOutput(index=index, invert=invert)
-    return _gpios[index]
+def _get_gpio_line(index, invert, direction):
+    key = (index, invert, direction)
+    if key not in _gpios:
+        _gpios[key] = GpioDigitalOutput(index=index, invert=invert, direction=direction)
+    else:
+        _gpios[key].configure(index, invert, direction)
+    return _gpios[key]
 
 def handle_set(index, invert, status):
-    gpio_line = _get_gpio_line(index, invert)
+    gpio_line = _get_gpio_line(index, invert, 'out')
     gpio_line.set(status)
 
 
-def handle_get(index, invert):
-    gpio_line = _get_gpio_line(index, invert)
+def handle_get(index, invert, direction='out'):
+    gpio_line = _get_gpio_line(index, invert, direction)
     return gpio_line.get()
 
 methods = {

tests/test_gpiodriver.py

@@ -0,0 +1,92 @@
+import types
+
+import labgrid.driver.gpiodriver as gpiodriver
+from labgrid.driver.gpiodriver import GpioDigitalInputDriver, GpioDigitalOutputDriver
+from labgrid.remote.client import ClientSession
+from labgrid.resource.common import ResourceManager
+from labgrid.resource.remote import NetworkSysfsGPIO
+from labgrid.resource import SysfsGPIO
+
+
+class FakeWrapper:
+    def __init__(self, host, proxy):
+        self.host = host
+        self.proxy = proxy
+
+    def load(self, name):
+        assert name == 'sysfsgpio'
+        return self.proxy
+
+    def close(self):
+        pass
+
+
+def test_gpio_input_driver_get(target, monkeypatch):
+    proxy = types.SimpleNamespace(calls=[])
+
+    def proxy_get(index, invert, direction):
+        proxy.calls.append((index, invert, direction))
+        return True
+
+    proxy.get = proxy_get
+
+    monkeypatch.setattr(gpiodriver, "AgentWrapper", lambda host: FakeWrapper(host, proxy))
+
+    SysfsGPIO(target, name=None, index=13, invert=True)
+    driver = GpioDigitalInputDriver(target, name=None)
+
+    target.activate(driver)
+
+    assert driver.get() is True
+    assert proxy.calls == [(13, True, 'in')]
+
+    target.deactivate(driver)
+
+
+def test_gpio_output_driver_implements_digital_input_protocol(target):
+    SysfsGPIO(target, name=None, index=13, invert=False)
+    driver = GpioDigitalOutputDriver(target, name=None)
+
+    assert target.get_driver("DigitalInputProtocol", activate=False) is driver
+
+
+def test_client_io_get_uses_configured_gpio_input_driver(target, monkeypatch, capsys):
+    proxy = types.SimpleNamespace(calls=[])
+
+    def proxy_get(index, invert, direction):
+        proxy.calls.append((index, invert, direction))
+        return True
+
+    proxy.get = proxy_get
+
+    monkeypatch.setattr(gpiodriver, "AgentWrapper", lambda host: FakeWrapper(host, proxy))
+
+    SysfsGPIO(target, name="gpio_in", index=13, invert=False)
+    GpioDigitalInputDriver(target, name="gpio_in")
+
+    session = object.__new__(ClientSession)
+    session.args = types.SimpleNamespace(action="get", name="gpio_in")
+    session.get_acquired_place = lambda: types.SimpleNamespace(name="test")
+    session._get_target = lambda place: target
+
+    session.digital_io()
+
+    assert "digital IO gpio_in for place test is high" in capsys.readouterr().out
+    assert proxy.calls == [(13, False, 'in')]
+
+
+def test_client_io_get_keeps_network_sysfs_output_fallback(target, monkeypatch, mocker):
+    monkeypatch.setattr(NetworkSysfsGPIO, "manager_cls", ResourceManager)
+    driver = types.SimpleNamespace(get=mocker.MagicMock(return_value=False))
+    session = object.__new__(ClientSession)
+    session.args = types.SimpleNamespace(action="get", name="gpio")
+    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)
+
+    NetworkSysfsGPIO(target, name="gpio", host="exporter", index=13, invert=False)
+
+    session.digital_io()
+
+    session._get_driver_or_new.assert_called_once_with(target, "GpioDigitalOutputDriver", name="gpio")
+    driver.get.assert_called_once_with()