feat: decide the classroom's audio output from what is plugged in
The room that prompted this had a television on HDMI with working PCM audio and nothing in the analog jack, and WirePlumber chose analog. A teacher who corrected it by hand then had to restart the browser before any sound came out. The policy is a pure function so it can be tested against that room rather than against a machine: an unavailable port is not a candidate, an analog port merely existing is not evidence of a cable, and an explicit AUDIO_OUTPUT overrides the lot. Twelve cases, the real room among them. The daemon follows PipeWire's own event stream rather than polling, and moves streams that are already playing, which is the part that stops a teacher having to close and reopen the browser. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
ekklesia
co-authored by
Claude Opus 5
Claude Opus 5
parent
87b55761c6
commit
2ff9d47be9
Executable
+218
@@ -0,0 +1,218 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""Keep a classroom playing through the right output.
|
||||
|
||||
The policy is in classroom_audio; this is the part that talks to PipeWire.
|
||||
It applies the decision, and then re-applies it whenever the sound hardware
|
||||
changes, which is what makes a television switched on after the computer
|
||||
work without anybody touching anything.
|
||||
|
||||
Three things happen on every change, and the third is the one that fixes
|
||||
the complaint from the room:
|
||||
|
||||
set the card profile so the output exists at all
|
||||
set the default sink so new sound goes there
|
||||
move existing streams so sound that is ALREADY PLAYING follows
|
||||
|
||||
Without the last step a browser that was started before the change keeps
|
||||
playing into the old sink, and the only fix a teacher can find is to close
|
||||
and reopen it. That is exactly what happened in the room this was written
|
||||
for.
|
||||
|
||||
Volume and mute are set once, when the output actually changes. After that
|
||||
the room's own volume is left alone.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, "/usr/local/lib/ekklesia-classroom")
|
||||
sys.path.insert(
|
||||
0,
|
||||
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lib"),
|
||||
)
|
||||
|
||||
import classroom_audio as policy # noqa: E402
|
||||
|
||||
try:
|
||||
from classroom_config import read_device_config
|
||||
except ImportError: # pragma: no cover - the daemon must not need it
|
||||
def read_device_config():
|
||||
return {}
|
||||
|
||||
|
||||
# Loud enough to be heard at the back of a classroom, quiet enough that the
|
||||
# first thing a teacher plays is not a shock. Only ever applied when the
|
||||
# output changes, never on top of a volume somebody has chosen.
|
||||
DEFAULT_VOLUME = "70%"
|
||||
|
||||
# Several udev events arrive for one cable being plugged in. Settle first.
|
||||
SETTLE_SECONDS = 1.5
|
||||
|
||||
|
||||
def pactl(*args, parse=False):
|
||||
command = ["pactl"]
|
||||
|
||||
if parse:
|
||||
command += ["-f", "json"]
|
||||
|
||||
command += list(args)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
command, capture_output=True, text=True, timeout=15, check=False
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as error:
|
||||
log(f"pactl {' '.join(args)}: {error}")
|
||||
return None
|
||||
|
||||
if result.returncode != 0:
|
||||
log(f"pactl {' '.join(args)}: {result.stderr.strip()}")
|
||||
return None
|
||||
|
||||
if not parse:
|
||||
return result.stdout
|
||||
|
||||
try:
|
||||
return json.loads(result.stdout or "[]")
|
||||
except ValueError as error:
|
||||
log(f"pactl {' '.join(args)}: unreadable output: {error}")
|
||||
return None
|
||||
|
||||
|
||||
def log(message):
|
||||
print(f"classroom-audio: {message}", flush=True)
|
||||
|
||||
|
||||
def preference():
|
||||
value = (read_device_config() or {}).get("AUDIO_OUTPUT", policy.AUTO)
|
||||
value = (value or "").strip().lower()
|
||||
|
||||
return value if value in (policy.AUTO, policy.HDMI, policy.ANALOG_ONLY) \
|
||||
else policy.AUTO
|
||||
|
||||
|
||||
def sink_for_card(card_name):
|
||||
for sink in pactl("list", "sinks", parse=True) or []:
|
||||
properties = sink.get("properties") or {}
|
||||
|
||||
if properties.get("alsa.card_name") and card_name and \
|
||||
properties.get("device.string"):
|
||||
pass
|
||||
|
||||
if (sink.get("card") == card_name
|
||||
or properties.get("device.name") == card_name
|
||||
or (card_name or "") in (sink.get("name") or "")):
|
||||
return sink
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def apply(state):
|
||||
cards = pactl("list", "cards", parse=True)
|
||||
|
||||
if cards is None:
|
||||
return
|
||||
|
||||
chosen = policy.choose(cards, preference())
|
||||
|
||||
if chosen is None:
|
||||
if state.get("port") is not None:
|
||||
log("no usable output")
|
||||
state["port"] = None
|
||||
return
|
||||
|
||||
profile = policy.profile_for(chosen)
|
||||
changed = state.get("port") != (chosen["card"], chosen["port"])
|
||||
|
||||
if changed and profile:
|
||||
pactl("set-card-profile", str(chosen["card"]), profile)
|
||||
# The sink appears only after the profile is in place.
|
||||
time.sleep(0.4)
|
||||
|
||||
sink = sink_for_card(chosen["card"])
|
||||
|
||||
if sink is None:
|
||||
log(f"no sink for {chosen['card']} after {profile}")
|
||||
return
|
||||
|
||||
name = sink.get("name")
|
||||
pactl("set-default-sink", name)
|
||||
|
||||
if changed:
|
||||
pactl("set-sink-mute", name, "0")
|
||||
pactl("set-sink-volume", name, DEFAULT_VOLUME)
|
||||
log(f"output is {chosen['port']} on {chosen['card']} ({name})")
|
||||
|
||||
# Anything already making sound follows, so a browser that was open
|
||||
# before the television was switched on does not have to be restarted.
|
||||
moved = 0
|
||||
|
||||
for stream in pactl("list", "sink-inputs", parse=True) or []:
|
||||
if stream.get("sink") == sink.get("index"):
|
||||
continue
|
||||
|
||||
if pactl("move-sink-input", str(stream.get("index")), name) is not None:
|
||||
moved += 1
|
||||
|
||||
if moved:
|
||||
log(f"moved {moved} playing stream(s) to {name}")
|
||||
|
||||
state["port"] = (chosen["card"], chosen["port"])
|
||||
|
||||
|
||||
def watch():
|
||||
"""Follow PipeWire's own event stream rather than polling it."""
|
||||
state = {}
|
||||
apply(state)
|
||||
|
||||
while True:
|
||||
process = subprocess.Popen(
|
||||
["pactl", "subscribe"],
|
||||
stdout=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
|
||||
try:
|
||||
for line in process.stdout:
|
||||
if not any(
|
||||
word in line for word in ("card", "server", "sink", "source")
|
||||
):
|
||||
continue
|
||||
|
||||
# Drain the rest of the burst before acting once.
|
||||
time.sleep(SETTLE_SECONDS)
|
||||
apply(state)
|
||||
except OSError as error:
|
||||
log(f"subscription ended: {error}")
|
||||
finally:
|
||||
process.terminate()
|
||||
|
||||
log("re-subscribing")
|
||||
time.sleep(3)
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "once":
|
||||
apply({})
|
||||
return 0
|
||||
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "show":
|
||||
cards = pactl("list", "cards", parse=True) or []
|
||||
for item in policy.candidates(cards):
|
||||
print(f"{item['kind']:6} {item['availability']:13} "
|
||||
f"prio={item['priority']:<6} {item['card']} {item['port']}")
|
||||
chosen = policy.choose(cards, preference())
|
||||
print(f"\npreference={preference()} chosen="
|
||||
f"{chosen['port'] if chosen else 'none'}")
|
||||
return 0
|
||||
|
||||
watch()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user