Merge pull request #36 from bhack/issue-35-auto-preset-startup-scope

Fix startup auto preset route recovery

Author: bhack

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.8.6 - 2026-06-04
+
+- Fix auto presets at login for outputs where PipeWire restores the selected
+  speaker sink before reporting the active port route.
+
 ## 0.8.5 - 2026-05-31
 
 - Fixed auto presets sometimes resetting to Neutral when PipeWire route

data/io.github.bhack.mini-eq.metainfo.xml

@@ -33,18 +33,23 @@
   </description>
   <screenshots>
     <screenshot type="default">
-      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.5/docs/screenshots/mini-eq.png</image>
+      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.6/docs/screenshots/mini-eq.png</image>
       <caption>Adjust sound output with equalizer controls</caption>
     </screenshot>
     <screenshot>
-      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.5/docs/screenshots/mini-eq-dark.png</image>
+      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.6/docs/screenshots/mini-eq-dark.png</image>
       <caption>Use the equalizer with dark style</caption>
     </screenshot>
   </screenshots>
   <url type="homepage">https://github.com/bhack/mini-eq</url>
   <url type="bugtracker">https://github.com/bhack/mini-eq/issues</url>
   <url type="vcs-browser">https://github.com/bhack/mini-eq</url>
   <releases>
+    <release version="0.8.6" date="2026-06-04">
+      <description>
+        <p>Fix auto presets at login for outputs where PipeWire restores the selected speaker sink before reporting the active port route.</p>
+      </description>
+    </release>
     <release version="0.8.5" date="2026-05-31">
       <description>
         <p>Fixed auto presets sometimes resetting to Neutral when PipeWire route metadata changed.</p>

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mini-eq"
-version = "0.8.5"
+version = "0.8.6"
 description = "Compact PipeWire system-wide parametric equalizer for Linux desktops."
 readme = "README.md"
 requires-python = ">=3.11"

src/mini_eq/core.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from functools import lru_cache
 from pathlib import Path
+from urllib.parse import unquote
 
 import numpy as np
 
@@ -23,6 +24,7 @@
 PRESET_FILE_SUFFIX = ".json"
 OUTPUT_PRESET_LINKS_VERSION = 1
 OUTPUT_PRESET_LINKS_FILE = "output-presets.json"
+OUTPUT_PRESET_ROUTE_KEY_PREFIX = "pipewire-route:v1:"
 EQ_MODE_APO = 6
 SAMPLE_RATE = 48000.0
 GRAPH_FREQ_MIN = 20.0
