testing around gain management in camilla + bug fixes

ezbeq/camilladsp.py

@@ -13,6 +13,8 @@
 from ezbeq.catalogue import CatalogueProvider, CatalogueEntry
 from ezbeq.device import SlotState, DeviceState, PersistentDevice
 
+BEQ_FILTER_NAME_PATTERN = r'^BEQ_(Gain_\d+|\d+_[a-zA-Z0-9]+)$'
+
 SLOT_ID = 'CamillaDSP'
 
 logger = logging.getLogger('ezbeq.camilladsp')
@@ -153,7 +155,7 @@ def __do_it() -> bool:
                                 if match:
                                     mv = 0.0
                                     if 'gains' in slot and slot['gains']:
-                                        mv = float(slot['gains'][0])
+                                        mv = float(slot['gains'][0]['value'])
                                     self.load_filter(SLOT_ID, match, mv)
                                     any_update = True
                             else:
@@ -162,13 +164,30 @@ def __do_it() -> bool:
                         elif 'gains' in slot or 'mutes' in slot:
                             merged = defaultdict(dict)
                             for g in slot.get('gains', []):
-                                merged[int(g['id'])]['gain'] = g['value']
+                                c_id = int(g['id'])
+                                if c_id in self.__beq_channels:
+                                    merged[c_id]['gain'] = g['value']
+                                else:
+                                    raise ValueError(f'Invalid channel id for gain setting {c_id}')
                             for g in slot.get('mutes', []):
-                                merged[int(g['id'])]['mute'] = g['value']
+                                c_id = int(g['id'])
+                                if c_id in self.__beq_channels:
+                                    merged[c_id]['mute'] = g['value']
+                                else:
+                                    raise ValueError(f'Invalid channel id for gain setting {c_id}')
 
                             def completed(success: bool):
                                 logger.log(logging.INFO if success else logging.WARNING,
                                            f'Completed gain update {success}')
+                                if success:
+                                    css = self._current_state.slot
+                                    for k, v in merged.items():
+                                        if 'gain' in v:
+                                            to_update = next(g for g in css.gains if g['id'] == k)
+                                            to_update['value'] = v['gain']
+                                        if 'mute' in v:
+                                            to_update = next(g for g in css.mutes if g['id'] == k)
+                                            to_update['value'] = v['mute']
 
                             self.__update_channel_levels(merged,
                                                          lambda b: self._hydrate_cache_broadcast(lambda: completed(b)))
