sdwire3: initial implementation

doc/configuration.rst

@@ -1103,6 +1103,29 @@ NetworkUSBSDWireDevice
 A :any:`NetworkUSBSDWireDevice` resource describes a `USBSDWireDevice`_
 resource available on a remote computer.
 
+USBSDWire3Device
+~~~~~~~~~~~~~~~~
+A :any:`USBSDWire3Device` resource describes a Tizen
+`SD Wire3 device <https://badgerd.nl/sdwire3/>`_.
+
+.. code-block:: yaml
+
+   USBSDWire3Device:
+     match:
+       '@ID_PATH': 'pci-0000:00:14.0-usb-0:1.2'
+
+Arguments:
+  - match (dict): key and value pairs for a udev match, see `udev Matching`_
+
+Used by:
+  - `USBSDWire3Driver`_
+  - `USBStorageDriver`_
+
+NetworkUSBSDWire3Device
+~~~~~~~~~~~~~~~~~~~~~~~
+A :any:`NetworkUSBSDWire3Device` resource describes a `USBSDWire3Device`_
+resource available on a remote computer.
+
 USBVideo
 ~~~~~~~~
 A :any:`USBVideo` resource describes a USB video camera which is supported by a
@@ -3009,6 +3032,23 @@ Arguments:
 The driver can be used in test cases by calling its ``set_mode()`` method with
 argument being "dut", "host", "off", or "client".
 
+USBSDWire3Driver
+~~~~~~~~~~~~~~~~
+The :any:`USBSDWire3Driver` uses a `USBSDWire3Device`_ resource to control a
+USB-SD-Wire3 device via `sdwire <https://github.com/Badger-Embedded/sdwire-cli>`_
+tool.
+
+Binds to:
+  mux:
+    - `USBSDWire3Device`_
+    - `NetworkUSBSDWire3Device`_
+
+Implements:
+  - None yet
+
+Arguments:
+  - None
+
 USBVideoDriver
 ~~~~~~~~~~~~~~
 The :any:`USBVideoDriver` is used to show a video stream from a remote USB

labgrid/driver/__init__.py

@@ -19,6 +19,7 @@
 from .usbloader import MXSUSBDriver, IMXUSBDriver, BDIMXUSBDriver, RKUSBDriver, UUUDriver
 from .usbsdmuxdriver import USBSDMuxDriver
 from .usbsdwiredriver import USBSDWireDriver
+from .usbsdwire3driver import USBSDWire3Driver
 from .common import Driver
 from .qemudriver import QEMUDriver
 from .modbusdriver import ModbusCoilDriver, WaveShareModbusCoilDriver

labgrid/driver/usbsdwire3driver.py

@@ -0,0 +1,77 @@
+import subprocess
+import re
+
+import attr
+
+from .common import Driver
+from ..factory import target_factory
+from ..step import step
+from .exception import ExecutionError
+from ..util.helper import processwrapper
+
+@target_factory.reg_driver
+@attr.s(eq=False)
+class USBSDWire3Driver(Driver):
+    """The USBSDWire3Driver uses the sdwire tool to control SDWire hardware
+
+    Args:
+        bindings (dict): driver to use with usbsdmux
+    """
+    bindings = {
+        "mux": {"USBSDWire3Device", "NetworkUSBSDWire3Device"},
+    }
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+        if self.target.env:
+            self.tool = self.target.env.config.get_tool('sdwire')
+        else:
+            self.tool = 'sdwire'
+        if self.mux.control_serial is None:
+            raise ExecutionError("USBSDWire3Driver requires 'control_serial' to be set in the resource")
+        self.control_serial = self.match_control_serial()
+
+    @Driver.check_active
+    @step(title='sdmux_set', args=['mode'])
+    def set_mode(self, mode):
+        if not mode.lower() in ['dut', 'host']:
+            raise ExecutionError(f"Setting mode '{mode}' not supported by USBSDWire3Driver")
+        cmd = self.mux.command_prefix + [
+            self.tool,
+            "switch",
+            "-s",
+            self.control_serial,
+            "dut" if mode.lower() == "dut" else "ts",
+        ]
+        processwrapper.check_output(cmd)
+
+    def match_control_serial(self):
+        cmd = self.mux.command_prefix + [
+            self.tool,
+            "list"
+        ]
+        proc = subprocess.run(
+            cmd,
+            stdout=subprocess.PIPE,
+            check=True
+        )
+        output = proc.stdout.strip().decode()
+        for line in output.splitlines():
+            if self.mux.control_serial is not None and line.find(self.mux.control_serial) >= 0:
+                return line.split()[0]
+        raise ExecutionError(f"Could not find control serial {self.mux.control_serial} in sdwire list output")
+
+    @Driver.check_active
+    @step(title='sdmux_get')
+    def get_mode(self):
+        cmd = self.mux.command_prefix + [
+            self.tool,
+            "state",
+            "-s",
+            self.control_serial,
+        ]
+        result = processwrapper.check_output(cmd)
+        for line in result.decode().splitlines():
+            if re.match(self.control_serial, line):
+                return line.split(" ", maxsplit=1)[1].strip()
+        raise ExecutionError(f"Could not find control serial {self.mux.control_serial} in sdwire list output")

labgrid/driver/usbstoragedriver.py

@@ -37,6 +37,8 @@ class USBStorageDriver(Driver):
             "NetworkUSBSDMuxDevice",
             "USBSDWireDevice",
             "NetworkUSBSDWireDevice",
