driver: Add TenmaControl power driver and integration support

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

- Implementation of TenmaControl in `powerdriver.py`
- Exporter and remote client support for the new driver
- Configuration documentation in `configuration.rst`
- New unit tests added in `test_tenmacontrol.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

@@ -410,6 +410,27 @@ Arguments:
 Used by:
   - `TasmotaPowerDriver`_
 
+TenmaControlPort
+++++++++++++++++
+A :any:`TenmaControlPort` describes a *Tenma* as supported by
+`tenca-control <https://multicomp-pro.com/>`_.
+
+.. code-block:: yaml
+
+     TenmaControlPort:
+        match:
+          ID_PATH: pci-0000:00:15.0-usb-0:3.2:1.0
+
+
+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:
+  - `TenmaControlDriver`_
+
 Digital Outputs
 ~~~~~~~~~~~~~~~
 

labgrid/driver/__init__.py

@@ -15,7 +15,7 @@
 from .powerdriver import ManualPowerDriver, ExternalPowerDriver, \
                          DigitalOutputPowerDriver, YKUSHPowerDriver, \
                          USBPowerDriver, SiSPMPowerDriver, NetworkPowerDriver, \
-                         PDUDaemonDriver
+                         PDUDaemonDriver, TenmaControlDriver
 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
@@ -109,6 +110,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 TenmaControlDriver(Driver, PowerResetMixin, PowerProtocol):
+    """TenmaControlDriver - Driver using a Single Output Programmable to control a
+       target's power using the tenma-control tool https://github.com/kxtells/tenma-serial/"""
+
+    bindings = {"port": {"TenmaControlPort", "NetworkTenmaControlPort"}, }
+    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_tenmacontrol_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_tenmacontrol_prefix() + cmd)
+
+    @Driver.check_active
+    @step()
+    def off(self):
+        cmd = ['--off']
+        processwrapper.check_output(self._get_tenmacontrol_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_tenmacontrol_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

@@ -883,7 +883,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,
+            NetworkTenmaControlPort,
+        )
         from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort
 
         drv = None
@@ -899,6 +904,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, NetworkTenmaControlPort):
+                    drv = self._get_driver_or_new(target, "TenmaControlDriver", 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 TenmaControlExport(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["TenmaControlPort"] = TenmaControlExport
+
+
 @attr.s(eq=False)
 class SerialPortExport(ResourceExport):
     """ResourceExport for a USB or Raw SerialPort"""
@@ -564,6 +585,7 @@ def __attrs_post_init__(self):
 exports["USBAudioInput"] = USBAudioInputExport
 exports["USBTMC"] = USBGenericExport
 exports["SiSPMPowerPort"] = SiSPMPowerPortExport
+exports["TenmaControlPort"] = TenmaControlExport
 exports["USBPowerPort"] = USBPowerPortExport
 exports["DeditecRelais8"] = USBDeditecRelaisExport
 exports["HIDRelay"] = USBHIDRelayExport

labgrid/resource/__init__.py

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

labgrid/resource/remote.py

@@ -260,6 +260,16 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
 
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class NetworkTenmaControlPort(RemoteUSBResource):
+    """The TenmaControlPort 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,
+    TenmaControlPort,
     USBSDMuxDevice,
     USBSDWireDevice,
     AlteraUSBBlaster,
@@ -45,6 +46,7 @@ def __init__(self, args):
         self.resources.append(IMXUSBLoader(**args))
         self.resources.append(AndroidUSBFastboot(**args))
         self.resources.append(DFUDevice(**args))
+        self.resources.append(TenmaControlPort(**args))
         self.resources.append(USBMassStorage(**args))
         self.resources.append(USBSDMuxDevice(**args))
         self.resources.append(USBSDWireDevice(**args))

labgrid/resource/udev.py

@@ -706,6 +706,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 TenmaControlPort(USBResource):
+    """This resource describes a tenma-control 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

@@ -156,6 +156,10 @@ See:  <https://github.com/openssh/openssh\-portable>
 Path to the sshfs binary, used by the SSHDriver.
 See:  <https://github.com/libfuse/sshfs> 
 .TP
+.B \fBtenma\-control\fP
+Path to the tenma\-control binary, used by the TenmaControlDriver.
+See:  <https://github.com/kxtells/tenma\-serial> 
+.TP
 .B \fBuhubctl\fP
 Path to the uhubctl binary, used by the USBPowerDriver.
 See:  <https://github.com/mvp/uhubctl>