CHANGELOG.md

@@ -11,24 +11,6 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
 
 -   [x]
 
-## [2.7.0] - 2026-03-01
-
-### Added
-
--   new kind `matrix` has been added, it does two things:
-    -   scales the interface according to `potato` kind, in practice this has no affect but it's required by the builder classes.
-    -   disables the rt listener threads since we aren't expecting to receive any from a Matrix VBAN server.
-        -   however, matrix responses may still be received with the {VbanCmd}.sendtext() method.
-
-### Changed
-
--   `outbound` kwarg has been renamed to `disable_rt_listeners`. Since it's job is to disable the listener threads for incoming RT packets this new name is more descriptive.
--   dataclasses representing packet headers and packets with ident:0 and ident:1 have been moved into an internal packet module.
-
-### Removed
-
--   {VbanCmd}.sendtext() @script decorator removed. It's purpose was to attempt to convert a dictionary to a script but it was poorly implemented and there exists the {VbanCmd}.apply() method already.
-
 ## [2.6.0] - 2026-02-26
 
 ### Added

README.md

@@ -8,19 +8,19 @@
 
 # VBAN CMD
 
-This python interface allows you to send Voicemeeter/Matrix commands over a network.
+This python interface allows you to transmit Voicemeeter parameters over a network.
 
