Fix output presets on PipeWire port changes

Author: bhack

src/mini_eq/deps.py

@@ -19,6 +19,7 @@
     "Device.enum_all_params",
     "Device.enum_params",
     "Device.new",
+    "Device.subscribe_params",
     "Param.new_props_controls",
     "RouteInfo.new_from_param",
     "Stream.set_pipewire_property",

src/mini_eq/pipewire_routes.py

@@ -94,9 +94,13 @@ def output_route_for_sink(self, sink) -> PipeWireOutputRoute | None:
             return None
 
         if sink.card_profile_device > 0:
-            for route in output_routes:
-                if route.route_device == sink.card_profile_device:
-                    return route
+            matching_device_routes = [
+                route for route in output_routes if route.route_device == sink.card_profile_device
+            ]
+            if len(matching_device_routes) == 1:
+                return matching_device_routes[0]
+            if len(matching_device_routes) > 1:
+                return None
 
         if len(output_routes) == 1:
             return output_routes[0]

src/mini_eq/window.py

@@ -71,6 +71,57 @@ def compact_warning_title(message: str) -> str:
     return COMPACT_WARNING_TITLES.get(message, message)
 
 
+def output_preset_target_identity(owner: object, fallback: str | None) -> str | None:
+    target = None
+    target_factory = getattr(owner, "output_preset_target", None)
+    if callable(target_factory):
+        try:
+            target = target_factory()
+        except Exception:
+            target = None
+
+    if target is not None:
+        try:
+            link_key = str(getattr(target, "link_key", "") or "").strip()
+        except Exception:
+            link_key = ""
+        if link_key:
+            return link_key
+
+        try:
+            keys = tuple(getattr(target, "keys", ()) or ())
+        except Exception:
+            keys = ()
+        for key in keys:
+            key_text = str(key or "").strip()
+            if key_text:
+                return key_text
+
+    controller = getattr(owner, "controller", None)
+    if controller is not None:
+        link_key_factory = getattr(controller, "output_preset_link_key", None)
+        if callable(link_key_factory):
+            try:
+                link_key = str(link_key_factory() or "").strip()
+            except Exception:
+                link_key = ""
+            if link_key:
+                return link_key
+
+        keys_factory = getattr(controller, "output_preset_keys", None)
+        if callable(keys_factory):
+            try:
+                keys = tuple(keys_factory() or ())
+            except Exception:
+                keys = ()
+            for key in keys:
+                key_text = str(key or "").strip()
+                if key_text:
+                    return key_text
+
+    return fallback
+
+
 class MiniEqWindow(
     MiniEqWindowPresetMixin,
     MiniEqWindowAnalyzerMixin,
@@ -128,6 +179,7 @@ def __init__(
         self.output_preset_curve_auto_loaded = False
         self.updating_output_preset_switch = False
         self.last_output_preset_sink_name: str | None = None
+        self.last_output_preset_target_identity: str | None = None
         self.preset_monitor: Gio.FileMonitor | None = None
         self.preset_refresh_source_id = 0
         self.analyzer_enabled = load_monitor_enabled()
@@ -763,6 +815,8 @@ def refresh_output_sinks(self, *, handle_observed_output_change: bool = True) ->
 
         active = self.controller.output_sink
         previous_output = self.last_output_preset_sink_name
+        previous_output_identity = getattr(self, "last_output_preset_target_identity", previous_output)
+        active_output_identity = output_preset_target_identity(self, active)
         previous_output_preset_auto_loaded = self.output_preset_curve_auto_loaded
         visible_sinks = self.list_visible_output_sinks()
         visible_sink_names = [sink.node_name for sink in visible_sinks if sink.node_name is not None]
@@ -789,11 +843,12 @@ def refresh_output_sinks(self, *, handle_observed_output_change: bool = True) ->
         finally:
             self.updating_output_combo = False
 
-        output_changed = previous_output is not None and previous_output != active
+        output_changed = previous_output_identity is not None and previous_output_identity != active_output_identity
         # App-originated selector refreshes should not consume the output
         # transition; the next PipeWire-observed refresh owns preset handling.
         if handle_observed_output_change or previous_output is None:
             self.last_output_preset_sink_name = active
+            self.last_output_preset_target_identity = active_output_identity
         self.update_preset_state()
         self.update_info_label()
         self.update_status_summary()

tests/test_mini_eq_deps.py

@@ -95,7 +95,12 @@ def test_pipewire_gobject_check_requires_current_library_version(monkeypatch) ->
     fake_pwg = SimpleNamespace(
         get_library_version=lambda: "0.3.4",
         Core=SimpleNamespace(set_pipewire_property=object()),
-        Device=SimpleNamespace(enum_all_params=object(), enum_params=object(), new=object()),
+        Device=SimpleNamespace(
+            enum_all_params=object(),
+            enum_params=object(),
+            new=object(),
+            subscribe_params=object(),
+        ),
         Param=SimpleNamespace(new_props_controls=object()),
         RouteInfo=SimpleNamespace(new_from_param=object()),
         Stream=SimpleNamespace(set_pipewire_property=object()),
@@ -123,7 +128,12 @@ def test_pipewire_gobject_check_requires_property_override_symbols(monkeypatch)
     fake_pwg = SimpleNamespace(
         get_library_version=lambda: "0.3.5",
         Core=SimpleNamespace(),
-        Device=SimpleNamespace(enum_all_params=object(), enum_params=object(), new=object()),
+        Device=SimpleNamespace(
+            enum_all_params=object(),
+            enum_params=object(),
+            new=object(),
+            subscribe_params=object(),
+        ),
         Param=SimpleNamespace(new_props_controls=object()),
         RouteInfo=SimpleNamespace(new_from_param=object()),
         Stream=SimpleNamespace(set_pipewire_property=object()),

tests/test_mini_eq_output_presets.py

@@ -806,6 +806,39 @@ def test_output_preset_actions_use_route_key_when_available(monkeypatch, tmp_pat
     assert test_window.statuses[-1] == "Auto preset cleared"
 
 
+def test_output_scope_state_follows_same_sink_route_change(monkeypatch, tmp_path) -> None:
+    monkeypatch.setattr(core, "PRESET_STORAGE_DIR", tmp_path / "presets")
+    monkeypatch.setattr(core, "OUTPUT_PRESET_LINKS_PATH", tmp_path / "output-presets.json")
+    controller = make_controller("alsa_output.internal")
+    route = SimpleNamespace(
+        description="Headphones",
+        name="analog-output-headphones",
+        output_preset_key="pipewire-route:v1:device=alsa_card.test;route=analog-output-headphones;route-device=6",
+    )
+
+    def output_preset_target():
+        return SimpleNamespace(
+            output_key=controller.output_sink,
+            route=route,
+            keys=(route.output_preset_key, controller.output_sink),
+            link_key=route.output_preset_key,
+            has_route_key=True,
+        )
+
+    controller.output_preset_target = output_preset_target
+    test_window = OutputPresetWindow(controller)
+
+    test_window.update_output_preset_state()
+    assert test_window.output_scope_state_label.text == "Headphones"
+
+    route.description = "Speakers"
+    route.name = "analog-output-speaker"
+    route.output_preset_key = "pipewire-route:v1:device=alsa_card.test;route=analog-output-speaker;route-device=6"
+
+    test_window.update_output_preset_state()
+    assert test_window.output_scope_state_label.text == "Speakers"
+
+
 def test_deleting_only_loaded_preset_keeps_curve_and_allows_neutral_reset(monkeypatch, tmp_path) -> None:
     monkeypatch.setattr(core, "PRESET_STORAGE_DIR", tmp_path / "presets")
     monkeypatch.setattr(core, "OUTPUT_PRESET_LINKS_PATH", tmp_path / "output-presets.json")
@@ -1134,6 +1167,54 @@ def test_pipewire_observed_output_change_runs_output_preset_handling() -> None:
     assert fake_window.last_output_preset_sink_name == "alsa_output.usb"
 
 
+def test_pipewire_observed_port_scope_change_runs_output_preset_handling() -> None:
+    calls: list[object] = []
+    old_route_key = "pipewire-route:v1:device=alsa_card.test;route=analog-output-headphones;route-device=6"
+    new_route_key = "pipewire-route:v1:device=alsa_card.test;route=analog-output-speaker;route-device=6"
+    target = SimpleNamespace(
+        link_key=new_route_key,
+        keys=(new_route_key, "alsa_output.internal"),
+    )
+    fake_window = SimpleNamespace(
+        ui_shutting_down=False,
+        controller=SimpleNamespace(
+            output_sink="alsa_output.internal",
+            follow_default_output=True,
+            get_default_output_sink_name=lambda: "alsa_output.internal",
+            get_sink=lambda _sink_name: None,
+        ),
+        last_output_preset_sink_name="alsa_output.internal",
+        last_output_preset_target_identity=old_route_key,
+        output_preset_auto_applied=True,
+        output_preset_curve_auto_loaded=True,
+        post_present_ready=True,
+        list_visible_output_sinks=lambda: [],
+        build_output_sink_labels=lambda _sinks: [],
+        follow_default_output_label=lambda: "Follow system output",
+        output_sink_names=[],
+        output_sink_labels=[],
+        output_sink_model=FakeModel(),
+        output_combo=FakeCombo(),
+        updating_output_combo=False,
+        output_preset_target=lambda: target,
+        update_preset_state=lambda: calls.append("preset-state"),
+        update_info_label=lambda: calls.append("info"),
+        update_status_summary=lambda: calls.append("summary"),
+        apply_output_preset_for_current_output=lambda **kwargs: calls.append(("auto", kwargs)),
+    )
+
+    window.MiniEqWindow.refresh_output_sinks(fake_window)
+
+    assert calls == [
+        "preset-state",
+        "info",
+        "summary",
+        ("auto", {"reset_auto_preset_without_link": True, "announce_no_output_preset": True}),
+    ]
+    assert fake_window.last_output_preset_sink_name == "alsa_output.internal"
+    assert fake_window.last_output_preset_target_identity == new_route_key
+
+
 def test_manual_output_refresh_updates_selector_without_handling_observed_output_change() -> None:
     calls: list[object] = []
     fake_window = SimpleNamespace(

tests/test_mini_eq_pipewire_backend.py

@@ -470,6 +470,93 @@ def test_output_preset_keys_prefer_matching_active_route(monkeypatch: pytest.Mon
     )
 
 
+def test_output_preset_keys_use_single_route_even_when_profile_device_is_stale(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    backend = pw_backend.PipeWireBackend()
+    backend._Pwg = SimpleNamespace(Device=FakeDeviceApi, RouteInfo=FakeRouteInfoApi)
+    sink = pw_backend.PipeWireNode(
+        bound_id=39,
+        object_serial="67",
+        media_class=pw_backend.AUDIO_SINK,
+        node_name="alsa_output.test",
+        node_description="Test Sink",
+        application_name=None,
+        node_dont_move=False,
+        device_id=72,
+        card_profile_device=8,
+    )
+    speakers = pw_routes.PipeWireOutputRoute(
+        device_bound_id=72,
+        device_name="alsa_card.test",
+        index=0,
+        route_device=6,
+        profile=0,
+        priority=100,
+        direction="Output",
+        name="analog-output-speaker",
+        description="Speakers",
+        availability="unknown",
+    )
+
+    monkeypatch.setattr(backend, "audio_sink_by_name", lambda _name: sink)
+    monkeypatch.setattr(backend, "_device_proxy_by_bound_id", lambda _bound_id: object())
+    monkeypatch.setattr(backend, "_enumerate_device_routes", lambda _device, _bound_id: [speakers])
+
+    assert backend.output_preset_keys_for_sink_name("alsa_output.test") == (
+        "pipewire-route:v1:device=alsa_card.test;route=analog-output-speaker;route-device=6",
+        "alsa_output.test",
+    )
+
+
+def test_output_preset_keys_do_not_guess_between_routes_sharing_profile_device(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    backend = pw_backend.PipeWireBackend()
+    backend._Pwg = SimpleNamespace(Device=FakeDeviceApi, RouteInfo=FakeRouteInfoApi)
+    sink = pw_backend.PipeWireNode(
+        bound_id=39,
+        object_serial="67",
+        media_class=pw_backend.AUDIO_SINK,
+        node_name="alsa_output.test",
+        node_description="Test Sink",
+        application_name=None,
+        node_dont_move=False,
+        device_id=72,
+        card_profile_device=6,
+    )
+    speakers = pw_routes.PipeWireOutputRoute(
+        device_bound_id=72,
+        device_name="alsa_card.test",
+        index=0,
+        route_device=6,
+        profile=0,
+        priority=100,
+        direction="Output",
+        name="analog-output-speaker",
+        description="Speakers",
+        availability="unknown",
+    )
+    headphones = pw_routes.PipeWireOutputRoute(
+        device_bound_id=72,
+        device_name="alsa_card.test",
+        index=1,
+        route_device=6,
+        profile=0,
+        priority=200,
+        direction="Output",
+        name="analog-output-headphones",
+        description="Headphones",
+        availability="yes",
+    )
+
+    monkeypatch.setattr(backend, "audio_sink_by_name", lambda _name: sink)
+    monkeypatch.setattr(backend, "_device_proxy_by_bound_id", lambda _bound_id: object())
+    monkeypatch.setattr(backend, "_enumerate_device_routes", lambda _device, _bound_id: [speakers, headphones])
+
+    assert backend.output_preset_keys_for_sink_name("alsa_output.test") == ("alsa_output.test",)
+
+
 def test_output_preset_keys_fall_back_to_sink_name_without_route_api(monkeypatch: pytest.MonkeyPatch) -> None:
     backend = pw_backend.PipeWireBackend()
     backend._Pwg = SimpleNamespace()