mirror of
https://github.com/kuhyx/wake-alarm.git
synced 2026-11-10 19:27:02 +01:00
wake_alarm: louder + typeable + multi-fan
- Drop overrideredirect on the dismiss UI: on X11 it bypassed the WM and silently killed keyboard focus on the Entry, so the user could not type the dismiss code. -fullscreen + -topmost still cover the whole screen. - Add _max_sink_volume() / _restore_sink_volume(): unmute the default PulseAudio/PipeWire sink and raise it to 100% at alarm start, restore the original volume + mute state on dismiss. This is the biggest audio lever — pcspkr is hardware-fixed and was already maxed. - pcspkr: 3 back-to-back 1.5s beeps per cycle (was 1x 0.8s) at near-S16 max amplitude. - Fans script: control every NCT pwm[1-9] channel, persist per-channel pre-alarm state to /run/wake-alarm-fans.state so restore needs no args. - Tests: 100% branch coverage on python_pkg/wake_alarm/ (140 passed).
This commit is contained in:
parent
3109d6053c
commit
7aebf8e333
@ -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("<h", sample))
|
||||
wav.writeframesraw(bytes(frames))
|
||||
_TONE_CACHE[frequency] = path
|
||||
return path
|
||||
|
||||
|
||||
def _try_player(binary: str, wav: Path) -> 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 "<none>",
|
||||
)
|
||||
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()
|
||||
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
|
||||
@ -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 <enable> <pwm> — 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 <old_enable> <old_pwm>" >&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 <old_enable> <old_pwm>" >&2
|
||||
echo "Usage: $0 max | $0 restore" >&2
|
||||
exit 1
|
||||
;;
|
||||
esac
|
||||
|
||||
Loading…
Reference in New Issue
Block a user