Merge pull request #1821 from Fomys/master

drivers: Add eth008 digital output

Author: Emantor

doc/configuration.rst

@@ -606,6 +606,28 @@ Arguments:
 Used by:
   - `HttpDigitalOutputDriver`_
 
+Eth008DigitalOutput
+++++++++++++++++++++
+An :any:`Eth008DigitalOutput` resource describes a digital output on a
+*Robot-Electronics eth008* relay board.
+
+.. code-block:: yaml
+
+   Eth008DigitalOutput:
+     host: '192.168.1.100'
+     index: 1
+     invert: false
+
+The example describes relay ``1`` on the ETH008 device at ``192.168.1.100``.
+
+Arguments:
+  - host (str): hostname or IP address of the ETH008 device
+  - index (int): number of the relay to use (1-8)
+  - invert (bool, default=False): whether to invert the output (active-low)
+
+Used by:
+  - `Eth008DigitalOutputDriver`_
+
 NetworkHIDRelay
 +++++++++++++++
 A :any:`NetworkHIDRelay` describes an `HIDRelay`_ resource available on a
@@ -3380,6 +3402,25 @@ Implements:
 Arguments:
   - None
 
+Eth008DigitalOutputDriver
+~~~~~~~~~~~~~~~~~~~~~~~~~
+A :any:`Eth008DigitalOutputDriver` binds to an `Eth008DigitalOutput`_ to set
+and get a digital output state on a Robot-Electronics eth008 relay board.
+
+Binds to:
+  output:
+    - `Eth008DigitalOutput`_
+
+.. code-block:: yaml
+
+   Eth008DigitalOutputDriver: {}
+
+Implements:
+  - :any:`DigitalOutputProtocol`
+
+Arguments:
+  - None
+
 PyVISADriver
 ~~~~~~~~~~~~
 The :any:`PyVISADriver` uses a `PyVISADevice`_ resource to control test

labgrid/driver/__init__.py

@@ -49,3 +49,4 @@
 from .deditecrelaisdriver import DeditecRelaisDriver
 from .dediprogflashdriver import DediprogFlashDriver
 from .httpdigitaloutput import HttpDigitalOutputDriver
+from .eth008digitaloutput import Eth008DigitalOutputDriver

labgrid/driver/eth008digitaloutput.py

@@ -0,0 +1,91 @@
+"""
+This driver implements a digital output driver for the robot electronics 8 relay
+outputs board (ETH008).
+
+Driver has been tested with:
+* ETH008 - 8 relay outputs
+"""
+
+import socket
+import attr
+
+from ..factory import target_factory
+from ..protocol import DigitalOutputProtocol
+from ..step import step
+from ..util.proxy import proxymanager
+from .common import Driver
+from .exception import ExecutionError
+
+PORT = 17494  # TCP port for ETH008 (0x4456)
+
+
+@target_factory.reg_driver
+@attr.s(eq=False)
+class Eth008DigitalOutputDriver(Driver, DigitalOutputProtocol):
+    """Eth008DigitalOutputDriver - Driver to control individual relays on ETH008 as digital outputs"""
+    bindings = {"output": {"Eth008DigitalOutput"}, }
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        self._host = None
+        self._port = None
+
+    def on_activate(self):
+        self._host, self._port = proxymanager.get_host_and_port(
+            self.output, force_port=PORT
+        )
+
+    def _send_tcp_command(self, command_bytes):
+        """Send a command over TCP and return the response."""
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.settimeout(5.0)
+            s.connect((self._host, self._port))
+            s.sendall(command_bytes)
+            response = s.recv(1)
+            return response
+
+    @Driver.check_active
+    @step(args=["status"])
+    def set(self, status):
+        index = int(self.output.index)
+        assert 1 <= index <= 8
+
+        if self.output.invert:
+            status = not status
+
+        # Use TCP command: 0x20 for active (on), 0x21 for inactive (off)
+        command = 0x20 if status else 0x21
+        # Permanent mode (time = 0)
+        command_bytes = bytes([command, index, 0])
+
+        response = self._send_tcp_command(command_bytes)
+
+        # Check response: 0 for success, 1 for failure
+        if response == b'\x01':
+            raise ExecutionError(f"failed to set port {index} to status {status}")
+        elif response != b'\x00':
+            raise ExecutionError(f"unexpected response from device: {response}")
+
+    @Driver.check_active
+    @step(result=["True"])
+    def get(self):
+        index = int(self.output.index)
+        assert 1 <= index <= 8
+
+        # Use TCP command: 0x24 to get all relay states
+        # The response is 1 byte where each bit represents a relay state
+        command_bytes = bytes([0x24])
+
+        response = self._send_tcp_command(command_bytes)
+
+        # Parse the response byte
+        # Each bit represents a relay: bit 0 = relay 1, bit 1 = relay 2, etc.
+        state_byte = response[0]
+        # Get the bit corresponding to the relay index (1-8)
+        relay_bit = (state_byte >> (index - 1)) & 0x01
+        state = bool(relay_bit)
+
+        if self.output.invert:
+            state = not state
+
+        return state

labgrid/remote/client.py

@@ -979,7 +979,13 @@ def digital_io(self):
         action = self.args.action
         name = self.args.name
         target = self._get_target(place)
-        from ..resource import ModbusTCPCoil, OneWirePIO, HttpDigitalOutput, WaveshareModbusTCPCoil
+        from ..resource import (
+            ModbusTCPCoil,
+            OneWirePIO,
+            HttpDigitalOutput,
+            WaveshareModbusTCPCoil,
+            Eth008DigitalOutput,
+        )
         from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay
 
         drv = None
@@ -993,6 +999,8 @@ def digital_io(self):
                     drv = self._get_driver_or_new(target, "WaveShareModbusCoilDriver", name=name)
                 elif isinstance(resource, ModbusTCPCoil):
                     drv = self._get_driver_or_new(target, "ModbusCoilDriver", name=name)
+                elif isinstance(resource, Eth008DigitalOutput):
+                    drv = self._get_driver_or_new(target, "Eth008DigitalOutputDriver", name=name)
                 elif isinstance(resource, OneWirePIO):
                     drv = self._get_driver_or_new(target, "OneWirePIODriver", name=name)
                 elif isinstance(resource, HttpDigitalOutput):

labgrid/resource/__init__.py

@@ -48,3 +48,4 @@
 from .httpdigitalout import HttpDigitalOutput
 from .sigrok import SigrokDevice
 from .fastboot import AndroidNetFastboot
+from .eth008 import Eth008DigitalOutput

labgrid/resource/eth008.py

@@ -0,0 +1,20 @@
+import attr
+
+from ..factory import target_factory
+from .common import Resource
+
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class Eth008DigitalOutput(Resource):
+    """This resource describes a digital output on an ETH008 relay board.
+
+    Args:
+        host (str): host to connect to
+        index (str): index of the relay on the ETH008 board (1-8)
+        invert (bool): whether to invert the output (default: False)
+    """
+    host = attr.ib(validator=attr.validators.instance_of(str))
+    index = attr.ib(validator=attr.validators.instance_of(str),
+                    converter=lambda x: str(int(x)))
+    invert = attr.ib(default=False, validator=attr.validators.instance_of(bool))

tests/test_eth008digitaloutput.py

