From 7216c574752e508759d0b281c9005132ddd364c8 Mon Sep 17 00:00:00 2001 From: Ozan Durgut Date: Mon, 13 Jul 2026 14:20:02 +0200 Subject: [PATCH 1/4] protocol: add digital input protocol Add a dedicated DigitalInputProtocol for drivers which can read a digital signal without also exposing writes. Keep DigitalOutputProtocol as a specialization of the input protocol, matching existing output drivers which already implement get(). Co-developed-by: Felix Zwettler Signed-off-by: Felix Zwettler Signed-off-by: Ozan Durgut --- labgrid/protocol/__init__.py | 1 + labgrid/protocol/digitalinputprotocol.py | 10 ++++++++++ labgrid/protocol/digitaloutputprotocol.py | 9 +++------ 3 files changed, 14 insertions(+), 6 deletions(-) create mode 100644 labgrid/protocol/digitalinputprotocol.py diff --git a/labgrid/protocol/__init__.py b/labgrid/protocol/__init__.py index 0ac2256229..3069d4fb4c 100644 --- a/labgrid/protocol/__init__.py +++ b/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 diff --git a/labgrid/protocol/digitalinputprotocol.py b/labgrid/protocol/digitalinputprotocol.py new file mode 100644 index 0000000000..768a4387e1 --- /dev/null +++ b/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 diff --git a/labgrid/protocol/digitaloutputprotocol.py b/labgrid/protocol/digitaloutputprotocol.py index 6b1ce7e239..66f5e0ab9b 100644 --- a/labgrid/protocol/digitaloutputprotocol.py +++ b/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): From ee9b25f4cbf0880d982b9d63a601055cdf993fa3 Mon Sep 17 00:00:00 2001 From: Ozan Durgut Date: Mon, 13 Jul 2026 14:20:31 +0200 Subject: [PATCH 2/4] driver: add GPIO digital input support Add GpioDigitalInputDriver for sysfs GPIO resources. Extend the sysfsgpio agent to configure lines for input reads while keeping output as the default path for existing users. Reconfigure cached GPIO lines before reuse so switching between input and output access leaves the sysfs direction in the state requested by the active driver. Co-developed-by: Felix Zwettler Signed-off-by: Felix Zwettler Signed-off-by: Ozan Durgut --- labgrid/driver/__init__.py | 2 +- labgrid/driver/gpiodriver.py | 33 ++++++++++++++++- labgrid/util/agents/sysfsgpio.py | 61 +++++++++++++++++++++----------- 3 files changed, 73 insertions(+), 23 deletions(-) diff --git a/labgrid/driver/__init__.py b/labgrid/driver/__init__.py index d3cd6f55ef..e84db4f77d 100644 --- a/labgrid/driver/__init__.py +++ b/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 diff --git a/labgrid/driver/gpiodriver.py b/labgrid/driver/gpiodriver.py index 0fc046092c..a519933b9c 100644 --- a/labgrid/driver/gpiodriver.py +++ b/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): diff --git a/labgrid/util/agents/sysfsgpio.py b/labgrid/util/agents/sysfsgpio.py index 580b77f0e2..bc9925605b 100644 --- a/labgrid/util/agents/sysfsgpio.py +++ b/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 = { From 61643122deebc3f7884beb21aa8cae962447ebf1 Mon Sep 17 00:00:00 2001 From: Ozan Durgut Date: Mon, 13 Jul 2026 14:20:52 +0200 Subject: [PATCH 3/4] client: use digital input protocol for io get Resolve labgrid-client io get via DigitalInputProtocol when a matching driver is configured. This allows input-only drivers to use the existing io command. Keep the existing NetworkSysfsGPIO fallback on GpioDigitalOutputDriver so resource-only configurations continue to behave as before. Co-developed-by: Felix Zwettler Signed-off-by: Felix Zwettler Signed-off-by: Ozan Durgut --- labgrid/remote/client.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 069240b9e2..daed578a3a 100755 --- a/labgrid/remote/client.py +++ b/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: From 2727a7203963f3df1ea384818e6ef0dff9352c72 Mon Sep 17 00:00:00 2001 From: Ozan Durgut Date: Mon, 13 Jul 2026 14:21:10 +0200 Subject: [PATCH 4/4] tests: cover GPIO digital input support Document GpioDigitalInputDriver and add coverage for sysfsgpio input direction handling, cached direction reconfiguration, input driver reads, protocol compatibility, and labgrid-client io get routing. Co-developed-by: Felix Zwettler Signed-off-by: Felix Zwettler Signed-off-by: Ozan Durgut --- doc/configuration.rst | 26 ++++++++++ tests/test_gpiodriver.py | 92 ++++++++++++++++++++++++++++++++++++ tests/test_sysfsgpioagent.py | 44 ++++++++++++++++- 3 files changed, 160 insertions(+), 2 deletions(-) create mode 100644 tests/test_gpiodriver.py diff --git a/doc/configuration.rst b/doc/configuration.rst index 57b23f13c5..1a80ea8c0b 100644 --- a/doc/configuration.rst +++ b/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 `__ +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. diff --git a/tests/test_gpiodriver.py b/tests/test_gpiodriver.py new file mode 100644 index 0000000000..9dbd4411ba --- /dev/null +++ b/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() diff --git a/tests/test_sysfsgpioagent.py b/tests/test_sysfsgpioagent.py index 1f70bc198c..093980fcc3 100644 --- a/tests/test_sysfsgpioagent.py +++ b/tests/test_sysfsgpioagent.py @@ -1,6 +1,7 @@ import pytest import os +from labgrid.util.agents import sysfsgpio from labgrid.util.agents.sysfsgpio import GpioDigitalOutput from tempfile import TemporaryDirectory @@ -11,7 +12,6 @@ def __init__(self, **kwargs): index = kwargs['index'] self.sysfs_mock_directory = TemporaryDirectory() GpioDigitalOutput._gpio_sysfs_path_prefix = self.sysfs_mock_directory.name - GpioDigitalOutput._buffered_file_access = True export_file_path = os.path.join(self.sysfs_mock_directory.name, 'export') os.mknod(export_file_path) # Since there is no real device, writing to `export` does not create a corresponding @@ -23,7 +23,7 @@ def __init__(self, **kwargs): assert export_content == str(index) self.gpio_line_directory = os.path.join(self.sysfs_mock_directory.name, f'gpio{index}') os.mkdir(self.gpio_line_directory) - for control_file in ['direction', 'value']: + for control_file in ['active_low', 'direction', 'value']: control_file_path = os.path.join(self.gpio_line_directory, control_file) print(f'creating control file `{control_file_path}`') os.mknod(control_file_path) @@ -44,3 +44,43 @@ def test_set(self): for val in [True, False, True, False]: gpio_line.set(val) assert gpio_line.get() == val + + def test_output_direction(self): + gpio_line = TestGpioAgent.GpioDigitalOutputMock(index=13, invert=False) + direction_file_path = os.path.join(gpio_line.gpio_line_directory, 'direction') + with open(direction_file_path, mode='rb') as direction_file: + assert direction_file.read() == b'out' + + def test_input_direction(self): + gpio_line = TestGpioAgent.GpioDigitalOutputMock(index=13, invert=False, direction='in') + direction_file_path = os.path.join(gpio_line.gpio_line_directory, 'direction') + value_file_path = os.path.join(gpio_line.gpio_line_directory, 'value') + + with open(direction_file_path, mode='rb') as direction_file: + assert direction_file.read() == b'in' + + with open(value_file_path, mode='wb') as value_file: + value_file.write(b'1') + + assert gpio_line.get() is True + + def test_invalid_direction(self): + with pytest.raises(ValueError, match='direction'): + GpioDigitalOutput(index=13, invert=False, direction='invalid') + + def test_cached_line_reconfigures_direction(self): + with TemporaryDirectory() as sysfs_mock_directory: + GpioDigitalOutput._gpio_sysfs_path_prefix = sysfs_mock_directory + sysfsgpio._gpios.clear() + gpio_line_directory = os.path.join(sysfs_mock_directory, 'gpio13') + os.mkdir(gpio_line_directory) + for control_file in ['active_low', 'direction', 'value']: + os.mknod(os.path.join(gpio_line_directory, control_file)) + + sysfsgpio.handle_set(13, False, True) + sysfsgpio.handle_get(13, False, 'in') + sysfsgpio.handle_set(13, False, False) + + direction_file_path = os.path.join(gpio_line_directory, 'direction') + with open(direction_file_path, mode='rb') as direction_file: + assert direction_file.read() == b'out'