Add VU-meter command

Author: HEnquist

README.md

@@ -15,6 +15,12 @@ Read the playback signal level continuously and print in the terminal, until sto
 python read_rms.py 1234
 ```
 
+## subscribe_events
+Subscribe to capture signal level events and state change events, and print incoming events in the terminal until stopped by Ctrl+c.
+```sh
+python subscribe_events.py 1234
+```
+
 ## get_config
 Read the configuration and print some parameters. 
 ```sh

camilladsp/general.py

@@ -4,7 +4,7 @@
 This module contains commands of general nature.
 """
 
-from typing import Tuple, List, Optional
+from typing import Any, Callable, List, Optional, Tuple
 
 from .commandgroup import _CommandGroup
 from .datastructures import (
@@ -32,6 +32,22 @@ def state(self) -> Optional[ProcessingState]:
         state = self.client.query("GetState")
         return _state_from_string(state)
 
+    def subscribe_state(self, callback: Callable[[Any], Optional[bool]]):
+        """
+        Subscribe to state change events and call `callback` for each event.
+
+        This method blocks until `callback` returns `False`.
+
+        Args:
+            callback: Function that receives event payloads.
+                Typical payload keys are `state` and `stop_reason`.
+        """
+        self.client.subscribe_events(
+            command="SubscribeState",
+            event_name="StateEvent",
+            callback=callback,
+        )
+
     def stop_reason(self) -> StopReason:
         """
         Get reason why processing stopped.

camilladsp/levels.py

@@ -227,3 +227,34 @@ def subscribe_signal_levels(
             event_name="SignalLevelsEvent",
             callback=callback,
         )
+
+    def subscribe_vu_levels(
+        self,
+        callback: Callable[[Dict[str, Any]], Optional[bool]],
+        max_rate: float,
+        attack: float,
+        release: float,
+    ):
+        """
+        Subscribe to VU meter level events and call `callback` for each event.
+
+        This method blocks until `callback` returns `False`.
+
+        Args:
+            callback: Function that receives event payloads.
+                Typical payload keys are `playback_rms`, `playback_peak`,
+                `capture_rms`, and `capture_peak`.
+            max_rate (float): Maximum event rate in Hz. Use 0 to disable capping.
+            attack (float): Attack time constant in milliseconds.
+            release (float): Release time constant in milliseconds.
+        """
+        self.client.subscribe_events(
+            command="SubscribeVuLevels",
+            arg={
+                "max_rate": float(max_rate),
+                "attack": float(attack),
+                "release": float(release),
+            },
+            event_name="VuLevelsEvent",
+            callback=callback,
+        )

docs/general.md

@@ -2,6 +2,8 @@
 This class is accessed via the `general` property on a `CamillaClient` instance.
 
 It provides the basic methods such as starting and stopping processing.
+It also includes `subscribe_state` for listening to processing state changes.
+Subscription methods block while events are being received.
 
 ##  class: `General`
 ::: camilladsp.general.General

docs/index.md

