driver: Add TenmaSerial power driver and integration support

This patch introduces a new power driver, TenmaSerial, providing
control support for Tenma power devices. Integration includes:

- Implementation of TenmaSerial in powerdriver.py
- Exporter and remote client support for the new driver
- Configuration documentation in configuration.rst
- New unit tests added in test_tenmaserial.py
- Minor updates to pyproject.toml and man pages

This initial draft enables basic functionality and lays the
groundwork for full Tenma device intration

Signed-off-by: Nico Orlando <nico.orlando@dave.eu>

Author: Nico Orlando

doc/configuration.rst

@@ -420,6 +420,28 @@ Arguments:
 Used by:
   - `TasmotaPowerDriver`_
 
+TenmaSerialPort
+++++++++++++++++
+A :any:`TenmaSerialPort` describes a *Tenma* as supported by
+`tenma-serial <https://github.com/kxtells/tenma-serial>`_.
+
+.. code-block:: yaml
+
+     TenmaSerialPort:
+        match:
+          ID_PATH: pci-0000:00:15.0-usb-0:3.2:1.0
+        index: 1
+
+
+The example describes port 1 on the hub with the ID_PATH
+``pci-0000:00:15.0-usb-0:3.2:1.0``.
+
+Arguments:
+  - match (dict): key and value pairs for a udev match, see `udev Matching`_
+
+Used by:
+  - `TenmaSerialDriver`_
+
 Digital Outputs
 ~~~~~~~~~~~~~~~
 
@@ -2336,6 +2358,36 @@ Implements:
 Arguments:
   - delay (float, default=2.0): delay in seconds between off and on
 
+TenmaSerialDriver
+~~~~~~~~~~~~~~~~~~
+A :any:`TenmaSerialDriver` controls a `TenmaSerialPort`_, allowing control of the
+target power state without user interaction.
+
+Binds to:
+  port:
+    - `TenmaSerialPort`_
+    - NetworkTenmaSerialPort
+
+Implements:
+  - :any:`PowerProtocol`
+  - :any:`ResetProtocol`
+
+.. code-block:: yaml
+
+   TenmaSerialDriver:
+     delay: 10.0
+     ovp: true
+     ocp: true
+     voltage: 12000
+     current: 2000
+
+Arguments:
+  - delay (float, default=2.0): delay in seconds between off and on
+  - ovp (bool, default=False): overvoltage protection
+  - ocp (bool, default=False): overcurrent protection
+  - voltage (int, default=12000): set mV
+  - current (int, default=2000): set mA
+
 TasmotaPowerDriver
 ~~~~~~~~~~~~~~~~~~
 A :any:`TasmotaPowerDriver` controls a `TasmotaPowerPort`_, allowing the outlet

labgrid/driver/__init__.py

@@ -15,7 +15,7 @@
 from .powerdriver import ManualPowerDriver, ExternalPowerDriver, \
                          DigitalOutputPowerDriver, YKUSHPowerDriver, \
                          USBPowerDriver, SiSPMPowerDriver, NetworkPowerDriver, \
-                         PDUDaemonDriver
+                         PDUDaemonDriver, TenmaSerialDriver
 from .usbloader import MXSUSBDriver, IMXUSBDriver, BDIMXUSBDriver, RKUSBDriver, UUUDriver
 from .usbsdmuxdriver import USBSDMuxDriver
 from .usbsdwiredriver import USBSDWireDriver

labgrid/driver/powerdriver.py

@@ -1,6 +1,7 @@
 import shlex
 import time
 import math
+import ast
 from importlib import import_module
 
 import attr
@@ -110,6 +111,81 @@ def get(self):
             return False
         raise ExecutionError(f"Did not find port status in sispmctl output ({repr(output)})")
 
+@target_factory.reg_driver
+@attr.s(eq=False)
+class TenmaSerialDriver(Driver, PowerResetMixin, PowerProtocol):
+    """TenmaSerialDriver - Driver using a Single Output Programmable to control a
+       target's power using the tenma-serial tool https://github.com/kxtells/tenma-serial/"""
+
+    bindings = {"port": {"TenmaSerialPort", "NetworkTenmaSerialPort"}, }
+    delay = attr.ib(default=2.0, validator=attr.validators.instance_of(float))
+    ovp = attr.ib(default=False, validator=attr.validators.instance_of(bool))
+    ocp = attr.ib(default=False, validator=attr.validators.instance_of(bool))
+    voltage = attr.ib(default=12000, validator=attr.validators.instance_of(int))
+    current = attr.ib(default=2000, validator=attr.validators.instance_of(int))
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        if self.target.env:
+            self.tool = self.target.env.config.get_tool('tenma-control')
+        else:
+            self.tool = 'tenma-control'
+
+    def _get_tenmaserial_prefix(self):
+        options = []
+
+        # overvoltage protection (bool)
+        if self.ovp:
+            options.append('--ovp-enable')
+        else:
+            options.append('--ovp-disable')
+
+        # overcurrent protection (bool)
+        if self.ocp:
+            options.append('--ocp-enable')
+        else:
+            options.append('--ocp-disable')
+
+        # set mV (int)
+        options.append(f'-v {self.voltage}')
+
+        # set mA (int)
+        options.append(f'-c {self.current}')
+
+        return self.port.command_prefix + [
+            self.tool,
+            str(self.port.path),
+        ] + options
+
+    @Driver.check_active
+    @step()
+    def on(self):
+        cmd = ['--on']
+        processwrapper.check_output(self._get_tenmaserial_prefix() + cmd)
+
+    @Driver.check_active
+    @step()
+    def off(self):
+        cmd = ['--off']
+        processwrapper.check_output(self._get_tenmaserial_prefix() + cmd)
+
+    @Driver.check_active
+    @step()
+    def cycle(self):
+        self.off()
+        time.sleep(self.delay)
+        self.on()
+
+    @Driver.check_active
+    @step()
+    def get(self):
+        cmd = ['-S']
+        output = processwrapper.check_output(self._get_tenmaserial_prefix() + cmd)
+        status = ast.literal_eval(output.decode('utf-8').strip().splitlines()[1])
+        if status['outEnabled']:
+            return True
+        else:
+            return False
 
 @target_factory.reg_driver
 @attr.s(eq=False)

labgrid/remote/client.py

@@ -938,7 +938,12 @@ def power(self):
         name = self.args.name
         target = self._get_target(place)
         from ..resource.power import NetworkPowerPort, PDUDaemonPort
-        from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO
+        from ..resource.remote import (
+            NetworkUSBPowerPort,
+            NetworkSiSPMPowerPort,
+            NetworkSysfsGPIO,
+            NetworkTenmaSerialPort,
+        )
         from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort
 
         drv = None
@@ -954,6 +959,8 @@ def power(self):
                     drv = self._get_driver_or_new(target, "USBPowerDriver", name=name)
                 elif isinstance(resource, NetworkSiSPMPowerPort):
                     drv = self._get_driver_or_new(target, "SiSPMPowerDriver", name=name)
+                elif isinstance(resource, NetworkTenmaSerialPort):
+                    drv = self._get_driver_or_new(target, "TenmaSerialDriver", name=name)
                 elif isinstance(resource, PDUDaemonPort):
                     drv = self._get_driver_or_new(target, "PDUDaemonDriver", name=name)
                 elif isinstance(resource, TasmotaPowerPort):

labgrid/remote/exporter.py

@@ -184,6 +184,27 @@ def release(self, *args, **kwargs):
         self.poll()
 
 
+@attr.s(eq=False)
+class TenmaSerialExport(ResourceExport):
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+
+    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,
+        }
+
+
+exports["TenmaSerialPort"] = TenmaSerialExport
+
+
 @attr.s(eq=False)
 class SerialPortExport(ResourceExport):
     """ResourceExport for a USB or Raw SerialPort"""
@@ -586,6 +607,7 @@ def __attrs_post_init__(self):
 exports["USBAudioInput"] = USBAudioInputExport
 exports["USBTMC"] = USBGenericExport
 exports["SiSPMPowerPort"] = SiSPMPowerPortExport
+exports["TenmaSerialPort"] = TenmaSerialExport
 exports["USBPowerPort"] = USBPowerPortExport
 exports["DeditecRelais8"] = USBDeditecRelaisExport
 exports["HIDRelay"] = USBHIDRelayExport

labgrid/resource/__init__.py

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

labgrid/resource/remote.py

@@ -272,6 +272,16 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
 
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class NetworkTenmaSerialPort(RemoteUSBResource):
+    """The TenmaSerialPort describes a remotely accessible tenma-contro power port"""
+    index = attr.ib(default=None, validator=attr.validators.instance_of(int))
+    def __attrs_post_init__(self):
+        self.timeout = 10.0
+        super().__attrs_post_init__()
+
+
 @target_factory.reg_resource
 @attr.s(eq=False)
 class NetworkUSBPowerPort(RemoteUSBResource):

labgrid/resource/suggest.py

@@ -12,6 +12,7 @@
     IMXUSBLoader,
     AndroidUSBFastboot,
     DFUDevice,
+    TenmaSerialPort,
     USBSDMuxDevice,
     USBSDWireDevice,
     USBSDWire3Device,
@@ -46,6 +47,7 @@ def __init__(self, args):
         self.resources.append(IMXUSBLoader(**args))
         self.resources.append(AndroidUSBFastboot(**args))
         self.resources.append(DFUDevice(**args))
+        self.resources.append(TenmaSerialPort(**args))
         self.resources.append(USBMassStorage(**args))
         self.resources.append(USBSDMuxDevice(**args))
         self.resources.append(USBSDWireDevice(**args))

labgrid/resource/udev.py

@@ -755,6 +755,21 @@ def __attrs_post_init__(self):
         self.match['ID_MODEL'] = 'DEDITEC_USB-OPT_REL-8'
         super().__attrs_post_init__()
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class TenmaSerialPort(USBResource):
+    """This resource describes a tenma-serial power port"""
+
+    def __attrs_post_init__(self):
+        self.match['SUBSYSTEM'] = 'tty'
+        super().__attrs_post_init__()
+
+    @property
+    def path(self):
+        if self.device is not None:
+            return self.device.device_node
+
+        return None
 
 @target_factory.reg_resource
 @attr.s(eq=False)

man/labgrid-device-config.5

@@ -157,6 +157,10 @@ See: \X'tty: link https://github.com/openssh/openssh-portable'\fI\%https://githu
 Path to the sshfs binary, used by the SSHDriver.
 See: \X'tty: link https://github.com/libfuse/sshfs'\fI\%https://github.com/libfuse/sshfs\fP\X'tty: link'
 .TP
+.B \fBtenma\-serial\fP
+Path to the tenma\-control binary, used by the TenmaSerialDriver.
+See:  <https://github.com/kxtells/tenma\-serial> 
+.TP
 .B \fBuhubctl\fP
 Path to the uhubctl binary, used by the USBPowerDriver.
 See: \X'tty: link https://github.com/mvp/uhubctl'\fI\%https://github.com/mvp/uhubctl\fP\X'tty: link'