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/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/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): 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: 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 = { 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'