+            "USBSDWire3Device",
+            "NetworkUSBSDWire3Device",
         },
     }
     image = attr.ib(

labgrid/remote/client.py

@@ -1206,7 +1206,7 @@ def sd_mux(self):
         action = self.args.action
         target = self._get_target(place)
         name = self.args.name
-        from ..resource.remote import NetworkUSBSDMuxDevice, NetworkUSBSDWireDevice
+        from ..resource.remote import NetworkUSBSDMuxDevice, NetworkUSBSDWireDevice, NetworkUSBSDWire3Device
 
         drv = None
         for resource in target.resources:
@@ -1216,6 +1216,8 @@ def sd_mux(self):
                 drv = self._get_driver_or_new(target, "USBSDMuxDriver", name=name)
             elif isinstance(resource, NetworkUSBSDWireDevice):
                 drv = self._get_driver_or_new(target, "USBSDWireDriver", name=name)
+            elif isinstance(resource, NetworkUSBSDWire3Device):
+                drv = self._get_driver_or_new(target, "USBSDWire3Driver", name=name)
             if drv:
                 break
 

labgrid/remote/exporter.py

@@ -409,6 +409,25 @@ def _get_params(self):
 class USBSDWireExport(USBGenericExport):
     """ResourceExport for USB devices accessed directly from userspace"""
 
+    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,
+        }
+
+
+@attr.s(eq=False)
+class USBSDWire3Export(USBGenericExport):
+    """ResourceExport for USB devices accessed directly from userspace"""
+
     def __attrs_post_init__(self):
         super().__attrs_post_init__()
 
@@ -557,6 +576,7 @@ def __attrs_post_init__(self):
 exports["SigrokUSBSerialDevice"] = USBSigrokExport
 exports["USBSDMuxDevice"] = USBSDMuxExport
 exports["USBSDWireDevice"] = USBSDWireExport
+exports["USBSDWire3Device"] = USBSDWire3Export
 exports["USBDebugger"] = USBGenericExport
 exports["USBHub"] = USBGenericRemoteExport
 exports["USBMassStorage"] = USBGenericExport

labgrid/resource/__init__.py

@@ -29,6 +29,7 @@
     USBPowerPort,
     USBSDMuxDevice,
     USBSDWireDevice,
+    USBSDWire3Device,
     USBSerialPort,
     USBTMC,
     USBVideo,

labgrid/resource/remote.py

@@ -250,6 +250,18 @@ def __attrs_post_init__(self):
         self.timeout = 10.0
         super().__attrs_post_init__()
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class NetworkUSBSDWire3Device(RemoteUSBResource):
+    """The NetworkUSBSDWire3Device describes a remotely accessible USBSDWire3 device"""
+    control_serial = attr.ib(
+        default=None,
+        validator=attr.validators.optional(attr.validators.instance_of(str))
+    )
+    def __attrs_post_init__(self):
+        self.timeout = 10.0
+        super().__attrs_post_init__()
+
 @target_factory.reg_resource
 @attr.s(eq=False)
 class NetworkSiSPMPowerPort(RemoteUSBResource):

labgrid/resource/suggest.py

@@ -14,6 +14,7 @@
     DFUDevice,
     USBSDMuxDevice,
     USBSDWireDevice,
+    USBSDWire3Device,
     AlteraUSBBlaster,
     RKUSBLoader,
     USBNetworkInterface,
@@ -48,6 +49,7 @@ def __init__(self, args):
         self.resources.append(USBMassStorage(**args))
         self.resources.append(USBSDMuxDevice(**args))
         self.resources.append(USBSDWireDevice(**args))
+        self.resources.append(USBSDWire3Device(**args))
         self.resources.append(AlteraUSBBlaster(**args))
         self.resources.append(RKUSBLoader(**args))
         self.resources.append(USBNetworkInterface(**args))

labgrid/resource/udev.py

@@ -526,6 +526,59 @@ def update(self):
     def path(self):
         return self.disk_path
 
+@target_factory.reg_resource
+@attr.s(eq=False)
+class USBSDWire3Device(USBResource):
+    """The USBSDWire3Device describes an attached SDWire3 device,
+    it is identified via USB using udev
+    """
+
+    control_serial = attr.ib(
+        default=None,
+        validator=attr.validators.optional(str)
+    )
+    disk_path = attr.ib(
+        default=None,
+        validator=attr.validators.optional(str)
+    )
+
+    def __attrs_post_init__(self):
+        self.match['ID_VENDOR_ID'] = '0bda'
+        self.match['ID_MODEL_ID'] = '0316'
+        self.match['@ID_VENDOR_ID'] = '1d6b'
+        self.match['@ID_MODEL_ID'] = '0002'        
+        super().__attrs_post_init__()
+
+    # Overwrite the avail attribute with our internal property
+    @property
+    def avail(self):
+        return bool(self.control_serial)
+
+    # Forbid the USBResource super class to set the avail property
+    @avail.setter
+    def avail(self, prop):
+        pass
+
+    # Overwrite the poll function. Only mark the SDWire3 as available if both
+    # paths are available.
+    def poll(self):
+        super().poll()
+        if self.device is not None and not self.avail:
+            for child in self.device.parent.children:
+                if child.subsystem == 'block' and child.device_type == 'disk':
+                    self.disk_path = child.device_node
+            self.control_serial = self.device.properties.get('ID_SERIAL_SHORT')
+
+    def update(self):
+        super().update()
+        if self.device is None:
+            self.disk_path = None
+            self.control_serial = None
+
+    @property
+    def path(self):
+        return self.disk_path
+
 @target_factory.reg_resource
 @attr.s(eq=False)
 class USBSDMuxDevice(USBResource):