Commit bf90f377 by PLN (Algolia)

feat(monitor): pick a sink by its human name, and say when it is the TV

parent 18fea207
...@@ -22,6 +22,15 @@ been taking all audio off the box roughly once a minute. ...@@ -22,6 +22,15 @@ been taking all audio off the box roughly once a minute.
- **`gig-preflight`: `monitor path`** — stage (Ardour receiving on N Tidal - **`gig-preflight`: `monitor path`** — stage (Ardour receiving on N Tidal
tracks) or headphones (the fold present), asking fold-orbits rather than tracks) or headphones (the fold present), asking fold-orbits rather than
re-deriving the port math. re-deriving the port math.
- **`--sink` takes human words.** `--sink Headphones`, `--sink Speaker`,
`--sink "DisplayPort 1"` — the fragment is matched against the node
DESCRIPTION as well as its name, because "Headphones" appears nowhere in
`alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink`
and matching names alone made that flag ambiguous across nine sinks. An
ambiguous fragment lists the candidates with their labels instead of
guessing which speaker a live set comes out of. And when `auto` lands on an
HDMI/DisplayPort sink, the fold says so out loud: at a house party the TV
usually IS the PA, but that should be a decision, not a surprise.
- **`rig-doctor`: `wireplumber churn`** (starts per hour, and who asked for - **`rig-doctor`: `wireplumber churn`** (starts per hour, and who asked for
them) and **`perf-tray reachable`** (registered with the LIVE pid, menu them) and **`perf-tray reachable`** (registered with the LIVE pid, menu
populated over dbusmenu — the only part a click needs). 53 checks. populated over dbusmenu — the only part a click needs). 53 checks.
......
...@@ -67,6 +67,71 @@ def orbit_pairs(indices: list[int]) -> list[tuple[int, int]]: ...@@ -67,6 +67,71 @@ def orbit_pairs(indices: list[int]) -> list[tuple[int, int]]:
return pairs return pairs
def sinks_in(ports: list[str]) -> list[str]:
"""Every node owning `playback_*` inputs — i.e. every sink you could fold onto."""
found = {}
for p in ports:
node, _, port = p.rpartition(":")
if node and port.startswith("playback_"):
found[node] = True
return sorted(found)
def sink_catalog(text: str) -> dict[str, str]:
"""{node.name: node.description} for Audio/Sink nodes in `pw-cli ls Node`.
The DESCRIPTION is where the human words live: "Raptor Lake-P/U/H cAVS
Headphones" is a sink whose node.name is
`alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink`
— no "headphones" anywhere in it. Matching names alone made `--sink
Headphones` ambiguous across nine sinks, which is precisely the moment,
in a dark room, when a tool must not be clever.
"""
catalog: dict[str, str] = {}
name = desc = None
for raw in text.splitlines():
if re.match(r"\s*id \d+, type PipeWire:Interface:Node", raw):
name = desc = None
continue
m = re.search(r'node\.name = "([^"]*)"', raw)
if m:
name = m.group(1)
continue
m = re.search(r'node\.description = "([^"]*)"', raw)
if m:
desc = m.group(1)
continue
if 'media.class = "Audio/Sink"' in raw and name:
catalog[name] = desc or name
return catalog
def resolve_sink(want: str, available: list[str],
catalog: dict[str, str] | None = None) -> tuple[str | None, list[str]]:
"""Exact node.name, else a unique case-insensitive fragment of name OR label.
Nobody types
`alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink`
at a house party. "Headphones", "HDMI 1", "Speaker" all pick a sink; an
ambiguous fragment returns the candidates rather than guessing which
speaker a live set comes out of.
"""
if want in available:
return want, []
catalog = catalog or {}
needle = want.lower()
hits = [s for s in available
if needle in s.lower() or needle in catalog.get(s, "").lower()]
if len(hits) == 1:
return hits[0], []
return None, hits or available
def looks_like_display(sink: str) -> bool:
"""Is this sink a TV/projector rather than the laptop's own output?"""
return bool(re.search(r"hdmi|displayport|\bdp\b|pro-output", sink, re.I))
def sink_stereo_ports(ports: list[str], sink: str) -> tuple[str, str] | None: def sink_stereo_ports(ports: list[str], sink: str) -> tuple[str, str] | None:
"""The sink's L/R input ports, by the naming PipeWire actually used.""" """The sink's L/R input ports, by the naming PipeWire actually used."""
owned = {p.split(":", 1)[1]: p for p in ports if p.startswith(sink + ":")} owned = {p.split(":", 1)[1]: p for p in ports if p.startswith(sink + ":")}
...@@ -171,7 +236,8 @@ def main() -> int: ...@@ -171,7 +236,8 @@ def main() -> int:
prog="fold-orbits", prog="fold-orbits",
description="Fold every SuperDirt orbit onto one stereo sink (no Ardour).") description="Fold every SuperDirt orbit onto one stereo sink (no Ardour).")
ap.add_argument("--sink", default="auto", ap.add_argument("--sink", default="auto",
help="PipeWire sink node.name, or 'auto' for the default sink") help="sink node.name or any unique fragment of it "
"(e.g. Headphones, hdmi); 'auto' = the PipeWire default sink")
ap.add_argument("--check", action="store_true", help="report only; exit 1 if not folded") ap.add_argument("--check", action="store_true", help="report only; exit 1 if not folded")
ap.add_argument("--undo", action="store_true", help="remove the fold links") ap.add_argument("--undo", action="store_true", help="remove the fold links")
ap.add_argument("--force", action="store_true", help="fold even while Ardour runs") ap.add_argument("--force", action="store_true", help="fold even while Ardour runs")
...@@ -194,12 +260,25 @@ def main() -> int: ...@@ -194,12 +260,25 @@ def main() -> int:
return emit("fail", "scsynth is not running — start the rig first " return emit("fail", "scsynth is not running — start the rig first "
"(systemctl --user start parvagues-sc)") "(systemctl --user start parvagues-sc)")
sink = default_sink() if args.sink == "auto" else args.sink
if not sink:
return emit("fail", "could not resolve the default sink (wpctl missing or no sink)")
_, outs = run(["pw-link", "-o"]) _, outs = run(["pw-link", "-o"])
_, ins = run(["pw-link", "-i"]) _, ins = run(["pw-link", "-i"])
available = sinks_in(parse_ports(ins))
if args.sink == "auto":
sink = default_sink()
if not sink:
return emit("fail", "could not resolve the default sink "
"(wpctl missing, or wireplumber is down and there are no devices)")
else:
_, catalog_text = run(["pw-cli", "ls", "Node"], timeout=8)
catalog = sink_catalog(catalog_text)
sink, candidates = resolve_sink(args.sink, available, catalog)
if not sink:
return emit("fail",
f"{args.sink!r} matches {len(candidates)} sinks — be more specific",
lines=[f"{catalog.get(c, '?')} [{c}]" for c in candidates[:8]],
sinks=candidates)
indices = sc_out_indices(parse_ports(outs)) indices = sc_out_indices(parse_ports(outs))
if not indices: if not indices:
return emit("fail", f"no {SC_CLIENT}:out_* ports in the graph — " return emit("fail", f"no {SC_CLIENT}:out_* ports in the graph — "
...@@ -248,11 +327,18 @@ def main() -> int: ...@@ -248,11 +327,18 @@ def main() -> int:
still = [f"{s} -> {d}" for s, d in want if (s, d) not in now] still = [f"{s} -> {d}" for s, d in want if (s, d) not in now]
if still: if still:
return emit("fail", f"{len(still)} of {len(want)} link(s) did not take", lines=still[:4]) return emit("fail", f"{len(still)} of {len(want)} link(s) did not take", lines=still[:4])
return emit("ok", notes = [f"summed at unity — {len(pairs)} orbits playing at once is "
f"folded {len(pairs)} orbit(s) (d1-d{len(pairs)}) onto {sink}", f"{len(pairs)}x one orbit's level; pull gains if it clips"]
lines=[f"summed at unity — {len(pairs)} orbits playing at once is " if args.sink == "auto" and looks_like_display(sink):
f"{len(pairs)}x one orbit's level; pull gains if it clips", # The party failure: plug into the TV for visuals and wireplumber moves
"re-run after plugging headphones (the default sink changes)"]) # the default sink to it, so `auto` sends the set out of the television.
# Which at a house party is often exactly right — so this states the
# fact rather than second-guessing it.
notes.append("NB this is a TV/projector's audio, not the laptop's — right if "
"the screen is the PA, otherwise: --sink Headphones")
else:
notes.append("re-run after plugging headphones or a screen in (the default sink moves)")
return emit("ok", f"folded {len(pairs)} orbit(s) (d1-d{len(pairs)}) onto {sink}", lines=notes)
if __name__ == "__main__": if __name__ == "__main__":
......
...@@ -184,3 +184,100 @@ def test_full_plan_for_the_real_rig_covers_every_orbit(): ...@@ -184,3 +184,100 @@ def test_full_plan_for_the_real_rig_covers_every_orbit():
want = fo.desired_links(pairs, (f"{SINK}:playback_FL", f"{SINK}:playback_FR")) want = fo.desired_links(pairs, (f"{SINK}:playback_FL", f"{SINK}:playback_FR"))
assert len(pairs) == 14 and len(want) == 28 assert len(pairs) == 14 and len(want) == 28
assert fo.orbits_of(want) == list(range(1, 15)) assert fo.orbits_of(want) == list(range(1, 15))
# --------------------------------------------------------------------------- #
# choosing a sink at a house party, in the dark
# --------------------------------------------------------------------------- #
PARTY_INPUTS = f"""
{SINK}:playback_FL
{SINK}:playback_FR
alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink:playback_FL
alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink:playback_FR
alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__HDMI1__sink:playback_FL
alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__HDMI1__sink:playback_FR
SuperCollider:in_1
"""
def test_sinks_in_lists_only_nodes_with_playback_inputs():
found = fo.sinks_in(fo.parse_ports(PARTY_INPUTS))
assert len(found) == 3
assert not any("SuperCollider" in s for s in found)
def test_resolve_sink_takes_an_exact_name():
available = fo.sinks_in(fo.parse_ports(PARTY_INPUTS))
assert fo.resolve_sink(SINK, available) == (SINK, [])
def test_resolve_sink_takes_a_unique_fragment():
available = fo.sinks_in(fo.parse_ports(PARTY_INPUTS))
assert fo.resolve_sink("HDMI1", available)[0].endswith("HDMI1__sink")
assert fo.resolve_sink("hdmi1", available)[0].endswith("HDMI1__sink")
def test_resolve_sink_refuses_an_ambiguous_fragment():
# "sof_sdw" is in all three device names, and "sofsoundwire" in the two
# analogue ones. Guessing here means guessing which speaker a live set
# comes out of, in front of a room.
available = fo.sinks_in(fo.parse_ports(PARTY_INPUTS))
assert fo.resolve_sink("sof_sdw", available) == (None, available)
sink, candidates = fo.resolve_sink("sofsoundwire", available)
assert sink is None and len(candidates) == 2
def test_looks_like_display_spots_a_tv_but_not_the_speakers():
assert fo.looks_like_display("alsa_output...HiFi__HDMI1__sink")
assert fo.looks_like_display("alsa_output.pci-0000_01_00.1.pro-output-3")
assert not fo.looks_like_display(SINK)
assert not fo.looks_like_display("bluez_output.88_C9_E8_9F_C1_49.1")
# Captured from `pw-cli ls Node` on the XPS24, 2026-09-07 (trimmed).
REAL_CATALOG = ''' id 29, type PipeWire:Interface:Node/3
node.name = "Dummy-Driver"
id 44, type PipeWire:Interface:Node/3
node.description = "Raptor Lake-P/U/H cAVS Speaker"
node.name = "alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire_2__sink"
media.class = "Audio/Sink"
id 47, type PipeWire:Interface:Node/3
node.description = "Raptor Lake-P/U/H cAVS Headphones"
node.name = "alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink"
media.class = "Audio/Sink"
id 60, type PipeWire:Interface:Node/3
node.description = "Integrated_Webcam_HD (V4L2)"
node.name = "v4l2_input.pci-0000_00_14.0-usb-0_6_1.2"
media.class = "Video/Source"
id 64, type PipeWire:Interface:Node/3
node.name = "Midi-Bridge"
media.class = "Midi/Bridge"
'''
def test_sink_catalog_keeps_only_audio_sinks():
cat = fo.sink_catalog(REAL_CATALOG)
assert len(cat) == 2
assert not any("v4l2" in k or k == "Midi-Bridge" for k in cat)
def test_sink_catalog_maps_name_to_the_human_label():
cat = fo.sink_catalog(REAL_CATALOG)
assert cat["alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__sink"] \
== "Raptor Lake-P/U/H cAVS Headphones"
def test_resolve_sink_matches_the_human_label_not_just_the_name():
# "Headphones" appears NOWHERE in the node.name — this is why matching
# names alone made `--sink Headphones` ambiguous across nine sinks.
cat = fo.sink_catalog(REAL_CATALOG)
sink, _ = fo.resolve_sink("Headphones", list(cat), cat)
assert sink.endswith("hw_sofsoundwire__sink")
sink, _ = fo.resolve_sink("speaker", list(cat), cat)
assert sink.endswith("hw_sofsoundwire_2__sink")
def test_resolve_sink_without_a_catalog_still_works_on_names():
available = ["alsa_output.x_2__sink", "alsa_output.y__sink"]
assert fo.resolve_sink("y__", available)[0] == "alsa_output.y__sink"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment