add ChannelState interface, use it in the meta functions.

reword busmodes bitwise logic.

comment out ratelimit, this will probably get permanently removed.

.github/workflows/publish.yml

@@ -0,0 +1,53 @@
+name: Publish to PyPI
+
+on:
+  release:
+    types: [published]
+  push:
+    tags:
+      - 'v*.*.*'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install Poetry
+        run: |
+          pip install poetry==2.3.1
+          poetry --version
+
+      - name: Build package
+        run: |
+          poetry install --only-root
+          poetry build
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: dist
+          path: ./dist
+
+  pypi-publish:
+    needs: build
+    name: Upload release to PyPI
+    runs-on: ubuntu-latest
+    environment:
+      name: pypi
+      url: https://pypi.org/project/vban-cmd/
+    permissions:
+      id-token: write
+    steps:
+      - uses: actions/download-artifact@v4
+        with:
+          name: dist
+          path: ./dist
+
+      - name: Publish package distributions to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          packages-dir: ./dist

.github/workflows/ruff.yml

@@ -0,0 +1,19 @@
+name: Ruff
+
+on:
+  push:
+    branches: [main]
+
+  pull_request:
+    branches: [main]
+
+  workflow_dispatch:
+
+jobs:
+  ruff:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - uses: astral-sh/ruff-action@v3
+        with:
+          args: 'format --check --diff'

vban_cmd/bus.py

@@ -4,7 +4,7 @@ from typing import Union
 
 from .enums import NBS, BusModes
 from .iremote import IRemote
-from .meta import bus_mode_prop, channel_bool_prop, channel_label_prop
+from .meta import bus_mode_prop, channel_bool_prop, channel_int_prop, channel_label_prop
 
 
 class Bus(IRemote):
@@ -90,20 +90,9 @@ class BusLevel(IRemote):
     def getter(self):
         """Returns a tuple of level values for the channel."""
 
-        def fget(i):
-            return round((((1 << 16) - 1) - i) * -0.01, 1)
-
         if not self._remote.stopped() and self._remote.event.ldirty:
-            return tuple(
+            return self._remote.cache['bus_level'][self.range[0] : self.range[-1]]
-                fget(i)
+        return self.public_packets[NBS.zero].levels.bus[self.range[0] : self.range[-1]]
-                for i in self._remote.cache['bus_level'][self.range[0] : self.range[-1]]
-            )
-        return tuple(
-            fget(i)
-            for i in self._remote._get_levels(self.public_packets[NBS.zero])[1][
-                self.range[0] : self.range[-1]
-            ]
-        )
 
     @property
     def identifier(self) -> str:
@@ -128,45 +117,49 @@ class BusLevel(IRemote):
 def _make_bus_mode_mixin():
     """Creates a mixin of Bus Modes."""
 
-    modestates = {
+    mode_names = [
-        'normal': [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+        'normal',
-        'amix': [1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1],
+        'amix',
-        'repeat': [0, 2, 2, 0, 0, 2, 2, 0, 0, 2, 2],
+        'repeat',
-        'bmix': [1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3],
+        'bmix',
-        'composite': [0, 0, 0, 4, 4, 4, 4, 0, 0, 0, 0],
+        'composite',
-        'tvmix': [1, 0, 1, 4, 5, 4, 5, 0, 1, 0, 1],
+        'tvmix',
-        'upmix21': [0, 2, 2, 4, 4, 6, 6, 0, 0, 2, 2],
+        'upmix21',
-        'upmix41': [1, 2, 3, 4, 5, 6, 7, 0, 1, 2, 3],
+        'upmix41',
-        'upmix61': [0, 0, 0, 0, 0, 0, 0, 8, 8, 8, 8],
+        'upmix61',
-        'centeronly': [1, 0, 1, 0, 1, 0, 1, 8, 9, 8, 9],
+        'centeronly',
-        'lfeonly': [0, 2, 2, 0, 0, 2, 2, 8, 8, 10, 10],
+        'lfeonly',
-        'rearonly': [1, 2, 3, 0, 1, 2, 3, 8, 9, 10, 11],
+        'rearonly',
-    }
+    ]
 
     def identifier(self) -> str:
         return f'bus[{self.index}].mode'
 
     def get(self):
-        states = [
+        """Get current bus mode using ChannelState for clean bit extraction."""
-            (
+        mode_cache_items = [
-                int.from_bytes(
+            (k, v)
-                    self.public_packets[NBS.zero].busstate[self.index], 'little'
+            for k, v in self._remote.cache.items()
-                )
+            if k.startswith(f'{self.identifier}.') and v == 1
-                & val
-            )
-            >> 4
-            for val in self._modes.modevals
         ]
-        for k, v in modestates.items():
+
-            if states == v:
+        if mode_cache_items:
-                return k
+            latest_cached = mode_cache_items[-1][0]
+            mode_name = latest_cached.split('.')[-1]
+            return mode_name
+
+        bus_state = self.public_packets[NBS.zero].states.bus[self.index]
+
+        # Extract bus mode from bits 4-7 (mask 0xF0, shift right by 4)
+        mode_value = (bus_state._state & 0x000000F0) >> 4
+
+        return mode_names[mode_value] if mode_value < len(mode_names) else 'normal'
 
     return type(
         'BusModeMixin',
         (IRemote,),
         {
             'identifier': property(identifier),
-            'modestates': modestates,
             **{mode.name: bus_mode_prop(mode.name) for mode in BusModes},
             'get': get,
         },
@@ -188,7 +181,8 @@ def bus_factory(phys_bus, remote, i) -> Union[PhysicalBus, VirtualBus]:
             'eq': BusEQ.make(remote, i),
             'levels': BusLevel(remote, i),
             'mode': BUSMODEMIXIN_cls(remote, i),
-            **{param: channel_bool_prop(param) for param in ['mute', 'mono']},
+            **{param: channel_bool_prop(param) for param in ('mute',)},
+            **{param: channel_int_prop(param) for param in ('mono',)},
             'label': channel_label_prop(),
         },
     )(remote, i)

vban_cmd/meta.py

@@ -1,7 +1,7 @@
 from functools import partial
 
-from .enums import NBS
+from .enums import NBS, BusModes
-from .util import cache_bool, cache_float, cache_string
+from .util import cache_bool, cache_float, cache_int, cache_string
 
 
 def channel_bool_prop(param):
@@ -11,17 +11,23 @@ def channel_bool_prop(param):
     def fget(self):
         cmd = self._cmd(param)
         self.logger.debug(f'getter: {cmd}')
-        return (
+
-            not int.from_bytes(
+        states = self.public_packets[NBS.zero].states
-                getattr(
+        channel_states = (
-                    self.public_packets[NBS.zero],
+            states.strip if 'strip' in type(self).__name__.lower() else states.bus
-                    f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}state',
-                )[self.index],
-                'little',
-            )
-            & getattr(self._modes, f'_{param.lower()}')
-            == 0
         )
+        channel_state = channel_states[self.index]
+
+        if param.lower() == 'mute':
+            return channel_state.mute
+        elif param.lower() == 'solo':
+            return channel_state.solo
+        elif param.lower() == 'mono':
+            return channel_state.mono
+        elif param.lower() == 'mc':
+            return channel_state.mc
+        else:
+            return channel_state.get_mode(getattr(self._modes, f'_{param.lower()}'))
 
     def fset(self, val):
         self.setter(param, 1 if val else 0)
@@ -29,18 +35,46 @@ def channel_bool_prop(param):
     return property(fget, fset)
 
 
+def channel_int_prop(param):
+    """meta function for channel integer parameters"""
+
+    @partial(cache_int, param=param)
+    def fget(self):
+        cmd = self._cmd(param)
+        self.logger.debug(f'getter: {cmd}')
+
+        states = self.public_packets[NBS.zero].states
+        channel_states = (
+            states.strip if 'strip' in type(self).__name__.lower() else states.bus
+        )
+        channel_state = channel_states[self.index]
+
+        # Special case: bus mono is an integer (0-2) encoded using bits 2 and 9
+        if param.lower() == 'mono' and 'bus' in type(self).__name__.lower():
+            bit_2 = (channel_state._state >> 2) & 1
+            bit_9 = (channel_state._state >> 9) & 1
+            return (bit_9 << 1) | bit_2
+        else:
+            return channel_state.get_mode_int(getattr(self._modes, f'_{param.lower()}'))
+
+    def fset(self, val):
+        self.setter(param, val)
+
+    return property(fget, fset)
+
+
 def channel_label_prop():
     """meta function for channel label parameters"""
 
     @partial(cache_string, param='label')
     def fget(self) -> str:
-        return getattr(
+        if 'strip' in type(self).__name__.lower():
-            self.public_packets[NBS.zero],
+            return self.public_packets[NBS.zero].labels.strip[self.index]
-            f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}labels',
+        else:
-        )[self.index]
+            return self.public_packets[NBS.zero].labels.bus[self.index]
 
     def fset(self, val: str):
-        self.setter('label', str(val))
+        self.setter('label', f'"{val}"')
 
     return property(fget, fset)
 
@@ -52,13 +86,10 @@ def strip_output_prop(param):
     def fget(self):
         cmd = self._cmd(param)
         self.logger.debug(f'getter: {cmd}')
-        return (
+
-            not int.from_bytes(
+        strip_state = self.public_packets[NBS.zero].states.strip[self.index]
-                self.public_packets[NBS.zero].stripstate[self.index], 'little'
+
-            )
+        return strip_state.get_mode(getattr(self._modes, f'_bus{param.lower()}'))
-            & getattr(self._modes, f'_bus{param.lower()}')
-            == 0
-        )
 
     def fset(self, val):
         self.setter(param, 1 if val else 0)
@@ -73,16 +104,15 @@ def bus_mode_prop(param):
     def fget(self):
         cmd = self._cmd(param)
         self.logger.debug(f'getter: {cmd}')
-        return [
+
-            (
+        bus_state = self.public_packets[NBS.zero].states.bus[self.index]
-                int.from_bytes(
+
-                    self.public_packets[NBS.zero].busstate[self.index], 'little'
+        # Extract current bus mode from bits 4-7
-                )
+        current_mode = (bus_state._state & 0x000000F0) >> 4
-                & val
+
-            )
+        expected_mode = getattr(BusModes, param.lower())
-            >> 4
+
-            for val in self._modes.modevals
+        return current_mode == expected_mode
-        ] == self.modestates[param]
 
     def fset(self, val):
         self.setter(param, 1 if val else 0)

vban_cmd/packet.py

@@ -19,8 +19,8 @@ VMPARAMSTRIP_SIZE = 174
 
 
 @dataclass
-class VbanRtPacket:
+class VbanPacket:
-    """Represents the body of a VBAN RT data packet"""
+    """Represents the header of a VBAN data packet"""
 
     nbs: NBS
     _kind: KindMapClass
@@ -31,10 +31,113 @@ class VbanRtPacket:
     _optionBits: bytes
     _samplerate: bytes
 
+    @property
+    def voicemeetertype(self) -> str:
+        """returns voicemeeter type as a string"""
+        return ['', 'basic', 'banana', 'potato'][
+            int.from_bytes(self._voicemeeterType, 'little')
+        ]
+
+    @property
+    def voicemeeterversion(self) -> tuple:
+        """returns voicemeeter version as a tuple"""
+        return tuple(self._voicemeeterVersion[i] for i in range(3, -1, -1))
+
+    @property
+    def samplerate(self) -> int:
+        """returns samplerate as an int"""
+        return int.from_bytes(self._samplerate, 'little')
+
+
+class Levels(NamedTuple):
+    strip: tuple[float, ...]
+    bus: tuple[float, ...]
+
+
+class ChannelState:
+    """Represents the processed state of a single strip or bus channel"""
+
+    def __init__(self, state_bytes: bytes):
+        # Convert 4-byte state to integer once for efficient lookups
+        self._state = int.from_bytes(state_bytes, 'little')
+
+    def get_mode(self, mode_value: int) -> bool:
+        """Get boolean state for a specific mode"""
+        return (self._state & mode_value) != 0
+
+    def get_mode_int(self, mode_value: int) -> int:
+        """Get integer state for a specific mode"""
+        return self._state & mode_value
+
+    # Common boolean modes
+    @property
+    def mute(self) -> bool:
+        return (self._state & 0x00000001) != 0
+
+    @property
+    def solo(self) -> bool:
+        return (self._state & 0x00000002) != 0
+
+    @property
+    def mono(self) -> bool:
+        return (self._state & 0x00000004) != 0
+
+    @property
+    def mc(self) -> bool:
+        return (self._state & 0x00000008) != 0
+
+    # EQ modes
+    @property
+    def eq_on(self) -> bool:
+        return (self._state & 0x00000100) != 0
+
+    @property
+    def eq_ab(self) -> bool:
+        return (self._state & 0x00000800) != 0
+
+    # Bus assignments (strip to bus routing)
+    @property
+    def busa1(self) -> bool:
+        return (self._state & 0x00001000) != 0
+
+    @property
+    def busa2(self) -> bool:
+        return (self._state & 0x00002000) != 0
+
+    @property
+    def busa3(self) -> bool:
+        return (self._state & 0x00004000) != 0
+
+    @property
+    def busa4(self) -> bool:
+        return (self._state & 0x00008000) != 0
+
+    @property
+    def busb1(self) -> bool:
+        return (self._state & 0x00010000) != 0
+
+    @property
+    def busb2(self) -> bool:
+        return (self._state & 0x00020000) != 0
+
+    @property
+    def busb3(self) -> bool:
+        return (self._state & 0x00040000) != 0
+
+
+class States(NamedTuple):
+    strip: tuple[ChannelState, ...]
+    bus: tuple[ChannelState, ...]
+
+
+class Labels(NamedTuple):
+    strip: tuple[str, ...]
+    bus: tuple[str, ...]
+
 
 @dataclass
-class VbanRtPacketNBS0(VbanRtPacket):
+class VbanPacketNBS0(VbanPacket):
-    """Represents the body of a VBAN RT data packet with NBS 0"""
+    """Represents the body of a VBAN data packet with ident:0"""
 
     _inputLeveldB100: bytes
     _outputLeveldB100: bytes
@@ -82,94 +185,86 @@ class VbanRtPacketNBS0(VbanRtPacket):
             _busLabelUTF8c60=data[932:1412],
         )
 
-    def _generate_levels(self, levelarray) -> tuple:
-        return tuple(
-            int.from_bytes(levelarray[i : i + 2], 'little')
-            for i in range(0, len(levelarray), 2)
-        )
-
-    @property
-    def strip_levels(self):
-        return self._generate_levels(self._inputLeveldB100)
-
-    @property
-    def bus_levels(self):
-        return self._generate_levels(self._outputLeveldB100)
-
     def pdirty(self, other) -> bool:
         """True iff any defined parameter has changed"""
 
