Add optional port filter based on usb info.

Author: karliss

inkcut/device/transports/qtserialport/plugin.py

@@ -85,8 +85,14 @@ class QtSerialTransport(DeviceTransport):
 
     def open_serial_port(self, config):
         try:
+            port_name = self.config.choose_filtered_port()
+            if not port_name:
+                raise Exception("{} | Could not find suitable port".format(config.port))
+            self.config.port = port_name  # might be updated if there is a filter
+
+
             serial_port = QSerialPort()
-            serial_port.setPortName(config.port)
+            serial_port.setPortName(port_name)
             #Setting the AllDirections flag is supported on all platforms. Windows supports only this mode.
             serial_port.setBaudRate(config.baudrate, QSerialPort.Direction.AllDirections)
             serial_port.setParity(config.map_parity())

inkcut/device/transports/serialport/plugin.py

@@ -20,6 +20,7 @@
 from twisted.internet.protocol import Protocol, connectionDone
 from twisted.internet.serialport import SerialPort
 from serial.tools.list_ports import comports
+import re
 
 from inkcut.device.transports.raw.plugin import RawFdTransport, RawFdProtocol
 
@@ -39,6 +40,11 @@ class SerialConfigBase(Model):
 
     #: Serial port config
     port = Str().tag(config=True)
+
+    port_filter_name = Str().tag(config=True)
+    port_filter_vid = Int().tag(config=True)
+    port_filter_pid = Int().tag(config=True)
+
     baudrate = Int(9600).tag(config=True)
     bytesize = Enum(serial.EIGHTBITS, serial.SEVENBITS, serial.SIXBITS,
                     serial.FIVEBITS).tag(config=True)
@@ -66,6 +72,71 @@ def _default_port(self):
     def refresh(self):
         self.ports = self._default_ports()
 
+    def has_port_filter(self):
+        return self.port_filter_name != "" or self.port_filter_vid > 0 or self.port_filter_pid > 0
+
+    def port_matches(self, port: SerialPortInfo):
+        if self.port_filter_name:
+            if not re.search(self.port_filter_name, port.description):
+                return False
+        if self.port_filter_vid > 0 and port.usb_vid != self.port_filter_vid:
+            return False
+        if self.port_filter_pid > 0 and port.usb_pid != self.port_filter_pid:
+            return False
+        return True
+
+    def clear_filter(self):
+        self.port_filter_name = ""
+        self.port_filter_vid = 0
+        self.port_filter_pid = 0
+
+    def make_filter(self, port_info: SerialPortInfo):
+        self.clear_filter()
+        if not port_info:
+            return
+        if port_info.usb_vid > 0:
+            # usb devices should have both vid and pid
+            self.port_filter_vid = port_info.usb_vid
+            self.port_filter_pid = port_info.usb_pid
+
+            # For now assume that only usb devices will have a meaningful description
+            #
+            # Physical serial ports can't know what's connected to them, but they are
+            # also more likely to have stable device path so there is less need for filter.
+            if port_info.description:
+                text = port_info.description
+                text = text.replace(port_info.device_path, '').strip()
+                text = text.removeprefix('-').removeprefix(':').strip()
+                if text:
+                    self.port_filter_name = re.escape(text)
+
+    def port_by_path(self, device_path):
+        for port in self.ports:
+            if port.device_path == device_path:
+                return port
+        return None
+
+    def get_matching_port(self):
+        for port in self.ports:
+            if self.port_matches(port):
+                return port
+        return None
+
+    def choose_filtered_port(self):
+        if not self.has_port_filter():
+            return self.port
+
+        self.refresh()
+        port_info = self.port_by_path(self.port)
+        if port_info and self.port_matches(port_info):
+            return port_info.device_path # prefer last used port when suitable
+
+        port_info = self.get_matching_port()
+        if port_info:
+            return port_info.device_path
+
+        return None
+
 
 class SerialConfig(SerialConfigBase):
     def _default_ports(self):
@@ -92,17 +163,22 @@ class SerialTransport(RawFdTransport):
 
     def connect(self):
         config = self.config
-        self.device_path = config.port
         try:
             #: Save a reference
             self.protocol.transport = self
 
             #: Make the wrapper
             self._protocol = RawFdProtocol(self, self.protocol)
 
