Merge pull request #102 from DCC-Lab/find-generic-ports

Discover generic USB/RS-232 adaptors and stop auto-probing devices behind them

Author: dccote

hardwarelibrary/communication/serialport.py

@@ -17,7 +17,7 @@ class MoreThanOneMatch(serial.SerialException):
 class SerialPort(CommunicationPort):
     """
     An implementation of CommunicationPort using BSD-style serial port
-    
+
     Two strategies to initialize the SerialPort:
     1. with a portPath/port name (i.e. "COM1" or "/dev/cu.serial")
     2. with an instance of pyserial.Serial() that will support the same
@@ -26,6 +26,17 @@ class SerialPort(CommunicationPort):
        or the find_url.py script from the distribution. More info: https://eblot.github.io/pyftdi/api/usbtools.html
        You have to add any custom VID/PID when using tools (but they are added here in SerialPort) @line 30.
     """
+
+    # USB idVendor of the common, generic RS-232/USB serial-converter chips.
+    # A device behind one of these bridges inherits the bridge's anonymous
+    # VID/PID and carries no identity of its own, so it can only be told apart
+    # from other adaptors by its serial number. Extend as needed.
+    genericSerialConverterVendors = {
+        0x0403: "FTDI",
+        0x067b: "Prolific",
+        0x10c4: "Silicon Labs (CP210x)",
+        0x1a86: "WCH (CH340/CH341)",
+    }
     def __init__(self, idVendor=None, idProduct=None, serialNumber=None, portPath=None, port=None, delay=0):
         CommunicationPort.__init__(self)
 
@@ -138,6 +149,27 @@ def ftdiPorts(cls):
 
         return urls
 
+    @classmethod
+    def isGenericSerialConverter(cls, idVendor) -> bool:
+        """Whether idVendor belongs to a known generic USB/RS-232 converter
+        chip (FTDI, Prolific, Silicon Labs, WCH, ...). A None idVendor (a
+        non-USB port such as a Bluetooth channel) is not a converter."""
+        return idVendor in cls.genericSerialConverterVendors
+
+    @classmethod
+    def genericSerialConverterPorts(cls):
+        """Return the connected serial ports that come from a generic USB/RS-232
+        converter chip, as pyserial ListPortInfo objects.
+
+        Useful when the instrument at the other end has no USB identity of its
+        own (it sits behind an FTDI/Prolific/CP210x/CH34x cable): this lists the
+        candidate adaptors so the caller can pick one, typically by its
+        `serial_number`. Open a match with its `.device` path. The recognized
+        chips are in `genericSerialConverterVendors`; read a port's chip name
+        with `genericSerialConverterVendors[port.vid]`.
+        """
+        return [port for port in comports() if cls.isGenericSerialConverter(port.vid)]
+
     @property
     def isOpen(self):
         if self.port is None:

hardwarelibrary/devicemanager.py

@@ -207,13 +207,23 @@ def stopMonitoring(self):
         else:
             raise RuntimeError("No monitoring loop running")
 
+    @classmethod
+    def candidateClassesForAutoDiscovery(cls, idVendor, idProduct):
+        """Device classes matching a plugged-in USB device that are safe to
+        instantiate automatically. Classes behind a generic serial converter are
+        excluded: their VID/PID identifies only the cable, so several instruments
+        share it and auto-probing them would send arbitrary protocol bytes to an
+        unknown device. Those must be constructed explicitly instead."""
+        candidates = utils.getCandidateDeviceClasses(PhysicalDevice, idVendor, idProduct)
+        return [aClass for aClass in candidates if not aClass.usesGenericSerialConverter]
+
     def usbDeviceConnected(self, usbDevice):
         descriptor = USBDeviceDescriptor.fromUSBDevice(usbDevice)
         NotificationCenter().postNotification(DeviceManagerNotification.usbDeviceDidConnect, notifyingObject=self, userInfo=descriptor)
         if descriptor not in self.usbDeviceDescriptors:
             self.usbDeviceDescriptors.append(descriptor)
 
