Classroom_Desktop/lib/classroom_audio.py

"""Deciding which output a classroom should be playing through.

This module is only the decision. It takes the state of the sound cards as
PipeWire reports it and returns the port that should be in use, so the
policy can be tested against the situation a real room was actually in
rather than against a running machine.

The room that prompted this had a Samsung television on HDMI with working
PCM audio, nothing at all plugged into the onboard analog jack, and
WirePlumber had chosen analog anyway. A teacher who fixed it by hand then
had to restart the browser before any sound came out.

Two rules came out of that:

* a port that reports itself unavailable is not a candidate, and "analog
  exists" is not evidence that a cable is in it; and
* the decision has to be re-made whenever the hardware situation changes,
  not once at boot, because a television that is switched on after the
  computer is the normal case in a classroom.

A third came out of the Sunday the rooms were silent on 0.2.2. The
decision was right and was never asked for: the part that applies it had
made it once, at startup, before PipeWire's PulseAudio server was
listening, and then waited for an event to tell it to try again. So this
module also answers whether the answer it just gave is the best one the
hardware could ever give, which is what lets the caller know whether it
still has something to wait for.

The shapes here are the ones real pactl emits, which are not the ones the
first set of fixtures assumed. `ports` is an object keyed by port name,
and a port carries no `direction` at all; availability is the phrase
"availability unknown" rather than an empty string. Measured against
PipeWire 1.4.2 and WirePlumber 0.5.8, which is what the rooms run.
"""

# Port name prefixes PipeWire uses for the digital display outputs.
DIGITAL = ("hdmi-output", "displayport-output", "hdmi", "dp-")
ANALOG = ("analog-output", "lineout", "headphones", "speaker")

AVAILABLE = "available"
UNAVAILABLE = "not available"
UNKNOWN = "unknown"

AUTO = "auto"
HDMI = "hdmi"
ANALOG_ONLY = "analog"


def _kind(port_name):
    name = (port_name or "").lower()

    for prefix in DIGITAL:
        if name.startswith(prefix):
            return HDMI

    for prefix in ANALOG:
        if name.startswith(prefix):
            return ANALOG_ONLY

    return None


def _availability(port):
    value = (port.get("availability") or "").strip().lower()

    if value in ("available", "yes"):
        return AVAILABLE

    if value in ("not available", "no"):
        return UNAVAILABLE

    return UNKNOWN


def candidates(cards):
    """Every output port, tagged with what it is and whether it is there.

    `cards` is the structure `pactl -f json list cards` produces.
    """
    found = []

    for card in cards or []:
        for name, port in (card.get("ports") or {}).items():
            if (port.get("direction") or "Output").lower() != "output":
                continue

            kind = _kind(name)

            if kind is None:
                continue

            found.append({
                "card": card.get("name"),
                "card_index": card.get("index"),
                "port": name,
                "kind": kind,
                "availability": _availability(port),
                "priority": port.get("priority") or 0,
                "profiles": list(port.get("profiles") or []),
            })

    return found


def choose(cards, preference=AUTO):
    """The port the room should be playing through, or None.

    An explicit preference is obeyed as far as the hardware allows, and
    falls back rather than leaving the room silent.
    """
    found = candidates(cards)

    if not found:
        return None

    def ranked(kind, accept):
        pool = [
            item for item in found
            if item["kind"] == kind and item["availability"] in accept
        ]
        # Highest priority first, then a stable order so a machine with two
        # identical outputs does not pick a different one each boot.
        return sorted(
            pool,
            key=lambda item: (-item["priority"], item["card"] or "", item["port"]),
        )

    HERE = (AVAILABLE,)
    SILENT = (AVAILABLE, UNKNOWN)
    ANYTHING = (AVAILABLE, UNKNOWN, UNAVAILABLE)

    # A television that is on says so. A port that reports nothing is still
    # worth using if there is no better answer, but never ahead of one that
    # reports itself connected. A port that reports itself absent is chosen
    # only when the alternative is a silent room — or when somebody has
    # named it explicitly, which is what an override is for.
    order = {
        AUTO: [
            (HDMI, HERE), (ANALOG_ONLY, HERE),
            (HDMI, SILENT), (ANALOG_ONLY, SILENT),
            (HDMI, ANYTHING), (ANALOG_ONLY, ANYTHING),
        ],
        HDMI: [
            (HDMI, HERE), (HDMI, SILENT), (HDMI, ANYTHING),
            (ANALOG_ONLY, HERE), (ANALOG_ONLY, SILENT), (ANALOG_ONLY, ANYTHING),
        ],
        ANALOG_ONLY: [
            (ANALOG_ONLY, HERE), (ANALOG_ONLY, SILENT), (ANALOG_ONLY, ANYTHING),
            (HDMI, HERE), (HDMI, SILENT), (HDMI, ANYTHING),
        ],
    }[preference]

    for kind, accept in order:
        pool = ranked(kind, accept)

        if pool:
            return pool[0]

    return None


def profile_for(port):
    """The card profile to switch to so that a port is usable.

    A port lists the profiles it appears in. The classroom wants the
    plainest output the card offers — stereo rather than one of the
    surround arrangements a television will not use — and, among equals,
    one that leaves an input in place.

    Dropping the input is what the narrower rule here used to do, and it
    is the difference between output:hdmi-stereo and the
    output:hdmi-stereo+input:analog-stereo an administrator reaches for by
    hand. Nothing in the room records sound today, but taking the
    microphone away is not this policy's business, and choosing the same
    profile a person would choose keeps the two from undoing each other.
    """
    profiles = [name for name in port.get("profiles") or [] if name != "off"]

    if not profiles:
        return None

    outputs = [name for name in profiles if "output" in name] or profiles

    def rank(name):
        # The output half decides first, so stereo beats surround however
        # long the rest of the name is.
        output = name.split("+", 1)[0]

        return (len(output), 0 if "+input:" in name else 1, len(name), name)

    return sorted(outputs, key=rank)[0]


def preferred_kind(preference):
    """The kind of output this preference is holding out for."""
    return ANALOG_ONLY if preference == ANALOG_ONLY else HDMI


def settled(cards, preference, chosen):
    """Whether waiting any longer could produce a better output.

    False means the room is playing through something, but the output it
    would rather have exists on this machine and is not usable yet — a
    television that is switched off, in practice. The caller uses this to
    decide whether it still has a reason to look again, so that a room
    which has what it wants is driven by events alone and never polls.
    """
    wanted = preferred_kind(preference)

    if chosen is not None and chosen["kind"] == wanted:
        return True

    # Nothing of the preferred kind is wired into this machine at all, so
    # there is nothing to wait for. A laptop with only a headphone jack is
    # settled the moment it picks that jack.
    return not any(item["kind"] == wanted for item in candidates(cards))