+            port = self.config.choose_filtered_port()
+            if not port:
+                raise Exception("{} | Could not find suitable port".format(config.port))
+            self.config.port = port  # might be updated if there is a filter
+            self.device_path = config.port
+
             self.connection = SerialPort(
                 self._protocol,
-                config.port,
+                port,
                 reactor,
                 baudrate=config.baudrate,
                 bytesize=config.bytesize,

inkcut/device/transports/serialport/settings.enaml

@@ -12,45 +12,153 @@ Created on Jul 12, 2015
 """
 import textwrap
 from inkcut.core.utils import load_icon
-from enaml.layout.api import hbox, align, spacer
+from enaml.layout.api import hbox, vbox, align, spacer
 from enaml.qt.QtWidgets import QApplication
-from enaml.widgets.api import Container, Form, Label, ObjectCombo, SpinBox, CheckBox, PushButton
+from enaml.widgets.api import Container, Form, Label, ObjectCombo, SpinBox, CheckBox, PushButton, Menu, Action, Field
+from enaml.validator import IntValidator
+from inkcut.core.api import log
+from enaml.stdlib.message_box import question
 import serial
 
+enamldef HexPair(Field):
+    attr number = 0
+    mask = "HHHH"
+    validator = IntValidator(minimum=0, maximum=0x888ffff, base=16)
+    text << "{:0>4x}".format(number)
+    text :: self.number = int(change['value'], 16)
+    max_length = 4
+
 enamldef SerialPortSettingsView(Container):
     attr model
     attr exclusive = False
+    attr show_all_ports = False
     padding = 0
 
     alias dsrdtr
     func selected_port(port, ports):
-        matches = [p for p in ports if p.device_path == port]
+        matches = [p for p in ports if p.device_path == port and model.port_matches(p)]
         return matches[0] if matches else None
 
+    func get_ports(candidates):
+        if show_all_ports:
+            return candidates
+        else:
+            return [port for port in candidates if model.port_matches(port)]
+
+    func port_string(port):
+        if model.port_matches(port):
+            return str(port)
+        return "{}{}".format(QApplication.translate("serialport", "(not compatible): "), (str(port)))
+
+    func filter_update_confirm(result):
+        if result and result.action == 'accept':
+            return True
+        return False
+
+    func refresh_filter_clear():
+        filter_clear.enabled = model.has_port_filter()
+
     Form:
         Label:
             text = QApplication.translate("serialport", "Port")
         Container:
             padding = 0
             constraints = [
-                hbox(cb, pb),
-                align('v_center', cb, pb)
+                vbox(
+                    hbox(cb, pb),
+                    hbox(filter_label, filter_descr, filter_label_vid, filter_vid,
+                         filter_label_pid, filter_pid, clear_group),
+                 ),
+                align('v_center', cb, pb),
+                align('v_center', filter_label, filter_label_vid, filter_label_pid, filter_descr),
+                (filter_vid.width == 50) | 'medium',
+                (filter_pid.width == 50) | 'medium',
+                filter_vid.width == filter_pid.width
             ]
             ObjectCombo: cb:
-                items << model.ports
+                items << get_ports(model.ports)
                 selected << selected_port(model.port, model.ports)
+                to_string = port_string
                 selected ::
                     port = change['value']
                     if port:
-                        model.port = port.device_path
+                        if not model.port_matches(port):
+                            if filter_update_confirm(question(
+                                    self, '', QApplication.translate("serialport",
+                                    "Port does not match the filter, do you want to update filter?"))):
+                                model.clear_filter()
+                                model.make_filter(port)
+                                model.port = port.device_path
+                                refresh_filter_clear()
+                            else:
+                                model.port = ""
+                                cb.selected = None
+                                cb._refresh_proxy(dict(type='update'))
+                        else:
+                            old_port = change['oldvalue']
+                            if not model.has_port_filter() and\
+                                 (old_port and old_port.device_path != port.device_path):
+                                model.make_filter(port)
+                            model.port = port.device_path
                 tool_tip = textwrap.dedent("""
                 List of serial ports detected by the system. If nothing is here you
                 must install the device driver for your machine.
                 """).strip()
-            PushButton: pb:
-                text = QApplication.translate("serialport", "Refresh")
-                icon = load_icon("arrow_refresh")
-                clicked :: model.refresh()
+            Container: pb:
+                padding = 0
+                constraints = [
+                    hbox(pb2, pb3, spacing=0),
+                    pb2.height == pb3.height,
+                    pb3.width == pb3.height
+                ]
+                hug_width = 'medium'
+                PushButton: pb2:
+                    text = QApplication.translate("serialport", "Refresh")
+                    icon = load_icon("arrow_refresh")
+                    clicked :: model.refresh()
+                PushButton: pb3:
+                    Menu:
+                        Action:
+                            text = QApplication.translate("serialport", "Show all")
+                            checkable = True
+                            checked := show_all_ports
+                            checked :: model.refresh()
+            Label: filter_label:
+                text = QApplication.translate("serialport", "Filter: ")
+            Field: filter_descr:
+                text := model.port_filter_name
+                text :: refresh_filter_clear()
+            Label: filter_label_vid:
+                text = QApplication.translate("serialport", "VID")
+            HexPair: filter_vid:
+                number := model.port_filter_vid
+                number :: refresh_filter_clear()
+                resist_width = 'weak'
+            Label: filter_label_pid:
+                text = QApplication.translate("serialport", "PID")
+            HexPair: filter_pid:
+                number := model.port_filter_pid
+                number :: refresh_filter_clear()
+                resist_width = 'weak'
+            Container: clear_group:
+                padding = 0
+                constraints = [
+                    hbox(filter_clear, pb4, spacing=0),
+                    pb4.height == filter_clear.height,
+                    pb4.width == pb4.height
+                ]
+                PushButton: filter_clear:
+                    text = QApplication.translate("serialport", "Clear")
+                    icon = load_icon("cancel")
+                    clicked ::
+                        model.clear_filter()
+                        model.refresh()
+                        refresh_filter_clear()
+                PushButton: pb4:
+                    Menu:
+                        Action:
+                            text = QApplication.translate("serialport", "Make filter")
+                            triggered :: model.make_filter(cb.selected)
         Label:
             text = QApplication.translate("serialport", "Baudrate")
         SpinBox:

tests/transport/test_serial_config.py

@@ -0,0 +1,81 @@
+"""
+Copyright (c) 2024, Karlis Senko
+
+Distributed under the terms of the GPL v3 License.
+
+The full license is in the file LICENSE, distributed with this software.
+
+Created on Dec 9, 2024
+
+@author: karliss
+"""
+import pytest
+from inkcut.device.transports.serialport.plugin import SerialConfigBase, SerialPortInfo
+
+
+def test_serial_filter():
+    info = SerialPortInfo(device_path='port1')
+    config = SerialConfigBase()
+
+    assert not config.has_port_filter()
+    assert config.port_matches(info)
+
+    config = SerialConfigBase(port_filter_name='ort')
+
+    assert config.has_port_filter()
+    assert not config.port_matches(info)
+
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=1, usb_pid=1))
+    assert not config.port_matches(SerialPortInfo(description='P_rt1', usb_vid=1, usb_pid=1))
+
+    config = SerialConfigBase(port_filter_vid=5)
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=5, usb_pid=1))
+    assert config.port_matches(SerialPortInfo(usb_vid=5, usb_pid=1))
+    assert not config.port_matches(SerialPortInfo(description='Port1', usb_vid=1, usb_pid=1))
+
+    config = SerialConfigBase(port_filter_pid=10)
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=5, usb_pid=10))
+    assert config.port_matches(SerialPortInfo(usb_vid=4, usb_pid=10))
+    assert not config.port_matches(SerialPortInfo(description='Port1', usb_vid=2, usb_pid=2))
+
+    config = SerialConfigBase(port_filter_name='Port1', port_filter_vid=0x1234, port_filter_pid=0x45ac)
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=0x1234, usb_pid=0x45ac))
+    assert not config.port_matches(SerialPortInfo(description='Port1', usb_vid=0x1234, usb_pid=0x45ad))
+    assert not config.port_matches(SerialPortInfo(description='Port1', usb_vid=0x1235, usb_pid=0x45ac))
+    assert not config.port_matches(SerialPortInfo(description='Port2', usb_vid=0x1234, usb_pid=0x45ac))
+
+    config = SerialConfigBase(port_filter_name='P.*1')
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=1, usb_pid=1))
+    assert config.port_matches(SerialPortInfo(description='Pooooooort1', usb_vid=1, usb_pid=1))
+    assert not config.port_matches(SerialPortInfo(description='P..', usb_vid=1, usb_pid=1))
+
+    config = SerialConfigBase(port_filter_name='^Port1$')
+    assert config.port_matches(SerialPortInfo(description='Port1', usb_vid=1, usb_pid=1))
+    assert not config.port_matches(SerialPortInfo(description='Port12', usb_vid=1, usb_pid=1))
+
+
+def test_make_filter():
+    info = SerialPortInfo(device_path='port1', description='Port1 some plotter', usb_vid=123, usb_pid=321)
+    config = SerialConfigBase()
+
+    config.make_filter(info)
+    assert config.has_port_filter()
+    assert config.port_matches(info)
+
+    info_2 = SerialPortInfo(device_path='port1', description='Port1 other plotter', usb_vid=123, usb_pid=321)
+    assert not config.port_matches(info_2)
+
+    info_2 = SerialPortInfo(device_path='port1', description='Port1 some plotter', usb_vid=124, usb_pid=321)
+    assert not config.port_matches(info_2)
+
+    info_2 = SerialPortInfo(device_path='port1', description='Port1 some plotter', usb_vid=123, usb_pid=421)
+    assert not config.port_matches(info_2)
+
+    config.clear_filter()
+    assert not config.has_port_filter()
+
+    info = SerialPortInfo(device_path='port1', description=r"a+.*?(){},{^$|\\", usb_vid=123, usb_pid=321)
+    config = SerialConfigBase()
+    config.make_filter(info)
+    assert config.port_matches(info)
+    assert not config.port_matches(SerialPortInfo(description='a', usb_vid=123, usb_pid=321))