"""Persistent audio-output selection via WirePlumber's wpctl. Two identifiers are in play and confusing them is the main failure mode here. `wpctl` takes a numeric object ID, which WirePlumber reassigns on every boot and every device hotplug — useless for persistence. `node.name` is stable across both, which is what audio-config.json stores. Home Assistant is shown neither: the select lists the human descriptions ("Built-in Audio Analog Stereo"), because those are what someone picking an output in a mobile app can actually recognise. Per the security note in mqtt_discovery.py, the payload from HA is only ever used to look up an entry in the sink table parsed from `wpctl status`. It never reaches a command line: what does is the integer ID that lookup returns. """ from __future__ import annotations import logging import re import subprocess from dataclasses import dataclass from .runtime_state import ensure_runtime_copy, load_json, save_json log = logging.getLogger(__name__) CONFIG_FILENAME = "audio-config.json" SYSTEM_DEFAULT = "system-default" # `wpctl status` prints a tree; sink rows look like # │ * 47. Built-in Audio Analog Stereo [vol: 0.65] # with a leading "*" on the current default. The box-drawing prefix varies between # WirePlumber versions, so the row is matched from the ID onwards rather than anchored. _SINK_ROW = re.compile(r"(\*?)\s*(\d+)\.\s+(.*?)(?:\s+\[vol:.*)?$") _SECTION = re.compile(r"^\s*[^\w]*\s*(\w[\w /]*):\s*$") @dataclass(frozen=True) class Sink: node_id: int description: str node_name: str is_default: bool class AudioControl: def __init__(self, env_provider): # wpctl talks to the user's PipeWire session, so it needs the same XDG_RUNTIME_DIR # derivation SwayControl uses for swaymsg and playerctl. self._env_provider = env_provider self.config_path = ensure_runtime_copy(CONFIG_FILENAME) config = load_json(self.config_path) self.preferred_sink = str(config.get("preferred_sink") or "") self.fallback = str(config.get("fallback") or SYSTEM_DEFAULT) self._sinks: list[Sink] = [] # --- wpctl -------------------------------------------------------------- def _wpctl(self, *args: str) -> str | None: try: result = subprocess.run( ["wpctl", *args], env=self._env_provider(), capture_output=True, text=True, timeout=10, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: log.warning("wpctl %s failed: %s", " ".join(args), exc) return None if result.returncode != 0: log.warning("wpctl %s: %s", " ".join(args), result.stderr.strip()) return None return result.stdout def _node_name(self, node_id: int) -> str: output = self._wpctl("inspect", str(node_id)) or "" for line in output.splitlines(): if "node.name" in line: _, _, value = line.partition("=") return value.strip().strip('"') return "" def list_sinks(self) -> list[Sink]: output = self._wpctl("status") if output is None: self._sinks = [] return self._sinks sinks: list[Sink] = [] in_sinks = False for line in output.splitlines(): section = _SECTION.match(line) if section: # Sources, Filters and Streams also carry numbered rows, so the parser # has to stop at the next heading rather than read to end of output. in_sinks = section.group(1).strip() == "Sinks" continue if not in_sinks: continue match = _SINK_ROW.search(line) if not match: continue node_id = int(match.group(2)) sinks.append( Sink( node_id=node_id, description=match.group(3).strip() or f"Sink {node_id}", node_name=self._node_name(node_id), is_default=match.group(1) == "*", ) ) self._sinks = sinks return sinks # --- entity surface ----------------------------------------------------- def options(self) -> list[str]: """Select options for HA: descriptions, plus the "let WirePlumber decide" entry.""" seen: dict[str, int] = {} result = [SYSTEM_DEFAULT] for sink in self._sinks: label = sink.description if label in seen: # Two identical descriptions (e.g. a pair of matched HDMI outputs) would # otherwise collapse into one unselectable option. seen[label] += 1 label = f"{label} ({seen[label]})" else: seen[label] = 1 result.append(label) return result def current_option(self) -> str: for sink in self._sinks: if self.preferred_sink and sink.node_name == self.preferred_sink: return sink.description if self.preferred_sink: # Configured but not present right now — say so rather than silently # reporting whatever WirePlumber happens to be using. return SYSTEM_DEFAULT for sink in self._sinks: if sink.is_default: return sink.description return SYSTEM_DEFAULT def _find(self, option: str) -> Sink | None: for sink in self._sinks: if sink.description == option: return sink # Match the disambiguating "(2)" suffix options() may have added. base = re.sub(r"\s+\(\d+\)$", "", option) matches = [s for s in self._sinks if s.description == base] return matches[0] if matches else None def apply_preferred(self) -> None: """Called once at startup, after the sink list has been read.""" if not self.preferred_sink: log.info("no preferred audio sink configured; leaving WirePlumber's default") return for sink in self._sinks: if sink.node_name == self.preferred_sink: self._set_default(sink) return # Never fatal: a docked machine booted undocked, or an HDMI display that is off, # legitimately has no such sink. The preference stays on file for next boot. log.warning( "preferred audio sink %r is not currently available; falling back to %s", self.preferred_sink, self.fallback, ) def _set_default(self, sink: Sink) -> None: log.info("setting default audio sink to %s (id=%s)", sink.description, sink.node_id) self._wpctl("set-default", str(sink.node_id)) def select(self, option: str) -> str: """Handle the HA select. Returns the option to publish back as state.""" option = option.strip() self.list_sinks() if option == SYSTEM_DEFAULT: self.preferred_sink = "" self._save() log.info("audio output preference cleared; WirePlumber's default applies") return self.current_option() sink = self._find(option) if sink is None: log.warning("ignoring unknown audio output %r", option) return self.current_option() self._set_default(sink) if sink.node_name: self.preferred_sink = sink.node_name self._save() else: # Without a node.name there is nothing stable to persist; the change still # applies to this boot. log.warning( "sink %r has no node.name; the change applies now but will not survive a reboot", sink.description, ) return sink.description def _save(self) -> None: save_json( self.config_path, {"preferred_sink": self.preferred_sink, "fallback": self.fallback}, )