From 142f4484d822645bf5aecf4e6a63d1680ea59218 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 11 Jul 2026 18:12:39 +0200 Subject: [PATCH] Play alarm audio on every sink, not one hardcoded output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit At wake time PipeWire often has only the auto_null sink (verified from journalctl on 2026-07-11: "ONLY auto_null PipeWire sink available — alarm will be SILENT"). The old code forced only the single HDMI monitor card on and waited for its one sink; when that never enumerated, all audio drained into auto_null = silence. Now activation forces *every* card's output profile on, unmutes and maxes every sink, and the beep loop plays on *all* sinks each cycle (paplay --device), re-enumerating so late/added outputs are covered. A switched-off external speaker can't be revived by software, so every other output is tried in parallel instead of betting on one. Prior per-card profile and per-sink volume/mute are captured and restored on dismiss. Concurrent paplay children are now reaped each cycle (was 1/cycle, would have been N/cycle of over the 30-min loud phase). Extract the pactl sink/card management into wake_alarm/_sinks.py to keep _audio.py under the 500-line limit. 100% branch coverage retained. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_018SJ6HYXhBmYs3SiGBZv1rF --- wake_alarm/_alarm.py | 28 +- wake_alarm/_audio.py | 224 +++------- wake_alarm/_sinks.py | 354 +++++++++++++++ wake_alarm/tests/test_alarm.py | 50 --- wake_alarm/tests/test_alarm_part2.py | 6 +- wake_alarm/tests/test_alarm_part3.py | 6 +- wake_alarm/tests/test_alarm_part4.py | 21 +- wake_alarm/tests/test_alarm_playback.py | 136 ++++++ wake_alarm/tests/test_alarm_sinks.py | 556 ++++++++++++++++-------- 9 files changed, 966 insertions(+), 415 deletions(-) create mode 100644 wake_alarm/_sinks.py create mode 100644 wake_alarm/tests/test_alarm_playback.py diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index 1a4aec0..8604015 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -22,16 +22,13 @@ from gatelock import GateRoot, LockConfig, LockWindow from wake_alarm._alarm_display import _restore_display, _wake_display from wake_alarm._audio import ( - _activate_alarm_audio, _beep_loud, _beep_medium, _beep_soft, _max_fans, - _play_on_extra_devices, - _restore_alarm_audio, + _play_on_all_sinks, _restore_fans, _set_max_brightness, - _warn_if_no_real_sink, ) from wake_alarm._challenges import ( _Challenge, @@ -51,6 +48,12 @@ from wake_alarm._constants import ( SOFT_BEEP_INTERVAL, ) from wake_alarm._logging_setup import configure_logging +from wake_alarm._sinks import ( + _activate_all_outputs, + _AudioState, + _restore_audio_state, + _warn_if_no_real_sink, +) from wake_alarm._smart_plug import turn_off_plug, turn_on_plug from wake_alarm._state import ( save_wake_state, @@ -96,7 +99,7 @@ class _AlarmHardware: """Hardware state captured at alarm start, restored when it closes.""" fan_state: bool - audio_restore: str | None + audio_restore: _AudioState | None class WakeAlarm: @@ -131,7 +134,7 @@ class WakeAlarm: self._schedule_skip_window_close() self._start_beep_thread() self._hardware = _AlarmHardware( - fan_state=_max_fans(), audio_restore=_activate_alarm_audio() + fan_state=_max_fans(), audio_restore=_activate_all_outputs() ) self._start_screen_flash() self._lock.grab_input() @@ -153,7 +156,7 @@ class WakeAlarm: """Restore fans/audio/display/plug; runs on every exit path, even SIGTERM.""" self._stop_beep.set() _restore_fans(active=self._hardware.fan_state) - _restore_alarm_audio(self._hardware.audio_restore) + _restore_audio_state(self._hardware.audio_restore) _restore_display() turn_off_plug() @@ -414,16 +417,16 @@ class WakeAlarm: elapsed_minutes = (time.monotonic() - self._alarm_start) / 60.0 if elapsed_minutes < PHASE_SOFT_END: - _play_on_extra_devices(440) + _play_on_all_sinks(440) _beep_soft() self._stop_beep.wait(SOFT_BEEP_INTERVAL) elif elapsed_minutes < PHASE_MEDIUM_END: - _play_on_extra_devices(1000) + _play_on_all_sinks(1000) _beep_medium() self._stop_beep.wait(MEDIUM_BEEP_INTERVAL) else: freq = 800 if int(elapsed_minutes * 10) % 2 == 0 else 1200 - _play_on_extra_devices(freq) + _play_on_all_sinks(freq) _beep_loud(freq) self._stop_beep.wait(LOUD_TOGGLE_INTERVAL) @@ -484,9 +487,8 @@ def main() -> None: ) _warn_if_no_real_sink() _wake_display() - # Wait for the G27Q to power on and enumerate its HDMI audio sink. - # Without this delay the sink often isn't visible yet when _activate_alarm_audio - # runs, making the alarm silent when the monitor was physically off at wake time. + # Wait for the G27Q to power on and enumerate its HDMI audio sink before + # _activate_all_outputs runs, else the alarm is silent if it was off. time.sleep(DISPLAY_WAKE_WAIT_SECONDS) _set_max_brightness() turn_on_plug() diff --git a/wake_alarm/_audio.py b/wake_alarm/_audio.py index 808aa1a..3b08a78 100644 --- a/wake_alarm/_audio.py +++ b/wake_alarm/_audio.py @@ -4,7 +4,6 @@ from __future__ import annotations import logging import math -import os from pathlib import Path import shutil import struct @@ -14,12 +13,10 @@ import tempfile import time import wave -from wake_alarm._constants import ( - ALARM_AUDIO_CARD, - ALARM_AUDIO_PROFILE, - ALARM_AUDIO_SINK, - ALARM_AUDIO_SINK_POLL_SECONDS, - ALARM_AUDIO_SINK_WAIT_SECONDS, +from wake_alarm._sinks import ( + _ensure_sink_audible, + _list_real_sinks, + _pactl_path, ) _logger = logging.getLogger(__name__) @@ -45,10 +42,6 @@ _PCSPKR_SND_TONE: int = 0x02 # struct input_event: timeval (long sec, long usec), u16 type, u16 code, s32 val _PCSPKR_EVENT_FMT: str = "llHHi" -# Extra PipeWire sinks to always play alarm audio on (alongside the default). -# alsa_output...hdmi-stereo = GA102 → G27Q (has built-in speaker, always on). -_EXTRA_PIPEWIRE_SINKS: tuple[str, ...] = ("alsa_output.pci-0000_01_00.1.hdmi-stereo",) - # NCT Super I/O chip names that expose a single pwm1 fan control channel. _NCT_CHIP_NAMES: frozenset[str] = frozenset( { @@ -222,28 +215,77 @@ def _play_tone(frequency: int) -> None: _beep_soft() -def _play_on_extra_devices(frequency: int) -> None: - """Fire-and-forget: play a sine tone on each extra PipeWire sink.""" - try: - path = _speaker_test_path() - except FileNotFoundError: - _logger.warning("speaker-test missing; skipping extra-device beep") +def _play_on_all_sinks(frequency: int) -> None: + """Play the tone WAV on every real PipeWire sink, simultaneously. + + Re-enumerates sinks on every call so outputs that appear late (the G27Q + HDMI sink can take ~15 s to power on) are automatically included without + re-running activation. All sinks are launched concurrently, then reaped so + the 30-minute alarm loop does not accumulate zombie ``paplay`` children. + """ + pactl = _pactl_path() + if pactl is None: + return + paplay = shutil.which("paplay") + if paplay is None: + _logger.warning("paplay not on PATH; skipping per-sink alarm playback") return - for sink in _EXTRA_PIPEWIRE_SINKS: - _play_tone_on_sink(path, sink, frequency) - - -def _play_tone_on_sink(path: str, sink: str, frequency: int) -> None: - """Launch speaker-test for *sink*; log a warning on OSError.""" try: - subprocess.Popen( - [path, "-t", "sine", "-f", str(frequency), "-l", "1"], - env={**os.environ, "PIPEWIRE_NODE": sink}, + wav = _ensure_tone_wav(frequency) + except OSError: + _logger.warning( + "Could not generate tone WAV at %d Hz for per-sink playback", + frequency, + exc_info=True, + ) + return + procs: list[subprocess.Popen[bytes]] = [] + for sink in _list_real_sinks(pactl): + # Re-assert audibility every cycle: a source that is muted, low, or only + # just appeared (a monitor powering on mid-alarm) is made audible before + # we play on it — we never bet on one sink staying the chosen output. + _ensure_sink_audible(pactl, sink) + proc = _play_wav_on_sink(paplay, sink, wav) + if proc is not None: + procs.append(proc) + _reap_players(procs) + + +def _play_wav_on_sink( + paplay: str, + sink: str, + wav: Path, +) -> subprocess.Popen[bytes] | None: + """Launch ``paplay --device=`` for *wav*; return the process or None.""" + try: + return subprocess.Popen( + [paplay, f"--device={sink}", str(wav)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) except OSError: _logger.warning("Failed to play tone on sink %s", sink, exc_info=True) + return None + + +def _reap_players(procs: list[subprocess.Popen[bytes]]) -> None: + """Wait for each concurrent paplay to finish, killing any that hang. + + The tone is ~1.5 s, so all launched players finish quickly and are reaped + here rather than left as ```` children for the alarm's lifetime. + """ + for proc in procs: + _reap_player(proc) + + +def _reap_player(proc: subprocess.Popen[bytes]) -> None: + """Wait for one paplay process, killing and reaping it if it hangs.""" + try: + proc.wait(timeout=_TONE_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + _logger.warning("paplay hung; killing pid %d", proc.pid) + proc.kill() + proc.wait() def _find_fan_hwmon() -> str | None: @@ -361,133 +403,3 @@ def _beep_medium(frequency: int = 1000) -> None: def _beep_loud(frequency: int = 1000) -> None: """Play a loud sine tone via paplay/aplay/speaker-test.""" _play_tone(frequency) - - -def _pactl_path() -> str | None: - """Return the absolute path to pactl, or None when not installed.""" - return shutil.which("pactl") - - -def _alarm_sink_present(pactl: str) -> bool: - """Return True when the dedicated alarm HDMI sink exists in PipeWire.""" - try: - result = subprocess.run( - [pactl, "list", "short", "sinks"], - capture_output=True, - timeout=3, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning("pactl list sinks failed", exc_info=True) - return False - return ALARM_AUDIO_SINK in result.stdout.decode(errors="replace") - - -def _current_default_sink(pactl: str) -> str | None: - """Return the current default sink name, or None on failure / empty.""" - try: - result = subprocess.run( - [pactl, "get-default-sink"], - capture_output=True, - timeout=3, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning("pactl get-default-sink failed", exc_info=True) - return None - name = result.stdout.decode(errors="replace").strip() - return name or None - - -def _activate_alarm_audio() -> str | None: - """Force the monitor's HDMI output on and route the alarm to it. - - At wake time the Bluetooth speaker is disconnected and PipeWire only has the - ``auto_null`` sink, so the alarm is silent. This forces the HDMI card - profile on, waits for its sink to appear, makes it the default sink, and - raises it to full volume - empirically the only output audible on this - machine at wake time (the G27Q monitor's built-in speaker). - - Returns: - The previous default sink name (to restore on close), or ``None`` when - the alarm audio sink could not be activated. - """ - pactl = _pactl_path() - if pactl is None: - _logger.warning("pactl not on PATH; cannot activate alarm audio") - return None - subprocess.run( - [pactl, "set-card-profile", ALARM_AUDIO_CARD, ALARM_AUDIO_PROFILE], - capture_output=True, - timeout=3, - check=False, - ) - attempts = max( - 1, - int(ALARM_AUDIO_SINK_WAIT_SECONDS / ALARM_AUDIO_SINK_POLL_SECONDS), - ) - for _ in range(attempts): - if _alarm_sink_present(pactl): - break - time.sleep(ALARM_AUDIO_SINK_POLL_SECONDS) - else: - _logger.warning( - "Alarm audio sink %s did not appear after %.0fs; alarm may be silent", - ALARM_AUDIO_SINK, - ALARM_AUDIO_SINK_WAIT_SECONDS, - ) - return None - old_default = _current_default_sink(pactl) - for cmd in ( - [pactl, "set-default-sink", ALARM_AUDIO_SINK], - [pactl, "set-sink-mute", ALARM_AUDIO_SINK, "0"], - [pactl, "set-sink-volume", ALARM_AUDIO_SINK, "100%"], - ): - subprocess.run(cmd, capture_output=True, timeout=3, check=False) - _logger.warning("Alarm audio routed to %s at 100%%", ALARM_AUDIO_SINK) - return old_default - - -def _restore_alarm_audio(old_default: str | None) -> None: - """Restore the default sink captured by :func:`_activate_alarm_audio`.""" - if old_default is None: - return - pactl = _pactl_path() - if pactl is None: - return - subprocess.run( - [pactl, "set-default-sink", old_default], - capture_output=True, - timeout=3, - check=False, - ) - - -def _warn_if_no_real_sink() -> None: - """Log a loud warning if PipeWire only has the auto_null sink.""" - pactl = _pactl_path() - if pactl is None: - _logger.warning("pactl not on PATH; cannot verify audio sinks") - return - try: - result = subprocess.run( - [pactl, "list", "short", "sinks"], - capture_output=True, - timeout=5, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning("pactl list sinks failed", exc_info=True) - return - sinks_text = result.stdout.decode(errors="replace").strip() - sink_names = [ - line.split("\t")[1] for line in sinks_text.splitlines() if "\t" in line - ] - real_sinks = [s for s in sink_names if s != "auto_null"] - if not real_sinks: - _logger.warning( - "ONLY auto_null PipeWire sink available — alarm will be SILENT. Sinks: %s", - sink_names or "", - ) - else: - _logger.info("Audio sinks available: %s", sink_names) diff --git a/wake_alarm/_sinks.py b/wake_alarm/_sinks.py new file mode 100644 index 0000000..fc16170 --- /dev/null +++ b/wake_alarm/_sinks.py @@ -0,0 +1,354 @@ +"""PipeWire sink/card enumeration, activation, and restore for the wake alarm. + +At wake time the Bluetooth speaker is gone and WirePlumber leaves most card +profiles ``off``, so the machine is silent. This module forces every card's +output on, unmutes and maxes every sink, and captures the prior state so the +alarm can restore it on dismiss. The alarm plays on *all* sinks every beep +cycle (see :func:`wake_alarm._audio._play_on_all_sinks`) rather than betting on +a single chosen output. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import logging +import re +import shutil +import subprocess +import time + +from wake_alarm._constants import ( + ALARM_AUDIO_SINK, + ALARM_AUDIO_SINK_POLL_SECONDS, + ALARM_AUDIO_SINK_WAIT_SECONDS, +) + +_logger = logging.getLogger(__name__) + + +def _pactl_path() -> str | None: + """Return the absolute path to pactl, or None when not installed.""" + return shutil.which("pactl") + + +def _pactl_text(pactl: str, *args: str, timeout: float = 5.0) -> str | None: + """Run ``pactl `` and return decoded stdout, or None on failure.""" + try: + result = subprocess.run( + [pactl, *args], + capture_output=True, + timeout=timeout, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("pactl %s failed", " ".join(args), exc_info=True) + return None + return result.stdout.decode(errors="replace") + + +def _list_real_sinks(pactl: str) -> list[str]: + """Return every non-``auto_null`` sink name from ``pactl list short sinks``.""" + text = _pactl_text(pactl, "list", "short", "sinks") + if text is None: + return [] + names = [line.split("\t")[1] for line in text.splitlines() if "\t" in line] + return [name for name in names if name != "auto_null"] + + +def _list_cards(pactl: str) -> list[str]: + """Return every card name from ``pactl list short cards``.""" + text = _pactl_text(pactl, "list", "short", "cards") + if text is None: + return [] + return [line.split("\t")[1] for line in text.splitlines() if "\t" in line] + + +def _current_default_sink(pactl: str) -> str | None: + """Return the current default sink name, or None on failure / empty.""" + try: + result = subprocess.run( + [pactl, "get-default-sink"], + capture_output=True, + timeout=3, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("pactl get-default-sink failed", exc_info=True) + return None + name = result.stdout.decode(errors="replace").strip() + return name or None + + +def _ensure_sink_audible(pactl: str, sink: str) -> None: + """Unmute *sink* and raise it to 100%, so playback on it is actually heard.""" + for args in ( + ["set-sink-mute", sink, "0"], + ["set-sink-volume", sink, "100%"], + ): + subprocess.run( + [pactl, *args], + capture_output=True, + timeout=3, + check=False, + ) + + +def _parse_profile_line(line: str) -> tuple[str, int] | None: + """Parse one ``output:*`` profile line into ``(name, priority)``. + + Returns None for non-output profiles and for profiles marked + ``available: no`` (an output whose sink cannot currently be produced, + e.g. an HDMI port with nothing plugged in). + """ + if not line.startswith("output:"): + return None + name, sep, rest = line.partition(": ") + if not sep or "available: no" in rest: + return None + match = re.search(r"priority:\s*(\d+)", rest) + priority = int(match.group(1)) if match else 0 + return name, priority + + +@dataclass +class _CardInfo: + """A card's active profile and its available ``output:*`` profiles.""" + + active_profile: str | None + # Available output profiles, best first (stereo preferred, then priority). + output_profiles: list[str] + + +@dataclass +class _CardAccum: + """Mutable accumulator for one card block while parsing ``list cards``.""" + + name: str | None = None + active: str | None = None + profiles: list[tuple[str, int]] = field(default_factory=list) + + +def _parse_cards(text: str) -> dict[str, _CardInfo]: + """Parse ``pactl list cards`` (full) into ``{card_name: _CardInfo}``.""" + cards: dict[str, _CardInfo] = {} + cur: _CardAccum | None = None + in_profiles = False + for raw in text.splitlines(): + if raw.startswith("Card #"): + _store_card(cards, cur) + cur = _CardAccum() + in_profiles = False + continue + if cur is None: + continue + stripped = raw.strip() + if stripped.startswith("Name:"): + cur.name = stripped.split(":", 1)[1].strip() + in_profiles = False + elif stripped.startswith("Active Profile:"): + cur.active = stripped.split(":", 1)[1].strip() + in_profiles = False + elif stripped == "Profiles:": + in_profiles = True + elif in_profiles: + parsed = _parse_profile_line(stripped) + if parsed is not None: + cur.profiles.append(parsed) + _store_card(cards, cur) + return cards + + +def _store_card(cards: dict[str, _CardInfo], cur: _CardAccum | None) -> None: + """Finalise one parsed card block and add it to *cards* (skip if unnamed).""" + if cur is None or cur.name is None: + return + # Prefer a stereo profile, then the highest priority. + ordered = sorted( + cur.profiles, key=lambda np: (0 if "stereo" in np[0] else 1, -np[1]) + ) + cards[cur.name] = _CardInfo( + active_profile=cur.active, + output_profiles=[profile for profile, _ in ordered], + ) + + +def _parse_sinks(text: str) -> tuple[dict[str, str], dict[str, bool]]: + """Parse ``pactl list sinks`` into ``({sink: raw_volume}, {sink: muted})``.""" + volumes: dict[str, str] = {} + mutes: dict[str, bool] = {} + name: str | None = None + for raw in text.splitlines(): + if raw.startswith("Sink #"): + name = None + continue + stripped = raw.strip() + if stripped.startswith("Name:"): + name = stripped.split(":", 1)[1].strip() + elif name is None: + continue + elif stripped.startswith("Mute:"): + mutes[name] = stripped.split(":", 1)[1].strip() == "yes" + elif stripped.startswith("Volume:"): + match = re.search(r"(\d+)", stripped) + if match: + volumes[name] = match.group(1) + return volumes, mutes + + +@dataclass +class _AudioState: + """Snapshot of audio state to restore after the alarm is dismissed.""" + + default_sink: str | None + card_profiles: dict[str, str] # card name -> active profile + sink_volumes: dict[str, str] # sink name -> raw first-channel volume + sink_mutes: dict[str, bool] # sink name -> muted? + + +def _capture_audio_state(pactl: str) -> _AudioState: + """Capture every card profile, sink volume/mute, and the default sink.""" + cards = _parse_cards(_pactl_text(pactl, "list", "cards") or "") + card_profiles = { + name: info.active_profile + for name, info in cards.items() + if info.active_profile is not None + } + sink_volumes, sink_mutes = _parse_sinks(_pactl_text(pactl, "list", "sinks") or "") + return _AudioState( + default_sink=_current_default_sink(pactl), + card_profiles=card_profiles, + sink_volumes=sink_volumes, + sink_mutes=sink_mutes, + ) + + +def _activate_all_outputs() -> _AudioState | None: + """Turn on every card's output and make every sink audible for the alarm. + + At wake time the Bluetooth speaker is gone and WirePlumber leaves most card + profiles ``off``, so the alarm is silent. This forces every card's best + output profile on, then unmutes and raises every sink to 100%. The alarm + then plays on *all* sinks every beep cycle + (:func:`wake_alarm._audio._play_on_all_sinks`) — it does not pick one output + and hope: a switched-off external speaker cannot be revived by software, so + every *other* output is tried in parallel instead. + + It waits (bounded) for the known-slow ``ALARM_AUDIO_SINK`` (the G27Q + monitor's HDMI speaker, ~15 s to enumerate after power-on) so that output is + included in the first pass; the default sink is set to it when present but is + only a convenience — per-cycle all-sink playback is what guarantees audio. + + Returns: + The prior audio state (to restore on close), or ``None`` only when + pactl is unavailable (nothing was changed, so nothing to restore). + """ + pactl = _pactl_path() + if pactl is None: + _logger.warning("pactl not on PATH; cannot activate alarm audio") + return None + state = _capture_audio_state(pactl) + for card, info in _parse_cards(_pactl_text(pactl, "list", "cards") or "").items(): + if info.output_profiles: + subprocess.run( + [pactl, "set-card-profile", card, info.output_profiles[0]], + capture_output=True, + timeout=3, + check=False, + ) + # Wait for the known always-on monitor sink specifically, not just "any" + # sink: a switched-off external speaker often leaves a live-but-silent sink + # already present, and breaking on that would route the alarm to the dead + # output (the exact silent-alarm failure this change fixes). Fall back to + # whatever sinks exist if the monitor never enumerates. + attempts = max( + 1, + int(ALARM_AUDIO_SINK_WAIT_SECONDS / ALARM_AUDIO_SINK_POLL_SECONDS), + ) + sinks: list[str] = [] + for _ in range(attempts): + sinks = _list_real_sinks(pactl) + if ALARM_AUDIO_SINK in sinks: + break + time.sleep(ALARM_AUDIO_SINK_POLL_SECONDS) + else: + _logger.warning( + "Preferred sink %s did not appear after %.0fs; using whatever exists", + ALARM_AUDIO_SINK, + ALARM_AUDIO_SINK_WAIT_SECONDS, + ) + sinks = _list_real_sinks(pactl) + if not sinks: + _logger.warning("No real sink available; alarm relies on PC speaker") + return state + for sink in sinks: + _ensure_sink_audible(pactl, sink) + default = ALARM_AUDIO_SINK if ALARM_AUDIO_SINK in sinks else sinks[0] + subprocess.run( + [pactl, "set-default-sink", default], + capture_output=True, + timeout=3, + check=False, + ) + _logger.warning( + "Alarm audio activated on %d sink(s) at 100%%, default=%s", + len(sinks), + default, + ) + return state + + +def _restore_audio_state(state: _AudioState | None) -> None: + """Restore card profiles, sink volumes/mutes, and default captured earlier.""" + if state is None: + return + pactl = _pactl_path() + if pactl is None: + return + for card, profile in state.card_profiles.items(): + subprocess.run( + [pactl, "set-card-profile", card, profile], + capture_output=True, + timeout=3, + check=False, + ) + for sink, volume in state.sink_volumes.items(): + subprocess.run( + [pactl, "set-sink-volume", sink, volume], + capture_output=True, + timeout=3, + check=False, + ) + for sink, muted in state.sink_mutes.items(): + subprocess.run( + [pactl, "set-sink-mute", sink, "1" if muted else "0"], + capture_output=True, + timeout=3, + check=False, + ) + if state.default_sink is not None: + subprocess.run( + [pactl, "set-default-sink", state.default_sink], + capture_output=True, + timeout=3, + check=False, + ) + + +def _warn_if_no_real_sink() -> None: + """Log a warning when PipeWire has no real (non-auto_null) sink yet. + + Informational only: :func:`_activate_all_outputs` runs afterwards and forces + card profiles on, so a missing sink here is not necessarily fatal. + """ + pactl = _pactl_path() + if pactl is None: + _logger.warning("pactl not on PATH; cannot verify audio sinks") + return + real_sinks = _list_real_sinks(pactl) + if not real_sinks: + _logger.warning( + "No real PipeWire sink yet — alarm will force card profiles on and " + "fall back to the PC speaker.", + ) + else: + _logger.info("Audio sinks available: %s", real_sinks) diff --git a/wake_alarm/tests/test_alarm.py b/wake_alarm/tests/test_alarm.py index 39616f4..677232e 100644 --- a/wake_alarm/tests/test_alarm.py +++ b/wake_alarm/tests/test_alarm.py @@ -23,7 +23,6 @@ from wake_alarm._audio import ( _beep_soft, _find_fan_hwmon, _max_fans, - _play_on_extra_devices, _restore_fans, _speaker_test_path, ) @@ -295,55 +294,6 @@ class TestShouldRunAlarm: assert _should_run_alarm() is False -class TestPlayOnExtraDevices: - """Tests for _play_on_extra_devices.""" - - def test_popen_called_for_each_device(self) -> None: - """_play_on_extra_devices spawns speaker-test with PIPEWIRE_NODE set.""" - with ( - patch( - "wake_alarm._audio._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch("wake_alarm._audio.subprocess.Popen") as mock_popen, - ): - _play_on_extra_devices(1000) - mock_popen.assert_called_once() - args = mock_popen.call_args[0][0] - env = mock_popen.call_args.kwargs["env"] - assert "/usr/bin/speaker-test" in args - assert "-D" not in args - assert "1000" in args - assert "PIPEWIRE_NODE" in env - assert "alsa_output" in env["PIPEWIRE_NODE"] - - def test_noop_when_speaker_test_missing(self) -> None: - """_play_on_extra_devices does nothing when speaker-test is absent.""" - with ( - patch( - "wake_alarm._audio._speaker_test_path", - side_effect=FileNotFoundError("not found"), - ), - patch("wake_alarm._audio.subprocess.Popen") as mock_popen, - ): - _play_on_extra_devices(1000) - mock_popen.assert_not_called() - - def test_ignores_oserror_on_popen(self) -> None: - """_play_on_extra_devices silently ignores OSError from Popen.""" - with ( - patch( - "wake_alarm._audio._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "wake_alarm._audio.subprocess.Popen", - side_effect=OSError("device busy"), - ), - ): - _play_on_extra_devices(1000) # must not raise - - class TestFindFanHwmon: """Tests for _find_fan_hwmon.""" diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index 77983cd..a88d8fa 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -54,15 +54,15 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( - patch("wake_alarm._alarm._play_on_extra_devices") as mock, + patch("wake_alarm._alarm._play_on_all_sinks") as mock, patch("wake_alarm._alarm._max_fans", return_value=False), patch("wake_alarm._alarm._restore_fans"), patch("wake_alarm._alarm._set_max_brightness"), patch("wake_alarm._alarm._wake_display"), patch("wake_alarm._alarm._restore_display"), patch("wake_alarm._alarm._warn_if_no_real_sink"), - patch("wake_alarm._alarm._activate_alarm_audio", return_value=None), - patch("wake_alarm._alarm._restore_alarm_audio"), + patch("wake_alarm._alarm._activate_all_outputs", return_value=None), + patch("wake_alarm._alarm._restore_audio_state"), patch("wake_alarm._alarm.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): diff --git a/wake_alarm/tests/test_alarm_part3.py b/wake_alarm/tests/test_alarm_part3.py index b23e0fc..86484bd 100644 --- a/wake_alarm/tests/test_alarm_part3.py +++ b/wake_alarm/tests/test_alarm_part3.py @@ -53,15 +53,15 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess calls for extra ALSA devices and hardware.""" with ( - patch("wake_alarm._alarm._play_on_extra_devices") as mock, + patch("wake_alarm._alarm._play_on_all_sinks") as mock, patch("wake_alarm._alarm._max_fans", return_value=False), patch("wake_alarm._alarm._restore_fans"), patch("wake_alarm._alarm._set_max_brightness"), patch("wake_alarm._alarm._wake_display"), patch("wake_alarm._alarm._restore_display"), patch("wake_alarm._alarm._warn_if_no_real_sink"), - patch("wake_alarm._alarm._activate_alarm_audio", return_value=None), - patch("wake_alarm._alarm._restore_alarm_audio"), + patch("wake_alarm._alarm._activate_all_outputs", return_value=None), + patch("wake_alarm._alarm._restore_audio_state"), patch("wake_alarm._alarm.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): diff --git a/wake_alarm/tests/test_alarm_part4.py b/wake_alarm/tests/test_alarm_part4.py index 20df5b8..10857db 100644 --- a/wake_alarm/tests/test_alarm_part4.py +++ b/wake_alarm/tests/test_alarm_part4.py @@ -12,6 +12,7 @@ if TYPE_CHECKING: from collections.abc import Generator from wake_alarm._alarm import WakeAlarm +from wake_alarm._sinks import _AudioState # --------------------------------------------------------------------------- # Helpers (duplicated from part 1 so this file is self-contained) @@ -51,15 +52,15 @@ def _block_real_tk() -> Generator[MagicMock]: def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( - patch("wake_alarm._alarm._play_on_extra_devices") as mock, + patch("wake_alarm._alarm._play_on_all_sinks") as mock, patch("wake_alarm._alarm._max_fans", return_value=False), patch("wake_alarm._alarm._restore_fans"), patch("wake_alarm._alarm._set_max_brightness"), patch("wake_alarm._alarm._wake_display"), patch("wake_alarm._alarm._restore_display"), patch("wake_alarm._alarm._warn_if_no_real_sink"), - patch("wake_alarm._alarm._activate_alarm_audio", return_value=None), - patch("wake_alarm._alarm._restore_alarm_audio"), + patch("wake_alarm._alarm._activate_all_outputs", return_value=None), + patch("wake_alarm._alarm._restore_audio_state"), patch("wake_alarm._alarm.turn_on_plug"), patch("wake_alarm._alarm.turn_off_plug"), ): @@ -142,13 +143,19 @@ class TestClose: self, mock_tk_module: MagicMock, ) -> None: - """on_close restores the default sink captured at activation.""" + """on_close restores the audio state captured at activation.""" del mock_tk_module alarm = WakeAlarm(demo_mode=True) - alarm._hardware.audio_restore = "jbl_sink" + state = _AudioState( + default_sink="jbl_sink", + card_profiles={}, + sink_volumes={}, + sink_mutes={}, + ) + alarm._hardware.audio_restore = state with patch( - "wake_alarm._alarm._restore_alarm_audio", + "wake_alarm._alarm._restore_audio_state", ) as mock_restore: alarm.on_close() - mock_restore.assert_called_once_with("jbl_sink") + mock_restore.assert_called_once_with(state) alarm._stop_beep.set() diff --git a/wake_alarm/tests/test_alarm_playback.py b/wake_alarm/tests/test_alarm_playback.py new file mode 100644 index 0000000..2cd3f02 --- /dev/null +++ b/wake_alarm/tests/test_alarm_playback.py @@ -0,0 +1,136 @@ +"""Tests for per-sink alarm playback (_play_on_all_sinks and helpers).""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +from wake_alarm._audio import _play_on_all_sinks, _play_wav_on_sink, _reap_players +from wake_alarm._sinks import _ensure_sink_audible + +if TYPE_CHECKING: + from pathlib import Path + + +class TestPlayOnAllSinks: + """Tests for _play_on_all_sinks.""" + + def test_noop_when_pactl_missing(self) -> None: + """No pactl → returns without looking up paplay or spawning Popen.""" + with ( + patch("wake_alarm._audio._pactl_path", return_value=None), + patch("wake_alarm._audio.shutil.which") as mock_which, + patch("wake_alarm._audio.subprocess.Popen") as mock_popen, + ): + _play_on_all_sinks(1000) + mock_which.assert_not_called() + mock_popen.assert_not_called() + + def test_warns_when_paplay_missing(self) -> None: + """pactl present but paplay absent → warning, no Popen.""" + with ( + patch("wake_alarm._audio._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._audio.shutil.which", return_value=None), + patch("wake_alarm._audio.subprocess.Popen") as mock_popen, + ): + _play_on_all_sinks(1000) + mock_popen.assert_not_called() + + def test_returns_when_wav_generation_fails(self) -> None: + """OSError generating the tone WAV → returns, no Popen.""" + with ( + patch("wake_alarm._audio._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._audio.shutil.which", return_value="/usr/bin/paplay"), + patch( + "wake_alarm._audio._ensure_tone_wav", + side_effect=OSError("no tempdir"), + ), + patch("wake_alarm._audio.subprocess.Popen") as mock_popen, + ): + _play_on_all_sinks(1000) + mock_popen.assert_not_called() + + def test_plays_and_maxes_each_sink(self, tmp_path: Path) -> None: + """Each real sink is made audible, then played on via paplay --device.""" + wav = tmp_path / "tone.wav" + with ( + patch("wake_alarm._audio._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._audio.shutil.which", return_value="/usr/bin/paplay"), + patch("wake_alarm._audio._ensure_tone_wav", return_value=wav), + patch( + "wake_alarm._audio._list_real_sinks", + return_value=["sink_a", "sink_b"], + ), + patch("wake_alarm._audio._ensure_sink_audible") as mock_audible, + patch("wake_alarm._audio.subprocess.Popen") as mock_popen, + ): + _play_on_all_sinks(1000) + assert mock_audible.call_count == 2 + devices = [call.args[0][1] for call in mock_popen.call_args_list] + assert devices == ["--device=sink_a", "--device=sink_b"] + + def test_skips_sink_whose_launch_failed(self) -> None: + """A sink whose paplay fails to launch (None) is not reaped.""" + with ( + patch("wake_alarm._audio._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._audio.shutil.which", return_value="/usr/bin/paplay"), + patch("wake_alarm._audio._ensure_tone_wav", return_value=MagicMock()), + patch("wake_alarm._audio._list_real_sinks", return_value=["sink_a"]), + patch("wake_alarm._audio._ensure_sink_audible"), + patch("wake_alarm._audio._play_wav_on_sink", return_value=None), + patch("wake_alarm._audio._reap_players") as mock_reap, + ): + _play_on_all_sinks(1000) + mock_reap.assert_called_once_with([]) + + +class TestReapPlayers: + """Tests for _reap_players.""" + + def test_waits_for_each(self) -> None: + """Each launched paplay is waited on (reaped).""" + procs = [MagicMock(), MagicMock()] + _reap_players(procs) + for proc in procs: + proc.wait.assert_called_once() + + def test_kills_hung_player(self) -> None: + """A paplay that exceeds the timeout is killed and then reaped.""" + proc = MagicMock() + proc.wait.side_effect = [subprocess.TimeoutExpired("paplay", 6), None] + _reap_players([proc]) + proc.kill.assert_called_once() + assert proc.wait.call_count == 2 + + +class TestEnsureSinkAudible: + """Tests for _ensure_sink_audible.""" + + def test_unmutes_and_maxes(self) -> None: + """Runs set-sink-mute 0 then set-sink-volume 100% for the sink.""" + with patch("wake_alarm._sinks.subprocess.run") as mock_run: + _ensure_sink_audible("/usr/bin/pactl", "sink_a") + cmds = [call.args[0] for call in mock_run.call_args_list] + assert ["/usr/bin/pactl", "set-sink-mute", "sink_a", "0"] in cmds + assert ["/usr/bin/pactl", "set-sink-volume", "sink_a", "100%"] in cmds + + +class TestPlayWavOnSink: + """Tests for _play_wav_on_sink.""" + + def test_spawns_paplay_with_device(self, tmp_path: Path) -> None: + """Popen is called with paplay --device= and the wav path.""" + with patch("wake_alarm._audio.subprocess.Popen") as mock_popen: + _play_wav_on_sink("/usr/bin/paplay", "sink_a", tmp_path / "tone.wav") + args = mock_popen.call_args.args[0] + assert args[0] == "/usr/bin/paplay" + assert args[1] == "--device=sink_a" + + def test_ignores_oserror(self, tmp_path: Path) -> None: + """OSError from Popen is swallowed (must not raise).""" + with patch( + "wake_alarm._audio.subprocess.Popen", + side_effect=OSError("device busy"), + ): + _play_wav_on_sink("/usr/bin/paplay", "sink_a", tmp_path / "tone.wav") diff --git a/wake_alarm/tests/test_alarm_sinks.py b/wake_alarm/tests/test_alarm_sinks.py index ba6909c..8c1065d 100644 --- a/wake_alarm/tests/test_alarm_sinks.py +++ b/wake_alarm/tests/test_alarm_sinks.py @@ -1,4 +1,4 @@ -"""Tests for sink management and parse_args in wake alarm.""" +"""Tests for pactl sink/card enumeration, activation, and restore.""" from __future__ import annotations @@ -6,114 +6,183 @@ import subprocess from unittest.mock import MagicMock, patch from wake_alarm._alarm import _parse_args -from wake_alarm._audio import ( - _activate_alarm_audio, - _alarm_sink_present, +from wake_alarm._constants import ALARM_AUDIO_SINK +from wake_alarm._sinks import ( + _activate_all_outputs, + _AudioState, + _capture_audio_state, + _CardInfo, _current_default_sink, - _restore_alarm_audio, + _list_cards, + _list_real_sinks, + _pactl_text, + _parse_cards, + _parse_profile_line, + _parse_sinks, + _restore_audio_state, _warn_if_no_real_sink, ) +# A representative `pactl list cards` block covering every parse branch: +# an HDMI card (off, with an available stereo output profile), an input-only +# card (no output profile), a card block with no Name line (must be skipped). +_CARDS_TEXT = """Card #54 +\tName: alsa_card.hdmi +\tDriver: module-alsa-card.c +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\t\toutput:hdmi-stereo: Stereo (sinks: 1, priority: 5900, available: yes) +\t\toutput:hdmi-surround: Surround (sinks: 1, priority: 600, available: no) +\t\tbroken (available: yes) +\tActive Profile: off +Card #55 +\tName: alsa_card.mic +\tProfiles: +\t\toff: Off (sinks: 0, sources: 0, priority: 0, available: yes) +\tActive Profile: off +Card #56 +\tProfiles: +\t\toutput:noname: NoName (available: yes) +\tActive Profile: off +""" -class TestWarnIfNoRealSink: - """Tests for _warn_if_no_real_sink.""" - - def test_logs_when_pactl_missing(self) -> None: - """No pactl on PATH → warns and returns.""" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value=None, - ), - patch( - "wake_alarm._audio.subprocess.run", - ) as mock_run, - ): - _warn_if_no_real_sink() - mock_run.assert_not_called() - - def test_warns_when_only_auto_null(self) -> None: - """Only auto_null sink → warning is emitted.""" - result = MagicMock() - result.stdout = b"4319\tauto_null\tPipeWire\tfloat32le 2ch 48000Hz\tIDLE\n" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch( - "wake_alarm._audio.subprocess.run", - return_value=result, - ), - patch("wake_alarm._audio._logger") as mock_log, - ): - _warn_if_no_real_sink() - mock_log.warning.assert_called() - - def test_info_when_real_sink_present(self) -> None: - """A non-auto_null sink → info log, no warning.""" - result = MagicMock() - result.stdout = b"1\talsa_output.pci-0000_01_00.1.hdmi-stereo\tPipeWire\t-\t-\n" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch( - "wake_alarm._audio.subprocess.run", - return_value=result, - ), - patch("wake_alarm._audio._logger") as mock_log, - ): - _warn_if_no_real_sink() - mock_log.info.assert_called() - mock_log.warning.assert_not_called() - - def test_handles_pactl_failure(self) -> None: - """OSError/TimeoutExpired running pactl → warning, no raise.""" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch( - "wake_alarm._audio.subprocess.run", - side_effect=subprocess.TimeoutExpired("pactl", 5), - ), - ): - _warn_if_no_real_sink() # must not raise +# A `pactl list sinks` block: a normal sink, plus one whose Volume line has +# no integer (to cover the "no digit" branch), plus a leading non-Name line. +_SINKS_TEXT = """Sink #62 +\tState: SUSPENDED +\tName: alsa_output.analog +\tMute: no +\tVolume: front-left: 65536 / 100% +\tDescription: Built-in Analog +Sink #63 +\tName: alsa_output.weird +\tMute: yes +\tVolume: n/a +""" -class TestAlarmSinkPresent: - """Tests for _alarm_sink_present.""" +class TestPactlText: + """Tests for _pactl_text.""" - def test_true_when_sink_listed(self) -> None: - """Returns True when the alarm sink name appears in pactl output.""" - from wake_alarm._constants import ALARM_AUDIO_SINK + def test_returns_decoded_stdout(self) -> None: + """Success → decoded stdout string.""" + proc = MagicMock(stdout=b"hello\n") + with patch("wake_alarm._sinks.subprocess.run", return_value=proc): + assert _pactl_text("/usr/bin/pactl", "info") == "hello\n" - proc = MagicMock(stdout=ALARM_AUDIO_SINK.encode() + b"\tPipeWire\n") + def test_returns_none_on_error(self) -> None: + """OSError/TimeoutExpired → None, no raise.""" with patch( - "wake_alarm._audio.subprocess.run", - return_value=proc, + "wake_alarm._sinks.subprocess.run", + side_effect=subprocess.TimeoutExpired("pactl", 5), ): - assert _alarm_sink_present("/usr/bin/pactl") is True + assert _pactl_text("/usr/bin/pactl", "info") is None - def test_false_when_sink_absent(self) -> None: - """Returns False when the alarm sink is not in pactl output.""" - proc = MagicMock(stdout=b"auto_null\tPipeWire\n") - with patch( - "wake_alarm._audio.subprocess.run", - return_value=proc, - ): - assert _alarm_sink_present("/usr/bin/pactl") is False - def test_false_on_subprocess_error(self) -> None: - """OSError while listing sinks → False, no raise.""" +class TestPactlPath: + """Tests for _pactl_path.""" + + def test_returns_which_result(self) -> None: + """Delegates to shutil.which('pactl').""" + from wake_alarm._sinks import _pactl_path + with patch( - "wake_alarm._audio.subprocess.run", - side_effect=OSError("boom"), + "wake_alarm._sinks.shutil.which", + return_value="/usr/bin/pactl", ): - assert _alarm_sink_present("/usr/bin/pactl") is False + assert _pactl_path() == "/usr/bin/pactl" + + +class TestListRealSinks: + """Tests for _list_real_sinks.""" + + def test_filters_auto_null(self) -> None: + """auto_null is dropped; real sink names are returned.""" + text = ( + "62\talsa_output.analog\tPipeWire\t-\t-\n4319\tauto_null\tPipeWire\t-\t-\n" + ) + with patch("wake_alarm._sinks._pactl_text", return_value=text): + assert _list_real_sinks("/usr/bin/pactl") == ["alsa_output.analog"] + + def test_returns_empty_on_none(self) -> None: + """pactl failure (None text) → empty list.""" + with patch("wake_alarm._sinks._pactl_text", return_value=None): + assert _list_real_sinks("/usr/bin/pactl") == [] + + +class TestListCards: + """Tests for _list_cards.""" + + def test_parses_names(self) -> None: + """Returns the card name column of `list short cards`.""" + text = "54\talsa_card.hdmi\talsa\n55\talsa_card.mic\talsa\n" + with patch("wake_alarm._sinks._pactl_text", return_value=text): + assert _list_cards("/usr/bin/pactl") == [ + "alsa_card.hdmi", + "alsa_card.mic", + ] + + def test_returns_empty_on_none(self) -> None: + """pactl failure (None text) → empty list.""" + with patch("wake_alarm._sinks._pactl_text", return_value=None): + assert _list_cards("/usr/bin/pactl") == [] + + +class TestParseProfileLine: + """Tests for _parse_profile_line.""" + + def test_non_output_returns_none(self) -> None: + """A non-output profile line → None.""" + assert _parse_profile_line("off: Off (available: yes)") is None + + def test_unavailable_returns_none(self) -> None: + """An output profile marked available: no → None.""" + assert _parse_profile_line("output:x: X (available: no)") is None + + def test_missing_colon_space_returns_none(self) -> None: + """An output line with no ': ' separator → None.""" + assert _parse_profile_line("output:x") is None + + def test_defaults_priority_zero(self) -> None: + """An available output with no priority field → priority 0.""" + assert _parse_profile_line("output:x: X (available: yes)") == ("output:x", 0) + + def test_parses_priority(self) -> None: + """Priority is parsed from the line.""" + assert _parse_profile_line( + "output:hdmi: HDMI (priority: 5900, available: yes)", + ) == ("output:hdmi", 5900) + + +class TestParseCards: + """Tests for _parse_cards.""" + + def test_parses_and_skips_edge_cases(self) -> None: + """Stereo preferred, input-only card empty, unnamed card skipped.""" + cards = _parse_cards(_CARDS_TEXT) + assert set(cards) == {"alsa_card.hdmi", "alsa_card.mic"} + assert cards["alsa_card.hdmi"].active_profile == "off" + assert cards["alsa_card.hdmi"].output_profiles == ["output:hdmi-stereo"] + assert cards["alsa_card.mic"].output_profiles == [] + + def test_ignores_lines_before_first_card(self) -> None: + """Leading noise before the first `Card #` is ignored.""" + assert _parse_cards("junk\nmore junk\n") == {} + + +class TestParseSinks: + """Tests for _parse_sinks.""" + + def test_parses_volume_and_mute(self) -> None: + """Captures first raw volume and mute state per sink.""" + volumes, mutes = _parse_sinks(_SINKS_TEXT) + assert volumes["alsa_output.analog"] == "65536" + assert mutes == {"alsa_output.analog": False, "alsa_output.weird": True} + + def test_volume_without_digit_is_skipped(self) -> None: + """A Volume line with no integer is not recorded.""" + volumes, _ = _parse_sinks(_SINKS_TEXT) + assert "alsa_output.weird" not in volumes class TestCurrentDefaultSink: @@ -122,145 +191,266 @@ class TestCurrentDefaultSink: def test_returns_sink_name(self) -> None: """Returns the trimmed default sink name.""" proc = MagicMock(stdout=b"jbl_sink\n") - with patch( - "wake_alarm._audio.subprocess.run", - return_value=proc, - ): + with patch("wake_alarm._sinks.subprocess.run", return_value=proc): assert _current_default_sink("/usr/bin/pactl") == "jbl_sink" def test_returns_none_when_empty(self) -> None: """Empty output → None.""" proc = MagicMock(stdout=b"\n") - with patch( - "wake_alarm._audio.subprocess.run", - return_value=proc, - ): + with patch("wake_alarm._sinks.subprocess.run", return_value=proc): assert _current_default_sink("/usr/bin/pactl") is None def test_returns_none_on_error(self) -> None: """TimeoutExpired → None, no raise.""" with patch( - "wake_alarm._audio.subprocess.run", + "wake_alarm._sinks.subprocess.run", side_effect=subprocess.TimeoutExpired("pactl", 3), ): assert _current_default_sink("/usr/bin/pactl") is None -class TestActivateAlarmAudio: - """Tests for _activate_alarm_audio.""" +class TestCaptureAudioState: + """Tests for _capture_audio_state.""" + + def test_captures_profiles_volumes_mutes_default(self) -> None: + """Combines card profiles, sink volumes/mutes, and the default sink.""" + with ( + patch( + "wake_alarm._sinks._pactl_text", + side_effect=[_CARDS_TEXT, _SINKS_TEXT], + ), + patch( + "wake_alarm._sinks._current_default_sink", + return_value="alsa_output.analog", + ), + ): + state = _capture_audio_state("/usr/bin/pactl") + assert state.default_sink == "alsa_output.analog" + assert state.card_profiles["alsa_card.hdmi"] == "off" + assert state.sink_volumes["alsa_output.analog"] == "65536" + assert state.sink_mutes["alsa_output.analog"] is False + + +class TestActivateAllOutputs: + """Tests for _activate_all_outputs.""" def test_returns_none_when_pactl_missing(self) -> None: """No pactl on PATH → returns None without touching audio.""" with ( - patch("wake_alarm._audio.shutil.which", return_value=None), - patch("wake_alarm._audio.subprocess.run") as mock_run, + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks.subprocess.run") as mock_run, ): - assert _activate_alarm_audio() is None + assert _activate_all_outputs() is None mock_run.assert_not_called() - def test_activates_and_returns_old_default(self) -> None: - """Sink present → routes audio there and returns prior default sink.""" - with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch( - "wake_alarm._audio._alarm_sink_present", - return_value=True, - ), - patch( - "wake_alarm._audio._current_default_sink", - return_value="jbl_sink", - ), - patch("wake_alarm._audio.subprocess.run") as mock_run, - ): - result = _activate_alarm_audio() - assert result == "jbl_sink" - cmds = [call.args[0] for call in mock_run.call_args_list] - from wake_alarm._constants import ( - ALARM_AUDIO_CARD, - ALARM_AUDIO_PROFILE, - ALARM_AUDIO_SINK, + def _sentinel_state(self) -> _AudioState: + return _AudioState( + default_sink="old", + card_profiles={}, + sink_volumes={}, + sink_mutes={}, ) + def test_activates_and_prefers_alarm_sink(self) -> None: + """Forces card profiles on, maxes sinks, defaults to ALARM_AUDIO_SINK.""" + state = self._sentinel_state() + cards = { + "card_hdmi": _CardInfo(None, ["output:hdmi-stereo"]), + "card_mic": _CardInfo(None, []), + } + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value=cards), + patch( + "wake_alarm._sinks._list_real_sinks", + return_value=[ALARM_AUDIO_SINK, "other"], + ), + patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is state + cmds = [call.args[0] for call in mock_run.call_args_list] assert [ "/usr/bin/pactl", "set-card-profile", - ALARM_AUDIO_CARD, - ALARM_AUDIO_PROFILE, + "card_hdmi", + "output:hdmi-stereo", ] in cmds + # The input-only card (no output profile) is never switched. + assert not any(c[2] == "card_mic" for c in cmds if c[1] == "set-card-profile") + assert ["/usr/bin/pactl", "set-default-sink", ALARM_AUDIO_SINK] in cmds + assert mock_audible.call_count == 2 + + def test_waits_for_slow_monitor_sink(self) -> None: + """The dead analog sink is present first; activation waits for the + monitor HDMI sink to enumerate on a later poll, then routes to it. + + This is the exact wake-time scenario the change fixes: a live-but-silent + sink already present must not short-circuit the wait for the always-on + monitor speaker. + """ + state = self._sentinel_state() + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch( + "wake_alarm._sinks._list_real_sinks", + side_effect=[["dead_analog"], ["dead_analog", ALARM_AUDIO_SINK]], + ), + patch("wake_alarm._sinks._ensure_sink_audible"), + patch("wake_alarm._sinks.time.sleep") as mock_sleep, + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + assert _activate_all_outputs() is state + mock_sleep.assert_called_once() + cmds = [call.args[0] for call in mock_run.call_args_list] assert ["/usr/bin/pactl", "set-default-sink", ALARM_AUDIO_SINK] in cmds - def test_returns_none_when_sink_never_appears(self) -> None: - """Sink never shows up → returns None after polling (no raise).""" + def test_falls_back_to_first_sink_when_alarm_sink_absent(self) -> None: + """ALARM_AUDIO_SINK never appears → default is the first real sink.""" + state = self._sentinel_state() with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch("wake_alarm._sinks.ALARM_AUDIO_SINK_WAIT_SECONDS", 1.0), + patch("wake_alarm._sinks.ALARM_AUDIO_SINK_POLL_SECONDS", 0.5), patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", + "wake_alarm._sinks._list_real_sinks", + side_effect=[["other"], ["other"], ["other"]], ), - patch( - "wake_alarm._audio._alarm_sink_present", - return_value=False, - ), - patch("wake_alarm._audio.time.sleep") as mock_sleep, - patch("wake_alarm._audio.subprocess.run"), + patch("wake_alarm._sinks._ensure_sink_audible"), + patch("wake_alarm._sinks.time.sleep") as mock_sleep, + patch("wake_alarm._sinks.subprocess.run") as mock_run, ): - assert _activate_alarm_audio() is None - mock_sleep.assert_called() + assert _activate_all_outputs() is state + mock_sleep.assert_called() + cmds = [call.args[0] for call in mock_run.call_args_list] + assert ["/usr/bin/pactl", "set-default-sink", "other"] in cmds - def test_waits_then_succeeds(self) -> None: - """Sink absent then present → sleeps once, then routes audio.""" + def test_returns_state_when_no_sink_at_all(self) -> None: + """No real sink ever appears → returns state, no default set.""" + state = self._sentinel_state() with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch( - "wake_alarm._audio._alarm_sink_present", - side_effect=[False, True], - ), - patch( - "wake_alarm._audio._current_default_sink", - return_value="old", - ), - patch("wake_alarm._audio.time.sleep") as mock_sleep, - patch("wake_alarm._audio.subprocess.run"), + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._capture_audio_state", return_value=state), + patch("wake_alarm._sinks._pactl_text", return_value=""), + patch("wake_alarm._sinks._parse_cards", return_value={}), + patch("wake_alarm._sinks.ALARM_AUDIO_SINK_WAIT_SECONDS", 1.0), + patch("wake_alarm._sinks.ALARM_AUDIO_SINK_POLL_SECONDS", 0.5), + patch("wake_alarm._sinks._list_real_sinks", return_value=[]), + patch("wake_alarm._sinks._ensure_sink_audible") as mock_audible, + patch("wake_alarm._sinks.time.sleep"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, ): - assert _activate_alarm_audio() == "old" - mock_sleep.assert_called_once() + assert _activate_all_outputs() is state + mock_audible.assert_not_called() + cmds = [call.args[0] for call in mock_run.call_args_list] + assert not any(c[1] == "set-default-sink" for c in cmds) -class TestRestoreAlarmAudio: - """Tests for _restore_alarm_audio.""" +class TestRestoreAudioState: + """Tests for _restore_audio_state.""" def test_none_is_noop(self) -> None: - """None default → does nothing, no pactl lookup.""" - with patch("wake_alarm._audio.shutil.which") as mock_which: - _restore_alarm_audio(None) - mock_which.assert_not_called() + """None state → does nothing, no pactl lookup.""" + with patch("wake_alarm._sinks._pactl_path") as mock_path: + _restore_audio_state(None) + mock_path.assert_not_called() def test_no_pactl_returns_silently(self) -> None: - """Default present but pactl missing → no raise, no run.""" + """State present but pactl missing → no raise, no run.""" + state = _AudioState( + default_sink="s", + card_profiles={}, + sink_volumes={}, + sink_mutes={}, + ) with ( - patch("wake_alarm._audio.shutil.which", return_value=None), - patch("wake_alarm._audio.subprocess.run") as mock_run, + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks.subprocess.run") as mock_run, ): - _restore_alarm_audio("jbl_sink") + _restore_audio_state(state) mock_run.assert_not_called() - def test_restores_default_sink(self) -> None: - """Calls set-default-sink with the captured prior default.""" + def test_restores_everything(self) -> None: + """Restores card profiles, sink volumes, mutes, and the default sink.""" + state = _AudioState( + default_sink="analog", + card_profiles={"card_hdmi": "off"}, + sink_volumes={"analog": "65536"}, + sink_mutes={"analog": True}, + ) with ( - patch( - "wake_alarm._audio.shutil.which", - return_value="/usr/bin/pactl", - ), - patch("wake_alarm._audio.subprocess.run") as mock_run, + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, ): - _restore_alarm_audio("jbl_sink") + _restore_audio_state(state) cmds = [call.args[0] for call in mock_run.call_args_list] - assert ["/usr/bin/pactl", "set-default-sink", "jbl_sink"] in cmds + assert ["/usr/bin/pactl", "set-card-profile", "card_hdmi", "off"] in cmds + assert ["/usr/bin/pactl", "set-sink-volume", "analog", "65536"] in cmds + assert ["/usr/bin/pactl", "set-sink-mute", "analog", "1"] in cmds + assert ["/usr/bin/pactl", "set-default-sink", "analog"] in cmds + + def test_skips_default_when_none(self) -> None: + """default_sink None → no set-default-sink command.""" + state = _AudioState( + default_sink=None, + card_profiles={}, + sink_volumes={"analog": "100"}, + sink_mutes={"analog": False}, + ) + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks.subprocess.run") as mock_run, + ): + _restore_audio_state(state) + cmds = [call.args[0] for call in mock_run.call_args_list] + assert not any(c[1] == "set-default-sink" for c in cmds) + # Unmuted sink restores with "0". + assert ["/usr/bin/pactl", "set-sink-mute", "analog", "0"] in cmds + + +class TestWarnIfNoRealSink: + """Tests for _warn_if_no_real_sink.""" + + def test_logs_when_pactl_missing(self) -> None: + """No pactl on PATH → warns and returns (no sink lookup).""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value=None), + patch("wake_alarm._sinks._list_real_sinks") as mock_list, + ): + _warn_if_no_real_sink() + mock_list.assert_not_called() + + def test_warns_when_no_real_sink(self) -> None: + """No real sink → warning is emitted.""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch("wake_alarm._sinks._list_real_sinks", return_value=[]), + patch("wake_alarm._sinks._logger") as mock_log, + ): + _warn_if_no_real_sink() + mock_log.warning.assert_called() + + def test_info_when_real_sink_present(self) -> None: + """A real sink → info log, no warning.""" + with ( + patch("wake_alarm._sinks._pactl_path", return_value="/usr/bin/pactl"), + patch( + "wake_alarm._sinks._list_real_sinks", + return_value=["alsa_output.analog"], + ), + patch("wake_alarm._sinks._logger") as mock_log, + ): + _warn_if_no_real_sink() + mock_log.info.assert_called() + mock_log.warning.assert_not_called() class TestParseArgs: