212 lines
7.9 KiB
Python
212 lines
7.9 KiB
Python
"""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},
|
|
)
|