diff --git a/wake_alarm/_alarm.py b/wake_alarm/_alarm.py index 0b09e8c..dbd7833 100644 --- a/wake_alarm/_alarm.py +++ b/wake_alarm/_alarm.py @@ -8,18 +8,23 @@ workout-free day via HMAC-signed wake state. from __future__ import annotations +import argparse from datetime import datetime, timezone import logging +import math import os from pathlib import Path import secrets import shutil import string +import struct import subprocess import sys +import tempfile import threading import time import tkinter as tk +import wave from python_pkg.wake_alarm._constants import ( ALARM_DAYS, @@ -93,6 +98,165 @@ def _speaker_test_path() -> str: return path +_TONE_CACHE: dict[int, Path] = {} +_TONE_TIMEOUT_SECONDS: float = 6.0 +_TONE_DURATION_SECONDS: float = 1.5 +_TONE_FRAMERATE: int = 48000 +_TONE_AMPLITUDE: int = 32760 # near s16 max for the loudest sine we can emit + +# Number of back-to-back PC-speaker beeps per _play_tone call. +# pcspkr volume is hardware-fixed, so we lean on repetition + duration to be +# loud enough to actually wake the user. +_PCSPKR_REPEATS: int = 3 +_PCSPKR_GAP_SECONDS: float = 0.12 + +# Motherboard PC speaker exposed by the pcspkr kernel module. +# Writing EV_SND/SND_TONE input_event structs makes it beep — bypasses +# PipeWire/ALSA entirely, so it stays audible even when no real sink exists. +_PCSPKR_DEVICE: str = "/dev/input/by-path/platform-pcspkr-event-spkr" +_PCSPKR_EV_SND: int = 0x12 +_PCSPKR_SND_TONE: int = 0x02 +# struct input_event: timeval (long sec, long usec), u16 type, u16 code, s32 val +_PCSPKR_EVENT_FMT: str = "llHHi" + + +def _beep_pcspkr(frequency: int, duration_seconds: float) -> None: + """Beep the motherboard PC speaker via evdev (audible without any sink). + + Silently no-ops when the device is missing or unwritable so the call is + always safe from the alarm hot path. + """ + try: + # buffering=0 so the write hits the device immediately. + with Path(_PCSPKR_DEVICE).open("wb", buffering=0) as dev: + dev.write( + struct.pack( + _PCSPKR_EVENT_FMT, + 0, + 0, + _PCSPKR_EV_SND, + _PCSPKR_SND_TONE, + int(frequency), + ), + ) + time.sleep(duration_seconds) + dev.write( + struct.pack( + _PCSPKR_EVENT_FMT, + 0, + 0, + _PCSPKR_EV_SND, + _PCSPKR_SND_TONE, + 0, + ), + ) + except OSError: + _logger.warning( + "PC speaker beep at %d Hz failed (device %s)", + frequency, + _PCSPKR_DEVICE, + exc_info=True, + ) + + +def _ensure_tone_wav(frequency: int) -> Path: + """Generate (and cache) a mono 48 kHz sine WAV at *frequency* Hz.""" + cached = _TONE_CACHE.get(frequency) + if cached is not None and cached.exists(): + return cached + path = Path(tempfile.gettempdir()) / f"wake_alarm_tone_{frequency}.wav" + n_frames = int(_TONE_FRAMERATE * _TONE_DURATION_SECONDS) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(_TONE_FRAMERATE) + frames = bytearray() + for i in range(n_frames): + sample = int( + _TONE_AMPLITUDE + * math.sin(2 * math.pi * frequency * i / _TONE_FRAMERATE), + ) + frames.extend(struct.pack(" bool: + """Run *binary* on *wav* with a generous timeout. Return True on success.""" + path = shutil.which(binary) + if path is None: + return False + try: + result = subprocess.run( + [path, str(wav)], + capture_output=True, + timeout=_TONE_TIMEOUT_SECONDS, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("%s failed playing %s", binary, wav.name, exc_info=True) + return False + if result.returncode != 0: + _logger.warning( + "%s exited %d for %s: %s", + binary, + result.returncode, + wav.name, + result.stderr.decode(errors="replace").strip()[:200], + ) + return False + return True + + +def _play_tone(frequency: int) -> None: + """Play a sine tone via paplay/aplay/speaker-test, fall back to soft beep. + + Always also beeps the motherboard PC speaker (multiple times) so the + alarm stays loud and audible even when PipeWire only has the auto_null + sink. + """ + for i in range(_PCSPKR_REPEATS): + _beep_pcspkr(frequency, _TONE_DURATION_SECONDS) + if i < _PCSPKR_REPEATS - 1: + time.sleep(_PCSPKR_GAP_SECONDS) + try: + wav = _ensure_tone_wav(frequency) + except OSError: + _logger.warning( + "Could not generate tone WAV at %d Hz; using soft beep", + frequency, + exc_info=True, + ) + _beep_soft() + return + for binary in ("paplay", "aplay"): + if _try_player(binary, wav): + return + try: + subprocess.run( + [ + _speaker_test_path(), + "-t", + "sine", + "-f", + str(frequency), + "-l", + "1", + ], + capture_output=True, + timeout=_TONE_TIMEOUT_SECONDS, + check=False, + ) + except (FileNotFoundError, OSError, subprocess.TimeoutExpired): + _logger.warning( + "All tone players failed at %d Hz; falling back to soft beep", + frequency, + exc_info=True, + ) + _beep_soft() + + # 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",) @@ -159,29 +323,18 @@ def _find_fan_hwmon() -> str | None: return None -def _max_fans() -> tuple[str, str, str] | None: - """Ramp all NCT fans to 100% speed for maximum audible noise. +def _max_fans() -> bool: + """Ramp every NCT pwm channel to 100% speed via the helper script. - Saves the current pwm1 values so they can be restored after the alarm. - Safe: higher fan speed only lowers temperatures, never damages hardware. + The helper records prior state under /run/wake-alarm-fans.state so + _restore_fans() can put things back without arguments. Safe: higher fan + speed only lowers temperatures, never damages hardware. Returns: - (hwmon_dir, old_enable, old_pwm) tuple to pass to _restore_fans(), - or None when fan control is unavailable. + True when the ramp script ran successfully, False otherwise. """ - hwmon = _find_fan_hwmon() - if hwmon is None: - return None - try: - old_enable = (Path(hwmon) / "pwm1_enable").read_text().strip() - old_pwm = (Path(hwmon) / "pwm1").read_text().strip() - except OSError: - _logger.warning( - "Could not read pwm1/pwm1_enable in %s; skipping fan ramp", - hwmon, - exc_info=True, - ) - return None + if _find_fan_hwmon() is None: + return False try: result = subprocess.run( [_SUDO_BIN, "-n", _FAN_SCRIPT, "max"], @@ -195,7 +348,7 @@ def _max_fans() -> tuple[str, str, str] | None: _FAN_SCRIPT, exc_info=True, ) - return None + return False if result.returncode != 0: _logger.warning( "Fan script %s exited %d: %s", @@ -203,18 +356,17 @@ def _max_fans() -> tuple[str, str, str] | None: result.returncode, result.stderr.decode(errors="replace").strip(), ) - return None - return (hwmon, old_enable, old_pwm) + return False + return True -def _restore_fans(state: tuple[str, str, str] | None) -> None: - """Restore fan speed to the values saved by _max_fans().""" - if state is None: +def _restore_fans(*, active: bool) -> None: + """Restore fan speed if _max_fans() previously succeeded.""" + if not active: return - _hwmon, old_enable, old_pwm = state try: subprocess.run( - [_SUDO_BIN, "-n", _FAN_SCRIPT, "restore", old_enable, old_pwm], + [_SUDO_BIN, "-n", _FAN_SCRIPT, "restore"], check=False, capture_output=True, timeout=5, @@ -263,53 +415,13 @@ def _set_max_brightness() -> None: def _beep_medium(frequency: int = 1000) -> None: - """Play a medium beep via speaker-test (sine wave, short).""" - try: - subprocess.run( - [ - _speaker_test_path(), - "-t", - "sine", - "-f", - str(frequency), - "-l", - "1", - ], - capture_output=True, - timeout=3, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning( - "_beep_medium subprocess failed; falling back to soft beep", - exc_info=True, - ) - _beep_soft() + """Play a medium beep (sine tone via paplay/aplay/speaker-test).""" + _play_tone(frequency) def _beep_loud(frequency: int = 1000) -> None: - """Play a loud sine tone via speaker-test.""" - try: - subprocess.run( - [ - _speaker_test_path(), - "-t", - "sine", - "-f", - str(frequency), - "-l", - "1", - ], - capture_output=True, - timeout=3, - check=False, - ) - except (OSError, subprocess.TimeoutExpired): - _logger.warning( - "_beep_loud subprocess failed; falling back to soft beep", - exc_info=True, - ) - _beep_soft() + """Play a loud sine tone via paplay/aplay/speaker-test.""" + _play_tone(frequency) class WakeAlarm: @@ -332,23 +444,29 @@ class WakeAlarm: self.root.title("Wake Alarm" + (" [DEMO]" if demo_mode else "")) self.root.configure(bg="#1a1a1a") - if demo_mode: - self.root.geometry("800x600") - else: - screen_w = self.root.winfo_screenwidth() - screen_h = self.root.winfo_screenheight() - fullscreen = True - self.root.overrideredirect(boolean=fullscreen) - self.root.geometry(f"{screen_w}x{screen_h}+0+0") - self.root.attributes("-fullscreen", fullscreen) - self.root.attributes("-topmost", fullscreen) + # Always hijack the full screen — demo_mode only controls timers. + # NOTE: we intentionally do NOT call overrideredirect(True): on X11 it + # removes WM management and the Entry widget can't receive keyboard + # focus, so the user can't type the dismiss code. -fullscreen + + # -topmost is enough to take over the screen while staying typeable. + screen_w = self.root.winfo_screenwidth() + screen_h = self.root.winfo_screenheight() + fullscreen = True + self.root.geometry(f"{screen_w}x{screen_h}+0+0") + self.root.attributes("-fullscreen", fullscreen) + self.root.attributes("-topmost", fullscreen) + + self.root.lift() + self.root.focus_force() + self.root.update_idletasks() self._current_code = _generate_code() self._build_ui() self._schedule_code_refresh() self._schedule_dismiss_window_close() self._start_beep_thread() - self._fan_state: tuple[str, str, str] | None = _max_fans() + self._fan_state: bool = _max_fans() + self._sink_volume_state: tuple[str, str, bool] | None = _max_sink_volume() self._flash_on: bool = False self._start_screen_flash() @@ -453,7 +571,8 @@ class WakeAlarm: def _close(self) -> None: """Close the alarm window.""" self._stop_beep.set() - _restore_fans(self._fan_state) + _restore_fans(active=self._fan_state) + _restore_sink_volume(self._sink_volume_state) _restore_display() turn_off_plug() self.root.destroy() @@ -497,7 +616,8 @@ class WakeAlarm: def _close_and_schedule_fallback(self) -> None: """Close the window and schedule the 1 PM fallback alarm.""" - _restore_fans(self._fan_state) + _restore_fans(active=self._fan_state) + _restore_sink_volume(self._sink_volume_state) _restore_display() turn_off_plug() self.root.destroy() @@ -572,6 +692,153 @@ def _should_run_alarm() -> bool: return True +def _pactl_path() -> str | None: + """Return the absolute path to pactl, or None when not installed.""" + return shutil.which("pactl") + + +def _max_sink_volume() -> tuple[str, str, bool] | None: + """Unmute the default sink, raise it to 100%, return state for restore. + + Returns ``(sink_name, original_volume_pct, original_mute)`` or ``None`` + when pactl is unavailable / the call fails. The alarm is loud only if the + user's actual sink (e.g. a Bluetooth speaker) is also turned up, so this + is the single biggest lever we have. + """ + pactl = _pactl_path() + if pactl is None: + _logger.warning("pactl not on PATH; cannot raise sink volume") + return None + try: + sink_proc = subprocess.run( + [pactl, "get-default-sink"], + capture_output=True, + timeout=3, + check=False, + ) + sink = sink_proc.stdout.decode(errors="replace").strip() + if not sink: + return None + vol_proc = subprocess.run( + [pactl, "get-sink-volume", sink], + capture_output=True, + timeout=3, + check=False, + ) + mute_proc = subprocess.run( + [pactl, "get-sink-mute", sink], + capture_output=True, + timeout=3, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("pactl volume query failed", exc_info=True) + return None + # "Volume: front-left: 20641 / 31% / ..." — grab the first percent token. + vol_text = vol_proc.stdout.decode(errors="replace") + pct = "100%" + for tok in vol_text.replace(",", " ").split(): + if tok.endswith("%"): + pct = tok + break + muted = b"yes" in mute_proc.stdout + try: + subprocess.run( + [pactl, "set-sink-mute", sink, "0"], + capture_output=True, + timeout=3, + check=False, + ) + subprocess.run( + [pactl, "set-sink-volume", sink, "100%"], + capture_output=True, + timeout=3, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("pactl volume set failed", exc_info=True) + return None + _logger.info("Raised sink %s volume %s\u2192100%%", sink, pct) + return (sink, pct, muted) + + +def _restore_sink_volume(state: tuple[str, str, bool] | None) -> None: + """Restore the sink volume + mute captured by :func:`_max_sink_volume`.""" + if state is None: + return + sink, pct, muted = state + pactl = _pactl_path() + if pactl is None: + return + try: + subprocess.run( + [pactl, "set-sink-volume", sink, pct], + capture_output=True, + timeout=3, + check=False, + ) + subprocess.run( + [pactl, "set-sink-mute", sink, "1" if muted else "0"], + capture_output=True, + timeout=3, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + _logger.warning("pactl volume restore failed", exc_info=True) + + +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 \u2014 alarm will be SILENT. " + "Sinks: %s", + sink_names or "", + ) + else: + _logger.info("Audio sinks available: %s", sink_names) + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + """Parse CLI arguments for the alarm daemon.""" + parser = argparse.ArgumentParser(description="Wake alarm daemon.") + parser.add_argument( + "--production", + action="store_true", + help="Production mode (default; kept for systemd compatibility).", + ) + parser.add_argument( + "--demo", + action="store_true", + help="Run with a smaller window and shorter timers.", + ) + parser.add_argument( + "--trigger-now", + action="store_true", + help="Bypass the day/dismiss gate and fire the alarm immediately.", + ) + return parser.parse_args(argv) + + def main() -> None: """Entry point for the wake alarm daemon.""" logging.basicConfig( @@ -579,14 +846,22 @@ def main() -> None: format="%(asctime)s %(name)s %(levelname)s %(message)s", ) - if not _should_run_alarm(): + args = _parse_args(sys.argv[1:]) + + if not args.trigger_now and not _should_run_alarm(): return - demo_mode = "--demo" in sys.argv + _logger.warning( + "ALARM TRIGGERED at %s (demo=%s, trigger_now=%s)", + datetime.now(tz=timezone.utc).isoformat(timespec="seconds"), + args.demo, + args.trigger_now, + ) + _warn_if_no_real_sink() _wake_display() _set_max_brightness() turn_on_plug() - alarm = WakeAlarm(demo_mode=demo_mode) + alarm = WakeAlarm(demo_mode=args.demo) alarm.run() diff --git a/wake_alarm/tests/test_alarm.py b/wake_alarm/tests/test_alarm.py index 2a5ecec..b4567bf 100644 --- a/wake_alarm/tests/test_alarm.py +++ b/wake_alarm/tests/test_alarm.py @@ -11,23 +11,31 @@ from unittest.mock import MagicMock, patch import pytest if TYPE_CHECKING: - from collections.abc import Generator + from collections.abc import Generator, Iterator from python_pkg.wake_alarm._alarm import ( _beep_loud, _beep_medium, + _beep_pcspkr, _beep_soft, + _ensure_tone_wav, _find_fan_hwmon, _generate_code, _is_alarm_day, _max_fans, + _max_sink_volume, + _parse_args, _play_on_extra_devices, + _play_tone, _restore_display, _restore_fans, + _restore_sink_volume, _set_max_brightness, _should_run_alarm, _speaker_test_path, + _try_player, _wake_display, + _warn_if_no_real_sink, ) from python_pkg.wake_alarm._constants import ( DISMISS_CODE_LENGTH, @@ -195,122 +203,17 @@ class TestBeepFunctions: mock_sys.stdout.write.assert_called_once_with("\a") mock_sys.stdout.flush.assert_called_once() - def test_beep_medium_calls_speaker_test(self) -> None: - """_beep_medium runs speaker-test subprocess.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "python_pkg.wake_alarm._alarm.subprocess.run", - ) as mock_run, - ): + def test_beep_medium_delegates_to_play_tone(self) -> None: + """_beep_medium just delegates to _play_tone.""" + with patch("python_pkg.wake_alarm._alarm._play_tone") as mock_play: _beep_medium(frequency=800) - mock_run.assert_called_once() - args = mock_run.call_args[0][0] - assert "/usr/bin/speaker-test" in args - assert "800" in args + mock_play.assert_called_once_with(800) - def test_beep_medium_falls_back_on_error(self) -> None: - """_beep_medium falls back to soft beep on OSError.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "python_pkg.wake_alarm._alarm.subprocess.run", - side_effect=OSError("no speaker-test"), - ), - patch( - "python_pkg.wake_alarm._alarm._beep_soft", - ) as mock_soft, - ): - _beep_medium() - mock_soft.assert_called_once() - - def test_beep_medium_falls_back_on_timeout(self) -> None: - """_beep_medium falls back on TimeoutExpired.""" - from subprocess import TimeoutExpired - - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "python_pkg.wake_alarm._alarm.subprocess.run", - side_effect=TimeoutExpired("cmd", 3), - ), - patch( - "python_pkg.wake_alarm._alarm._beep_soft", - ) as mock_soft, - ): - _beep_medium() - mock_soft.assert_called_once() - - def test_beep_medium_falls_back_on_missing_binary(self) -> None: - """_beep_medium falls back when speaker-test binary not found.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - side_effect=FileNotFoundError("not found"), - ), - patch( - "python_pkg.wake_alarm._alarm._beep_soft", - ) as mock_soft, - ): - _beep_medium() - mock_soft.assert_called_once() - - def test_beep_loud_calls_speaker_test(self) -> None: - """_beep_loud runs speaker-test subprocess.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "python_pkg.wake_alarm._alarm.subprocess.run", - ) as mock_run, - ): + def test_beep_loud_delegates_to_play_tone(self) -> None: + """_beep_loud just delegates to _play_tone.""" + with patch("python_pkg.wake_alarm._alarm._play_tone") as mock_play: _beep_loud(frequency=1200) - mock_run.assert_called_once() - args = mock_run.call_args[0][0] - assert "1200" in args - - def test_beep_loud_falls_back_on_error(self) -> None: - """_beep_loud falls back to soft beep on OSError.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - return_value="/usr/bin/speaker-test", - ), - patch( - "python_pkg.wake_alarm._alarm.subprocess.run", - side_effect=OSError("fail"), - ), - patch( - "python_pkg.wake_alarm._alarm._beep_soft", - ) as mock_soft, - ): - _beep_loud() - mock_soft.assert_called_once() - - def test_beep_loud_falls_back_on_missing_binary(self) -> None: - """_beep_loud falls back when speaker-test binary not found.""" - with ( - patch( - "python_pkg.wake_alarm._alarm._speaker_test_path", - side_effect=FileNotFoundError("not found"), - ), - patch( - "python_pkg.wake_alarm._alarm._beep_soft", - ) as mock_soft, - ): - _beep_loud() - mock_soft.assert_called_once() + mock_play.assert_called_once_with(1200) class TestShouldRunAlarm: @@ -368,6 +271,21 @@ class TestDisplayHelpers: _wake_display() mock_run.assert_not_called() + def test_wake_display_runs_xset_commands(self) -> None: + """_wake_display runs xset dpms force on + xset s off.""" + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/xset", + ), + patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run, + ): + _wake_display() + assert mock_run.call_count == 2 + call_args = [call[0][0] for call in mock_run.call_args_list] + assert ["/usr/bin/xset", "dpms", "force", "on"] in call_args + assert ["/usr/bin/xset", "s", "off"] in call_args + def test_restore_display_skips_when_xset_missing(self) -> None: """_restore_display does nothing when xset is not on PATH.""" with ( @@ -468,23 +386,13 @@ class TestFindFanHwmon: class TestMaxFans: """Tests for _max_fans.""" - def test_returns_none_when_no_hwmon(self) -> None: - """No fan controller → returns None immediately.""" + def test_returns_false_when_no_hwmon(self) -> None: + """No fan controller → returns False immediately.""" with patch("python_pkg.wake_alarm._alarm._find_fan_hwmon", return_value=None): - assert _max_fans() is None + assert _max_fans() is False - def test_returns_none_on_oserror_reading_pwm(self, tmp_path: pathlib.Path) -> None: - """Missing pwm files → returns None.""" - hwmon_dir = str(tmp_path) - with patch( - "python_pkg.wake_alarm._alarm._find_fan_hwmon", return_value=hwmon_dir - ): - assert _max_fans() is None - - def test_returns_none_on_script_oserror(self, tmp_path: pathlib.Path) -> None: - """OSError running fan script → returns None.""" - (tmp_path / "pwm1_enable").write_text("5\n") - (tmp_path / "pwm1").write_text("165\n") + def test_returns_false_on_script_oserror(self, tmp_path: pathlib.Path) -> None: + """OSError running fan script → returns False.""" with ( patch( "python_pkg.wake_alarm._alarm._find_fan_hwmon", @@ -495,12 +403,10 @@ class TestMaxFans: side_effect=OSError("not found"), ), ): - assert _max_fans() is None + assert _max_fans() is False - def test_returns_none_on_script_timeout(self, tmp_path: pathlib.Path) -> None: - """TimeoutExpired running fan script → returns None.""" - (tmp_path / "pwm1_enable").write_text("5\n") - (tmp_path / "pwm1").write_text("165\n") + def test_returns_false_on_script_timeout(self, tmp_path: pathlib.Path) -> None: + """TimeoutExpired running fan script → returns False.""" with ( patch( "python_pkg.wake_alarm._alarm._find_fan_hwmon", @@ -511,12 +417,10 @@ class TestMaxFans: side_effect=subprocess.TimeoutExpired("fan", 5), ), ): - assert _max_fans() is None + assert _max_fans() is False - def test_returns_none_on_nonzero_returncode(self, tmp_path: pathlib.Path) -> None: - """Fan script exits non-zero → returns None.""" - (tmp_path / "pwm1_enable").write_text("5\n") - (tmp_path / "pwm1").write_text("165\n") + def test_returns_false_on_nonzero_returncode(self, tmp_path: pathlib.Path) -> None: + """Fan script exits non-zero → returns False.""" mock_result = MagicMock() mock_result.returncode = 1 with ( @@ -525,15 +429,14 @@ class TestMaxFans: return_value=str(tmp_path), ), patch( - "python_pkg.wake_alarm._alarm.subprocess.run", return_value=mock_result + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=mock_result, ), ): - assert _max_fans() is None + assert _max_fans() is False - def test_returns_state_on_success(self, tmp_path: pathlib.Path) -> None: - """Successful run → returns (hwmon, old_enable, old_pwm).""" - (tmp_path / "pwm1_enable").write_text("5\n") - (tmp_path / "pwm1").write_text("165\n") + def test_returns_true_on_success(self, tmp_path: pathlib.Path) -> None: + """Successful run → returns True (state is saved by the helper).""" mock_result = MagicMock() mock_result.returncode = 0 with ( @@ -542,32 +445,30 @@ class TestMaxFans: return_value=str(tmp_path), ), patch( - "python_pkg.wake_alarm._alarm.subprocess.run", return_value=mock_result + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=mock_result, ), ): - result = _max_fans() - assert result == (str(tmp_path), "5", "165") + assert _max_fans() is True class TestRestoreFans: """Tests for _restore_fans.""" - def test_noop_when_state_is_none(self) -> None: - """None state → subprocess.run is never called.""" + def test_noop_when_inactive(self) -> None: + """False state → subprocess.run is never called.""" with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run: - _restore_fans(None) + _restore_fans(active=False) mock_run.assert_not_called() - def test_calls_fan_script_with_saved_values(self) -> None: - """Saved state → fan script called with restore + old values.""" + def test_calls_fan_script_restore(self) -> None: + """Active state → fan script called with restore (no args).""" with patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run: mock_run.return_value.returncode = 0 - _restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) + _restore_fans(active=True) mock_run.assert_called_once() args = mock_run.call_args[0][0] assert "restore" in args - assert "5" in args - assert "165" in args def test_ignores_oserror_on_restore(self) -> None: """OSError from fan script is silently suppressed.""" @@ -575,7 +476,7 @@ class TestRestoreFans: "python_pkg.wake_alarm._alarm.subprocess.run", side_effect=OSError("no script"), ): - _restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) # must not raise + _restore_fans(active=True) # must not raise def test_ignores_timeout_on_restore(self) -> None: """TimeoutExpired from fan script is silently suppressed.""" @@ -583,7 +484,7 @@ class TestRestoreFans: "python_pkg.wake_alarm._alarm.subprocess.run", side_effect=subprocess.TimeoutExpired("fan", 5), ): - _restore_fans(("/sys/class/hwmon/hwmon6", "5", "165")) # must not raise + _restore_fans(active=True) # must not raise class TestSetMaxBrightness: @@ -694,3 +595,549 @@ class TestSetMaxBrightness: ), ): _set_max_brightness() # must not raise + + +class TestEnsureToneWav: + """Tests for _ensure_tone_wav (sine WAV generator + cache).""" + + def test_generates_and_caches(self, tmp_path: pathlib.Path) -> None: + """First call generates the WAV; second call returns the cached path.""" + from python_pkg.wake_alarm import _alarm as alarm_mod + + alarm_mod._TONE_CACHE.clear() + with patch( + "python_pkg.wake_alarm._alarm.tempfile.gettempdir", + return_value=str(tmp_path), + ): + path1 = _ensure_tone_wav(440) + assert path1.exists() + size = path1.stat().st_size + assert size > 0 + # Second call must hit the cache (no regeneration). + with patch("python_pkg.wake_alarm._alarm.wave.open") as mock_open: + path2 = _ensure_tone_wav(440) + mock_open.assert_not_called() + assert path2 == path1 + alarm_mod._TONE_CACHE.clear() + + def test_regenerates_when_cached_file_missing( + self, + tmp_path: pathlib.Path, + ) -> None: + """If the cached file was deleted, regenerate it.""" + from python_pkg.wake_alarm._alarm import _TONE_CACHE + + _TONE_CACHE.clear() + with patch( + "python_pkg.wake_alarm._alarm.tempfile.gettempdir", + return_value=str(tmp_path), + ): + path1 = _ensure_tone_wav(880) + path1.unlink() + path2 = _ensure_tone_wav(880) + assert path2.exists() + _TONE_CACHE.clear() + + +class TestTryPlayer: + """Tests for _try_player.""" + + def test_returns_false_when_binary_missing( + self, + tmp_path: pathlib.Path, + ) -> None: + """Missing binary returns False without raising.""" + wav = tmp_path / "x.wav" + wav.write_bytes(b"\x00") + with patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value=None, + ): + assert _try_player("paplay", wav) is False + + def test_returns_true_on_success(self, tmp_path: pathlib.Path) -> None: + """Zero exit code returns True.""" + wav = tmp_path / "x.wav" + wav.write_bytes(b"\x00") + result = MagicMock() + result.returncode = 0 + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/paplay", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=result, + ), + ): + assert _try_player("paplay", wav) is True + + def test_returns_false_on_nonzero_exit( + self, + tmp_path: pathlib.Path, + ) -> None: + """Non-zero exit code returns False and logs.""" + wav = tmp_path / "x.wav" + wav.write_bytes(b"\x00") + result = MagicMock() + result.returncode = 1 + result.stderr = b"boom" + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/paplay", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=result, + ), + ): + assert _try_player("paplay", wav) is False + + def test_returns_false_on_timeout(self, tmp_path: pathlib.Path) -> None: + """TimeoutExpired returns False and logs.""" + wav = tmp_path / "x.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/paplay", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=subprocess.TimeoutExpired("paplay", 6), + ), + ): + assert _try_player("paplay", wav) is False + + def test_returns_false_on_oserror(self, tmp_path: pathlib.Path) -> None: + """OSError returns False and logs.""" + wav = tmp_path / "x.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/paplay", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=OSError("nope"), + ), + ): + assert _try_player("paplay", wav) is False + + +class TestBeepPcspkr: + """Tests for _beep_pcspkr (evdev PC speaker).""" + + def test_writes_tone_then_zero_to_device(self) -> None: + """Successful path writes start-frequency then stop event.""" + + mock_dev = MagicMock() + mock_open_ctx = MagicMock() + mock_open_ctx.__enter__.return_value = mock_dev + mock_open_ctx.__exit__.return_value = False + with ( + patch( + "python_pkg.wake_alarm._alarm.Path.open", + return_value=mock_open_ctx, + ), + patch("python_pkg.wake_alarm._alarm.time.sleep"), + ): + _beep_pcspkr(1000, 0.05) + # First write carries the frequency, second write carries 0 (stop). + assert mock_dev.write.call_count == 2 + + def test_oserror_is_swallowed(self) -> None: + """OSError opening the device must not raise.""" + + with patch( + "python_pkg.wake_alarm._alarm.Path.open", + side_effect=OSError("no device"), + ): + _beep_pcspkr(1000, 0.05) # must not raise + + +class TestPlayTone: + """Tests for _play_tone.""" + + @pytest.fixture(autouse=True) + def _silence_pcspkr(self) -> Iterator[None]: + """Stop tests from hitting the real /dev/input PC speaker device.""" + with patch("python_pkg.wake_alarm._alarm._beep_pcspkr"): + yield + + def test_paplay_success_short_circuits(self, tmp_path: pathlib.Path) -> None: + """If paplay succeeds, no further players are tried.""" + wav = tmp_path / "tone.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm._ensure_tone_wav", + return_value=wav, + ), + patch( + "python_pkg.wake_alarm._alarm._try_player", + return_value=True, + ) as mock_try, + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + ) as mock_run, + ): + _play_tone(440) + mock_try.assert_called_once_with("paplay", wav) + mock_run.assert_not_called() + + def test_falls_back_to_aplay_then_speaker_test( + self, + tmp_path: pathlib.Path, + ) -> None: + """paplay+aplay fail → speaker-test is tried.""" + wav = tmp_path / "tone.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm._ensure_tone_wav", + return_value=wav, + ), + patch( + "python_pkg.wake_alarm._alarm._try_player", + return_value=False, + ), + patch( + "python_pkg.wake_alarm._alarm._speaker_test_path", + return_value="/usr/bin/speaker-test", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + ) as mock_run, + ): + _play_tone(1000) + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert "/usr/bin/speaker-test" in args + assert "1000" in args + + def test_soft_beep_when_speaker_test_missing( + self, + tmp_path: pathlib.Path, + ) -> None: + """All players fail → soft beep.""" + wav = tmp_path / "tone.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm._ensure_tone_wav", + return_value=wav, + ), + patch( + "python_pkg.wake_alarm._alarm._try_player", + return_value=False, + ), + patch( + "python_pkg.wake_alarm._alarm._speaker_test_path", + side_effect=FileNotFoundError("missing"), + ), + patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft, + ): + _play_tone(800) + mock_soft.assert_called_once() + + def test_soft_beep_when_speaker_test_times_out( + self, + tmp_path: pathlib.Path, + ) -> None: + """speaker-test TimeoutExpired → soft beep.""" + wav = tmp_path / "tone.wav" + wav.write_bytes(b"\x00") + with ( + patch( + "python_pkg.wake_alarm._alarm._ensure_tone_wav", + return_value=wav, + ), + patch( + "python_pkg.wake_alarm._alarm._try_player", + return_value=False, + ), + patch( + "python_pkg.wake_alarm._alarm._speaker_test_path", + return_value="/usr/bin/speaker-test", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=subprocess.TimeoutExpired("speaker-test", 6), + ), + patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft, + ): + _play_tone(800) + mock_soft.assert_called_once() + + def test_soft_beep_when_wav_generation_fails(self) -> None: + """OSError generating WAV → soft beep.""" + with ( + patch( + "python_pkg.wake_alarm._alarm._ensure_tone_wav", + side_effect=OSError("disk full"), + ), + patch("python_pkg.wake_alarm._alarm._beep_soft") as mock_soft, + ): + _play_tone(440) + mock_soft.assert_called_once() + + +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( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value=None, + ), + patch( + "python_pkg.wake_alarm._alarm.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( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=result, + ), + patch("python_pkg.wake_alarm._alarm._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( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=result, + ), + patch("python_pkg.wake_alarm._alarm._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( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=subprocess.TimeoutExpired("pactl", 5), + ), + ): + _warn_if_no_real_sink() # must not raise + + +class TestMaxSinkVolume: + """Tests for _max_sink_volume and _restore_sink_volume.""" + + def test_returns_none_when_pactl_missing(self) -> None: + """No pactl on PATH → returns None, logs warning.""" + with patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None): + assert _max_sink_volume() is None + + def test_returns_none_when_default_sink_empty(self) -> None: + """Empty get-default-sink output → returns None.""" + sink_proc = MagicMock(stdout=b"") + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + return_value=sink_proc, + ), + ): + assert _max_sink_volume() is None + + def test_query_failure_returns_none(self) -> None: + """OSError during query → returns None, no raise.""" + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=OSError("boom"), + ), + ): + assert _max_sink_volume() is None + + def test_set_failure_returns_none(self) -> None: + """OSError during set-sink-volume → returns None.""" + sink_proc = MagicMock(stdout=b"my_sink\n") + vol_proc = MagicMock(stdout=b"Volume: front-left: 20641 / 31% / -30.10 dB") + mute_proc = MagicMock(stdout=b"Mute: no\n") + + def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock: + if "get-default-sink" in cmd: + return sink_proc + if "get-sink-volume" in cmd: + return vol_proc + if "get-sink-mute" in cmd: + return mute_proc + raise subprocess.TimeoutExpired(cmd, 3) + + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=fake_run, + ), + ): + assert _max_sink_volume() is None + + def test_happy_path_returns_state(self) -> None: + """Successful query+set returns the captured state tuple.""" + sink_proc = MagicMock(stdout=b"my_sink\n") + vol_proc = MagicMock(stdout=b"Volume: front-left: 20641 / 31% / -30.10 dB") + mute_proc = MagicMock(stdout=b"Mute: yes\n") + ok = MagicMock(stdout=b"", returncode=0) + + def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock: + if "get-default-sink" in cmd: + return sink_proc + if "get-sink-volume" in cmd: + return vol_proc + if "get-sink-mute" in cmd: + return mute_proc + return ok + + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=fake_run, + ), + ): + state = _max_sink_volume() + assert state == ("my_sink", "31%", True) + + def test_happy_path_no_percent_token(self) -> None: + """Missing % token → falls back to 100%, not None.""" + sink_proc = MagicMock(stdout=b"s\n") + vol_proc = MagicMock(stdout=b"weird output") + mute_proc = MagicMock(stdout=b"Mute: no\n") + ok = MagicMock(stdout=b"", returncode=0) + + def fake_run(cmd: list[str], **_kwargs: object) -> MagicMock: + if "get-default-sink" in cmd: + return sink_proc + if "get-sink-volume" in cmd: + return vol_proc + if "get-sink-mute" in cmd: + return mute_proc + return ok + + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=fake_run, + ), + ): + state = _max_sink_volume() + assert state == ("s", "100%", False) + + +class TestRestoreSinkVolume: + """Tests for _restore_sink_volume.""" + + def test_none_state_is_noop(self) -> None: + """None state → does nothing, no pactl call.""" + with patch("python_pkg.wake_alarm._alarm.shutil.which") as mock_which: + _restore_sink_volume(None) + mock_which.assert_not_called() + + def test_no_pactl_returns_silently(self) -> None: + """State present but pactl missing → no raise, no call.""" + with ( + patch("python_pkg.wake_alarm._alarm.shutil.which", return_value=None), + patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run, + ): + _restore_sink_volume(("sink", "42%", False)) + mock_run.assert_not_called() + + def test_restores_volume_and_mute(self) -> None: + """Calls set-sink-volume and set-sink-mute with captured values.""" + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run, + ): + _restore_sink_volume(("sink", "42%", True)) + cmds = [call.args[0] for call in mock_run.call_args_list] + assert ["/usr/bin/pactl", "set-sink-volume", "sink", "42%"] in cmds + assert ["/usr/bin/pactl", "set-sink-mute", "sink", "1"] in cmds + + def test_oserror_during_restore_is_swallowed(self) -> None: + """OSError during restore → no raise.""" + with ( + patch( + "python_pkg.wake_alarm._alarm.shutil.which", + return_value="/usr/bin/pactl", + ), + patch( + "python_pkg.wake_alarm._alarm.subprocess.run", + side_effect=OSError("boom"), + ), + ): + _restore_sink_volume(("sink", "50%", False)) # must not raise + + +class TestParseArgs: + """Tests for _parse_args.""" + + def test_default_flags_are_false(self) -> None: + """No CLI args means every flag is False.""" + ns = _parse_args([]) + assert ns.demo is False + assert ns.trigger_now is False + assert ns.production is False + + def test_flags_parse(self) -> None: + """Each flag flips to True when passed.""" + ns = _parse_args(["--production", "--demo", "--trigger-now"]) + assert ns.production is True + assert ns.demo is True + assert ns.trigger_now is True diff --git a/wake_alarm/tests/test_alarm_part2.py b/wake_alarm/tests/test_alarm_part2.py index a65ac29..725e354 100644 --- a/wake_alarm/tests/test_alarm_part2.py +++ b/wake_alarm/tests/test_alarm_part2.py @@ -53,9 +53,13 @@ def _block_extra_devices() -> Generator[MagicMock]: """Prevent real subprocess.Popen calls for extra ALSA devices.""" with ( patch("python_pkg.wake_alarm._alarm._play_on_extra_devices") as mock, - patch("python_pkg.wake_alarm._alarm._max_fans", return_value=None), + patch("python_pkg.wake_alarm._alarm._max_fans", return_value=False), patch("python_pkg.wake_alarm._alarm._restore_fans"), patch("python_pkg.wake_alarm._alarm._set_max_brightness"), + patch("python_pkg.wake_alarm._alarm._wake_display"), + patch("python_pkg.wake_alarm._alarm._warn_if_no_real_sink"), + patch("python_pkg.wake_alarm._alarm._max_sink_volume", return_value=None), + patch("python_pkg.wake_alarm._alarm._restore_sink_volume"), patch("python_pkg.wake_alarm._alarm.turn_on_plug"), patch("python_pkg.wake_alarm._alarm.turn_off_plug"), ): @@ -82,10 +86,20 @@ class TestWakeAlarmInit: self, mock_tk_module: MagicMock, ) -> None: - """Demo mode creates a smaller window.""" + """Demo mode still hijacks the full screen — only timers differ.""" alarm = WakeAlarm(demo_mode=True) assert alarm.demo_mode is True assert alarm.dismissed is False + mock_root = mock_tk_module.Tk.return_value + # We deliberately drop overrideredirect (X11 focus bug); fullscreen+topmost + # are what take over the screen now. + mock_root.overrideredirect.assert_not_called() + fs_calls = [ + c + for c in mock_root.attributes.call_args_list + if c.args and c.args[0] == "-fullscreen" + ] + assert fs_calls, "-fullscreen attribute must be set" alarm._stop_beep.set() # Stop beep thread def test_production_mode_fullscreen( @@ -96,7 +110,13 @@ class TestWakeAlarmInit: alarm = WakeAlarm(demo_mode=False) assert alarm.demo_mode is False mock_root = mock_tk_module.Tk.return_value - mock_root.overrideredirect.assert_called_once() + mock_root.overrideredirect.assert_not_called() + fs_calls = [ + c + for c in mock_root.attributes.call_args_list + if c.args and c.args[0] == "-fullscreen" + ] + assert fs_calls, "-fullscreen attribute must be set" alarm._stop_beep.set() @@ -181,10 +201,14 @@ class TestMain: def test_exits_when_not_alarm_day(self) -> None: """main() returns early when not an alarm day.""" - with patch( - "python_pkg.wake_alarm._alarm._should_run_alarm", - return_value=False, + with ( + patch( + "python_pkg.wake_alarm._alarm._should_run_alarm", + return_value=False, + ), + patch("python_pkg.wake_alarm._alarm.sys") as mock_sys, ): + mock_sys.argv = ["alarm"] main() # Should just return without error def test_creates_alarm_when_should_run( @@ -192,6 +216,7 @@ class TestMain: mock_tk_module: MagicMock, ) -> None: """main() creates a WakeAlarm when conditions are met.""" + del mock_tk_module with ( patch( "python_pkg.wake_alarm._alarm._should_run_alarm", @@ -203,10 +228,30 @@ class TestMain: patch.object(WakeAlarm, "run") as mock_run, patch.object(WakeAlarm, "__init__", return_value=None), ): - mock_sys.argv = [] + mock_sys.argv = ["alarm"] main() mock_run.assert_called_once() + def test_trigger_now_bypasses_gate( + self, + mock_tk_module: MagicMock, + ) -> None: + """--trigger-now bypasses _should_run_alarm.""" + del mock_tk_module + with ( + patch( + "python_pkg.wake_alarm._alarm._should_run_alarm", + return_value=False, + ) as mock_gate, + patch("python_pkg.wake_alarm._alarm.sys") as mock_sys, + patch.object(WakeAlarm, "run") as mock_run, + patch.object(WakeAlarm, "__init__", return_value=None), + ): + mock_sys.argv = ["alarm", "--trigger-now"] + main() + mock_gate.assert_not_called() + mock_run.assert_called_once() + class TestCodeRefreshAndTimer: """Tests for code refresh and timer update methods.""" @@ -286,11 +331,10 @@ class TestCloseAndFallback: ) -> None: """_close calls _restore_fans with the saved fan state.""" alarm = WakeAlarm(demo_mode=True) - saved_state = ("/sys/class/hwmon/hwmon6", "5", "165") - alarm._fan_state = saved_state + alarm._fan_state = True with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore: alarm._close() - mock_restore.assert_called_once_with(saved_state) + mock_restore.assert_called_once_with(active=True) def test_close_and_schedule_fallback( self, @@ -308,11 +352,10 @@ class TestCloseAndFallback: ) -> None: """_close_and_schedule_fallback calls _restore_fans with the saved state.""" alarm = WakeAlarm(demo_mode=True) - saved_state = ("/sys/class/hwmon/hwmon6", "5", "165") - alarm._fan_state = saved_state + alarm._fan_state = True with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore: alarm._close_and_schedule_fallback() - mock_restore.assert_called_once_with(saved_state) + mock_restore.assert_called_once_with(active=True) alarm._stop_beep.set() diff --git a/wake_alarm/wake-alarm-fans.sh b/wake_alarm/wake-alarm-fans.sh index 6745695..8f014da 100755 --- a/wake_alarm/wake-alarm-fans.sh +++ b/wake_alarm/wake-alarm-fans.sh @@ -1,15 +1,19 @@ #!/bin/bash -# Control CPU/case fan speed for the wake alarm. +# Control ALL NCT pwm fan channels for the wake alarm. # # Usage: -# wake-alarm-fans.sh max — ramp all NCT fans to 100% -# wake-alarm-fans.sh restore — restore saved values +# wake-alarm-fans.sh max — ramp every pwm[1-9] channel to 100% +# wake-alarm-fans.sh restore — restore the state captured by the last `max` # # Must be run as root (installed in /etc/sudoers.d/wake-alarm via install.sh). # Safe: fans are designed to run at max speed indefinitely. +# +# State is stored at $STATE_FILE so `restore` doesn't need any arguments. set -euo pipefail +STATE_FILE="/run/wake-alarm-fans.state" + # Locate the hwmon directory for any NCT Super I/O fan controller. HWMON="" for name_file in /sys/class/hwmon/hwmon*/name; do @@ -28,25 +32,32 @@ if [[ -z "$HWMON" ]]; then exit 0 fi -PWM_PATH="$HWMON/pwm1" -ENABLE_PATH="$HWMON/pwm1_enable" - case "${1:-}" in max) - echo 1 > "$ENABLE_PATH" # Switch to manual mode - echo 255 > "$PWM_PATH" # 255/255 = 100% speed + : > "$STATE_FILE" + for pwm in "$HWMON"/pwm[0-9]; do + [[ -w "$pwm" ]] || continue + enable="${pwm}_enable" + [[ -w "$enable" ]] || continue + old_pwm=$(cat "$pwm") + old_enable=$(cat "$enable") + printf '%s %s %s\n' "$pwm" "$old_enable" "$old_pwm" >> "$STATE_FILE" + echo 1 > "$enable" # Switch to manual mode. + echo 255 > "$pwm" # 255/255 = 100% speed. + done ;; restore) - if [[ $# -ne 3 ]]; then - echo "Usage: $0 restore " >&2 - exit 1 - fi - # Restore pwm value first, then restore the control mode. - echo "${3}" > "$PWM_PATH" - echo "${2}" > "$ENABLE_PATH" + [[ -f "$STATE_FILE" ]] || exit 0 + while read -r pwm old_enable old_pwm; do + [[ -w "$pwm" && -w "${pwm}_enable" ]] || continue + # Restore pwm value first, then restore the control mode. + echo "$old_pwm" > "$pwm" + echo "$old_enable" > "${pwm}_enable" + done < "$STATE_FILE" + rm -f "$STATE_FILE" ;; *) - echo "Usage: $0 max | $0 restore " >&2 + echo "Usage: $0 max | $0 restore" >&2 exit 1 ;; esac