Classroom_Desktop/scripts/classroom-audio

#!/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())