@@ -196,7 +215,10 @@ def __update_channel_levels(self, values: Dict[int, dict], on_complete: Callable
 
     def __send_filter(self, entry: Optional[CatalogueEntry], mv_adjust: float, on_complete: Callable[[bool], None]):
         if self.__config_updater is None:
-            logger.info(f"Sending {len(entry.filters)} filters for {entry.formatted_title}")
+            if entry:
+                logger.info(f"Sending {len(entry.filters)} filters for {entry.formatted_title}")
+            else:
+                logger.info(f"Clearing filters")
             self.__config_updater = LoadConfig(entry, self.__beq_channels, mv_adjust, self.ws_server, self.__ws_client,
                                                on_complete)
             self.__config_updater.send_get_config()
@@ -232,6 +254,10 @@ def __do_load_filter(self, entry: Optional[CatalogueEntry], mv_adjust: float = 0
             def completed(success: bool):
                 if success:
                     self._current_state.slot.last = entry.formatted_title if entry else 'Empty'
+                    for g in self._current_state.slot.gains:
+                        g['value'] = mv_adjust
+                    for g in self._current_state.slot.mutes:
+                        g['value'] = False
                 else:
                     self._current_state.slot.last = 'ERROR'
 
@@ -476,22 +502,22 @@ def create_cfg_for_entry(entry: Optional[CatalogueEntry], base_cfg: dict, beq_ch
                          mute: bool) -> dict:
     from copy import deepcopy
     new_cfg = deepcopy(base_cfg)
-    if 'filters' not in new_cfg:
-        new_cfg['filters'] = {}
     beq_filters = entry.filters if entry else []
-    filters = {k: v for k, v in new_cfg['filters'].items() if not k.startswith('BEQ_')}
+    filters = {k: v for k, v in new_cfg.get('filters', {}).items() if not k.startswith('BEQ_')}
+    new_cfg['filters'] = filters
     filter_names = []
-    for c in beq_channels:
-        name = f'BEQ_Gain_{c}'
-        filter_names.append(name)
-        filters[name] = {
-            'type': 'Gain',
-            'parameters': {
-                'gain': mv_adjust,
-                'inverted': False,
-                'mute': mute
+    if entry or not math.isclose(mv_adjust, 0.0) or mute is True:
+        for c in beq_channels:
+            name = f'BEQ_Gain_{c}'
+            filter_names.append(name)
+            filters[name] = {
+                'type': 'Gain',
+                'parameters': {
+                    'gain': mv_adjust,
+                    'inverted': False,
+                    'mute': mute
+                }
             }
-        }
     for i, peq in enumerate(beq_filters):
         name = f'BEQ_{i}_{entry.digest}'
         filters[name] = {
@@ -516,7 +542,8 @@ def create_cfg_for_entry(entry: Optional[CatalogueEntry], base_cfg: dict, beq_ch
                 existing = empty_filter
                 pipeline.append(existing)
             import re
-            existing['names'] = [n for n in existing['names'] if re.match(r'^BEQ_.*_\d$', n) is None] + filter_names
+            new_names = [n for n in existing['names'] if re.match(BEQ_FILTER_NAME_PATTERN, n) is None] + filter_names
+            existing['names'] = new_names
     else:
         raise ValueError(f'Unable to load BEQ, dsp config has no pipeline declared')
     return new_cfg
@@ -562,7 +589,7 @@ def create_cfg_for_gains(values: Dict[int, dict], base_cfg: dict) -> dict:
                     pipeline.append(existing)
                 import re
                 if gain_filter_name not in existing['names']:
-                    insert_at = next((i for i, n in enumerate(existing['names']) if re.match(r'^BEQ_.*_\d$', n) is not None), -1)
+                    insert_at = next((i for i, n in enumerate(existing['names']) if re.match(BEQ_FILTER_NAME_PATTERN, n) is not None), -1)
                     if insert_at == -1:
                         existing['names'].append(gain_filter_name)
                     else:

tests/conftest.py

@@ -149,7 +149,7 @@ def minidsp_shd_client(minidsp_shd_app):
 @pytest.fixture
 def camilladsp_app(httpserver: HTTPServer, tmp_path):
     """Create and configure a new app instance for each test."""
-    cfg = CamillaDspSpyConfig(httpserver.host, httpserver.port, tmp_path, channels=[4])
+    cfg = CamillaDspSpyConfig(httpserver.host, httpserver.port, tmp_path, channels=[1])
     app, ws = main.create_app(cfg, cfg.msg_spy)
     yield app
 
@@ -342,7 +342,7 @@ def send(self, msg: str):
             self.__cmd_queue.put(payload)
             self.__mv = payload['SetVolume']
         elif 'SetConfigJson' in payload:
-            self.__cmd_queue.put('SetConfigJson')
+            self.__cmd_queue.put(payload)
             self.__current_cfg = json.loads(payload['SetConfigJson'])
             self.listener.on_set_config('Ok')
         elif 'SetUpdateInterval' in payload:

tests/test_camilladsp_api.py

@@ -3,7 +3,7 @@
 import pytest
 from busypie import wait, SECOND, MILLISECOND
 
-from conftest import CamillaDspSpyConfig, CamillaDspSpy
+from conftest import CamillaDspSpyConfig
 
 
 @pytest.mark.parametrize("mute_op", [True, False])
@@ -49,7 +49,8 @@ def volume_changed():
         assert cmds[0] == {'SetVolume': volume}
         assert cmds[1] == 'GetVolume'
 
-    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('volume_changed').until_asserted(volume_changed)
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('volume_changed').until_asserted(
+        volume_changed)
 
     def ui_updated():
         device_states = take_device_states(config)
@@ -58,6 +59,160 @@ def ui_updated():
     wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('ui_updated').until_asserted(ui_updated)
 
 
+def test_load_known_entry_and_then_clear(camilladsp_client, camilladsp_app):
+    config: CamillaDspSpyConfig = camilladsp_app.config['APP_CONFIG']
+    assert isinstance(config, CamillaDspSpyConfig)
+    ensure_inited(config)
+
+    r = camilladsp_client.put(f"/api/1/devices/master/filter/CamillaDSP", data=json.dumps({'entryId': '123456_0'}),
+                              content_type='application/json')
+    assert r.status_code == 200
+
+    def beq_loaded():
+        cmds = config.spy.take_commands()
+        assert len(cmds) == 3
+        assert cmds[0] == 'GetConfigJson'
+        assert isinstance(cmds[1], dict)
+        assert 'SetConfigJson' in cmds[1]
+        beq_is_loaded(json.loads(cmds[1]['SetConfigJson']), 0.0)
+        assert cmds[2] == 'Reload'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('beq_loaded').until_asserted(beq_loaded)
+
+    def entry_is_shown():
+        device_states = take_device_states(config)
+        assert device_states[-1]['slots'][0]['last'] == 'Alien Resurrection'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('entry_is_shown').until_asserted(
+        entry_is_shown)
+
+    r = camilladsp_client.delete(f"/api/1/devices/master/filter/CamillaDSP")
+    assert r.status_code == 200
+
+    def beq_unloaded():
+        cmds = config.spy.take_commands()
+        assert len(cmds) == 3
+        assert cmds[0] == 'GetConfigJson'
+        assert isinstance(cmds[1], dict)
+        assert 'SetConfigJson' in cmds[1]
+        beq_is_unloaded(json.loads(cmds[1]['SetConfigJson']))
+        assert cmds[2] == 'Reload'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('beq_unloaded').until_asserted(
+        beq_unloaded)
+
+    def entry_is_removed():
+        device_states = take_device_states(config)
+        assert device_states[-1]['slots'][0]['last'] == 'Empty'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('entry_is_removed').until_asserted(
+        entry_is_removed)
+
+
+def test_load_known_entry_with_gain_and_then_clear(camilladsp_client, camilladsp_app):
+    config: CamillaDspSpyConfig = camilladsp_app.config['APP_CONFIG']
+    assert isinstance(config, CamillaDspSpyConfig)
+    ensure_inited(config)
+
+    payload = {"slots": [{"id": "CamillaDSP", "gains": [{"id": "0", "value": -1.5}], "mutes": [], "entry": "123456_0"}]}
+    r = camilladsp_client.patch(f"/api/2/devices/master", data=json.dumps(payload), content_type='application/json')
+    assert r.status_code == 200
+
+    def beq_loaded():
+        cmds = config.spy.take_commands()
+        assert len(cmds) == 3
+        assert cmds[0] == 'GetConfigJson'
+        assert isinstance(cmds[1], dict)
+        assert 'SetConfigJson' in cmds[1]
+        beq_is_loaded(json.loads(cmds[1]['SetConfigJson']), -1.5)
+        assert cmds[2] == 'Reload'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('beq_loaded').until_asserted(beq_loaded)
+
+    def entry_is_shown():
+        device_states = take_device_states(config)
+        assert device_states[-1]['slots'][0]['last'] == 'Alien Resurrection'
+        assert device_states[-1]['slots'][0]['gains'] == [{'id': 1, 'value': -1.5}]
+        assert device_states[-1]['slots'][0]['mutes'] == [{'id': 1, 'value': False}]
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('entry_is_shown').until_asserted(
+        entry_is_shown)
+
+    r = camilladsp_client.delete(f"/api/1/devices/master/filter/CamillaDSP")
+    assert r.status_code == 200
+
+    def beq_unloaded():
+        cmds = config.spy.take_commands()
+        assert len(cmds) == 3
+        assert cmds[0] == 'GetConfigJson'
+        assert isinstance(cmds[1], dict)
+        assert 'SetConfigJson' in cmds[1]
+        beq_is_unloaded(json.loads(cmds[1]['SetConfigJson']))
+        assert cmds[2] == 'Reload'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('beq_unloaded').until_asserted(
+        beq_unloaded)
+
+    def entry_is_removed():
+        device_states = take_device_states(config)
+        assert device_states[-1]['slots'][0]['last'] == 'Empty'
+        assert device_states[-1]['slots'][0]['gains'] == [{'id': 1, 'value': 0.0}]
+        assert device_states[-1]['slots'][0]['mutes'] == [{'id': 1, 'value': False}]
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('entry_is_removed').until_asserted(
+        entry_is_removed)
+
+
+def test_input_gain_and_mute(camilladsp_client, camilladsp_app):
+    config: CamillaDspSpyConfig = camilladsp_app.config['APP_CONFIG']
+    assert isinstance(config, CamillaDspSpyConfig)
+    ensure_inited(config)
+
+    payload = {"slots": [{"id": "CamillaDSP", "gains": [{"id": "1", "value": -1.5}], "mutes": [{"id": "1", "value": True}]}]}
+    r = camilladsp_client.patch(f"/api/2/devices/master", data=json.dumps(payload), content_type='application/json')
+    assert r.status_code == 200
+
+    def gain_changed():
+        cmds = config.spy.take_commands()
+        assert len(cmds) == 3
+        assert cmds[0] == 'GetConfigJson'
+        assert isinstance(cmds[1], dict)
+        assert 'SetConfigJson' in cmds[1]
+        new_config = json.loads(cmds[1]['SetConfigJson'])
+        assert next(f for f in new_config['pipeline'] if f['type'] == 'Filter' and f['channel'] == 1)['names'] == [
+            "vol",
+            "BEQ_Gain_1"
+        ]
+        new_filters = new_config['filters']
+        assert 'vol' in new_filters
+        assert 'BEQ_Gain_1' in new_filters
+        assert new_filters['BEQ_Gain_1'] == {'parameters': {'gain': -1.5, 'inverted': False, 'mute': True},
+                                             'type': 'Gain'}
+        assert len(list(new_filters.keys())) == 2
+        assert cmds[2] == 'Reload'
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('beq_loaded').until_asserted(gain_changed)
+
+    def entry_is_shown():
+        device_states = take_device_states(config)
+        assert device_states[-1]['slots'][0]['last'] == 'Empty'
+        assert device_states[-1]['slots'][0]['gains'] == [{'id': 1, 'value': -1.5}]
+        assert device_states[-1]['slots'][0]['mutes'] == [{'id': 1, 'value': True}]
+
+    wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('entry_is_shown').until_asserted(
+        entry_is_shown)
+
+
+def test_input_gain_and_mute_on_unsupported_channel(camilladsp_client, camilladsp_app):
+    config: CamillaDspSpyConfig = camilladsp_app.config['APP_CONFIG']
+    assert isinstance(config, CamillaDspSpyConfig)
+    ensure_inited(config)
+
+    payload = {"slots": [{"id": "CamillaDSP", "gains": [{"id": "0", "value": -1.5}], "mutes": [{"id": "0", "value": True}]}]}
+    r = camilladsp_client.patch(f"/api/2/devices/master", data=json.dumps(payload), content_type='application/json')
+    assert r.status_code == 500
+
+
 def take_device_states(config):
     msgs = config.msg_spy.take_messages()
     assert len(msgs) != 0
@@ -80,3 +235,49 @@ def inited():
 
     wait().at_most(2 * SECOND).poll_interval(10 * MILLISECOND).with_description('inited').until_asserted(inited)
     config.msg_spy.take_messages()
+
+
+def beq_is_loaded(new_config, expected_gain):
+    assert next(f for f in new_config['pipeline'] if f['type'] == 'Filter' and f['channel'] == 1)['names'] == [
+        "vol",
+        "BEQ_Gain_1",
+        "BEQ_0_abcdefghijklm",
+        "BEQ_1_abcdefghijklm",
+        "BEQ_2_abcdefghijklm",
+        "BEQ_3_abcdefghijklm",
+        "BEQ_4_abcdefghijklm"
+    ]
+    new_filters = new_config['filters']
+    assert 'vol' in new_filters
+    assert 'BEQ_Gain_1' in new_filters
+    assert new_filters['BEQ_Gain_1'] == {'parameters': {'gain': expected_gain, 'inverted': False, 'mute': False},
+                                         'type': 'Gain'}
+    assert 'BEQ_0_abcdefghijklm' in new_filters
+    assert new_filters['BEQ_0_abcdefghijklm'] == {
+        'parameters': {'freq': 33.0, 'gain': 5.0, 'q': 0.9, 'type': 'Lowshelf'}, 'type': 'Biquad'}
+    assert 'BEQ_1_abcdefghijklm' in new_filters
+    assert new_filters['BEQ_1_abcdefghijklm'] == {
+        'parameters': {'freq': 33.0, 'gain': 5.0, 'q': 0.9, 'type': 'Lowshelf'}, 'type': 'Biquad'}
+    assert 'BEQ_2_abcdefghijklm' in new_filters
+    assert new_filters['BEQ_2_abcdefghijklm'] == {
+        'parameters': {'freq': 33.0, 'gain': 5.0, 'q': 0.9, 'type': 'Lowshelf'}, 'type': 'Biquad'}
+    assert 'BEQ_3_abcdefghijklm' in new_filters
+    assert new_filters['BEQ_3_abcdefghijklm'] == {
+        'parameters': {'freq': 33.0, 'gain': 5.0, 'q': 0.9, 'type': 'Lowshelf'}, 'type': 'Biquad'}
+    assert 'BEQ_4_abcdefghijklm' in new_filters
+    assert new_filters['BEQ_4_abcdefghijklm'] == {
+        'parameters': {'freq': 33.0, 'gain': 5.0, 'q': 0.9, 'type': 'Lowshelf'}, 'type': 'Biquad'}
+    assert len(list(new_filters.keys())) == 7
+
+
+def beq_is_unloaded(new_config):
+    assert next(f for f in new_config['pipeline'] if f['type'] == 'Filter' and f['channel'] == 1)['names'] == ["vol"]
+    new_filters = new_config['filters']
+    assert 'vol' in new_filters
+    assert 'BEQ_Gain_1' not in new_filters
+    assert 'BEQ_0_abcdefghijklm' not in new_filters
+    assert 'BEQ_1_abcdefghijklm' not in new_filters
+    assert 'BEQ_2_abcdefghijklm' not in new_filters
+    assert 'BEQ_3_abcdefghijklm' not in new_filters
+    assert 'BEQ_4_abcdefghijklm' not in new_filters
+    assert len(list(new_filters.keys())) == 1