@@ -0,0 +1,179 @@
+import pytest
+import socket
+
+from labgrid.driver import Eth008DigitalOutputDriver
+from labgrid.resource import Eth008DigitalOutput
+
+
+@pytest.fixture(scope="function")
+def mock_tcp_server(mocker):
+    """Mock the ETH008 TCP server responses."""
+    state = 0x00  # Initial state: all relays OFF
+
+    # Track sent commands for verification
+    sent_commands = []
+
+    def mock_sendall(data):
+        """Mock socket.sendall."""
+        nonlocal state, sent_commands
+
+        sent_commands.append(data)
+
+        if len(data) == 1 and data[0] == 0x24:
+            return
+        elif len(data) == 3:
+            command = data[0]
+            relay_index = data[1]
+
+            if command == 0x20:  # Digital Active (ON)
+                state |= (1 << (relay_index - 1))
+            elif command == 0x21:  # Digital Inactive (OFF)
+                state &= ~(1 << (relay_index - 1))
+
+    def mock_recv(bufsize):
+        """Mock socket.recv to return appropriate responses."""
+        nonlocal state, sent_commands
+
+        last_sent = sent_commands[-1] if sent_commands else None
+
+        if last_sent and len(last_sent) == 1 and last_sent[0] == 0x24:
+            return bytes([state])
+        else:
+            return b'\x00'
+
+    mock_socket = mocker.MagicMock()
+    mock_socket.sendall.side_effect = mock_sendall
+    mock_socket.recv.side_effect = mock_recv
+    mock_socket.__enter__ = mocker.MagicMock(return_value=mock_socket)
+    mock_socket.__exit__ = mocker.MagicMock(return_value=None)
+
+    def mock_socket_factory(*args, **kwargs):
+        return mock_socket
+
+    mock_socket_class = mocker.patch("socket.socket")
+    mock_socket_class.side_effect = mock_socket_factory
+    mock_socket_class.return_value = mock_socket
+
+    return mock_socket
+
+
+def _make_eth008_driver(target, index, invert):
+    """Helper function to create an ETH008 digital output driver with resource."""
+    eth008_res = Eth008DigitalOutput(
+        target,
+        name=None,
+        host='192.168.1.100',
+        index=str(index),
+        invert=invert
+    )
+
+    driver = Eth008DigitalOutputDriver(target, name=None)
+    target.activate(driver)
+
+    return driver
+
+
+def test_eth008_instance(target, mocker):
+    """Test that Eth008DigitalOutputDriver can be instantiated and activated."""
+    mocker.patch("socket.socket")
+    driver = _make_eth008_driver(target, 1, False)
+    assert isinstance(driver, Eth008DigitalOutputDriver)
+
+
+def test_eth008_set_true(target, mock_tcp_server):
+    """Test setting ETH008 relay to True (asserted)."""
+    driver = _make_eth008_driver(target, 1, False)
+
+    driver.set(True)
+
+    # Verify the correct command was sent (0x20 = Digital Active, 0x01 = relay 1, 0x00 = permanent)
+    mock_tcp_server.sendall.assert_any_call(bytes([0x20, 0x01, 0x00]))
+
+
+def test_eth008_set_false(target, mock_tcp_server):
+    """Test setting ETH008 relay to False (de-asserted)."""
+    driver = _make_eth008_driver(target, 2, False)
+
+    driver.set(False)
+
+    # Verify the correct command was sent (0x21 = Digital Inactive, 0x02 = relay 2, 0x00 = permanent)
+    mock_tcp_server.sendall.assert_any_call(bytes([0x21, 0x02, 0x00]))
+
+
+def test_eth008_get_true(target, mock_tcp_server):
+    """Test getting ETH008 relay state when True."""
+    driver = _make_eth008_driver(target, 1, False)
+
+    # Set the relay to True first
+    driver.set(True)
+
+    # Get the state
+    result = driver.get()
+
+    assert result is True
+
+
+def test_eth008_get_false(target, mock_tcp_server):
+    """Test getting ETH008 relay state when False."""
+    driver = _make_eth008_driver(target, 2, False)
+
+    # Set the relay to False first
+    driver.set(False)
+
+    # Get the state
+    result = driver.get()
+
+    assert result is False
+
+
+def test_eth008_invert_true(target, mock_tcp_server):
+    """Test ETH008 with inversion enabled."""
+    driver = _make_eth008_driver(target, 1, True)  # Invert the logic
+
+    # When invert=True, setting True should actually set the relay to False
+    driver.set(True)
+
+    # Verify that 0x21 (Digital Inactive) was sent instead of 0x20 (Digital Active)
+    mock_tcp_server.sendall.assert_any_call(bytes([0x21, 0x01, 0x00]))
+
+
+def test_eth008_invert_get(target, mock_tcp_server):
+    """Test getting ETH008 relay state with inversion enabled."""
+    driver = _make_eth008_driver(target, 1, True)  # Invert the logic
+
+    # Set the relay to True (which will actually set it to False due to inversion)
+    driver.set(True)
+
+    # When invert=True, get() should return the opposite of the actual state
+    result = driver.get()
+
+    # The actual relay state is False, but with inversion it should return True
+    assert result is True
+
+
+def test_eth008_invalid_index(target, mocker):
+    """Test that invalid relay indices are handled correctly."""
+    mocker.patch("socket.socket")
+
+    driver = _make_eth008_driver(target, 9, False)  # Invalid index (must be 1-8)
+
+    # This should raise an AssertionError due to invalid index
+    with pytest.raises(AssertionError):
+        driver.set(True)
+
+
+def test_eth008_different_indices(target, mock_tcp_server):
+    """Test that different relay indices work correctly."""
+    # Test first, middle, and last relays
+    # Create separate targets for each test to avoid resource conflicts
+    from labgrid.target import Target
+    for index in [1, 4, 8]:
+        test_target = Target(name=f'Test-{index}', env=None)
+
+        driver = _make_eth008_driver(test_target, index, False)
+
+        # Set and get should work for all indices
+        driver.set(True)
+        result = driver.get()
+
+        assert result is True