ble.lwp3: add virtual port commands

Author: dlech

pybricksdev/ble/lwp3/bytecodes.py

@@ -995,3 +995,8 @@ class DataFormat(IntEnum):
 
     DATAF = 0x03
     """32-bit floating point, little-endian."""
+
+
+class VirtualPortSetupCommand(IntEnum):
+    DISCONNECT = 0x00
+    CONNECT = 0x01

pybricksdev/ble/lwp3/messages.py

@@ -45,6 +45,7 @@
     PortID,
     PortInfoFormatSetupCommand,
     Version,
+    VirtualPortSetupCommand,
 )
 
 
@@ -1314,6 +1315,56 @@ def __repr__(self) -> str:
         return f"{self.__class__.__name__}({repr(self.port)}, {repr(self.combo)}, {repr(self.multi_update)}, {repr(self.modes_and_datasets)})"
 
 
+###############################################################################
+# Virtual port messages
+###############################################################################
+
+
+class AbstractVirtualPortSetupMessage(AbstractMessage):
+    @abc.abstractmethod
+    def __init__(self, length: int, command: VirtualPortSetupCommand) -> None:
+        super().__init__(length, MessageKind.VIRTUAL_PORT_SETUP)
+
+        self._data[3] = command
+
+    @property
+    def command(self) -> VirtualPortSetupCommand:
+        return VirtualPortSetupCommand(self._data[3])
+
+
+class VirtualPortSetupDisconnectMessage(AbstractVirtualPortSetupMessage):
+    def __init__(self, port: PortID) -> None:
+        super().__init__(5, VirtualPortSetupCommand.DISCONNECT)
+
+        self._data[4] = port
+
+    @property
+    def port(self) -> PortID:
+        return PortID(self._data[4])
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({repr(self.port)})"
+
+
+class VirtualPortSetupConnectMessage(AbstractVirtualPortSetupMessage):
+    def __init__(self, port_a: PortID, port_b: PortID) -> None:
+        super().__init__(6, VirtualPortSetupCommand.CONNECT)
+
+        self._data[4] = port_a
+        self._data[5] = port_b
+
+    @property
+    def port_a(self) -> PortID:
+        return PortID(self._data[4])
+
+    @property
+    def port_b(self) -> PortID:
+        return PortID(self._data[5])
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({repr(self.port_a)}, {repr(self.port_b)})"
+
+
 ###############################################################################
 # Message parsing
 ###############################################################################
@@ -1399,6 +1450,11 @@ class _Lookup(NamedTuple):
     ModeInfoKind.FORMAT: PortModeInfoFormatMessage,
 }
 
+_VIRTUAL_PORT_SETUP_CLASS_MAP = {
+    VirtualPortSetupCommand.DISCONNECT: VirtualPortSetupDisconnectMessage,
+    VirtualPortSetupCommand.CONNECT: VirtualPortSetupConnectMessage,
+}
+
 # base type descriminator for messages
 _MESSAGE_CLASS_MAP = {
     MessageKind.HUB_PROPERTY: _Lookup(4, _HUB_PROPERTY_OP_CLASS_MAP),
@@ -1420,6 +1476,7 @@ class _Lookup(NamedTuple):
     MessageKind.PORT_VALUE_COMBO: PortValueComboMessage,
     MessageKind.PORT_INPUT_FMT: PortInputFormatMessage,
     MessageKind.PORT_INPUT_FMT_COMBO: PortInputFormatComboMessage,
+    MessageKind.VIRTUAL_PORT_SETUP: _Lookup(3, _VIRTUAL_PORT_SETUP_CLASS_MAP),
 }
 
 

tests/ble/test_lwp3_messages.py

@@ -27,6 +27,7 @@
     PortID,
     PortInfoFormatSetupCommand,
     Version,
+    VirtualPortSetupCommand,
 )
 from pybricksdev.ble.lwp3.messages import (
     AbstractHubAlertMessage,
@@ -36,6 +37,7 @@
     AbstractMessage,
     AbstractPortInfoMessage,
     AbstractPortModeInfoMessage,
+    AbstractVirtualPortSetupMessage,
     ErrorMessage,
     FirmwareUpdateMessage,
     HubActionMessage,
@@ -88,6 +90,8 @@
     PortModeInfoSymbolMessage,
     PortValueComboMessage,
     PortValueMessage,
+    VirtualPortSetupConnectMessage,
+    VirtualPortSetupDisconnectMessage,
     parse_message,
 )
 
@@ -1292,3 +1296,48 @@ def test_parse_message(self):
             assert msg.combo == 2
             assert msg.multi_update is True
             assert msg.modes_and_datasets == [0, 1]
+
+
+class TestVirtualPortMessages:
+    class TestVirtualPortSetupMessages:
+        def test_is_abstract(self):
+            assert inspect.isabstract(AbstractVirtualPortSetupMessage)
+
+        class TestVirtualPortSetupDisconnectMessage:
+            def test_constructor(self):
+                msg = VirtualPortSetupDisconnectMessage(PortID(3))
+                assert msg.length == 5
+                assert msg.kind is MessageKind.VIRTUAL_PORT_SETUP
+                assert msg.command is VirtualPortSetupCommand.DISCONNECT
+                assert msg.port is PortID(3)
+                assert repr(msg) == "VirtualPortSetupDisconnectMessage(<PortID.3: 3>)"
+
+            def test_parse_message(self):
+                msg = parse_message(b"\x05\x00\x61\x00\x03")
+                assert isinstance(msg, VirtualPortSetupDisconnectMessage)
+                assert msg.length == 5
+                assert msg.kind is MessageKind.VIRTUAL_PORT_SETUP
+                assert msg.command is VirtualPortSetupCommand.DISCONNECT
+                assert msg.port is PortID(3)
+
+        class TestVirtualPortSetupConnectMessage:
+            def test_constructor(self):
+                msg = VirtualPortSetupConnectMessage(PortID(1), PortID(2))
+                assert msg.length == 6
+                assert msg.kind is MessageKind.VIRTUAL_PORT_SETUP
+                assert msg.command is VirtualPortSetupCommand.CONNECT
+                assert msg.port_a is PortID(1)
+                assert msg.port_b is PortID(2)
+                assert (
+                    repr(msg)
+                    == "VirtualPortSetupConnectMessage(<PortID.1: 1>, <PortID.2: 2>)"
+                )
+
+            def test_parse_message(self):
+                msg = parse_message(b"\x06\x00\x61\x01\x01\x02")
+                assert isinstance(msg, VirtualPortSetupConnectMessage)
+                assert msg.length == 6
+                assert msg.kind is MessageKind.VIRTUAL_PORT_SETUP
+                assert msg.command is VirtualPortSetupCommand.CONNECT
+                assert msg.port_a is PortID(1)
+                assert msg.port_b is PortID(2)