-        candidates = utils.getCandidateDeviceClasses(PhysicalDevice, descriptor.idVendor, descriptor.idProduct)
+        candidates = self.candidateClassesForAutoDiscovery(descriptor.idVendor, descriptor.idProduct)
         for candidateClass in candidates:
             # This may throw if incompat:
             #                 deviceInstanceible

hardwarelibrary/echodevice.py

@@ -14,6 +14,7 @@
 class EchoDevice(PhysicalDevice):
     classIdProduct = 0x6001
     classIdVendor = 0x0403
+    usesGenericSerialConverter = True
     commands = {
         "ECHO1": TextCommand(name="ECHO1", requestEncoder="someText", replyDecoder="someText"),
         "ECHO2": TextCommand(name="ECHO2", requestEncoder="someOtherText", replyDecoder="someOtherText"),

hardwarelibrary/motion/intellidrivedevice.py

@@ -22,6 +22,7 @@ class State(Enum):
 class IntellidriveDevice(RotationDevice):
     classIdVendor = 0x0403
     classIdProduct = 0x6001
+    usesGenericSerialConverter = True
 
     commands = {
         "SET_REGISTER": TextCommand(name="SET_REGISTER",

hardwarelibrary/oscilloscope/oscilloscopedevice.py

@@ -26,6 +26,7 @@ def __init__(self, code, msg, esr, stb, underlyingErrors=None):
 class OscilloscopeDevice(PhysicalDevice):
     classIdVendor = 0x0403
     classIdProduct = 0x6001
+    usesGenericSerialConverter = True
 
     def __init__(self, serialNumber:str = None, idProduct = 0x6001, idVendor = 0x0403):
         super().__init__(serialNumber, idProduct=self.classIdProduct, idVendor=self.classIdVendor)

hardwarelibrary/physicaldevice.py

@@ -38,6 +38,14 @@ class NotInitialized(Exception):
     classIdProduct = None
     commands = None
 
+    # True when classIdVendor/classIdProduct are those of a generic, off-the-shelf
+    # USB/RS-232 converter (a stock FTDI/Prolific/CP210x/CH34x cable) rather than
+    # the instrument's own identity. Such a device shares its VID/PID with every
+    # other instrument behind the same cable model, so it cannot be identified by
+    # VID/PID alone: it matches any generic converter (see vidpids) and must be
+    # disambiguated by serial number, and DeviceManager will not auto-probe it.
+    usesGenericSerialConverter = False
+
     def __init__(self, serialNumber:str, idProduct:int, idVendor:int):
         if serialNumber == "*" or serialNumber is None:
             serialNumber = ".*"
@@ -65,13 +73,22 @@ def __init__(self, serialNumber:str, idProduct:int, idVendor:int):
 
     @classmethod
     def vidpids(cls):
+        if cls.usesGenericSerialConverter:
+            # Identity is not in the VID/PID: match any generic converter chip,
+            # PID wildcarded (None). Local import avoids a physicaldevice <->
+            # communication import cycle.
+            from hardwarelibrary.communication.serialport import SerialPort
+            return [(idVendor, None) for idVendor in SerialPort.genericSerialConverterVendors]
         return [(cls.classIdVendor, cls.classIdProduct)]
 
     @classmethod
     def isCompatibleWith(cls, serialNumber, idProduct, idVendor):
         for compatibleIdVendor, compatibleIdProduct in cls.vidpids():
-            if idVendor == compatibleIdVendor and idProduct == compatibleIdProduct:
-                return True
+            if idVendor == compatibleIdVendor:
+                # A None product id in the pair is a wildcard: it matches any
+                # product behind that vendor (used for generic converters).
+                if compatibleIdProduct is None or idProduct == compatibleIdProduct:
+                    return True
 
         return False
 

hardwarelibrary/powermeters/fieldmasterdevice.py

@@ -43,6 +43,7 @@ class FieldMasterDevice(PowerMeterDevice):
 
     classIdVendor = 0x0403
     classIdProduct = 0x6001
+    usesGenericSerialConverter = True
 
     def __init__(self, serialNumber: str = None, idProduct: int = 0x6001,
                  idVendor: int = 0x0403, portPath: str = None,
@@ -152,6 +153,7 @@ class DebugFieldMasterDevice(FieldMasterDevice):
 
     classIdVendor = 0xFFFF
     classIdProduct = 0xFFF1
+    usesGenericSerialConverter = False   # debug device has its own fake identity
 
     def __init__(self, serialNumber='debug'):
         PhysicalDevice.__init__(self, serialNumber=serialNumber,

hardwarelibrary/tests/testGenericSerialConverter.py

@@ -0,0 +1,73 @@
+import env
+import unittest
+
+from hardwarelibrary.physicaldevice import PhysicalDevice
+from hardwarelibrary.communication.serialport import SerialPort
+from hardwarelibrary.powermeters import (
+    FieldMasterDevice, DebugFieldMasterDevice, IntegraDevice,
+)
+from hardwarelibrary.devicemanager import DeviceManager
+import hardwarelibrary.utils as utils
+
+
+class TestGenericConverterFlag(unittest.TestCase):
+    def testFieldMasterIsGeneric(self):
+        self.assertTrue(FieldMasterDevice.usesGenericSerialConverter)
+
+    def testDebugFieldMasterIsNotGeneric(self):
+        self.assertFalse(DebugFieldMasterDevice.usesGenericSerialConverter)
+
+    def testIntegraIsNotGeneric(self):
+        self.assertFalse(IntegraDevice.usesGenericSerialConverter)
+
+    def testBaseDefaultIsNotGeneric(self):
+        self.assertFalse(PhysicalDevice.usesGenericSerialConverter)
+
+
+class TestGenericConverterVidpids(unittest.TestCase):
+    def testGenericExpandsToAllConverterVendors(self):
+        expected = [(idVendor, None) for idVendor in SerialPort.genericSerialConverterVendors]
+        self.assertEqual(FieldMasterDevice.vidpids(), expected)
+
+    def testNonGenericKeepsSinglePair(self):
+        self.assertEqual(IntegraDevice.vidpids(), [(0x1ad5, 0x0300)])
+
+    def testDebugKeepsItsOwnPair(self):
+        self.assertEqual(DebugFieldMasterDevice.vidpids(), [(0xFFFF, 0xFFF1)])
+
+
+class TestGenericConverterCompatibility(unittest.TestCase):
+    def testMatchesAnyGenericConverterVendorWithAnyProduct(self):
+        # Same instrument behind FTDI, Prolific, or CP210x cables (different PIDs)
+        self.assertTrue(FieldMasterDevice.isCompatibleWith("*", 0x6001, 0x0403))
+        self.assertTrue(FieldMasterDevice.isCompatibleWith("*", 0x2303, 0x067b))
+        self.assertTrue(FieldMasterDevice.isCompatibleWith("*", 0xEA60, 0x10c4))
+
+    def testRejectsUnknownVendor(self):
+        self.assertFalse(FieldMasterDevice.isCompatibleWith("*", 0x0001, 0x9999))
+
+    def testNonGenericStillRequiresExactProduct(self):
+        self.assertTrue(IntegraDevice.isCompatibleWith("*", 0x0300, 0x1ad5))
+        self.assertFalse(IntegraDevice.isCompatibleWith("*", 0x9999, 0x1ad5))
+        # No wildcard leak: a None product must not match a concrete pair.
+        self.assertFalse(IntegraDevice.isCompatibleWith("*", None, 0x1ad5))
+
+    def testDebugFieldMasterStillConstructs(self):
+        device = DebugFieldMasterDevice()  # would raise if the flag broke its vidpids
+        self.assertEqual(device.idVendor, 0xFFFF)
+        self.assertEqual(device.idProduct, 0xFFF1)
+
+
+class TestAutoDiscoveryGuard(unittest.TestCase):
+    def testGenericClassesMatchButAreExcludedFromAutoDiscovery(self):
+        candidates = utils.getCandidateDeviceClasses(PhysicalDevice, 0x0403, 0x6001)
+        self.assertIn(FieldMasterDevice, candidates)  # it does match the FTDI cable
+
+        discoverable = DeviceManager.candidateClassesForAutoDiscovery(0x0403, 0x6001)
+        self.assertNotIn(FieldMasterDevice, discoverable)  # ...but is not auto-probed
+        for aClass in discoverable:
+            self.assertFalse(aClass.usesGenericSerialConverter)
+
+
+if __name__ == "__main__":
+    unittest.main()

hardwarelibrary/tests/testSerialPortConverters.py

@@ -0,0 +1,51 @@
+import env
+import unittest
+from unittest.mock import patch
+
+from hardwarelibrary.communication.serialport import SerialPort
+
+
+class FakePort:
+    """Minimal stand-in for pyserial's ListPortInfo."""
+
+    def __init__(self, device, vid, pid=0x0001, serial_number="SN"):
+        self.device = device
+        self.vid = vid
+        self.pid = pid
+        self.serial_number = serial_number
+
+
+class TestIsGenericSerialConverter(unittest.TestCase):
+    def testKnownVendorsAreConverters(self):
+        for idVendor in (0x0403, 0x067b, 0x10c4, 0x1a86):
+            self.assertTrue(SerialPort.isGenericSerialConverter(idVendor))
+
+    def testUnknownVendorIsNotConverter(self):
+        self.assertFalse(SerialPort.isGenericSerialConverter(0x0483))  # STM32 CDC
+
+    def testNoneVendorIsNotConverter(self):
+        self.assertFalse(SerialPort.isGenericSerialConverter(None))
+
+
+class TestGenericSerialConverterPorts(unittest.TestCase):
+    def testFiltersToConvertersOnly(self):
+        fakePorts = [
+            FakePort("/dev/cu.usbserial-FTDI", 0x0403, 0x6001, "FTFDLOTS"),
+            FakePort("/dev/cu.usbserial-CP210", 0x10c4, 0xea60, "CP01"),
+            FakePort("/dev/cu.usbmodem-STM32", 0x0483, 0x5740, "STM"),
+            FakePort("/dev/cu.Bluetooth", None, None, None),
+        ]
+        with patch("hardwarelibrary.communication.serialport.comports", return_value=fakePorts):
+            found = SerialPort.genericSerialConverterPorts()
+
+        devices = [port.device for port in found]
+        self.assertEqual(devices, ["/dev/cu.usbserial-FTDI", "/dev/cu.usbserial-CP210"])
+
+    def testReturnsEmptyWhenNoConverters(self):
+        fakePorts = [FakePort("/dev/cu.usbmodem-STM32", 0x0483)]
+        with patch("hardwarelibrary.communication.serialport.comports", return_value=fakePorts):
+            self.assertEqual(SerialPort.genericSerialConverterPorts(), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

hardwarelibrary/utils.py

@@ -77,7 +77,12 @@ def connectedUSBDevices(vidpids = None, serialNumberPattern=None):
     devices = []
     if vidpids is not None:
         for (idVendor, idProduct) in vidpids:
-            devices.extend(list(usb.core.find(find_all=True, idVendor=idVendor, idProduct=idProduct)))
+            if idProduct is None:
+                # A None product id means "any product from this vendor". pyusb's
+                # find matches on equality, so idProduct must be omitted, not None.
+                devices.extend(list(usb.core.find(find_all=True, idVendor=idVendor)))
+            else:
+                devices.extend(list(usb.core.find(find_all=True, idVendor=idVendor, idProduct=idProduct)))
     else:
         devices.extend(list(usb.core.find(find_all=True)))