Merge pull request #1464 from jluebbe/lcus-hid

add support for the LCTech LCUS USB

Author: Emantor

doc/configuration.rst

@@ -480,7 +480,7 @@ HIDRelay
 ++++++++
 An :any:`HIDRelay` resource describes a single output of an HID protocol based
 USB relays.
-It currently supports the widely used *dcttech USBRelay*.
+It currently supports the widely used *dcttech USBRelay* and *lctech LCUS*
 
 .. code-block:: yaml
 

labgrid/resource/udev.py

@@ -671,10 +671,15 @@ class HIDRelay(USBResource):
     index = attr.ib(default=1, validator=attr.validators.instance_of(int))
     invert = attr.ib(default=False, validator=attr.validators.instance_of(bool))
 
-    def __attrs_post_init__(self):
-        self.match['ID_VENDOR_ID'] = '16c0'
-        self.match['ID_MODEL_ID'] = '05df'
-        super().__attrs_post_init__()
+    def filter_match(self, device):
+        match = (device.properties.get('ID_VENDOR_ID'), device.properties.get('ID_MODEL_ID'))
+
+        if match not in [("16c0", "05df"),  # dcttech USBRelay2
+                         ("5131", "2007"),  # LC-US8
+                         ]:
+            return False
+
+        return super().filter_match(device)
 
 @target_factory.reg_resource
 @attr.s(eq=False)

labgrid/util/agents/usb_hid_relay.py

@@ -10,6 +10,7 @@
 
 - Turn digital output on and off
 """
+
 import usb.core
 import usb.util
 
@@ -23,46 +24,79 @@ def __init__(self, **args):
         self._dev = usb.core.find(**args)
         if self._dev is None:
             raise ValueError("Device not found")
+
+        if self._dev.idVendor == 0x16C0:
+            self.set_output = self.set_output_dcttech
+            self.get_output = self.get_output_dcttech
+        elif self._dev.idVendor == 0x5131:
+            self.set_output = self.set_output_lcus
+            self.get_output = self.get_output_lcus
+        else:
+            raise ValueError(f"Unknown vendor/protocol for VID {self._dev.idVendor:x}")
+
         if self._dev.is_kernel_driver_active(0):
             self._dev.detach_kernel_driver(0)
 
-    def set_output(self, number, status):
+    def set_output_dcttech(self, number, status):
         assert 1 <= number <= 8
         req = [0xFF if status else 0xFD, number]
         self._dev.ctrl_transfer(
             usb.util.CTRL_TYPE_CLASS | usb.util.CTRL_RECIPIENT_DEVICE | usb.util.ENDPOINT_OUT,
             SET_REPORT,
-            (REPORT_TYPE_FEATURE << 8) | 0, # no report ID
+            (REPORT_TYPE_FEATURE << 8) | 0,  # no report ID
             0,
-            req, # payload
+            req,  # payload
         )
 
-    def get_output(self, number):
+    def get_output_dcttech(self, number):
         assert 1 <= number <= 8
         resp = self._dev.ctrl_transfer(
             usb.util.CTRL_TYPE_CLASS | usb.util.CTRL_RECIPIENT_DEVICE | usb.util.ENDPOINT_IN,
             GET_REPORT,
-            (REPORT_TYPE_FEATURE << 8) | 0, # no report ID
+            (REPORT_TYPE_FEATURE << 8) | 0,  # no report ID
             0,
-            8, # size
+            8,  # size
         )
-        return bool(resp[7] & (1 << (number-1)))
+        return bool(resp[7] & (1 << (number - 1)))
+
+    def set_output_lcus(self, number, status):
+        assert 1 <= number <= 8
+        ep_in = self._dev[0][(0, 0)][0]
+        ep_out = self._dev[0][(0, 0)][1]
+        req = [0xA0, number, 0x01 if status else 0x00, 0x00]
+        req[3] = sum(req) & 0xFF
+        ep_out.write(req)
+        ep_in.read(64)
+
+    def get_output_lcus(self, number):
+        assert 1 <= number <= 8
+        # we have no information on how to read the current value
+        return False
 
     def __del__(self):
         usb.util.release_interface(self._dev, 0)
 
 
+_relays = {}
+
+
+def _get_relay(busnum, devnum):
+    if (busnum, devnum) not in _relays:
+        _relays[(busnum, devnum)] = USBHIDRelay(bus=busnum, address=devnum)
+    return _relays[(busnum, devnum)]
+
+
 def handle_set(busnum, devnum, number, status):
-    relay = USBHIDRelay(bus=busnum, address=devnum)
+    relay = _get_relay(busnum, devnum)
     relay.set_output(number, status)
 
 
 def handle_get(busnum, devnum, number):
-    relay = USBHIDRelay(bus=busnum, address=devnum)
+    relay = _get_relay(busnum, devnum)
     return relay.get_output(number)
 
 
 methods = {
-    'set': handle_set,
-    'get': handle_get,
+    "set": handle_set,
+    "get": handle_get,
 }