Replace sleep-and-hope audio waits with per-cycle presence checks
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Tests / test (3.10) (push) Waiting to run
Tests / test (3.11) (push) Waiting to run
Tests / test (3.12) (push) Waiting to run

Two spots blocked the alarm on a timer, guessing when the monitor's HDMI
sink would be "ready":

- _activate_all_outputs polled _list_real_sinks in a loop with
  time.sleep, waiting up to 20s for ALARM_AUDIO_SINK to appear.
- main() slept DISPLAY_WAKE_WAIT_SECONDS (5s) after waking the display
  before audio setup.

Both are gone. Activation now forces every card's output profile on and
activates whatever sink is present at that instant; the beep loop's
_play_on_all_sinks already re-lists real sinks every cycle, so a slow
monitor is swept in the moment it actually enumerates — a real presence
check, never a timed guess. main() wakes the display and proceeds
immediately; the monitor powers on in parallel.

Removes now-dead constants ALARM_AUDIO_CARD, ALARM_AUDIO_PROFILE,
ALARM_AUDIO_SINK_WAIT_SECONDS, ALARM_AUDIO_SINK_POLL_SECONDS,
DISPLAY_WAKE_WAIT_SECONDS. Activation returns in ~0.26s (was up to 20s);
the test suite drops from ~12s to ~2s (two main() tests were each really
sleeping 5s). 100% branch coverage retained.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SJ6HYXhBmYs3SiGBZv1rF
This commit is contained in:
Krzysztof Rudnicki 2026-07-11 18:51:03 +02:00
parent 142f4484d8
commit d2f75f7f21
4 changed files with 40 additions and 80 deletions

View File

@ -40,7 +40,6 @@ from wake_alarm._constants import (
DISMISS_FLASH_SECONDS,
DISMISS_ROUNDS_REQUIRED,
DISMISS_WINDOW_MINUTES,
DISPLAY_WAKE_WAIT_SECONDS,
LOUD_TOGGLE_INTERVAL,
MEDIUM_BEEP_INTERVAL,
PHASE_MEDIUM_END,
@ -486,10 +485,11 @@ def main() -> None:
args.trigger_now,
)
_warn_if_no_real_sink()
# Wake the display, then proceed immediately — no sleep hoping the monitor
# is "up". The monitor powers on in parallel; brightness and audio don't
# block on it, and the beep loop's per-cycle sink check picks up the HDMI
# output the moment it enumerates.
_wake_display()
# 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()
alarm = WakeAlarm(demo_mode=args.demo)

View File

@ -44,25 +44,12 @@ WAKE_STATE_FILE: Path = Path(__file__).resolve().parent / "wake_state.json"
# rtcwake binary path
RTCWAKE_BIN: str = "/usr/sbin/rtcwake"
# Alarm audio output (machine-specific, empirically verified 2026-05-25).
# At wake time the Bluetooth speaker is disconnected and PipeWire only has the
# auto_null sink, so the alarm is silent unless we activate a real output. The
# only audible always-present output on this machine is the G27Q monitor's
# built-in speaker on the NVidia GPU's HDMI audio. WirePlumber leaves the card
# profile "off", so the alarm must force the profile on and wait for the sink.
ALARM_AUDIO_CARD: str = "alsa_card.pci-0000_01_00.1"
ALARM_AUDIO_PROFILE: str = "output:hdmi-stereo"
# Preferred alarm audio output (machine-specific, empirically verified).
# At wake time the Bluetooth speaker is disconnected and PipeWire often has only
# the auto_null sink, so the alarm activates every card and plays on all sinks.
# This is the G27Q monitor's built-in HDMI speaker — the one always-present
# output on this machine — used as the preferred default sink when present.
ALARM_AUDIO_SINK: str = "alsa_output.pci-0000_01_00.1.hdmi-stereo"
# Seconds to wait for the HDMI sink to appear after forcing the profile on.
# The G27Q takes up to ~15 s to power on from a hard-off state and enumerate
# its HDMI audio; 6 s was too short when the monitor was physically off.
ALARM_AUDIO_SINK_WAIT_SECONDS: float = 20.0
# Poll interval while waiting for the sink.
ALARM_AUDIO_SINK_POLL_SECONDS: float = 0.5
# Seconds to pause after waking the display (xset dpms force on) before
# attempting audio setup. Gives the G27Q time to come out of power-off
# and re-enumerate its HDMI audio sink under PipeWire.
DISPLAY_WAKE_WAIT_SECONDS: float = 5.0
# Path to the workout log written by the companion screen_locker package.
# Dict keyed by YYYY-MM-DD date strings; presence of today's key means the

View File

@ -15,12 +15,9 @@ 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__)
@ -233,10 +230,11 @@ def _activate_all_outputs() -> _AudioState | None:
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.
It does not block waiting for the known-slow ``ALARM_AUDIO_SINK`` (the G27Q
monitor's HDMI speaker, ~15 s to enumerate); it activates whatever is present
now and lets the per-cycle presence check in ``_play_on_all_sinks`` sweep in
the rest. The default sink, when set, 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
@ -255,30 +253,18 @@ def _activate_all_outputs() -> _AudioState | None:
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)
# No sleep-and-hope wait. Forcing the profiles on above makes sinks appear
# asynchronously (the monitor's HDMI sink can take ~15 s), but we don't block
# on a timer guessing when it's "ready". We activate whatever is present
# right now; the beep loop's _play_on_all_sinks re-checks _list_real_sinks
# every cycle, so each output is made audible and played on the moment it
# actually exists — a real presence check, never a timed guess.
sinks = _list_real_sinks(pactl)
if not sinks:
_logger.warning("No real sink available; alarm relies on PC speaker")
_logger.warning(
"No real sink present yet; the beep loop will pick sinks up as they "
"enumerate (PC speaker covers the gap).",
)
return state
for sink in sinks:
_ensure_sink_audible(pactl, sink)
@ -290,7 +276,8 @@ def _activate_all_outputs() -> _AudioState | None:
check=False,
)
_logger.warning(
"Alarm audio activated on %d sink(s) at 100%%, default=%s",
"Alarm audio activated on %d present sink(s) at 100%%, default=%s; more "
"are added by the beep loop as they enumerate.",
len(sinks),
default,
)

View File

@ -283,13 +283,11 @@ class TestActivateAllOutputs:
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.
def test_does_not_block_on_sink_enumeration(self) -> None:
"""Activation checks present sinks once — no polling/sleeping loop.
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.
The per-cycle presence check in _play_on_all_sinks is what covers a
slow monitor sink, so activation must not spin waiting for it.
"""
state = self._sentinel_state()
with (
@ -299,53 +297,41 @@ class TestActivateAllOutputs:
patch("wake_alarm._sinks._parse_cards", return_value={}),
patch(
"wake_alarm._sinks._list_real_sinks",
side_effect=[["dead_analog"], ["dead_analog", ALARM_AUDIO_SINK]],
),
return_value=[ALARM_AUDIO_SINK],
) as mock_list,
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,
patch("wake_alarm._sinks.subprocess.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
# Enumerated exactly once — no retry/poll loop.
mock_list.assert_called_once()
def test_falls_back_to_first_sink_when_alarm_sink_absent(self) -> None:
"""ALARM_AUDIO_SINK never appears → default is the first real sink."""
"""ALARM_AUDIO_SINK not present → default is the first present 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._sinks._list_real_sinks",
side_effect=[["other"], ["other"], ["other"]],
),
patch("wake_alarm._sinks._list_real_sinks", return_value=["other"]),
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()
cmds = [call.args[0] for call in mock_run.call_args_list]
assert ["/usr/bin/pactl", "set-default-sink", "other"] in cmds
def test_returns_state_when_no_sink_at_all(self) -> None:
"""No real sink ever appears → returns state, no default set."""
def test_returns_state_when_no_sink_present(self) -> None:
"""No real sink present yet → returns state, no default, no maxing."""
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._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_all_outputs() is state