#!/bin/bash
# The audio policy daemon, driven through the situations a classroom is
# actually in, with the real script and a pactl that answers the way
# PipeWire really does.
#
# These are the properties a snapshot cannot show: that the room converges
# when the sound server was not ready at boot, that a television switched
# on later is picked up whether or not an event announces it, and that
# nothing is switched over and over when everything is already right.
set -uo pipefail

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

PYTHONPATH="$ROOT/lib" CLASSROOM_ROOT="$ROOT" python3 - <<'PY'
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time

ROOT = os.environ["CLASSROOM_ROOT"]
DAEMON = os.path.join(ROOT, "scripts", "classroom-audio")
MOCK = os.path.join(ROOT, "tests", "mock-pactl")

passed = failed = 0


def check(label, condition, detail=""):
    global passed, failed
    if condition:
        print(f"  ok   {label}")
        passed += 1
    else:
        print(f"  FAIL {label}")
        if detail:
            print(f"       {detail}")
        failed += 1


ANALOG_PROFILES = [
    "output:analog-stereo",
    "output:analog-stereo+input:analog-stereo",
    "output:analog-surround-40+input:analog-stereo",
]
HDMI_PROFILES = [
    "output:hdmi-stereo",
    "output:hdmi-stereo+input:analog-stereo",
    "output:hdmi-surround+input:analog-stereo",
]
ALL_PROFILES = ["off"] + ANALOG_PROFILES + HDMI_PROFILES

ANALOG_SINK = "alsa_output.pci-0000_00_1f.3.analog-stereo"
HDMI_SINK = "alsa_output.pci-0000_00_1f.3.hdmi-stereo"


def room(hdmi="not available", jack="not available",
         active="output:analog-stereo+input:analog-stereo",
         server_up=True):
    """The River Kids machine, as pactl described it over SSH.

    `jack` is what the onboard analog output reports. River Kids has
    nothing plugged into it, which is why it says "not available" and why
    that room prefers HDMI even with the television off: there is nowhere
    better for the sound to go. A room with speakers in the jack is a
    different room, and it is the one that can fall back.
    """
    return {
        "server_up": server_up,
        "default_sink": ANALOG_SINK,
        "levels": {},
        "streams": [],
        "cards": [{
            "name": "alsa_card.pci-0000_00_1f.3",
            "index": 47,
            "description": "Built-in Audio",
            "active_profile": active,
            "profiles": ALL_PROFILES,
            "ports": {
                "analog-output-lineout": {
                    "availability": jack,
                    "priority": 9900,
                    "profiles": ANALOG_PROFILES,
                },
                "hdmi-output-0": {
                    "availability": hdmi,
                    "priority": 5900,
                    "profiles": HDMI_PROFILES,
                },
            },
        }],
    }


class Lab:
    def __init__(self, state):
        self.dir = tempfile.mkdtemp(prefix="classroom-audio-")
        self.state_path = os.path.join(self.dir, "state.json")
        self.events = os.path.join(self.dir, "events")
        self.journal = os.path.join(self.dir, "journal")
        self.log = os.path.join(self.dir, "daemon.log")
        self.write(state)
        open(self.events, "w").close()
        open(self.journal, "w").close()

        # A pactl that is this mock and nothing else.
        self.bin = os.path.join(self.dir, "bin")
        os.makedirs(self.bin)
        os.symlink(MOCK, os.path.join(self.bin, "pactl"))

        self.process = None

    def write(self, state):
        with open(self.state_path, "w") as handle:
            json.dump(state, handle, indent=2)

    def read(self):
        with open(self.state_path) as handle:
            return json.load(handle)

    def edit(self, change):
        state = self.read()
        change(state)
        self.write(state)

    def start(self):
        environment = dict(os.environ)
        environment["PATH"] = self.bin + os.pathsep + environment["PATH"]
        environment["MOCK_AUDIO_STATE"] = self.state_path
        environment["MOCK_AUDIO_EVENTS"] = self.events
        environment["MOCK_AUDIO_JOURNAL"] = self.journal
        environment["XDG_STATE_HOME"] = os.path.join(self.dir, "state-home")
        # device.conf is not readable here, so the daemon falls back to
        # auto, which is what the classroom machines are set to.
        environment["PYTHONPATH"] = os.path.join(ROOT, "lib")

        self.process = subprocess.Popen(
            [sys.executable, DAEMON],
            stdout=open(self.log, "w"), stderr=subprocess.STDOUT,
            env=environment,
        )

    def stop(self):
        if self.process:
            self.process.terminate()
            try:
                self.process.wait(timeout=10)
            except subprocess.SubprocessError:
                self.process.kill()

    def journal_lines(self):
        with open(self.journal) as handle:
            return [line.strip() for line in handle if line.strip()]

    def daemon_log(self):
        with open(self.log) as handle:
            return handle.read()

    def await_log(self, text, timeout=40):
        """Wait for the daemon to SAY it has done something.

        Waiting on the sink cannot distinguish "the daemon chose this"
        from "the fixture already said this", and a test that cannot tell
        those apart races the daemon's first decision.
        """
        deadline = time.monotonic() + timeout

        while time.monotonic() < deadline:
            if text in self.daemon_log():
                return True
            time.sleep(0.25)

        return False

    def await_sink(self, name, timeout=40):
        """Wait for the room to be playing through a particular sink."""
        deadline = time.monotonic() + timeout

        while time.monotonic() < deadline:
            if self.read().get("default_sink") == name:
                return True
            time.sleep(0.25)

        return False

    def clean(self):
        self.stop()
        shutil.rmtree(self.dir, ignore_errors=True)


def profile_of(state):
    return state["cards"][0]["active_profile"]


def switches(lab):
    return [line for line in lab.journal_lines()
            if line.startswith("set-card-profile")]


# ---------------------------------------------------------------- 1
print("1. the computer starts with the television already on")
lab = Lab(room(hdmi="available"))
try:
    lab.start()
    check("HDMI is selected without anyone touching anything",
          lab.await_sink(HDMI_SINK), lab.daemon_log())
    check("and on a profile that keeps the card's input",
          profile_of(lab.read()) == "output:hdmi-stereo+input:analog-stereo",
          profile_of(lab.read()))
finally:
    lab.clean()