-        return not (
+        self_gains = (
-            self._stripState == other._stripState
+            self._stripGaindB100Layer1
-            and self._busState == other._busState
+            + self._stripGaindB100Layer2
-            and self._stripGaindB100Layer1 == other._stripGaindB100Layer1
+            + self._stripGaindB100Layer3
-            and self._stripGaindB100Layer2 == other._stripGaindB100Layer2
+            + self._stripGaindB100Layer4
-            and self._stripGaindB100Layer3 == other._stripGaindB100Layer3
+            + self._stripGaindB100Layer5
-            and self._stripGaindB100Layer4 == other._stripGaindB100Layer4
+            + self._stripGaindB100Layer6
-            and self._stripGaindB100Layer5 == other._stripGaindB100Layer5
+            + self._stripGaindB100Layer7
-            and self._stripGaindB100Layer6 == other._stripGaindB100Layer6
+            + self._stripGaindB100Layer8
-            and self._stripGaindB100Layer7 == other._stripGaindB100Layer7
+        )
-            and self._stripGaindB100Layer8 == other._stripGaindB100Layer8
+        other_gains = (
-            and self._busGaindB100 == other._busGaindB100
+            other._stripGaindB100Layer1
-            and self._stripLabelUTF8c60 == other._stripLabelUTF8c60
+            + other._stripGaindB100Layer2
-            and self._busLabelUTF8c60 == other._busLabelUTF8c60
+            + other._stripGaindB100Layer3
+            + other._stripGaindB100Layer4
+            + other._stripGaindB100Layer5
+            + other._stripGaindB100Layer6
+            + other._stripGaindB100Layer7
+            + other._stripGaindB100Layer8
+        )
+
+        return (
+            self._stripState != other._stripState
+            or self._busState != other._busState
+            or self_gains != other_gains
+            or self._busGaindB100 != other._busGaindB100
+            or self._stripLabelUTF8c60 != other._stripLabelUTF8c60
+            or self._busLabelUTF8c60 != other._busLabelUTF8c60
         )
 
     def ldirty(self, strip_cache, bus_cache) -> bool:
+        """True iff any level has changed, ignoring changes when levels are very quiet"""
         self._strip_comp, self._bus_comp = (
             tuple(not val for val in comp(strip_cache, self.strip_levels)),
             tuple(not val for val in comp(bus_cache, self.bus_levels)),
         )
-        return any(any(li) for li in (self._strip_comp, self._bus_comp))
+        return any(self._strip_comp) or any(self._bus_comp)
 
     @property
-    def voicemeetertype(self) -> str:
+    def strip_levels(self) -> tuple[float, ...]:
-        """returns voicemeeter type as a string"""
+        """Returns strip levels in dB"""
-        type_ = ('basic', 'banana', 'potato')
-        return type_[int.from_bytes(self._voicemeeterType, 'little') - 1]
-
-    @property
-    def voicemeeterversion(self) -> tuple:
-        """returns voicemeeter version as a tuple"""
         return tuple(
-            reversed(
+            round(
-                tuple(
+                int.from_bytes(self._inputLeveldB100[i : i + 2], 'little', signed=True)
-                    int.from_bytes(self._voicemeeterVersion[i : i + 1], 'little')
+                * 0.01,
-                    for i in range(4)
+                1,
-                )
             )
+            for i in range(0, len(self._inputLeveldB100), 2)
+        )[: self._kind.num_strip_levels]
+
+    @property
+    def bus_levels(self) -> tuple[float, ...]:
+        """Returns bus levels in dB"""
+        return tuple(
+            round(
+                int.from_bytes(self._outputLeveldB100[i : i + 2], 'little', signed=True)
+                * 0.01,
+                1,
+            )
+            for i in range(0, len(self._outputLeveldB100), 2)
+        )[: self._kind.num_bus_levels]
+
+    @property
+    def levels(self) -> Levels:
+        """Returns strip and bus levels as a namedtuple"""
+        return Levels(strip=self.strip_levels, bus=self.bus_levels)
+
+    @property
+    def states(self) -> States:
+        """returns States object with processed strip and bus channel states"""
+        return States(
+            strip=tuple(
+                ChannelState(self._stripState[i : i + 4]) for i in range(0, 32, 4)
+            ),
+            bus=tuple(ChannelState(self._busState[i : i + 4]) for i in range(0, 32, 4)),
         )
 
-    @property
-    def samplerate(self) -> int:
-        """returns samplerate as an int"""
-        return int.from_bytes(self._samplerate, 'little')
-
-    @property
-    def inputlevels(self) -> tuple:
-        """returns the entire level array across all inputs for a kind"""
-        return self.strip_levels[0 : self._kind.num_strip_levels]
-
-    @property
-    def outputlevels(self) -> tuple:
-        """returns the entire level array across all outputs for a kind"""
-        return self.bus_levels[0 : self._kind.num_bus_levels]
-
-    @property
-    def stripstate(self) -> tuple:
-        """returns tuple of strip states accessable through bit modes"""
-        return tuple(self._stripState[i : i + 4] for i in range(0, 32, 4))
-
-    @property
-    def busstate(self) -> tuple:
-        """returns tuple of bus states accessable through bit modes"""
-        return tuple(self._busState[i : i + 4] for i in range(0, 32, 4))
-
-    """ 
-    these functions return an array of gainlayers[i] across all strips 
-    ie stripgainlayer1 = [strip[0].gainlayer[0], strip[1].gainlayer[0], strip[2].gainlayer[0]...]
-    """
-
     @property
     def gainlayers(self) -> tuple:
         """returns tuple of all strip gain layers as tuples"""
@@ -202,19 +297,35 @@ class VbanRtPacketNBS0(VbanRtPacket):
         )
 
     @property
-    def striplabels(self) -> tuple:
+    def labels(self) -> Labels:
-        """returns tuple of strip labels"""
+        """returns Labels namedtuple of strip and bus labels"""
-        return tuple(
-            self._stripLabelUTF8c60[i : i + 60].decode().split('\x00')[0]
-            for i in range(0, 480, 60)
-        )
 
-    @property
+        def _extract_labels_from_bytes(label_bytes: bytes) -> tuple[str, ...]:
-    def buslabels(self) -> tuple:
+            """Extract null-terminated UTF-8 labels from 60-byte chunks"""
-        """returns tuple of bus labels"""
+            labels = []
-        return tuple(
+            for i in range(0, len(label_bytes), 60):
-            self._busLabelUTF8c60[i : i + 60].decode().split('\x00')[0]
+                chunk = label_bytes[i : i + 60]
-            for i in range(0, 480, 60)
+                null_pos = chunk.find(b'\x00')
+                if null_pos == -1:
+                    try:
+                        label = chunk.decode('utf-8', errors='replace').rstrip('\x00')
+                    except UnicodeDecodeError:
+                        label = ''
+                else:
+                    try:
+                        label = (
+                            chunk[:null_pos].decode('utf-8', errors='replace')
+                            if null_pos > 0
+                            else ''
+                        )
+                    except UnicodeDecodeError:
+                        label = ''
+                labels.append(label)
+            return tuple(labels)
+
+        return Labels(
+            strip=_extract_labels_from_bytes(self._stripLabelUTF8c60),
+            bus=_extract_labels_from_bytes(self._busLabelUTF8c60),
         )
 
 
@@ -535,8 +646,8 @@ class VbanVMParamStrip:
 
 
 @dataclass
-class VbanRtPacketNBS1(VbanRtPacket):
+class VbanPacketNBS1(VbanPacket):
-    """Represents the body of a VBAN RT data packet with NBS 1"""
+    """Represents the body of a VBAN data packet with ident:1"""
 
     strips: tuple[VbanVMParamStrip, ...]
 

vban_cmd/strip.py

@@ -540,22 +540,11 @@ class StripLevel(IRemote):
     def getter(self):
         """Returns a tuple of level values for the channel."""
 
-        def fget(i):
-            return round((((1 << 16) - 1) - i) * -0.01, 1)
-
         if not self._remote.stopped() and self._remote.event.ldirty:
-            return tuple(
+            return self._remote.cache['strip_level'][self.range[0] : self.range[-1]]
-                fget(i)
+        return self.public_packets[NBS.zero].levels.strip[
-                for i in self._remote.cache['strip_level'][
+            self.range[0] : self.range[-1]
-                    self.range[0] : self.range[-1]
+        ]
-                ]
-            )
-        return tuple(
-            fget(i)
-            for i in self._remote._get_levels(self.public_packets[NBS.zero])[0][
-                self.range[0] : self.range[-1]
-            ]
-        )
 
     @property
     def identifier(self) -> str:

vban_cmd/util.py

@@ -15,13 +15,27 @@ def cache_bool(func, param):
     return wrapper
 
 
+def cache_int(func, param):
+    """Check cache for an int prop"""
+
+    def wrapper(*args, **kwargs):
+        self, *rem = args
+        if self._cmd(param) in self._remote.cache:
+            return self._remote.cache.pop(self._cmd(param))
+        if self._remote.sync:
+            self._remote.clear_dirty()
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
 def cache_string(func, param):
     """Check cache for a string prop"""
 
     def wrapper(*args, **kwargs):
         self, *rem = args
         if self._cmd(param) in self._remote.cache:
-            return self._remote.cache.pop(self._cmd(param))
+            return self._remote.cache.pop(self._cmd(param)).strip('"')
         if self._remote.sync:
             self._remote.clear_dirty()
         return func(*args, **kwargs)
@@ -72,16 +86,18 @@ def script(func):
 
 def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
     """
-    Generator function, accepts two tuples.
+    Generator function, accepts two tuples of dB values.
 
     Evaluates equality of each member in both tuples.
+    Only ignores changes when levels are very quiet (below -72 dB).
     """
 
     for a, b in zip(t0, t1):
-        if ((1 << 16) - 1) - b <= 7200:
+        # If both values are very quiet (below -72dB), ignore small changes
-            yield a == b
+        if a <= -72.0 and b <= -72.0:
+            yield a == b  # Both quiet, check if they're equal
         else:
-            yield True
+            yield a != b  # At least one has significant level, detect changes
 
 
 def deep_merge(dict1, dict2):

vban_cmd/vbancmd.py

@@ -5,7 +5,7 @@ import threading
 import time
 from pathlib import Path
 from queue import Queue
-from typing import Iterable, Union
+from typing import Union
 
 from .enums import NBS
 from .error import VBANCMDError
@@ -184,17 +184,6 @@ class VbanCmd(abc.ABC):
         while self.pdirty:
             time.sleep(self.DELAY)
 
-    def _get_levels(self, packet) -> Iterable:
-        """
-        returns both level arrays (strip_levels, bus_levels) BEFORE math conversion
-
-        strip levels in PREFADER mode.
-        """
-        return (
-            packet.inputlevels,
-            packet.outputlevels,
-        )
-
     def apply(self, data: dict):
         """
         Sets all parameters of a dict

vban_cmd/worker.py

@@ -10,10 +10,10 @@ from .packet import (
     VBAN_PROTOCOL_SERVICE,
     VBAN_SERVICE_RTPACKET,
     SubscribeHeader,
-    VbanRtPacket,
+    VbanPacket,
+    VbanPacketNBS0,
+    VbanPacketNBS1,
     VbanRtPacketHeader,
-    VbanRtPacketNBS0,
-    VbanRtPacketNBS1,
 )
 from .util import bump_framecounter
 
@@ -75,16 +75,16 @@ class Producer(threading.Thread):
         (
             self._remote.cache['strip_level'],
             self._remote.cache['bus_level'],
-        ) = self._remote._get_levels(self._remote.public_packets[NBS.zero])
+        ) = self._remote.public_packets[NBS.zero].levels
 
-    def _get_rt(self) -> VbanRtPacket:
+    def _get_rt(self) -> VbanPacket:
         """Attempt to fetch data packet until a valid one found"""
 
         while True:
             if resp := self._fetch_rt_packet():
                 return resp
 
-    def _fetch_rt_packet(self) -> VbanRtPacket | None:
+    def _fetch_rt_packet(self) -> VbanPacket | None:
         try:
             data, _ = self._remote.sock.recvfrom(2048)
             if len(data) < HEADER_SIZE:
@@ -99,12 +99,12 @@ class Producer(threading.Thread):
 
             match response_header.format_nbs:
                 case NBS.zero:
-                    return VbanRtPacketNBS0.from_bytes(
+                    return VbanPacketNBS0.from_bytes(
                         nbs=NBS.zero, kind=self._remote.kind, data=data
                     )
 
                 case NBS.one:
-                    return VbanRtPacketNBS1.from_bytes(
+                    return VbanPacketNBS1.from_bytes(
                         nbs=NBS.one, kind=self._remote.kind, data=data
                     )
             return None
@@ -140,7 +140,7 @@ class Producer(threading.Thread):
                 self.queue.put('pdirty')
             if self._remote.event.ldirty:
                 self.queue.put('ldirty')
-            time.sleep(self._remote.ratelimit)
+            # time.sleep(self._remote.ratelimit)
         self.logger.debug(f'terminating {self.name} thread')
         self.queue.put(None)
 
@@ -177,9 +177,6 @@ class Updater(threading.Thread):
                 (
                     self._remote.cache['strip_level'],
                     self._remote.cache['bus_level'],
-                ) = (
+                ) = self._remote.public_packets[NBS.zero].levels
-                    self._remote._public_packets[NBS.zero].inputlevels,
-                    self._remote._public_packets[NBS.zero].outputlevels,
-                )
                 self._remote.subject.notify(event)
         self.logger.debug(f'terminating {self.name} thread')