#!/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 when the output changes: 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. WHY THIS WAS REWRITTEN ---------------------- On 0.2.2 the rooms came up silent, and the reason was not the decision — which was right — but when it was asked for. The old shape was: apply once at startup then apply only when an event arrives `pactl` talks to PipeWire's PulseAudio server, and this service was ordered after pipewire and wireplumber but not after that. At boot it therefore raced the server and lost: the one startup apply failed with "Connection refused", the subscription then attached successfully a few seconds later, and from that moment the daemon sat waiting for an event that had already happened. It looked perfectly healthy — systemd said active, the subscription was live — and it had never applied its policy at all. The room was left on whatever WirePlumber had chosen, which is how an administrator came to find output:analog-stereo+input:analog-stereo on a card whose HDMI port was available: that profile is WirePlumber's own pick, and not one this policy would ever produce. Reproduced in the lab by starting this daemon with pipewire-pulse stopped and then starting it, which leaves exactly the state the room was found in. So the shape is now: wait for the server, within reason apply subscribe, and apply again the moment the subscription attaches apply on events, coalescing each burst into one decision while the room does NOT have the output it would prefer, look again on a backing-off timer; once it does, stop looking and wait The re-apply on attach is the fix for the race. The timer is the fix for the larger version of the same problem: this policy no longer depends on an event arriving at the one moment it happens to be listening. A television switched on hours later is picked up whether or not ALSA's jack change reaches us as an event. Nothing polls once the room has the output it wants. A machine with no digital output at all never polls either, because there is nothing it could be waiting for. VOLUME AND MUTE --------------- This no longer touches either, except once on a machine it has never configured before. The old code unmuted the sink and forced it to 70% on every output change, which meant every television power cycle reset a volume the teacher had set, and could unmute an output that was muted on purpose. WirePlumber already remembers volume per device, so leaving it alone is both less code and the behaviour the room wants. """ import errno import json import os import select 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. Applied once in the life of a # machine, when this policy has never configured it before; after that the # room's own volume is never touched again. DEFAULT_VOLUME = "70%" # Several events arrive for one television being switched on. The burst is # drained to quiet before deciding, so one change is one decision rather # than one decision per event — which is what the old sleep-per-line did, # and it turned a single change into some forty seconds of churn. SETTLE_SECONDS = 1.5 # However chatty the server is, service it eventually. SETTLE_LIMIT_SECONDS = 6.0 # How long to wait for PipeWire's PulseAudio server at startup before # carrying on regardless. Bounded: a room with no sound server must still # reach the watch loop rather than spin here. SERVER_WAIT_SECONDS = 60 # While the room has not got the output it would prefer — a television # that is off, in practice — look again on a backing-off timer. Read-only # unless the answer actually changes, and capped, so a set that is never # switched on costs one `pactl list cards` a minute forever and changes # nothing. RECHECK_START_SECONDS = 5 RECHECK_LIMIT_SECONDS = 60 # A subscription that will not stay up is retried, but at a rate that # cannot become a spin on a machine with no sound server at all. RESUBSCRIBE_SECONDS = 3 RESUBSCRIBE_LIMIT_SECONDS = 60 # A subscription that lasted this long was a real one, so the next time it # drops it is worth reconnecting quickly again. Anything shorter is a # server that is not answering, and is backed off from. SUBSCRIPTION_HEALTHY_SECONDS = 10 # A card profile is only re-asserted so many times before the daemon stops # trying, so that a sink which never appears cannot become a profile # switching loop. PROFILE_ATTEMPTS = 3 # Remembers that this machine has been configured once, so the default # volume is a first-run courtesy rather than something applied at every # boot on top of a volume somebody chose. STATE_DIRECTORY = os.path.join( os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state"), "ekklesia-classroom", ) STATE_FILE = os.path.join(STATE_DIRECTORY, "audio-configured") def log(message): print(f"classroom-audio: {message}", flush=True) def pactl(*args, parse=False, quiet=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: if not quiet: log(f"pactl {' '.join(args)}: {error}") return None if result.returncode != 0: if not quiet: 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 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 configured_before(): return os.path.exists(STATE_FILE) def remember_configured(): try: os.makedirs(STATE_DIRECTORY, exist_ok=True) with open(STATE_FILE, "w") as handle: handle.write("This machine's audio output has been set up once.\n") except OSError as error: # Losing this only means the first-run volume is offered again # later. It is never worth failing the policy for. log(f"could not record first run: {error}") def sink_for(sinks, chosen): """The sink that carries the chosen port. A sink is tied to its card by properties["device.name"], which is the card's name — there is no top-level card field on a sink, whatever the older code here believed. Two of its three conditions could never match; the one that did returned whichever sink of that card came first, which is not necessarily the one for the port just chosen. """ card = chosen["card"] port = chosen["port"] on_card = [ sink for sink in sinks or [] if (sink.get("properties") or {}).get("device.name") == card ] if not on_card: return None for sink in on_card: if sink.get("active_port") == port: return sink for sink in on_card: names = { entry.get("name") for entry in sink.get("ports") or [] if isinstance(entry, dict) } if port in names: return sink return on_card[0] def active_profile_of(cards, card_name): for card in cards or []: if card.get("name") == card_name: return card.get("active_profile") return None def move_streams(name, index): """Send sound that is already playing to the new output. Changing the default sink does move streams that never asked for a particular one — measured on PipeWire 1.4.2, WirePlumber does it — but a stream WirePlumber has pinned to a device stays where it is, so the move is still made explicitly. """ moved = 0 for stream in pactl("list", "sink-inputs", parse=True) or []: if stream.get("sink") == index: continue if pactl("move-sink-input", str(stream.get("index")), name) is not None: moved += 1 return moved def apply(state): """Make the room match the decision. Returns True once it is settled.""" cards = pactl("list", "cards", parse=True, quiet=state.get("offline", False)) if cards is None: # Said once, not every few seconds for as long as the machine has # no sound server. if not state.get("offline"): log("cannot reach the sound server") state["offline"] = True return False if state.get("offline"): log("the sound server is back") state["offline"] = False want = preference() chosen = policy.choose(cards, want) if chosen is None: if state.get("port") is not None: log("no usable output") state["port"] = None return policy.settled(cards, want, None) key = (chosen["card"], chosen["port"]) changed = state.get("port") != key if changed: profile = policy.profile_for(chosen) active = active_profile_of(cards, chosen["card"]) # Only ever asked for when it is not already in place, so a room # that is working is never disturbed and the card is not switched # in a circle. if profile and profile != active: attempts = state.get("attempts", {}) tried = attempts.get(key, 0) if tried >= PROFILE_ATTEMPTS: log(f"giving up on {profile} for {chosen['card']}") return True attempts[key] = tried + 1 state["attempts"] = attempts pactl("set-card-profile", str(chosen["card"]), profile) # The sink only exists once the profile is in place. time.sleep(0.6) sinks = pactl("list", "sinks", parse=True) sink = sink_for(sinks, chosen) if sink is None: # Deliberately leaves state["port"] alone so the next look tries # again, and deliberately counted, so it cannot try forever. log(f"no sink yet for {chosen['port']} on {chosen['card']}") return False name = sink.get("name") if sink.get("name") != (pactl("get-default-sink") or "").strip(): pactl("set-default-sink", name) if changed: if not configured_before(): # The one time this machine's volume is set for it. pactl("set-sink-volume", name, DEFAULT_VOLUME) log(f"first run: {name} set to {DEFAULT_VOLUME}") remember_configured() moved = move_streams(name, sink.get("index")) log(f"output is {chosen['port']} on {chosen['card']} ({name})" + (f", moved {moved} playing stream(s)" if moved else "")) state["attempts"] = {} else: # A television coming back can leave a stream stranded on a sink # that has gone; nothing is logged when there is nothing to do. move_streams(name, sink.get("index")) state["port"] = key return policy.settled(cards, want, chosen) def wait_for_server(): """Let PipeWire's PulseAudio server finish starting. The race this avoids is the whole reason the rooms were silent, and ordering the unit after pipewire-pulse is not by itself enough: the socket can be there before the server will answer. """ deadline = time.monotonic() + SERVER_WAIT_SECONDS complained = False while time.monotonic() < deadline: if pactl("info", quiet=True) is not None: return True if not complained: log("waiting for the sound server") complained = True time.sleep(1) log("carrying on without having reached the sound server") return False def _interesting(line): return any(word in line for word in ("card", "server", "sink", "source")) def watch(): state = {} wait_for_server() settled = apply(state) delay = RECHECK_START_SECONDS backoff = RESUBSCRIBE_SECONDS complained = False while True: try: started = time.monotonic() process = subprocess.Popen( ["pactl", "subscribe"], stdout=subprocess.PIPE, # pactl says "Connection refused" on its own stderr every # time it cannot reach the server. Saying so once is this # daemon's job; letting pactl say it into the journal # every few seconds is not. stderr=subprocess.DEVNULL, ) except OSError as error: log(f"could not subscribe: {error}") time.sleep(backoff) backoff = min(backoff * 2, RESUBSCRIBE_LIMIT_SECONDS) continue stream = process.stdout pending = b"" # Everything that happened before this attached is invisible to it, # including everything that happened at boot. This is the line the # silent rooms were missing. settled = apply(state) try: while process.poll() is None: timeout = None if settled else delay ready, _, _ = select.select([stream], [], [], timeout) if not ready: # The room has not got the output it would prefer, so # look again without waiting to be told. settled = apply(state) delay = (RECHECK_START_SECONDS if settled else min(delay * 2, RECHECK_LIMIT_SECONDS)) continue try: chunk = os.read(stream.fileno(), 65536) except OSError as error: if error.errno == errno.EINTR: continue raise if not chunk: break pending += chunk lines = pending.split(b"\n") pending = lines.pop() if not any( _interesting(line.decode("utf-8", "replace")) for line in lines ): continue # Drain to quiet before deciding, so one television being # switched on is one decision and not a dozen. limit = time.monotonic() + SETTLE_LIMIT_SECONDS while time.monotonic() < limit: quiet = select.select([stream], [], [], SETTLE_SECONDS)[0] if not quiet: break if not os.read(stream.fileno(), 65536): break pending = b"" settled = apply(state) # Whatever it was waiting for, it starts looking eagerly # again rather than at whatever rate it had backed off to. delay = RECHECK_START_SECONDS except OSError as error: log(f"subscription ended: {error}") finally: process.terminate() try: process.wait(timeout=5) except subprocess.SubprocessError: process.kill() if time.monotonic() - started >= SUBSCRIPTION_HEALTHY_SECONDS: # It ran for a while, so this is an ordinary reconnect rather # than a server that is not there. backoff = RESUBSCRIBE_SECONDS complained = False if not complained: log("re-subscribing") complained = True time.sleep(backoff) backoff = min(backoff * 2, RESUBSCRIBE_LIMIT_SECONDS) 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'} " f"profile={policy.profile_for(chosen) if chosen else 'none'} " f"settled={policy.settled(cards, preference(), chosen)}") return 0 watch() return 0 if __name__ == "__main__": sys.exit(main())