Commit 560e0e95 by PLN (Algolia)

feat(rig): one supervised surface tap, and an output that follows a preference list

Two pieces of the cockpit (#155, #41), both written the same way: resolve by
IDENTITY every time, never cache an address.

tools/bridge/surface.py — THE control-surface tap
-------------------------------------------------
Four tools each spawned their own aseqdump. This morning three of them were
bound to ALSA client 24 (an Arturia KeyStep) instead of 20 (the Launch Control
XL) because a replug had renumbered the clients, and each resolves its port
once, at start. I "fixed" them by moving them to 20:0.

Then PLN replugged again this evening and the numbers SWAPPED BACK:

    client 20: 'Arturia KeyStep 32'     <- was the LCXL
    client 24: 'Launch Control XL'      <- was the KeyStep

So the two I fixed are wrong again and the one I called broken is now right.
An address is a lease, not an identity. This module resolves "Launch Control
XL" by name on EVERY respawn, so a replug costs one backoff interval instead of
a silent session. First run against the live rig, with zero configuration:

    starting   port=None
    ok         port=24:0

It also keeps the last value of every CC, and — the part that matters — records
`seen: false` distinctly from `value: 0`. An untouched control is what silences
a Tidal stream (a bare "^NN" with no value yields NO EVENTS), so a cockpit that
rendered untouched-as-zero would hide the exact failure it exists to show.

And it does not litter. Three smoke runs of this file each orphaned an
aseqdump to pid 1 — precisely the mess PLN complained about this morning
("i see 4 aseqdumps... you had one job"). try/finally is not enough, because
SIGKILL, OOM and `timeout` all skip it. PR_SET_PDEATHSIG puts the guarantee in
the kernel: the tap dies with its parent however the parent dies.

tools/master-out.py — Ardour Master follows a preference list
-------------------------------------------------------------
PLN: "can we control this tedious routing... ideally theres a tie breaking list
of choice, from uhc to headphones to speakers". Ardour's master does not follow
the PipeWire default sink, so every rig move means re-patching in qjackctl.

Not "route to X" but "route to the best X actually present": UMC -> Headphones
-> Speakers, matched as substrings of the node name so serial numbers and
profile suffixes do not matter.

Two properties make it safe to run mid-set:
  * IDEMPOTENT — computes wanted links, diffs against existing, applies only
    the delta. Already correct = zero pw-link calls = cannot glitch the audio.
    The naive unlink-then-relink is a guaranteed dropout even when nothing
    changed.
  * EXCLUSIVE BUT NARROW — removes Master links to other sinks (or plugging the
    UMC back in gives you doubled output), and touches nothing else. The limit
    is a predicate, `_is_ours`, not a good intention: apply() refuses to unlink
    any source that is not a Master output.

Verified against KNOWN-BAD states, not just against agreement with reality —
the whole lesson of this session. Ten checks: preference order at each of the
three tiers, the "master stranded on headphones when the UMC appears" case
(2 adds + 2 removes), idempotency (0 and 0), and the guard refusing both a
foreign link in plan() and a foreign source handed straight to apply().
All pass. Live dry-run reports "already correct — 0 changes".
parent 098d5712
#!/usr/bin/env python3
"""master-out — point Ardour's Master at the best available output, by preference.
#41. Ardour's master does NOT follow the PipeWire default sink, so every time
the rig moves — studio to kitchen table, UMC plugged or not, headphones in —
the Master output has to be re-patched by hand in qjackctl. PLN, 2026-08-14:
"can we control this tedious routing... ideally theres a tie breaking list of
choice, from uhc to headphones to speakers".
That is exactly the shape: not "route to X", but "route to the best X that is
actually here right now". So this is a PREFERENCE LIST plus a converger.
tools/master-out.py # show state + the plan, change nothing
tools/master-out.py --apply # converge
tools/master-out.py --watch 5 # converge whenever the hardware changes
THE TWO PROPERTIES THAT MAKE THIS SAFE TO RUN WHILE PLAYING
IDEMPOTENT — it computes the links it WANTS, diffs against the links that
EXIST, and applies only the difference. Already correct means literally zero
pw-link calls, so re-running mid-set is a no-op and cannot glitch the audio.
Every previous version of this idea in this repo failed by unlinking first
and relinking after, which is a guaranteed dropout even when nothing changed.
EXCLUSIVE, BUT NARROWLY — it removes Master links to sinks OTHER than the
chosen one (otherwise you get doubled output the first time you plug the UMC
back in), and it touches nothing else. SuperDirt->Ardour track links, the
capture chain, and anything not originating at Master/audio_out are left
alone. The blast radius is a predicate, not a promise: see `_is_ours`.
DELIBERATELY NOT AUTOMATIC BY DEFAULT. Ripping the output from under a live set
because a headphone jack was nudged is worse than the manual patching this
replaces. Run --watch when you want it; the gig path should pin the device and
verify it, not chase it.
"""
from __future__ import annotations
import argparse
import subprocess
import sys
import time
# In preference order. Matched as a case-insensitive substring against the
# PipeWire node name, so it survives the serial number and the profile suffix
# ("...UMC202HD_192k_12345678-00.Direct__Direct__sink").
PRIORITY = [
("UMC / Behringer", "umc202hd"),
("Headphones", "headphones"),
("Speakers", "speaker"),
]
SOURCE_NODE = "ardour"
SOURCE_PORTS = ("Master/audio_out 1", "Master/audio_out 2")
# playback_FL/FR is the near-universal PipeWire naming for a stereo sink. Kept
# explicit rather than "first two ports" — guessing port order is how you end
# up with a silently swapped stereo image.
SINK_PORTS = ("playback_FL", "playback_FR")
def _run(args: list[str]) -> str:
try:
r = subprocess.run(args, capture_output=True, text=True, timeout=10)
return r.stdout
except (OSError, subprocess.SubprocessError):
return ""
def sinks() -> list[str]:
"""Distinct sink node names that currently exist, from their input ports."""
out, seen = [], set()
for line in _run(["pw-link", "-i"]).splitlines():
line = line.strip()
if not line.startswith("alsa_output") or ":" not in line:
continue
node = line.rsplit(":", 1)[0]
if node not in seen:
seen.add(node)
out.append(node)
return out
def choose(available: list[str], priority=PRIORITY) -> tuple[str | None, str | None]:
"""First (label, node) in preference order that is actually present. Pure."""
for label, needle in priority:
for node in available:
if needle in node.lower():
return label, node
return None, None
def links() -> list[tuple[str, str]]:
"""Existing (output_port, input_port) pairs, parsed from `pw-link -l`.
pw-link prints a port, then its connections indented with |-> or |<-. Only
the `|->` (outgoing) lines under an output port are ours to reason about.
"""
out: list[tuple[str, str]] = []
current = None
for raw in _run(["pw-link", "-l"]).splitlines():
stripped = raw.strip()
if not stripped:
continue
if not raw.startswith((" ", "\t")):
current = stripped
continue
if stripped.startswith("|->") and current:
out.append((current, stripped[3:].strip()))
return out
def _is_ours(src: str) -> bool:
"""Guard: only links leaving Ardour's Master outputs may ever be removed."""
return any(src == f"{SOURCE_NODE}:{p}" for p in SOURCE_PORTS)
def plan(node: str, existing: list[tuple[str, str]]):
"""(to_add, to_remove) to make Master feed exactly `node`. Pure."""
want = {(f"{SOURCE_NODE}:{sp}", f"{node}:{dp}")
for sp, dp in zip(SOURCE_PORTS, SINK_PORTS)}
have = {(s, d) for s, d in existing if _is_ours(s)}
return sorted(want - have), sorted(have - want)
def apply(to_add, to_remove, dry=True) -> list[str]:
log = []
for s, d in to_remove:
if not _is_ours(s): # belt and braces
log.append(f" REFUSED to unlink {s} — not a Master output")
continue
log.append(f" - {s} -> {d}")
if not dry:
_run(["pw-link", "-d", s, d])
for s, d in to_add:
log.append(f" + {s} -> {d}")
if not dry:
_run(["pw-link", s, d])
return log
def converge(dry=True, quiet=False) -> bool:
"""True when Master ends up (or already is) on the preferred sink."""
avail = sinks()
label, node = choose(avail)
if node is None:
if not quiet:
print("no known sink present — nothing to do")
for a in avail:
print(f" (saw: {a})")
return False
to_add, to_remove = plan(node, links())
if not quiet:
print(f"preferred: {label}")
print(f" {node}")
if not to_add and not to_remove:
if not quiet:
print("already correct — 0 changes (safe to re-run mid-set)")
return True
steps = apply(to_add, to_remove, dry=dry)
if not quiet:
print("plan:" if dry else "applied:")
print("\n".join(steps))
if dry:
print("\n(dry run — pass --apply to make it so)")
return True
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--apply", action="store_true", help="actually re-patch")
ap.add_argument("--watch", type=float, metavar="SEC",
help="re-converge every SEC (implies --apply)")
a = ap.parse_args()
if a.watch:
print(f"master-out — converging every {a.watch}s · Ctrl-C to stop")
last = None
try:
while True:
sig = (tuple(sinks()), tuple(sorted(links())))
if sig != last: # only speak when something moved
converge(dry=False)
print()
last = (tuple(sinks()), tuple(sorted(links())))
time.sleep(a.watch)
except KeyboardInterrupt:
print("\nstopped")
return 0
return 0 if converge(dry=not a.apply) else 1
if __name__ == "__main__":
sys.exit(main())
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