@@ -27,6 +27,12 @@ state = client.general.state()
 capture_levels = client.levels.capture_rms()
 ```
 
+Subscriptions are also available for streaming updates. For example,
+`client.levels.subscribe_signal_levels(...)` can be used to receive signal
+level events, and `client.general.subscribe_state(...)` can be used to
+receive processing state changes. Subscription calls block while listening for
+events, so use a separate client connection for each concurrent subscription.
+
 ## Command group classes
 |      Class   | Via property | Description |
 |--------------|----------|-------------|
@@ -44,6 +50,8 @@ capture_levels = client.levels.capture_rms()
 ### [General][camilladsp.general.General]
 These commands are accessed via the [general][camilladsp.CamillaClient.general]
 property of a [CamillaClient][camilladsp.CamillaClient] instance.
+This group also includes `subscribe_state`, which listens for processing state
+change events.
 ::: camilladsp.general.General
     options:
       show_bases: false
@@ -84,6 +92,8 @@ property of a [CamillaClient][camilladsp.CamillaClient] instance.
 ### [Levels][camilladsp.levels.Levels]
 These commands are accessed via the [levels][camilladsp.CamillaClient.levels]
 property of a [CamillaClient][camilladsp.CamillaClient] instance.
+This group also includes `subscribe_signal_levels`, which listens for signal
+level events on the selected side.
 ::: camilladsp.levels.Levels
     options:
       show_bases: false

docs/levels.md

@@ -2,6 +2,10 @@
 This class is accessed via the `levels` property on a `CamillaClient` instance.
 
 It provides methods for reading signal levels.
+It also includes `subscribe_signal_levels` for listening to signal level
+events from the websocket server, and `subscribe_vu_levels` for listening to
+pre-smoothed, rate-limited VU meter events. Subscription methods block while
+events are being received.
 
 ##  class: `Levels`
 ::: camilladsp.levels.Levels

examples/subscribe_events/subscribe_events.py

@@ -0,0 +1,58 @@
+# subscribe to events
+from camilladsp import CamillaClient
+import sys
+import threading
+import time
+
+
+def print_capture_levels(event):
+    print(f"capture levels: {event}")
+    return True
+
+
+def print_state(event):
+    print(f"state: {event}")
+    return True
+
+
+def subscribe_capture_levels(port):
+    cdsp = CamillaClient("127.0.0.1", port)
+    cdsp.connect()
+    try:
+        cdsp.levels.subscribe_signal_levels(print_capture_levels, side="capture")
+    finally:
+        cdsp.disconnect()
+
+
+def subscribe_state(port):
+    cdsp = CamillaClient("127.0.0.1", port)
+    cdsp.connect()
+    try:
+        cdsp.general.subscribe_state(print_state)
+    finally:
+        cdsp.disconnect()
+
+
+try:
+    port = int(sys.argv[1])
+except:
+    print("Usage: start CamillaDSP with the socketserver enabled:")
+    print("> camilladsp -p4321 yourconfig.yml")
+    print("Then run this script to print capture signal levels and state events.")
+    print("> python subscribe_events.py 4321")
+    sys.exit()
+
+
+level_thread = threading.Thread(target=subscribe_capture_levels, args=(port,), daemon=True)
+state_thread = threading.Thread(target=subscribe_state, args=(port,), daemon=True)
+
+level_thread.start()
+state_thread.start()
+
+print("Listening for capture signal level and state events, press Ctrl+C to stop")
+
+try:
+    while True:
+        time.sleep(1)
+except KeyboardInterrupt:
+    pass

tests/test_camillaws.py

@@ -164,6 +164,56 @@ def test_connect(camilla_mockws):
         camilla_mockws.general.state()
     camilla_mockws.connect()
     assert camilla_mockws.is_connected()
+    def test_subscribe_vu_events(camilla_mockws):
+        camilla_mockws.connect()
+        sent = []
+        replies = iter(
+            [
+                json.dumps({"SubscribeVuLevels": {"result": "Ok"}}),
+                json.dumps(
+                    {
+                        "VuLevelsEvent": {
+                            "result": "Ok",
+                            "value": {
+                                "playback_rms": [-20.0, -21.0],
+                                "playback_peak": [-10.0, -11.0],
+                                "capture_rms": [-30.0, -31.0],
+                                "capture_peak": [-12.0, -13.0],
+                            },
+                        }
+                    }
+                ),
+                json.dumps({"StopSubscription": {"result": "Ok"}}),
+            ]
+        )
+
+        camilla_mockws.mockconnection.send = MagicMock(side_effect=lambda msg: sent.append(msg))
+        camilla_mockws.mockconnection.recv = MagicMock(side_effect=lambda: next(replies))
+
+        events = []
+
+        def on_event(event_data):
+            events.append(event_data)
+            return False
+
+        camilla_mockws.levels.subscribe_vu_levels(
+            on_event, max_rate=30, attack=10, release=200
+        )
+
+        assert events == [
+            {
+                "playback_rms": [-20.0, -21.0],
+                "playback_peak": [-10.0, -11.0],
+                "capture_rms": [-30.0, -31.0],
+                "capture_peak": [-12.0, -13.0],
+            }
+        ]
+        assert sent == [
+            json.dumps(
+                {"SubscribeVuLevels": {"max_rate": 30.0, "attack": 10.0, "release": 200.0}}
+            ),
+            json.dumps("StopSubscription"),
+        ]
     assert camilla_mockws.general.state() == camilladsp.ProcessingState.INACTIVE
     assert camilla_mockws.versions.camilladsp() == ("0", "3", "2")
     assert camilla_mockws.versions.library() == tuple(camilladsp.VERSION.split("."))
@@ -424,6 +474,38 @@ def test_subscribe_signal_levels(camilla_mockquery):
     )
 
 
+def test_subscribe_vu_levels(camilla_mockquery):
+    callback = MagicMock(return_value=False)
+    camilla_mockquery.subscribe_events = MagicMock()
+
+    camilla_mockquery.levels.subscribe_vu_levels(
+        callback, max_rate=30, attack=10, release=200
+    )
+
+    camilla_mockquery.subscribe_events.assert_called_with(
+        command="SubscribeVuLevels",
+        arg={"max_rate": 30.0, "attack": 10.0, "release": 200.0},
+        event_name="VuLevelsEvent",
+        callback=callback,
+    )
+
+
+def test_subscribe_vu_levels(camilla_mockquery):
+    callback = MagicMock(return_value=False)
+    camilla_mockquery.subscribe_events = MagicMock()
+
+    camilla_mockquery.levels.subscribe_vu_levels(
+        callback, max_rate=30, attack=10, release=200
+    )
+
+    camilla_mockquery.subscribe_events.assert_called_with(
+        command="SubscribeVuLevels",
+        arg={"max_rate": 30.0, "attack": 10.0, "release": 200.0},
+        event_name="VuLevelsEvent",
+        callback=callback,
+    )
+
+
 def test_subscribe_events(camilla_mockws):
     camilla_mockws.connect()
     sent = []
@@ -470,3 +552,55 @@ def on_event(event_data):
             "peak": [-39.4, -38.9],
         }
     ]
+
+
+def test_subscribe_vu_events(camilla_mockws):
+    camilla_mockws.connect()
+    sent = []
+    replies = iter(
+        [
+            json.dumps({"SubscribeVuLevels": {"result": "Ok"}}),
+            json.dumps(
+                {
+                    "VuLevelsEvent": {
+                        "result": "Ok",
+                        "value": {
+                            "playback_rms": [-20.0, -21.0],
+                            "playback_peak": [-10.0, -11.0],
+                            "capture_rms": [-30.0, -31.0],
+                            "capture_peak": [-12.0, -13.0],
+                        },
+                    }
+                }
+            ),
+            json.dumps({"StopSubscription": {"result": "Ok"}}),
+        ]
+    )
+
+    camilla_mockws.mockconnection.send = MagicMock(side_effect=lambda msg: sent.append(msg))
+    camilla_mockws.mockconnection.recv = MagicMock(side_effect=lambda: next(replies))
+
+    events = []
+
+    def on_event(event_data):
+        events.append(event_data)
+        return False
+
+    camilla_mockws.levels.subscribe_vu_levels(
+        on_event, max_rate=30, attack=10, release=200
+    )
+
+    assert events == [
+        {
+            "playback_rms": [-20.0, -21.0],
+            "playback_peak": [-10.0, -11.0],
+            "capture_rms": [-30.0, -31.0],
+            "capture_peak": [-12.0, -13.0],
+        }
+    ]
+    assert sent == [
+        json.dumps(
+            {"SubscribeVuLevels": {"max_rate": 30.0, "attack": 10.0, "release": 200.0}}
+        ),
+        json.dumps("StopSubscription"),
+    ]