# ---------------------------------------------------------------- 2
print("2. the television is off at boot, then switched on (with an event)")
lab = Lab(room(hdmi="not available", jack="available"))
try:
    lab.start()
    check("the room falls back to the jack rather than going silent",
          lab.await_log("output is analog-output-lineout"), lab.daemon_log())

    def switch_on(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(switch_on)
    with open(lab.events, "a") as handle:
        handle.write("Event 'change' on card #47\n")

    check("the television is picked up when it announces itself",
          lab.await_sink(HDMI_SINK), lab.daemon_log())
finally:
    lab.clean()

# ---------------------------------------------------------------- 3
print("3. the television is switched on and NOTHING announces it")
# ALSA's jack change does not always surface as a pactl event. The old
# policy could only ever act on an event, so this was a room that stayed
# silent all morning.
lab = Lab(room(hdmi="not available", jack="available"))
try:
    lab.start()
    check("the room starts on the jack",
          lab.await_log("output is analog-output-lineout"), lab.daemon_log())

    def switch_on(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(switch_on)
    # Deliberately no event written.
    check("the television is still picked up, by looking again",
          lab.await_sink(HDMI_SINK), lab.daemon_log())
finally:
    lab.clean()

# ---------------------------------------------------------------- 4
print("4. the boot race that silenced the rooms on 0.2.2")
# The daemon starts before PipeWire's PulseAudio server will answer. The
# old code failed its single startup apply, subscribed successfully a
# moment later, and then waited forever for an event that had already
# happened — leaving the card on WirePlumber's own profile, which is
# exactly what was found in River Kids.
lab = Lab(room(hdmi="available", server_up=False))
try:
    lab.start()
    time.sleep(3)
    check("nothing is applied while the server refuses connections",
          lab.read().get("default_sink") == ANALOG_SINK)

    def server_up(state):
        state["server_up"] = True

    lab.edit(server_up)
    # No event is written: at a real boot the card's events all happen
    # while the daemon cannot connect, and none arrive afterwards.
    check("the room converges on HDMI once the server answers",
          lab.await_sink(HDMI_SINK), lab.daemon_log())
    check("and the card is left on a profile the policy chose",
          profile_of(lab.read()) == "output:hdmi-stereo+input:analog-stereo",
          profile_of(lab.read()))
finally:
    lab.clean()

# ---------------------------------------------------------------- 5
print("5. the television is switched off and on again while a lesson plays")
# Speakers in the jack, so there is somewhere to fall back TO. The room
# that has nothing in the jack is scenario 5b below.
lab = Lab(room(hdmi="available", jack="available"))
try:
    lab.start()
    check("the lesson starts out on the television", lab.await_sink(HDMI_SINK))

    def add_stream(state):
        state["streams"] = [{
            "index": 700,
            "sink": 41,
            "sink_name": HDMI_SINK,
            "properties": {"application.name": "Chromium"},
        }]

    lab.edit(add_stream)

    def switch_off(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = \
            "not available"

    lab.edit(switch_off)
    with open(lab.events, "a") as handle:
        handle.write("Event 'change' on card #47\n")

    check("it falls back to the jack while the set is off",
          lab.await_sink(ANALOG_SINK), lab.daemon_log())

    def switch_on(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(switch_on)
    with open(lab.events, "a") as handle:
        handle.write("Event 'change' on card #47\n")

    check("and comes back to the television when it returns",
          lab.await_sink(HDMI_SINK), lab.daemon_log())

    time.sleep(2)
    stream = lab.read()["streams"][0]
    check("the lesson that never stopped playing is on the television too",
          stream["sink_name"] == HDMI_SINK,
          f"stream left on {stream['sink_name']}")
finally:
    lab.clean()

# ---------------------------------------------------------------- 5b
print("5b. River Kids: the same power cycle with nothing in the jack")
# This room has no speakers in the analog socket, so when the television
# goes off there is nowhere better for the sound to go and the policy is
# right to stay pointed at the set. What it must not do is thrash between
# two outputs that are both unusable, or lose the television when it comes
# back.
lab = Lab(room(hdmi="available", jack="not available"))
try:
    lab.start()
    check("the lesson starts on the television", lab.await_sink(HDMI_SINK))

    def switch_off(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = \
            "not available"

    lab.edit(switch_off)
    with open(lab.events, "a") as handle:
        handle.write("Event 'change' on card #47\n")
    time.sleep(8)

    check("it stays on the set rather than moving to an empty socket",
          lab.read()["default_sink"] == HDMI_SINK,
          lab.read()["default_sink"])
    check("and does not switch the card back and forth while it is off",
          len(switches(lab)) <= 2, f"profile switches: {switches(lab)}")

    def switch_on(state):
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(switch_on)
    check("and the television still works when it comes back",
          lab.await_sink(HDMI_SINK), lab.daemon_log())
finally:
    lab.clean()

# ---------------------------------------------------------------- 6
print("6. a television that is never switched on")
lab = Lab(room(hdmi="not available", jack="available"))
try:
    lab.start()
    check("the room settles on the jack",
          lab.await_log("output is analog-output-lineout"), lab.daemon_log())
    time.sleep(25)
    moves = switches(lab)
    check("the card is not switched over and over while it waits",
          len(moves) <= 2, f"{len(moves)} profile switches: {moves}")
    check("and the daemon is still alive rather than having given up",
          lab.process.poll() is None)
    check("and it never went silent",
          lab.read()["default_sink"] == ANALOG_SINK)
finally:
    lab.clean()

# ---------------------------------------------------------------- 7
print("7. a room that is already right is left alone")
lab = Lab(room(hdmi="available"))
try:
    lab.start()
    check("it selects the television once", lab.await_sink(HDMI_SINK))
    time.sleep(20)
    check("and then changes nothing at all, however long it runs",
          len(switches(lab)) == 1,
          f"profile switches: {switches(lab)}")
    check("no repeated default-sink writes either",
          len([line for line in lab.journal_lines()
               if line.startswith("set-default-sink")]) == 1,
          str(lab.journal_lines()))
finally:
    lab.clean()

# ---------------------------------------------------------------- 8
print("8. the teacher's volume and mute are left alone")
lab = Lab(room(hdmi="not available", jack="available"))
try:
    lab.start()
    check("the room starts on the jack",
          lab.await_log("output is analog-output-lineout"), lab.daemon_log())

    # The teacher turns it down and mutes it on purpose.
    def teacher(state):
        state["levels"] = {
            ANALOG_SINK: {"volume": 35, "mute": True},
            HDMI_SINK: {"volume": 55, "mute": True},
        }
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(teacher)
    check("the television is picked up", lab.await_sink(HDMI_SINK),
          lab.daemon_log())
    time.sleep(2)

    levels = lab.read()["levels"]
    check("the television's own volume is not overwritten",
          levels[HDMI_SINK]["volume"] == 55,
          f"volume became {levels[HDMI_SINK]['volume']}")
    check("an output muted on purpose is not unmuted",
          levels[HDMI_SINK]["mute"] is True)
    volume_commands = [line for line in lab.journal_lines()
                       if line.startswith("set-sink-volume")]
    check("the only volume ever set is the one first run on the jack",
          len(volume_commands) == 1
          and ANALOG_SINK in volume_commands[0]
          and "70%" in volume_commands[0],
          str(volume_commands))
    check("and the television's volume was never written",
          not [line for line in volume_commands if HDMI_SINK in line],
          str(volume_commands))
    check("and mute was never touched",
          not [line for line in lab.journal_lines()
               if line.startswith("set-sink-mute")],
          str(lab.journal_lines()))
finally:
    lab.clean()

# ---------------------------------------------------------------- 9
print("9. sound that is already playing follows the television")
lab = Lab(room(hdmi="not available", jack="available"))
try:
    lab.start()
    check("the room starts on the jack",
          lab.await_log("output is analog-output-lineout"), lab.daemon_log())

    def lesson_and_tv(state):
        state["streams"] = [
            {"index": 701, "sink": 41, "sink_name": ANALOG_SINK,
             "properties": {"application.name": "Wonder Ink"}},
            {"index": 702, "sink": 41, "sink_name": ANALOG_SINK,
             "properties": {"application.name": "Chromium"}},
        ]
        state["cards"][0]["ports"]["hdmi-output-0"]["availability"] = "available"

    lab.edit(lesson_and_tv)
    check("the television is picked up", lab.await_sink(HDMI_SINK),
          lab.daemon_log())
    time.sleep(2)

    streams = lab.read()["streams"]
    check("Wonder Ink follows without being restarted",
          streams[0]["sink_name"] == HDMI_SINK,
          f"left on {streams[0]['sink_name']}")
    check("and so does Chrome",
          streams[1]["sink_name"] == HDMI_SINK,
          f"left on {streams[1]['sink_name']}")
    check("a stream that starts afterwards lands on the television",
          lab.read()["default_sink"] == HDMI_SINK)
finally:
    lab.clean()

# ---------------------------------------------------------------- 10
print("10. a machine with no digital output never waits for one")
laptop = {
    "server_up": True,
    "default_sink": "alsa_output.pci-0000_00_1f.3.analog-stereo",
    "levels": {},
    "streams": [],
    "cards": [{
        "name": "alsa_card.pci-0000_00_1f.3",
        "index": 47,
        "description": "Built-in Audio",
        "active_profile": "output:analog-stereo+input:analog-stereo",
        "profiles": ["off"] + ANALOG_PROFILES,
        "ports": {
            "analog-output-speaker": {
                "availability": "available",
                "priority": 10000,
                "profiles": ANALOG_PROFILES,
            },
        },
    }],
}
lab = Lab(laptop)
try:
    lab.start()
    check("it uses the speakers it has",
          lab.await_sink("alsa_output.pci-0000_00_1f.3.analog-stereo"))
    time.sleep(15)
    check("and settles without hunting for a television that cannot exist",
          len(switches(lab)) <= 1, f"profile switches: {switches(lab)}")
finally:
    lab.clean()

print()
print(f"{passed} passed, {failed} failed")
raise SystemExit(1 if failed else 0)
PY