-It offers the same public API as [Voicemeeter Remote Python API](https://github.com/onyx-and-iris/voicemeeter-api-python).
+It may be used standalone or to extend the [Voicemeeter Remote Python API](https://github.com/onyx-and-iris/voicemeeter-api-python)
 
-Only the VBAN SERVICE/TEXT subprotocols are supported, there is no support for AUDIO or MIDI in this package.
+There is no support for audio transfer in this package, only parameters.
 
 For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
 
 ## Tested against
 
--   Basic 1.1.2.2
--   Banana 2.1.2.2
--   Potato 3.1.2.2
+-   Basic 1.0.8.8
+-   Banana 2.0.6.8
+-   Potato 3.0.2.8
 
 ## Requirements
 
@@ -29,9 +29,7 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
 
 ## Installation
 
-```console
-pip install vban-cmd
-```
+`pip install vban-cmd`
 
 ## `Use`
 
@@ -115,8 +113,6 @@ Pass the kind of Voicemeeter as an argument. KIND_ID may be:
 -   `banana`
 -   `potato`
 
-A fourth kind `matrix` has been added, if you pass it as a KIND_ID you are expected to use the [{VbanCmd}.sendtext()](https://github.com/onyx-and-iris/vban-cmd-python?tab=readme-ov-file#vbansendtextscript) method for sending text requests.
-
 ## `Available commands`
 
 ### Strip
@@ -515,8 +511,7 @@ You may pass the following optional keyword arguments:
 -   `pdirty`: boolean=False, parameter updates
 -   `ldirty`: boolean=False, level updates
 -   `timeout`: int=5, amount of time (seconds) to wait for an incoming RT data packet (parameter states).
--   `disable_rt_listeners`: boolean=False, set `True` if you don't wish to receive RT packets.
-    -   You can still send Matrix string requests ending with `?` and receive a response.
+-   `outbound`: boolean=False, set `True` if you are only interested in sending commands. (no rt packets will be received)
 
 #### `vban.pdirty`
 
@@ -534,14 +529,6 @@ Sends a script block as a string request, for example:
 vban.sendtext('Strip[0].Mute=1;Bus[0].Mono=1')
 ```
 
-You can even use it to send matrix commands:
-
-```python
-vban.sendtext('Point(ASIO128.IN[1..4],ASIO128.OUT[1]).dBGain = -3.0')
-
-vban.sendtext('Command.Version = ?')
-```
-
 ## Errors
 
 -   `errors.VBANCMDError`: Base VBANCMD Exception class.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vban-cmd"
-version = "2.7.0"
+version = "2.6.0"
 description = "Python interface for the VBAN RT Packet Service (Sendtext)"
 authors = [{ name = "Onyx and Iris", email = "code@onyxandiris.online" }]
 license = { text = "MIT" }

vban_cmd/factory.py

@@ -85,7 +85,7 @@ class FactoryBase(VbanCmd):
             'channel': 0,
             'ratelimit': 0.01,
             'timeout': 5,
-            'disable_rt_listeners': False,
+            'outbound': False,
             'sync': False,
             'pdirty': False,
             'ldirty': False,
@@ -202,13 +202,7 @@ def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd:
             _factory = BasicFactory
         case 'banana':
             _factory = BananaFactory
-        case 'potato' | 'matrix':
-            # matrix is a special kind where:
-            # - we don't need to scale the interface with the builder (in other words kind is arbitrary).
-            # - we don't ever need to use real-time listeners, so we disable them to avoid confusion
-            if kind_id == 'matrix':
-                kwargs['disable_rt_listeners'] = True
-                kind_id = 'potato'
+        case 'potato':
             _factory = PotatoFactory
         case _:
             raise ValueError(f"Unknown Voicemeeter kind '{kind_id}'")

vban_cmd/packet/headers.py

@@ -9,9 +9,6 @@ VBAN_PROTOCOL_SERVICE = 0x60
 VBAN_SERVICE_RTPACKETREGISTER = 32
 VBAN_SERVICE_RTPACKET = 33
 VBAN_SERVICE_MASK = 0xE0
-VBAN_PROTOCOL_MASK = 0xE0
-VBAN_SERVICE_REQUESTREPLY = 0x02
-VBAN_SERVICE_FNCT_REPLY = 0x02
 
 MAX_PACKET_SIZE = 1436
 HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16
@@ -95,38 +92,6 @@ class VbanSubscribeHeader:
         return bytes(data)
 
 
-def _parse_vban_service_header(data: bytes) -> dict:
-    """Common parsing and validation for VBAN service protocol headers."""
-    if len(data) < HEADER_SIZE:
-        raise ValueError('Data is too short to be a valid VBAN header')
-
-    if data[:4] != b'VBAN':
-        raise ValueError('Invalid VBAN magic bytes')
-
-    format_sr = data[4]
-    format_nbs = data[5]
-    format_nbc = data[6]
-    format_bit = data[7]
-
-    # Verify this is a service protocol packet
-    protocol = format_sr & VBAN_PROTOCOL_MASK
-    if protocol != VBAN_PROTOCOL_SERVICE:
-        raise ValueError(f'Not a service protocol packet: {protocol:02x}')
-
-    # Extract stream name and frame counter
-    name = data[8:24].rstrip(b'\x00').decode('utf-8', errors='ignore')
-    framecounter = int.from_bytes(data[24:28], 'little')
-
-    return {
-        'format_sr': format_sr,
-        'format_nbs': format_nbs,
-        'format_nbc': format_nbc,
-        'format_bit': format_bit,
-        'name': name,
-        'framecounter': framecounter,
-    }
-
-
 @dataclass
 class VbanResponseHeader:
     """Represents the header of a response packet"""
@@ -136,7 +101,6 @@ class VbanResponseHeader:
     format_nbs: int = 0
     format_nbc: int = VBAN_SERVICE_RTPACKET
     format_bit: int = 0
-    framecounter: int = 0
 
     @property
     def vban(self) -> bytes:
@@ -148,63 +112,17 @@ class VbanResponseHeader:
 
     @classmethod
     def from_bytes(cls, data: bytes):
-        """Parse a VbanResponseHeader from bytes."""
-        parsed = _parse_vban_service_header(data)
+        if len(data) < HEADER_SIZE:
+            raise ValueError('Data is too short to be a valid VbanResponseHeader')
 
-        # Validate this is an RTPacket response
-        if parsed['format_nbc'] != VBAN_SERVICE_RTPACKET:
-            raise ValueError(
-                f'Not a RTPacket response packet: {parsed["format_nbc"]:02x}'
-            )
-
-        return cls(**parsed)
-
-
-@dataclass
-class VbanMatrixResponseHeader:
-    """Represents the header of a matrix response packet"""
-
-    name: str = 'Request Reply'
-    format_sr: int = VBAN_PROTOCOL_SERVICE
-    format_nbs: int = VBAN_SERVICE_FNCT_REPLY
-    format_nbc: int = VBAN_SERVICE_REQUESTREPLY
-    format_bit: int = 0
-    framecounter: int = 0
-
-    @property
-    def vban(self) -> bytes:
-        return b'VBAN'
-
-    @property
-    def streamname(self) -> bytes:
-        return self.name.encode('ascii')[:16].ljust(16, b'\x00')
-
-    @classmethod
-    def from_bytes(cls, data: bytes):
-        """Parse a matrix response packet from bytes."""
-        parsed = _parse_vban_service_header(data)
-
-        # Validate this is a service reply packet
-        if parsed['format_nbs'] != VBAN_SERVICE_FNCT_REPLY:
-            raise ValueError(f'Not a service reply packet: {parsed["format_nbs"]:02x}')
-
-        return cls(**parsed)
-
-    @classmethod
-    def extract_payload(cls, data: bytes) -> str:
-        """Extract the text payload from a matrix response packet."""
-        if len(data) <= HEADER_SIZE:
-            return ''
-
-        payload_bytes = data[HEADER_SIZE:]
-        return payload_bytes.rstrip(b'\x00').decode('utf-8', errors='ignore')
-
-    @classmethod
-    def parse_response(cls, data: bytes) -> tuple['VbanMatrixResponseHeader', str]:
-        """Parse a complete matrix response packet returning header and payload."""
-        header = cls.from_bytes(data)
-        payload = cls.extract_payload(data)
-        return header, payload
+        name = data[8:24].rstrip(b'\x00').decode('utf-8')
+        return cls(
+            name=name,
+            format_sr=data[4] & VBAN_SERVICE_MASK,
+            format_nbs=data[5],
+            format_nbc=data[6],
+            format_bit=data[7],
+        )
 
 
 @dataclass

vban_cmd/util.py

@@ -63,6 +63,27 @@ def depth(d):
     return 0
 
 
+def script(func):
+    """Convert dictionary to script"""
+
+    def wrapper(*args):
+        remote, script = args
+        if isinstance(script, dict):
+            params = ''
+            for key, val in script.items():
+                obj, m2, *rem = key.split('-')
+                index = int(m2) if m2.isnumeric() else int(*rem)
+                params += ';'.join(
+                    f'{obj}{f".{m2}stream" if not m2.isnumeric() else ""}[{index}].{k}={int(v) if isinstance(v, bool) else v}'
+                    for k, v in val.items()
+                )
+                params += ';'
+            script = params
+        return func(remote, script)
+
+    return wrapper
+
+
 def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
     """
     Generator function, accepts two tuples of dB values.

vban_cmd/vbancmd.py

@@ -10,9 +10,9 @@ from typing import Union
 from .enums import NBS
 from .error import VBANCMDError
 from .event import Event
-from .packet.headers import VbanMatrixResponseHeader, VbanRequestHeader
+from .packet.headers import VbanRequestHeader
 from .subject import Subject
-from .util import bump_framecounter, deep_merge
+from .util import bump_framecounter, deep_merge, script
 from .worker import Producer, Subscriber, Updater
 
 logger = logging.getLogger(__name__)
@@ -86,8 +86,8 @@ class VbanCmd(abc.ABC):
         self.logout()
 
     def login(self) -> None:
-        """Starts the subscriber and updater threads (unless disable_rt_listeners is True) and logs into Voicemeeter."""
-        if not self.disable_rt_listeners:
+        """Starts the subscriber and updater threads (unless in outbound mode)"""
+        if not self.outbound:
             self.event.info()
 
             self.stop_event = threading.Event()
@@ -139,20 +139,11 @@ class VbanCmd(abc.ABC):
         self._send_request(f'{cmd}={val};')
         self.cache[cmd] = val
 
-    def sendtext(self, script) -> str | None:
+    @script
+    def sendtext(self, script):
         """Sends a multiple parameter string over a network."""
         self._send_request(script)
         self.logger.debug(f'sendtext: {script}')
-
-        if self.disable_rt_listeners and script.endswith(('?', '?;')):
-            try:
-                response = VbanMatrixResponseHeader.extract_payload(
-                    self.sock.recv(1024)
-                )
-                return response
-            except ValueError as e:
-                self.logger.warning(f'Error extracting matrix response: {e}')
-
         time.sleep(self.DELAY)
 
     @property

vban_cmd/worker.py

@@ -7,6 +7,8 @@ from .enums import NBS
 from .error import VBANCMDConnectionError
 from .packet.headers import (
     HEADER_SIZE,
+    VBAN_PROTOCOL_SERVICE,
+    VBAN_SERVICE_RTPACKET,
     VbanPacket,
     VbanResponseHeader,
     VbanSubscribeHeader,
@@ -87,31 +89,31 @@ class Producer(threading.Thread):
             data, _ = self._remote.sock.recvfrom(2048)
             if len(data) < HEADER_SIZE:
                 return
+
+            response_header = VbanResponseHeader.from_bytes(data[:HEADER_SIZE])
+            if (
+                response_header.format_sr != VBAN_PROTOCOL_SERVICE
+                or response_header.format_nbc != VBAN_SERVICE_RTPACKET
+            ):
+                return
+
+            match response_header.format_nbs:
+                case NBS.zero:
+                    return VbanPacketNBS0.from_bytes(
+                        nbs=NBS.zero, kind=self._remote.kind, data=data
+                    )
+
+                case NBS.one:
+                    return VbanPacketNBS1.from_bytes(
+                        nbs=NBS.one, kind=self._remote.kind, data=data
+                    )
+            return None
         except TimeoutError as e:
             self.logger.exception(f'{type(e).__name__}: {e}')
             raise VBANCMDConnectionError(
-                f'timeout waiting for response from {self._remote.ip}:{self._remote.port}'
+                f'timeout waiting for RtPacket from {self._remote.ip}'
             ) from e
 
-        try:
-            header = VbanResponseHeader.from_bytes(data[:HEADER_SIZE])
-        except ValueError as e:
-            self.logger.warning(f'Error parsing response packet: {e}')
-            return None
-
-        match header.format_nbs:
-            case NBS.zero:
-                return VbanPacketNBS0.from_bytes(
-                    nbs=NBS.zero, kind=self._remote.kind, data=data
-                )
-
-            case NBS.one:
-                return VbanPacketNBS1.from_bytes(
-                    nbs=NBS.one, kind=self._remote.kind, data=data
-                )
-
-        return None
-
     def stopped(self):
         return self.stop_event.is_set()