@@ -426,6 +428,35 @@ def normalize_output_preset_key_candidates(output_keys: str | Iterable[str | Non
     return normalized
 
 
+def parse_output_preset_route_key(output_key: str | None) -> dict[str, object] | None:
+    key = str(output_key or "").strip()
+    if not key.startswith(OUTPUT_PRESET_ROUTE_KEY_PREFIX):
+        return None
+
+    fields: dict[str, str] = {}
+    for part in key[len(OUTPUT_PRESET_ROUTE_KEY_PREFIX) :].split(";"):
+        if "=" not in part:
+            return None
+        name, value = part.split("=", 1)
+        fields[name] = unquote(value)
+
+    device_name = fields.get("device", "").strip()
+    route_name = fields.get("route", "").strip()
+    try:
+        route_device = int(fields.get("route-device", ""))
+    except ValueError:
+        return None
+
+    if not device_name or not route_name or route_device < 0:
+        return None
+
+    return {
+        "device": device_name,
+        "route": route_name,
+        "route_device": route_device,
+    }
+
+
 def load_output_preset_config() -> tuple[dict[str, str], str | None]:
     links_path = output_preset_links_path()
     if not links_path.exists():
@@ -491,6 +522,34 @@ def get_output_preset_link(output_keys: str | Iterable[str | None] | None) -> st
     return match[1] if match is not None else None
 
 
+def get_output_preset_route_device_link_match(
+    device_name: str | None,
+    route_device: int | None,
+) -> tuple[str, str] | None:
+    device = str(device_name or "").strip()
+    try:
+        route_device_id = int(route_device)
+    except (TypeError, ValueError):
+        return None
+
+    if not device or route_device_id < 0:
+        return None
+
+    links, _default_preset = load_output_preset_config()
+    matches: list[tuple[str, str]] = []
+    for output_key, preset_name in links.items():
+        route_key = parse_output_preset_route_key(output_key)
+        if route_key is None:
+            continue
+        if route_key["device"] == device and route_key["route_device"] == route_device_id:
+            matches.append((output_key, preset_name))
+
+    if len(matches) == 1:
+        return matches[0]
+
+    return None
+
+
 def get_default_preset_name() -> str | None:
     _links, default_preset = load_output_preset_config()
     return default_preset

src/mini_eq/diagnostics.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import json
+import os
+from datetime import UTC, datetime
+from pathlib import Path
+
+from gi.repository import GLib
+
+STARTUP_TRACE_ENV = "MINI_EQ_STARTUP_TRACE"
+STARTUP_TRACE_DIR_NAME = "mini-eq"
+STARTUP_TRACE_FILE_NAME = "startup-trace.log"
+
+
+def startup_trace_enabled() -> bool:
+    value = os.environ.get(STARTUP_TRACE_ENV, "")
+    return value.strip().lower() not in {"", "0", "false", "no", "off"}
+
+
+def startup_trace_path() -> Path:
+    return Path(GLib.get_user_state_dir()) / STARTUP_TRACE_DIR_NAME / STARTUP_TRACE_FILE_NAME
+
+
+def describe_output_preset_target(target: object | None) -> dict[str, object] | None:
+    if target is None:
+        return None
+
+    route = getattr(target, "route", None)
+    route_info = None
+    if route is not None:
+        route_info = {
+            "description": _json_safe(getattr(route, "description", None)),
+            "device_name": _json_safe(getattr(route, "device_name", None)),
+            "name": _json_safe(getattr(route, "name", None)),
+            "output_preset_key": _json_safe(getattr(route, "output_preset_key", None)),
+            "route_device": _json_safe(getattr(route, "route_device", None)),
+        }
+
+    return {
+        "device_name": _json_safe(getattr(target, "device_name", None)),
+        "has_route_key": bool(getattr(target, "has_route_key", False)),
+        "keys": _json_safe(tuple(getattr(target, "keys", ()) or ())),
+        "link_key": _json_safe(getattr(target, "link_key", "")),
+        "output_key": _json_safe(getattr(target, "output_key", None)),
+        "route": route_info,
+        "route_device": _json_safe(getattr(target, "route_device", None)),
+    }
+
+
+def describe_output_preset_snapshot(snapshot: object | None) -> dict[str, object] | None:
+    if snapshot is None:
+        return None
+
+    return {
+        "identity": _json_safe(getattr(snapshot, "identity", None)),
+        "sink_name": _json_safe(getattr(snapshot, "sink_name", None)),
+        "target": describe_output_preset_target(getattr(snapshot, "target", None)),
+    }
+
+
+def describe_output_preset_transition(transition: object | None) -> dict[str, object] | None:
+    if transition is None:
+        return None
+
+    return {
+        "changed": bool(getattr(transition, "changed", False)),
+        "current": describe_output_preset_snapshot(getattr(transition, "current", None)),
+        "previous": describe_output_preset_snapshot(getattr(transition, "previous", None)),
+    }
+
+
+def trace_startup_event(event: str, **fields: object) -> None:
+    if not startup_trace_enabled():
+        return
+
+    path = startup_trace_path()
+    record = {
+        "event": event,
+        "timestamp": datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z"),
+    }
+    record.update({key: _json_safe(value) for key, value in fields.items()})
+
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("a", encoding="utf-8") as trace_file:
+            trace_file.write(json.dumps(record, sort_keys=True, separators=(",", ":")))
+            trace_file.write("\n")
+    except Exception:
+        return
+
+
+def _json_safe(value: object) -> object:
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+
+    if isinstance(value, tuple | list | set):
+        return [_json_safe(item) for item in value]
+
+    if isinstance(value, dict):
+        return {str(key): _json_safe(item) for key, item in value.items()}
+
+    return str(value)