Support event subscriptions

Author: HEnquist

camilladsp/camillaws.py

@@ -4,7 +4,7 @@
 This module contains the websocket connection class.
 """
 
-from typing import Dict, Tuple, Optional, Union
+from typing import Any, Callable, Dict, Tuple, Optional, Union
 from threading import Lock
 import json
 from websocket import create_connection, WebSocket  # type: ignore
@@ -79,6 +79,21 @@ def _handle_reply(self, command: str, rawreply: Union[str, bytes]):
         except json.JSONDecodeError as err:
             raise IOError(f"Invalid response received: {rawreply!r}") from err
 
+    def _handle_event_reply(self, event_name: str, rawreply: Union[str, bytes]):
+        try:
+            reply = json.loads(rawreply)
+            if event_name not in reply:
+                return None
+            response_data = reply[event_name]
+            result = response_data["result"]
+            state, message = self._handle_result(result)
+            value = response_data.get("value")
+            if state == "Ok":
+                return value
+            _raise_error(state, message, value)
+        except json.JSONDecodeError as err:
+            raise IOError(f"Invalid response received: {rawreply!r}") from err
+
     def _handle_result(
         self, result: Union[str, Dict[str, str]]
     ) -> Tuple[str, Optional[str]]:
@@ -105,6 +120,73 @@ def connect(self):
             self._ws = None
             raise
 
+    def subscribe_events(
+        self,
+        command: str,
+        event_name: str,
+        callback: Callable[[Any], Optional[bool]],
+        arg=None,
+    ):
+        """
+        Start a subscription and call `callback` for each incoming event value.
+
+        This method blocks until the callback returns `False`, or an exception
+        is raised. A `StopSubscription` command is sent before returning.
+
+        Args:
+            command (str): Subscription command to send.
+            event_name (str): Name of event messages to listen for.
+            callback: Function called with each event payload.
+            arg: Optional parameter to send with the subscription command.
+        """
+        if not callable(callback):
+            raise TypeError("callback must be callable")
+        if self._ws is None:
+            raise IOError("Not connected to CamillaDSP")
+
+        subscribed = False
+        try:
+            with self._lock:
+                if arg is not None:
+                    query = json.dumps({command: arg})
+                else:
+                    query = json.dumps(command)
+                self._ws.send(query)
+                rawrepl = self._ws.recv()
+        except Exception as err:
+            self._ws = None
+            raise IOError("Lost connection to CamillaDSP") from err
+
+        self._handle_reply(command, rawrepl)
+        subscribed = True
+
+        try:
+            while True:
+                if self._ws is None:
+                    raise IOError("Lost connection to CamillaDSP")
+                try:
+                    with self._lock:
+                        raw_event = self._ws.recv()
+                except Exception as err:
+                    self._ws = None
+                    raise IOError("Lost connection to CamillaDSP") from err
+
+                event_data = self._handle_event_reply(event_name, raw_event)
+                if event_data is None:
+                    continue
+                should_continue = callback(event_data)
+                if should_continue is False:
+                    break
+        finally:
+            if subscribed and self._ws is not None:
+                try:
+                    with self._lock:
+                        self._ws.send(json.dumps("StopSubscription"))
+                        rawrepl = self._ws.recv()
+                    self._handle_reply("StopSubscription", rawrepl)
+                except Exception:
+                    self._ws = None
+
     def disconnect(self):
         """
         Close the connection to the websocket.

camilladsp/levels.py

@@ -4,7 +4,7 @@
 This module contains commands for reading levels.
 """
 
-from typing import Dict, List
+from typing import Any, Callable, Dict, List, Optional
 import math
 
 from .commandgroup import _CommandGroup
@@ -201,3 +201,29 @@ def labels(self):
         """
         labels = self.client.query("GetChannelLabels")
         return labels
+
+    def subscribe_signal_levels(
+        self,
+        callback: Callable[[Dict[str, Any]], Optional[bool]],
+        side: str = "both",
+    ):
+        """
+        Subscribe to signal 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 `side`, `rms`, and `peak`.
+            side (str): Which side to subscribe to. One of
+                `capture`, `playback`, or `both`.
+        """
+        if side not in ("capture", "playback", "both"):
+            raise ValueError("side must be one of: capture, playback, both")
+
+        self.client.subscribe_events(
+            command="SubscribeSignalLevels",
+            arg=side,
+            event_name="SignalLevelsEvent",
+            callback=callback,
+        )

tests/test_camillaws.py

@@ -408,3 +408,65 @@ def test_queries_customreplies(camilla_mockquery):
     camilla_mockquery.query.assert_called_with(
         "AdjustFaderVolume", arg=(0, (-5.0, -150.0, 3.0))
     )
+
+
+def test_subscribe_signal_levels(camilla_mockquery):
+    callback = MagicMock(return_value=False)
+    camilla_mockquery.subscribe_events = MagicMock()
+
+    camilla_mockquery.levels.subscribe_signal_levels(callback, side="playback")
+
+    camilla_mockquery.subscribe_events.assert_called_with(
+        command="SubscribeSignalLevels",
+        arg="playback",
+        event_name="SignalLevelsEvent",
+        callback=callback,
+    )
+
+
+def test_subscribe_events(camilla_mockws):
+    camilla_mockws.connect()
+    sent = []
+    replies = iter(
+        [
+            json.dumps({"SubscribeSignalLevels": {"result": "Ok"}}),
+            json.dumps(
+                {
+                    "SignalLevelsEvent": {
+                        "result": "Ok",
+                        "value": {
+                            "side": "capture",
+                            "rms": [-58.1, -57.6],
+                            "peak": [-39.4, -38.9],
+                        },
+                    }
+                }
+            ),
+            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.subscribe_events(
+        command="SubscribeSignalLevels",
+        arg="capture",
+        event_name="SignalLevelsEvent",
+        callback=on_event,
+    )
+
+    assert sent == [json.dumps({"SubscribeSignalLevels": "capture"}), '"StopSubscription"']
+    assert events == [
+        {
+            "side": "capture",
+            "rms": [-58.1, -57.6],
+            "peak": [-39.4, -38.9],
+        }
+    ]