mirror of
https://github.com/kuhyx/wake-alarm.git
synced 2026-07-04 13:43:01 +02:00
feat: split oversized modules for 500-line limit, fix kasa coverage gap
Split diet_guard/_gatelock.py, wake_alarm/_alarm.py, and the usage_report.py/_usage_report_parsing.py pair into focused sub-modules so every Python file is <= 500 lines, satisfying test_file_length.py. Install python-kasa into .venv (declared in requirements but missing after the 3.13->3.14 venv upgrade), fixing 8 failing smart_plug tests and restoring 100% coverage. Also includes prior in-progress work from the working tree: the wake_alarm Progress/View/Hardware field-grouping refactor, brother_printer query module + tests, diet_guard foodbank/state/cli updates, new shared coerce/logging_setup helpers, morning_routine orchestrator tweaks, dwm window-manager config, gaming scripts, and misc maintenance/digital-wellbeing script updates. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
dfe3fc6e27
commit
db2be1f6a1
@ -9,15 +9,16 @@ workout-free day via HMAC-signed wake state.
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import tkinter as tk
|
||||
|
||||
from python_pkg.shared.logging_setup import configure_logging
|
||||
from python_pkg.wake_alarm._alarm_display import _restore_display, _wake_display
|
||||
from python_pkg.wake_alarm._audio import (
|
||||
_activate_alarm_audio,
|
||||
_beep_loud,
|
||||
@ -40,6 +41,7 @@ from python_pkg.wake_alarm._constants import (
|
||||
DISMISS_FLASH_SECONDS,
|
||||
DISMISS_ROUNDS_REQUIRED,
|
||||
DISMISS_WINDOW_MINUTES,
|
||||
DISPLAY_WAKE_WAIT_SECONDS,
|
||||
LOUD_TOGGLE_INTERVAL,
|
||||
MEDIUM_BEEP_INTERVAL,
|
||||
PHASE_MEDIUM_END,
|
||||
@ -50,6 +52,7 @@ from python_pkg.wake_alarm._smart_plug import turn_off_plug, turn_on_plug
|
||||
from python_pkg.wake_alarm._state import (
|
||||
save_wake_state,
|
||||
was_alarm_dismissed_today,
|
||||
was_workout_logged_today,
|
||||
)
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
@ -60,31 +63,37 @@ def _is_alarm_day() -> bool:
|
||||
return datetime.now(tz=timezone.utc).weekday() in ALARM_DAYS
|
||||
|
||||
|
||||
def _wake_display() -> None:
|
||||
"""Force the display on and disable screensaver during alarm."""
|
||||
xset = shutil.which("xset")
|
||||
if xset is None:
|
||||
_logger.warning("xset not on PATH; skipping display wake")
|
||||
return
|
||||
for cmd in (
|
||||
[xset, "dpms", "force", "on"],
|
||||
[xset, "s", "off"],
|
||||
):
|
||||
subprocess.run(cmd, check=False, capture_output=True, timeout=5)
|
||||
@dataclass
|
||||
class _AlarmView:
|
||||
"""The Tk widgets that make up the alarm's dismiss-challenge screen."""
|
||||
|
||||
container: tk.Frame
|
||||
title_label: tk.Label
|
||||
round_label: tk.Label
|
||||
info_label: tk.Label
|
||||
code_label: tk.Label
|
||||
entry: tk.Entry
|
||||
status_label: tk.Label
|
||||
timer_label: tk.Label
|
||||
|
||||
|
||||
def _restore_display() -> None:
|
||||
"""Re-enable screensaver after the alarm ends."""
|
||||
xset = shutil.which("xset")
|
||||
if xset is None:
|
||||
_logger.warning("xset not on PATH; skipping display restore")
|
||||
return
|
||||
subprocess.run(
|
||||
[xset, "s", "on"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
timeout=5,
|
||||
)
|
||||
@dataclass
|
||||
class _AlarmProgress:
|
||||
"""Mutable dismiss-challenge progress state."""
|
||||
|
||||
current_challenge: _Challenge
|
||||
skip_earnable: bool = True
|
||||
rounds_completed: int = 0
|
||||
flash_remaining: int = 0
|
||||
flash_on: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class _AlarmHardware:
|
||||
"""Hardware state captured at alarm start, restored when it closes."""
|
||||
|
||||
fan_state: bool
|
||||
audio_restore: str | None
|
||||
|
||||
|
||||
class WakeAlarm:
|
||||
@ -123,92 +132,103 @@ class WakeAlarm:
|
||||
self.root.focus_force()
|
||||
self.root.update_idletasks()
|
||||
|
||||
self._current_challenge: _Challenge = _make_challenge()
|
||||
self._skip_earnable: bool = True
|
||||
self._rounds_completed: int = 0
|
||||
self._flash_remaining: int = 0
|
||||
self._build_ui()
|
||||
if self._current_challenge.kind == "flash":
|
||||
self._progress = _AlarmProgress(current_challenge=_make_challenge())
|
||||
self._view = self._build_ui()
|
||||
self._update_timer()
|
||||
if self._progress.current_challenge.kind == "flash":
|
||||
self._start_flash_countdown()
|
||||
self._schedule_code_refresh()
|
||||
self._schedule_skip_window_close()
|
||||
self._start_beep_thread()
|
||||
self._fan_state: bool = _max_fans()
|
||||
self._audio_restore: str | None = _activate_alarm_audio()
|
||||
self._flash_on: bool = False
|
||||
self._hardware = _AlarmHardware(
|
||||
fan_state=_max_fans(),
|
||||
audio_restore=_activate_alarm_audio(),
|
||||
)
|
||||
self._start_screen_flash()
|
||||
|
||||
def _build_ui(self) -> None:
|
||||
"""Build the dismiss challenge UI."""
|
||||
self._container = tk.Frame(self.root, bg="#1a1a1a")
|
||||
self._container.place(relx=0.5, rely=0.5, anchor="center")
|
||||
def _build_ui(self) -> _AlarmView:
|
||||
"""Build the dismiss-challenge UI and return its widgets as a view."""
|
||||
challenge = self._progress.current_challenge
|
||||
|
||||
self._title_label = tk.Label(
|
||||
self._container,
|
||||
container = tk.Frame(self.root, bg="#1a1a1a")
|
||||
container.place(relx=0.5, rely=0.5, anchor="center")
|
||||
|
||||
title_label = tk.Label(
|
||||
container,
|
||||
text="WAKE UP!",
|
||||
font=("Arial", 48, "bold"),
|
||||
fg="#ff4444",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._title_label.pack(pady=20)
|
||||
title_label.pack(pady=20)
|
||||
|
||||
self._round_label = tk.Label(
|
||||
self._container,
|
||||
round_label = tk.Label(
|
||||
container,
|
||||
text=f"Round 1 / {DISMISS_ROUNDS_REQUIRED}",
|
||||
font=("Arial", 24, "bold"),
|
||||
fg="#ffaa00",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._round_label.pack(pady=5)
|
||||
round_label.pack(pady=5)
|
||||
|
||||
self._info_label = tk.Label(
|
||||
self._container,
|
||||
text=self._current_challenge.hint,
|
||||
info_label = tk.Label(
|
||||
container,
|
||||
text=challenge.hint,
|
||||
font=("Arial", 18),
|
||||
fg="white",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._info_label.pack(pady=10)
|
||||
info_label.pack(pady=10)
|
||||
|
||||
# Math and sort use a smaller font because their display text is wider.
|
||||
code_font_size = 48 if self._current_challenge.kind in ("math", "sort") else 72
|
||||
self._code_label = tk.Label(
|
||||
self._container,
|
||||
text=self._current_challenge.display,
|
||||
code_font_size = 48 if challenge.kind in ("math", "sort") else 72
|
||||
code_label = tk.Label(
|
||||
container,
|
||||
text=challenge.display,
|
||||
font=("Courier", code_font_size, "bold"),
|
||||
fg="#00ff00",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._code_label.pack(pady=30)
|
||||
code_label.pack(pady=30)
|
||||
|
||||
self._entry = tk.Entry(
|
||||
self._container,
|
||||
entry = tk.Entry(
|
||||
container,
|
||||
font=("Courier", 36),
|
||||
justify="center",
|
||||
width=12,
|
||||
)
|
||||
self._entry.pack(pady=10)
|
||||
self._entry.focus_set()
|
||||
self._entry.bind("<Return>", self._on_submit)
|
||||
entry.pack(pady=10)
|
||||
entry.focus_set()
|
||||
entry.bind("<Return>", self._on_submit)
|
||||
|
||||
self._status_label = tk.Label(
|
||||
self._container,
|
||||
status_label = tk.Label(
|
||||
container,
|
||||
text="",
|
||||
font=("Arial", 18),
|
||||
fg="#ff4444",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._status_label.pack(pady=10)
|
||||
status_label.pack(pady=10)
|
||||
|
||||
self._timer_label = tk.Label(
|
||||
self._container,
|
||||
timer_label = tk.Label(
|
||||
container,
|
||||
text="",
|
||||
font=("Arial", 14),
|
||||
fg="#aaaaaa",
|
||||
bg="#1a1a1a",
|
||||
)
|
||||
self._timer_label.pack(pady=5)
|
||||
self._update_timer()
|
||||
timer_label.pack(pady=5)
|
||||
|
||||
return _AlarmView(
|
||||
container=container,
|
||||
title_label=title_label,
|
||||
round_label=round_label,
|
||||
info_label=info_label,
|
||||
code_label=code_label,
|
||||
entry=entry,
|
||||
status_label=status_label,
|
||||
timer_label=timer_label,
|
||||
)
|
||||
|
||||
def _on_submit(self, _event: object = None) -> None:
|
||||
"""Handle challenge submission.
|
||||
@ -218,57 +238,57 @@ class WakeAlarm:
|
||||
correct round generates a new random challenge so the user must stay
|
||||
awake and re-engage each time.
|
||||
"""
|
||||
entered = self._entry.get().strip().upper()
|
||||
if entered != self._current_challenge.answer:
|
||||
self._status_label.configure(text="Wrong! Try again.")
|
||||
self._entry.delete(0, tk.END)
|
||||
if self._current_challenge.kind == "flash":
|
||||
self._code_label.configure(
|
||||
text=self._current_challenge.display,
|
||||
entered = self._view.entry.get().strip().upper()
|
||||
if entered != self._progress.current_challenge.answer:
|
||||
self._view.status_label.configure(text="Wrong! Try again.")
|
||||
self._view.entry.delete(0, tk.END)
|
||||
if self._progress.current_challenge.kind == "flash":
|
||||
self._view.code_label.configure(
|
||||
text=self._progress.current_challenge.display,
|
||||
fg="#00ff00",
|
||||
)
|
||||
self._start_flash_countdown()
|
||||
return
|
||||
self._rounds_completed += 1
|
||||
if self._rounds_completed >= DISMISS_ROUNDS_REQUIRED:
|
||||
self._dismiss_alarm(earned_skip=self._skip_earnable)
|
||||
self._progress.rounds_completed += 1
|
||||
if self._progress.rounds_completed >= DISMISS_ROUNDS_REQUIRED:
|
||||
self._dismiss_alarm(earned_skip=self._progress.skip_earnable)
|
||||
return
|
||||
self._current_challenge = _make_challenge()
|
||||
self._code_label.configure(
|
||||
text=self._current_challenge.display,
|
||||
self._progress.current_challenge = _make_challenge()
|
||||
self._view.code_label.configure(
|
||||
text=self._progress.current_challenge.display,
|
||||
fg="#00ff00",
|
||||
)
|
||||
self._info_label.configure(text=self._current_challenge.hint)
|
||||
self._entry.delete(0, tk.END)
|
||||
next_round = self._rounds_completed + 1
|
||||
self._round_label.configure(
|
||||
self._view.info_label.configure(text=self._progress.current_challenge.hint)
|
||||
self._view.entry.delete(0, tk.END)
|
||||
next_round = self._progress.rounds_completed + 1
|
||||
self._view.round_label.configure(
|
||||
text=f"Round {next_round} / {DISMISS_ROUNDS_REQUIRED}",
|
||||
)
|
||||
self._status_label.configure(
|
||||
text=f"Round {self._rounds_completed} done — keep going!",
|
||||
self._view.status_label.configure(
|
||||
text=f"Round {self._progress.rounds_completed} done — keep going!",
|
||||
)
|
||||
if self._current_challenge.kind == "flash":
|
||||
if self._progress.current_challenge.kind == "flash":
|
||||
self._start_flash_countdown()
|
||||
|
||||
def _start_flash_countdown(self) -> None:
|
||||
"""Begin the flash countdown: show code then hide it."""
|
||||
self._flash_remaining = DISMISS_FLASH_SECONDS
|
||||
self._progress.flash_remaining = DISMISS_FLASH_SECONDS
|
||||
self._flash_tick()
|
||||
|
||||
def _flash_tick(self) -> None:
|
||||
"""Decrement flash countdown; replace the displayed code with placeholders."""
|
||||
if not self._active:
|
||||
return
|
||||
if self._flash_remaining > 0:
|
||||
self._status_label.configure(
|
||||
text=f"Memorise! Hiding in {self._flash_remaining}s…",
|
||||
if self._progress.flash_remaining > 0:
|
||||
self._view.status_label.configure(
|
||||
text=f"Memorise! Hiding in {self._progress.flash_remaining}s…",
|
||||
)
|
||||
self._flash_remaining -= 1
|
||||
self._progress.flash_remaining -= 1
|
||||
self.root.after(1000, self._flash_tick)
|
||||
else:
|
||||
hidden = "?" * len(self._current_challenge.display)
|
||||
self._code_label.configure(text=hidden, fg="#555555")
|
||||
self._status_label.configure(text="Now type the code from memory!")
|
||||
hidden = "?" * len(self._progress.current_challenge.display)
|
||||
self._view.code_label.configure(text=hidden, fg="#555555")
|
||||
self._view.status_label.configure(text="Now type the code from memory!")
|
||||
|
||||
def _dismiss_alarm(self, *, earned_skip: bool) -> None:
|
||||
"""Dismiss the alarm and save state."""
|
||||
@ -278,7 +298,7 @@ class WakeAlarm:
|
||||
now_iso = datetime.now(tz=timezone.utc).isoformat()
|
||||
save_wake_state(dismissed_at=now_iso, skip_workout=earned_skip)
|
||||
|
||||
for widget in self._container.winfo_children():
|
||||
for widget in self._view.container.winfo_children():
|
||||
widget.destroy()
|
||||
|
||||
msg = (
|
||||
@ -289,7 +309,7 @@ class WakeAlarm:
|
||||
color = "#00ff00" if earned_skip else "#ffaa00"
|
||||
|
||||
tk.Label(
|
||||
self._container,
|
||||
self._view.container,
|
||||
text=msg,
|
||||
font=("Arial", 36, "bold"),
|
||||
fg=color,
|
||||
@ -301,8 +321,8 @@ class WakeAlarm:
|
||||
def _close(self) -> None:
|
||||
"""Close the alarm window."""
|
||||
self._stop_beep.set()
|
||||
_restore_fans(active=self._fan_state)
|
||||
_restore_alarm_audio(self._audio_restore)
|
||||
_restore_fans(active=self._hardware.fan_state)
|
||||
_restore_alarm_audio(self._hardware.audio_restore)
|
||||
_restore_display()
|
||||
turn_off_plug()
|
||||
self.root.destroy()
|
||||
@ -315,14 +335,14 @@ class WakeAlarm:
|
||||
"""
|
||||
if not self._active:
|
||||
return
|
||||
self._current_challenge = _make_challenge()
|
||||
self._code_label.configure(
|
||||
text=self._current_challenge.display,
|
||||
self._progress.current_challenge = _make_challenge()
|
||||
self._view.code_label.configure(
|
||||
text=self._progress.current_challenge.display,
|
||||
fg="#00ff00",
|
||||
)
|
||||
self._info_label.configure(text=self._current_challenge.hint)
|
||||
self._entry.delete(0, tk.END)
|
||||
if self._current_challenge.kind == "flash":
|
||||
self._view.info_label.configure(text=self._progress.current_challenge.hint)
|
||||
self._view.entry.delete(0, tk.END)
|
||||
if self._progress.current_challenge.kind == "flash":
|
||||
self._start_flash_countdown()
|
||||
ms = DISMISS_CODE_REFRESH_SECONDS * 1000 if not self.demo_mode else 10_000
|
||||
self.root.after(ms, self._schedule_code_refresh)
|
||||
@ -341,11 +361,11 @@ class WakeAlarm:
|
||||
"""
|
||||
if not self._active:
|
||||
return
|
||||
self._skip_earnable = False
|
||||
self._info_label.configure(
|
||||
self._progress.skip_earnable = False
|
||||
self._view.info_label.configure(
|
||||
text="Skip window closed - type the code to stop the alarm",
|
||||
)
|
||||
self._status_label.configure(text="No workout skip today.")
|
||||
self._view.status_label.configure(text="No workout skip today.")
|
||||
_logger.info("Skip window expired - alarm continues until dismissed.")
|
||||
|
||||
def _update_timer(self) -> None:
|
||||
@ -355,14 +375,14 @@ class WakeAlarm:
|
||||
elapsed = time.monotonic() - self._alarm_start
|
||||
window = DISMISS_WINDOW_MINUTES * 60 if not self.demo_mode else 30
|
||||
remaining = max(0, window - elapsed)
|
||||
if self._skip_earnable and remaining > 0:
|
||||
if self._progress.skip_earnable and remaining > 0:
|
||||
minutes = int(remaining) // 60
|
||||
seconds = int(remaining) % 60
|
||||
self._timer_label.configure(
|
||||
self._view.timer_label.configure(
|
||||
text=f"Skip window: {minutes:02d}:{seconds:02d}",
|
||||
)
|
||||
else:
|
||||
self._timer_label.configure(
|
||||
self._view.timer_label.configure(
|
||||
text="No skip available - type the code to stop the alarm",
|
||||
)
|
||||
self.root.after(1000, self._update_timer)
|
||||
@ -383,8 +403,8 @@ class WakeAlarm:
|
||||
"""Alternate background colour every 750 ms (below seizure-risk 3 Hz)."""
|
||||
if not self._active:
|
||||
return
|
||||
self.root.configure(bg="#ff0000" if self._flash_on else "#1a1a1a")
|
||||
self._flash_on = not self._flash_on
|
||||
self.root.configure(bg="#ff0000" if self._progress.flash_on else "#1a1a1a")
|
||||
self._progress.flash_on = not self._progress.flash_on
|
||||
self.root.after(750, self._flash_step)
|
||||
|
||||
def _beep_loop(self) -> None:
|
||||
@ -419,6 +439,9 @@ def _should_run_alarm() -> bool:
|
||||
if was_alarm_dismissed_today():
|
||||
_logger.info("Alarm already dismissed today. Exiting.")
|
||||
return False
|
||||
if was_workout_logged_today():
|
||||
_logger.info("Workout already logged today. Skipping alarm.")
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@ -445,10 +468,7 @@ def _parse_args(argv: list[str]) -> argparse.Namespace:
|
||||
|
||||
def main() -> None:
|
||||
"""Entry point for the wake alarm daemon."""
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(name)s %(levelname)s %(message)s",
|
||||
)
|
||||
configure_logging()
|
||||
|
||||
args = _parse_args(sys.argv[1:])
|
||||
|
||||
@ -463,6 +483,10 @@ def main() -> None:
|
||||
)
|
||||
_warn_if_no_real_sink()
|
||||
_wake_display()
|
||||
# Wait for the G27Q to power on and enumerate its HDMI audio sink.
|
||||
# Without this delay the sink often isn't visible yet when _activate_alarm_audio
|
||||
# runs, making the alarm silent when the monitor was physically off at wake time.
|
||||
time.sleep(DISPLAY_WAKE_WAIT_SECONDS)
|
||||
_set_max_brightness()
|
||||
turn_on_plug()
|
||||
alarm = WakeAlarm(demo_mode=args.demo)
|
||||
|
||||
76
wake_alarm/_alarm_display.py
Normal file
76
wake_alarm/_alarm_display.py
Normal file
@ -0,0 +1,76 @@
|
||||
"""Display power and screensaver helpers for the wake alarm.
|
||||
|
||||
Wakes monitors that may be physically powered off (via DDC/CI) or in DPMS
|
||||
standby, and restores the screensaver once the alarm dismiss flow ends.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _ddcutil_power_on() -> None:
|
||||
"""Power on all connected monitors via DDC/CI VCP code D6.
|
||||
|
||||
This wakes monitors that were physically turned off with the power button
|
||||
and therefore ignore DPMS signals. Falls back silently when ddcutil is
|
||||
absent or returns an error (e.g. no i2c access yet).
|
||||
"""
|
||||
ddcutil = shutil.which("ddcutil")
|
||||
if ddcutil is None:
|
||||
_logger.warning("ddcutil not on PATH; skipping DDC/CI monitor power-on")
|
||||
return
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[ddcutil, "setvcp", "D6", "01"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
timeout=10,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
_logger.warning("ddcutil setvcp failed", exc_info=True)
|
||||
return
|
||||
if result.returncode != 0:
|
||||
_logger.warning(
|
||||
"ddcutil setvcp D6 01 exited %d: %s",
|
||||
result.returncode,
|
||||
result.stderr.decode(errors="replace").strip()[:200],
|
||||
)
|
||||
else:
|
||||
_logger.info("DDC/CI monitor power-on sent")
|
||||
|
||||
|
||||
def _wake_display() -> None:
|
||||
"""Force the display on and disable screensaver during alarm.
|
||||
|
||||
Sends both a DDC/CI hard power-on (for monitors powered off via the
|
||||
power button) and a DPMS force-on (for monitors in standby).
|
||||
"""
|
||||
_ddcutil_power_on()
|
||||
xset = shutil.which("xset")
|
||||
if xset is None:
|
||||
_logger.warning("xset not on PATH; skipping DPMS display wake")
|
||||
return
|
||||
for cmd in (
|
||||
[xset, "dpms", "force", "on"],
|
||||
[xset, "s", "off"],
|
||||
):
|
||||
subprocess.run(cmd, check=False, capture_output=True, timeout=5)
|
||||
|
||||
|
||||
def _restore_display() -> None:
|
||||
"""Re-enable screensaver after the alarm ends."""
|
||||
xset = shutil.which("xset")
|
||||
if xset is None:
|
||||
_logger.warning("xset not on PATH; skipping display restore")
|
||||
return
|
||||
subprocess.run(
|
||||
[xset, "s", "on"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
timeout=5,
|
||||
)
|
||||
@ -54,9 +54,22 @@ ALARM_AUDIO_CARD: str = "alsa_card.pci-0000_01_00.1"
|
||||
ALARM_AUDIO_PROFILE: str = "output:hdmi-stereo"
|
||||
ALARM_AUDIO_SINK: str = "alsa_output.pci-0000_01_00.1.hdmi-stereo"
|
||||
# Seconds to wait for the HDMI sink to appear after forcing the profile on.
|
||||
ALARM_AUDIO_SINK_WAIT_SECONDS: float = 6.0
|
||||
# The G27Q takes up to ~15 s to power on from a hard-off state and enumerate
|
||||
# its HDMI audio; 6 s was too short when the monitor was physically off.
|
||||
ALARM_AUDIO_SINK_WAIT_SECONDS: float = 20.0
|
||||
# Poll interval while waiting for the sink.
|
||||
ALARM_AUDIO_SINK_POLL_SECONDS: float = 0.5
|
||||
# Seconds to pause after waking the display (xset dpms force on) before
|
||||
# attempting audio setup. Gives the G27Q time to come out of power-off
|
||||
# and re-enumerate its HDMI audio sink under PipeWire.
|
||||
DISPLAY_WAKE_WAIT_SECONDS: float = 5.0
|
||||
|
||||
# Path to the workout log written by the companion screen_locker package.
|
||||
# Dict keyed by YYYY-MM-DD date strings; presence of today's key means the
|
||||
# workout was already completed and the alarm should not fire.
|
||||
WORKOUT_LOG_FILE: Path = (
|
||||
Path.home() / "screen-locker" / "screen_locker" / "workout_log.json"
|
||||
)
|
||||
|
||||
# TP-Link Tapo P110 smart-plug config file (JSON).
|
||||
# Create with mode 0600 and these keys: host, email, password.
|
||||
|
||||
@ -10,7 +10,7 @@ from python_pkg.shared.log_integrity import (
|
||||
compute_entry_hmac,
|
||||
verify_entry_hmac,
|
||||
)
|
||||
from python_pkg.wake_alarm._constants import WAKE_STATE_FILE
|
||||
from python_pkg.wake_alarm._constants import WAKE_STATE_FILE, WORKOUT_LOG_FILE
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
@ -103,3 +103,26 @@ def was_alarm_dismissed_today() -> bool:
|
||||
if state is None:
|
||||
return False
|
||||
return state.get("dismissed_at") is not None
|
||||
|
||||
|
||||
def was_workout_logged_today() -> bool:
|
||||
"""Check if the workout was already logged today via the screen locker.
|
||||
|
||||
Reads the companion screen_locker workout_log.json. The file is a
|
||||
dict keyed by YYYY-MM-DD date strings; presence of today's key means
|
||||
the workout was completed and the alarm is no longer needed.
|
||||
|
||||
Returns:
|
||||
True if today's workout entry exists, False on any error or absence.
|
||||
"""
|
||||
if not WORKOUT_LOG_FILE.exists():
|
||||
return False
|
||||
try:
|
||||
with WORKOUT_LOG_FILE.open() as f:
|
||||
log = json.load(f)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
_logger.warning("Cannot read workout log file %s", WORKOUT_LOG_FILE)
|
||||
return False
|
||||
if not isinstance(log, dict):
|
||||
return False
|
||||
return _today_str() in log
|
||||
|
||||
@ -89,7 +89,7 @@ else
|
||||
fi
|
||||
|
||||
# 7. Install python-kasa (AUR) for TP-Link Tapo P110 smart-plug control
|
||||
echo "[7/7] Installing python-kasa (AUR)..."
|
||||
echo "[7/8] Installing python-kasa (AUR)..."
|
||||
if python -c 'import kasa' 2>/dev/null; then
|
||||
echo " python-kasa already installed"
|
||||
elif command -v yay &>/dev/null; then
|
||||
@ -102,6 +102,28 @@ if [[ ! -f "$HOME/.config/wake_alarm/tapo.json" ]]; then
|
||||
echo " Create it (mode 0600) with keys: host, email, password."
|
||||
fi
|
||||
|
||||
# 8. Install ddcutil for DDC/CI monitor power control
|
||||
# ddcutil lets the alarm force the G27Q on via DDC/CI even when the monitor
|
||||
# was physically powered off (power button), bypassing DPMS limitations.
|
||||
echo "[8/8] Installing ddcutil (DDC/CI monitor power control)..."
|
||||
if command -v ddcutil &>/dev/null; then
|
||||
echo " ddcutil already installed"
|
||||
else
|
||||
sudo pacman -S --noconfirm ddcutil
|
||||
echo " ddcutil installed"
|
||||
fi
|
||||
# ddcutil needs access to /dev/i2c-* — add user to i2c group if it exists.
|
||||
if getent group i2c &>/dev/null; then
|
||||
if ! id -nG "$USER" | grep -qw i2c; then
|
||||
sudo usermod -aG i2c "$USER"
|
||||
echo " Added $USER to i2c group (re-login required for group to take effect)"
|
||||
else
|
||||
echo " $USER already in i2c group"
|
||||
fi
|
||||
else
|
||||
echo " i2c group not found — ddcutil will run via sudo"
|
||||
fi
|
||||
|
||||
echo "=== Installation complete ==="
|
||||
echo "The wake alarm will activate on boot for alarm days (Mon, Fri, Sat, Sun)."
|
||||
echo "After hibernate resume the sleep hook will restart the alarm service."
|
||||
|
||||
@ -15,9 +15,7 @@ if TYPE_CHECKING:
|
||||
|
||||
from python_pkg.wake_alarm._alarm import (
|
||||
_is_alarm_day,
|
||||
_restore_display,
|
||||
_should_run_alarm,
|
||||
_wake_display,
|
||||
)
|
||||
from python_pkg.wake_alarm._audio import (
|
||||
_beep_loud,
|
||||
@ -249,51 +247,30 @@ class TestShouldRunAlarm:
|
||||
"python_pkg.wake_alarm._alarm.was_alarm_dismissed_today",
|
||||
return_value=False,
|
||||
),
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm.was_workout_logged_today",
|
||||
return_value=False,
|
||||
),
|
||||
):
|
||||
assert _should_run_alarm() is True
|
||||
|
||||
|
||||
class TestDisplayHelpers:
|
||||
"""Tests for _wake_display and _restore_display when xset is absent."""
|
||||
|
||||
def test_wake_display_skips_when_xset_missing(self) -> None:
|
||||
"""_wake_display does nothing when xset is not on PATH."""
|
||||
def test_returns_false_when_workout_already_logged(self) -> None:
|
||||
"""Return False when workout was already logged today."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm.shutil.which",
|
||||
return_value=None,
|
||||
"python_pkg.wake_alarm._alarm._is_alarm_day",
|
||||
return_value=True,
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
|
||||
):
|
||||
_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",
|
||||
"python_pkg.wake_alarm._alarm.was_alarm_dismissed_today",
|
||||
return_value=False,
|
||||
),
|
||||
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 (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm.shutil.which",
|
||||
return_value=None,
|
||||
"python_pkg.wake_alarm._alarm.was_workout_logged_today",
|
||||
return_value=True,
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm.subprocess.run") as mock_run,
|
||||
):
|
||||
_restore_display()
|
||||
mock_run.assert_not_called()
|
||||
assert _should_run_alarm() is False
|
||||
|
||||
|
||||
class TestPlayOnExtraDevices:
|
||||
|
||||
129
wake_alarm/tests/test_alarm_display.py
Normal file
129
wake_alarm/tests/test_alarm_display.py
Normal file
@ -0,0 +1,129 @@
|
||||
"""Tests for _alarm_display.py — DDC/CI and DPMS display power helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from python_pkg.wake_alarm._alarm_display import (
|
||||
_ddcutil_power_on,
|
||||
_restore_display,
|
||||
_wake_display,
|
||||
)
|
||||
|
||||
|
||||
class TestDdcutilPowerOn:
|
||||
"""Tests for _ddcutil_power_on."""
|
||||
|
||||
def test_skips_when_ddcutil_missing(self) -> None:
|
||||
"""_ddcutil_power_on does nothing when ddcutil is not on PATH."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value=None,
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm_display.subprocess.run") as mock_run,
|
||||
):
|
||||
_ddcutil_power_on()
|
||||
mock_run.assert_not_called()
|
||||
|
||||
def test_runs_setvcp_when_ddcutil_present(self) -> None:
|
||||
"""_ddcutil_power_on sends setvcp D6 01 when ddcutil is found."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value="/usr/bin/ddcutil",
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm_display.subprocess.run") as mock_run,
|
||||
):
|
||||
_ddcutil_power_on()
|
||||
mock_run.assert_called_once()
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert cmd == ["/usr/bin/ddcutil", "setvcp", "D6", "01"]
|
||||
|
||||
def test_logs_success_when_returncode_zero(self) -> None:
|
||||
"""_ddcutil_power_on logs success when setvcp returns 0."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value="/usr/bin/ddcutil",
|
||||
),
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.subprocess.run",
|
||||
return_value=MagicMock(returncode=0),
|
||||
),
|
||||
):
|
||||
_ddcutil_power_on()
|
||||
|
||||
def test_handles_timeout(self) -> None:
|
||||
"""_ddcutil_power_on does not raise on TimeoutExpired."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value="/usr/bin/ddcutil",
|
||||
),
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.subprocess.run",
|
||||
side_effect=subprocess.TimeoutExpired(cmd="ddcutil", timeout=10),
|
||||
),
|
||||
):
|
||||
_ddcutil_power_on() # must not raise
|
||||
|
||||
def test_handles_oserror(self) -> None:
|
||||
"""_ddcutil_power_on does not raise on OSError."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value="/usr/bin/ddcutil",
|
||||
),
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.subprocess.run",
|
||||
side_effect=OSError("no device"),
|
||||
),
|
||||
):
|
||||
_ddcutil_power_on() # must not raise
|
||||
|
||||
|
||||
class TestDisplayHelpers:
|
||||
"""Tests for _wake_display and _restore_display when xset is absent."""
|
||||
|
||||
def test_wake_display_skips_when_xset_missing(self) -> None:
|
||||
"""_wake_display skips xset commands but still attempts ddcutil."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value=None,
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm_display.subprocess.run") as mock_run,
|
||||
):
|
||||
_wake_display()
|
||||
mock_run.assert_not_called()
|
||||
|
||||
def test_wake_display_runs_ddcutil_and_xset_commands(self) -> None:
|
||||
"""_wake_display runs ddcutil setvcp, xset dpms force on, xset s off."""
|
||||
with (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value="/usr/bin/xset",
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm_display.subprocess.run") as mock_run,
|
||||
):
|
||||
_wake_display()
|
||||
# 1 ddcutil setvcp call + 2 xset calls
|
||||
assert mock_run.call_count == 3
|
||||
call_args = [call[0][0] for call in mock_run.call_args_list]
|
||||
assert ["/usr/bin/xset", "setvcp", "D6", "01"] in call_args
|
||||
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 (
|
||||
patch(
|
||||
"python_pkg.wake_alarm._alarm_display.shutil.which",
|
||||
return_value=None,
|
||||
),
|
||||
patch("python_pkg.wake_alarm._alarm_display.subprocess.run") as mock_run,
|
||||
):
|
||||
_restore_display()
|
||||
mock_run.assert_not_called()
|
||||
@ -133,7 +133,7 @@ class TestWakeAlarmDismiss:
|
||||
"python_pkg.wake_alarm._alarm.save_wake_state",
|
||||
) as mock_save:
|
||||
for _ in range(DISMISS_ROUNDS_REQUIRED):
|
||||
mock_entry.get.return_value = alarm._current_challenge.answer
|
||||
mock_entry.get.return_value = alarm._progress.current_challenge.answer
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm.dismissed is True
|
||||
@ -148,12 +148,12 @@ class TestWakeAlarmDismiss:
|
||||
"""A single correct entry is not enough — DISMISS_ROUNDS_REQUIRED is 2+."""
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
mock_entry = mock_tk_module.Entry.return_value
|
||||
mock_entry.get.return_value = alarm._current_challenge.answer
|
||||
mock_entry.get.return_value = alarm._progress.current_challenge.answer
|
||||
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm.dismissed is False
|
||||
assert alarm._rounds_completed == 1
|
||||
assert alarm._progress.rounds_completed == 1
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_first_round_correct_non_flash_next_no_countdown(
|
||||
@ -165,14 +165,14 @@ class TestWakeAlarmDismiss:
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
mock_entry = mock_tk_module.Entry.return_value
|
||||
mock_entry.get.return_value = alarm._current_challenge.answer
|
||||
mock_entry.get.return_value = alarm._progress.current_challenge.answer
|
||||
next_math = _Challenge(kind="math", display="2 + 2 = ?", answer="4", hint="x")
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._alarm._make_challenge", return_value=next_math
|
||||
):
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm._current_challenge.kind == "math"
|
||||
assert alarm._progress.current_challenge.kind == "math"
|
||||
assert alarm.dismissed is False
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -185,7 +185,7 @@ class TestWakeAlarmDismiss:
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
# Use a pinned math challenge so the non-flash wrong-answer branch is covered.
|
||||
alarm._current_challenge = _Challenge(
|
||||
alarm._progress.current_challenge = _Challenge(
|
||||
kind="math", display="2 + 2 = ?", answer="4", hint="test"
|
||||
)
|
||||
mock_entry = mock_tk_module.Entry.return_value
|
||||
@ -210,12 +210,12 @@ class TestWakeAlarmDismiss:
|
||||
alarm._on_skip_window_expired()
|
||||
|
||||
# Alarm stays active and audible; only the skip reward is gone.
|
||||
assert alarm._skip_earnable is False
|
||||
assert alarm._progress.skip_earnable is False
|
||||
assert alarm._active is True
|
||||
assert alarm.dismissed is False
|
||||
assert not alarm._stop_beep.is_set()
|
||||
mock_save.assert_not_called()
|
||||
alarm._info_label.configure.assert_called()
|
||||
alarm._view.info_label.configure.assert_called()
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_skip_window_expired_noop_if_not_active(
|
||||
@ -230,7 +230,7 @@ class TestWakeAlarmDismiss:
|
||||
alarm._on_skip_window_expired()
|
||||
|
||||
# skip_earnable stays at its initial True (method returned early).
|
||||
assert alarm._skip_earnable is True
|
||||
assert alarm._progress.skip_earnable is True
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_dismiss_after_skip_window_earns_no_skip(
|
||||
@ -241,14 +241,14 @@ class TestWakeAlarmDismiss:
|
||||
from python_pkg.wake_alarm._constants import DISMISS_ROUNDS_REQUIRED
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._skip_earnable = False
|
||||
alarm._progress.skip_earnable = False
|
||||
mock_entry = mock_tk_module.Entry.return_value
|
||||
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._alarm.save_wake_state",
|
||||
) as mock_save:
|
||||
for _ in range(DISMISS_ROUNDS_REQUIRED):
|
||||
mock_entry.get.return_value = alarm._current_challenge.answer
|
||||
mock_entry.get.return_value = alarm._progress.current_challenge.answer
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm.dismissed is True
|
||||
@ -325,7 +325,7 @@ class TestCodeRefreshAndTimer:
|
||||
displays = set()
|
||||
for _ in range(50):
|
||||
alarm._schedule_code_refresh()
|
||||
displays.add(alarm._current_challenge.display)
|
||||
displays.add(alarm._progress.current_challenge.display)
|
||||
assert len(displays) > 1
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -336,9 +336,9 @@ class TestCodeRefreshAndTimer:
|
||||
"""Code refresh is a no-op when alarm is no longer active."""
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._active = False
|
||||
old_challenge = alarm._current_challenge
|
||||
old_challenge = alarm._progress.current_challenge
|
||||
alarm._schedule_code_refresh()
|
||||
assert alarm._current_challenge is old_challenge
|
||||
assert alarm._progress.current_challenge is old_challenge
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_update_timer_noop_when_not_active(
|
||||
@ -391,7 +391,7 @@ class TestClose:
|
||||
"""_close calls _restore_fans with the saved fan state."""
|
||||
del mock_tk_module
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._fan_state = True
|
||||
alarm._hardware.fan_state = True
|
||||
with patch("python_pkg.wake_alarm._alarm._restore_fans") as mock_restore:
|
||||
alarm._close()
|
||||
mock_restore.assert_called_once_with(active=True)
|
||||
@ -403,7 +403,7 @@ class TestClose:
|
||||
"""_close restores the default sink captured at activation."""
|
||||
del mock_tk_module
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._audio_restore = "jbl_sink"
|
||||
alarm._hardware.audio_restore = "jbl_sink"
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._alarm._restore_alarm_audio",
|
||||
) as mock_restore:
|
||||
@ -425,11 +425,11 @@ class TestScreenFlash:
|
||||
mock_root.configure.reset_mock()
|
||||
mock_root.after.reset_mock()
|
||||
|
||||
alarm._flash_on = False
|
||||
alarm._progress.flash_on = False
|
||||
alarm._flash_step()
|
||||
|
||||
mock_root.configure.assert_called_once_with(bg="#1a1a1a")
|
||||
assert alarm._flash_on is True
|
||||
assert alarm._progress.flash_on is True
|
||||
mock_root.after.assert_called_with(750, alarm._flash_step)
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -443,11 +443,11 @@ class TestScreenFlash:
|
||||
mock_root.configure.reset_mock()
|
||||
mock_root.after.reset_mock()
|
||||
|
||||
alarm._flash_on = True
|
||||
alarm._progress.flash_on = True
|
||||
alarm._flash_step()
|
||||
|
||||
mock_root.configure.assert_called_once_with(bg="#ff0000")
|
||||
assert alarm._flash_on is False
|
||||
assert alarm._progress.flash_on is False
|
||||
mock_root.after.assert_called_with(750, alarm._flash_step)
|
||||
alarm._stop_beep.set()
|
||||
|
||||
|
||||
@ -157,7 +157,7 @@ class TestUpdateTimerActive:
|
||||
del mock_tk_module
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._update_timer()
|
||||
text = alarm._timer_label.configure.call_args[1]["text"]
|
||||
text = alarm._view.timer_label.configure.call_args[1]["text"]
|
||||
assert text.startswith("Skip window:")
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -173,7 +173,7 @@ class TestUpdateTimerActive:
|
||||
alarm._alarm_start = time_mod.monotonic() - 60 * 60
|
||||
alarm.root.after.reset_mock()
|
||||
alarm._update_timer()
|
||||
text = alarm._timer_label.configure.call_args[1]["text"]
|
||||
text = alarm._view.timer_label.configure.call_args[1]["text"]
|
||||
assert "type the code" in text
|
||||
# The alarm keeps nagging: it always reschedules while active.
|
||||
alarm.root.after.assert_called_once()
|
||||
@ -187,9 +187,9 @@ class TestUpdateTimerActive:
|
||||
del mock_tk_module
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._active = False
|
||||
alarm._timer_label.configure.reset_mock()
|
||||
alarm._view.timer_label.configure.reset_mock()
|
||||
alarm._update_timer()
|
||||
alarm._timer_label.configure.assert_not_called()
|
||||
alarm._view.timer_label.configure.assert_not_called()
|
||||
alarm._stop_beep.set()
|
||||
|
||||
|
||||
@ -204,27 +204,28 @@ class TestFlashChallenge:
|
||||
from python_pkg.wake_alarm._alarm import _Challenge
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._current_challenge = _Challenge(
|
||||
alarm._progress.current_challenge = _Challenge(
|
||||
kind="flash",
|
||||
display="ABCDEFGH",
|
||||
answer="ABCDEFGH",
|
||||
hint="Memorise",
|
||||
)
|
||||
alarm._flash_remaining = 2
|
||||
alarm._status_label.configure.reset_mock()
|
||||
alarm._progress.flash_remaining = 2
|
||||
alarm._view.status_label.configure.reset_mock()
|
||||
|
||||
alarm._flash_tick()
|
||||
assert alarm._flash_remaining == 1
|
||||
alarm._status_label.configure.assert_called()
|
||||
assert alarm._progress.flash_remaining == 1
|
||||
alarm._view.status_label.configure.assert_called()
|
||||
|
||||
alarm._flash_tick()
|
||||
assert alarm._flash_remaining == 0
|
||||
assert alarm._progress.flash_remaining == 0
|
||||
|
||||
# Final tick hides the code.
|
||||
alarm._flash_tick()
|
||||
# _code_label and _status_label share the same mock; inspect all calls.
|
||||
all_texts = [
|
||||
c.kwargs.get("text", "") for c in alarm._code_label.configure.call_args_list
|
||||
c.kwargs.get("text", "")
|
||||
for c in alarm._view.code_label.configure.call_args_list
|
||||
]
|
||||
assert any("?" in t for t in all_texts)
|
||||
alarm._stop_beep.set()
|
||||
@ -236,12 +237,12 @@ class TestFlashChallenge:
|
||||
"""_flash_tick returns immediately when the alarm is no longer active."""
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._active = False
|
||||
alarm._flash_remaining = 3
|
||||
alarm._status_label.configure.reset_mock()
|
||||
alarm._progress.flash_remaining = 3
|
||||
alarm._view.status_label.configure.reset_mock()
|
||||
|
||||
alarm._flash_tick()
|
||||
|
||||
alarm._status_label.configure.assert_not_called()
|
||||
alarm._view.status_label.configure.assert_not_called()
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_wrong_flash_answer_reshows_code(
|
||||
@ -252,7 +253,7 @@ class TestFlashChallenge:
|
||||
from python_pkg.wake_alarm._alarm import _Challenge
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._current_challenge = _Challenge(
|
||||
alarm._progress.current_challenge = _Challenge(
|
||||
kind="flash",
|
||||
display="TESTCODE",
|
||||
answer="TESTCODE",
|
||||
@ -260,13 +261,13 @@ class TestFlashChallenge:
|
||||
)
|
||||
mock_entry = mock_tk_module.Entry.return_value
|
||||
mock_entry.get.return_value = "WRONGCODE"
|
||||
alarm._code_label.configure.reset_mock()
|
||||
alarm._view.code_label.configure.reset_mock()
|
||||
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm.dismissed is False
|
||||
# Code label should be reconfigured (code shown again + countdown restarted).
|
||||
alarm._code_label.configure.assert_called()
|
||||
alarm._view.code_label.configure.assert_called()
|
||||
alarm._stop_beep.set()
|
||||
|
||||
def test_next_round_flash_starts_countdown(
|
||||
@ -277,7 +278,7 @@ class TestFlashChallenge:
|
||||
from python_pkg.wake_alarm._alarm import _Challenge
|
||||
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
alarm._current_challenge = _Challenge(
|
||||
alarm._progress.current_challenge = _Challenge(
|
||||
kind="math", display="2 + 2 = ?", answer="4", hint="test"
|
||||
)
|
||||
next_flash = _Challenge(
|
||||
@ -291,7 +292,7 @@ class TestFlashChallenge:
|
||||
):
|
||||
alarm._on_submit()
|
||||
|
||||
assert alarm._current_challenge.kind == "flash"
|
||||
assert alarm._progress.current_challenge.kind == "flash"
|
||||
assert alarm.dismissed is False
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -307,7 +308,7 @@ class TestDismissWithoutSkip:
|
||||
del mock_tk_module
|
||||
alarm = WakeAlarm(demo_mode=True)
|
||||
mock_widget = MagicMock()
|
||||
alarm._container.winfo_children.return_value = [mock_widget]
|
||||
alarm._view.container.winfo_children.return_value = [mock_widget]
|
||||
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._alarm.save_wake_state",
|
||||
@ -334,7 +335,7 @@ class TestSkipWindowExpiredMessage:
|
||||
|
||||
alarm._on_skip_window_expired()
|
||||
|
||||
alarm._status_label.configure.assert_called_with(
|
||||
alarm._view.status_label.configure.assert_called_with(
|
||||
text="No workout skip today.",
|
||||
)
|
||||
alarm._stop_beep.set()
|
||||
|
||||
@ -14,6 +14,7 @@ from python_pkg.wake_alarm._state import (
|
||||
load_wake_state,
|
||||
save_wake_state,
|
||||
was_alarm_dismissed_today,
|
||||
was_workout_logged_today,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -259,3 +260,55 @@ class TestWasAlarmDismissedToday:
|
||||
return_value=True,
|
||||
):
|
||||
assert was_alarm_dismissed_today() is False
|
||||
|
||||
|
||||
class TestWasWorkoutLoggedToday:
|
||||
"""Tests for was_workout_logged_today."""
|
||||
|
||||
def test_returns_false_when_file_missing(self, tmp_path: Path) -> None:
|
||||
"""Return False when the workout log file does not exist."""
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._state.WORKOUT_LOG_FILE",
|
||||
tmp_path / "workout_log.json",
|
||||
):
|
||||
assert was_workout_logged_today() is False
|
||||
|
||||
def test_returns_false_when_file_is_invalid_json(self, tmp_path: Path) -> None:
|
||||
"""Return False when the workout log contains invalid JSON."""
|
||||
log_file = tmp_path / "workout_log.json"
|
||||
log_file.write_text("not json {{{")
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._state.WORKOUT_LOG_FILE",
|
||||
log_file,
|
||||
):
|
||||
assert was_workout_logged_today() is False
|
||||
|
||||
def test_returns_false_when_file_is_not_a_dict(self, tmp_path: Path) -> None:
|
||||
"""Return False when the workout log is not a JSON object."""
|
||||
log_file = tmp_path / "workout_log.json"
|
||||
log_file.write_text(json.dumps([1, 2, 3]))
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._state.WORKOUT_LOG_FILE",
|
||||
log_file,
|
||||
):
|
||||
assert was_workout_logged_today() is False
|
||||
|
||||
def test_returns_false_when_today_absent(self, tmp_path: Path) -> None:
|
||||
"""Return False when the workout log has no entry for today."""
|
||||
log_file = tmp_path / "workout_log.json"
|
||||
log_file.write_text(json.dumps({"1999-01-01": {"type": "old"}}))
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._state.WORKOUT_LOG_FILE",
|
||||
log_file,
|
||||
):
|
||||
assert was_workout_logged_today() is False
|
||||
|
||||
def test_returns_true_when_today_present(self, tmp_path: Path) -> None:
|
||||
"""Return True when today's date key exists in the workout log."""
|
||||
log_file = tmp_path / "workout_log.json"
|
||||
log_file.write_text(json.dumps({_today_str(): {"type": "phone_verified"}}))
|
||||
with patch(
|
||||
"python_pkg.wake_alarm._state.WORKOUT_LOG_FILE",
|
||||
log_file,
|
||||
):
|
||||
assert was_workout_logged_today() is True
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{
|
||||
"date": "2026-05-25",
|
||||
"dismissed_at": "2026-05-25T10:33:09.098156+00:00",
|
||||
"date": "2026-06-14",
|
||||
"dismissed_at": "2026-06-14T05:01:28.589654+00:00",
|
||||
"skip_workout": true,
|
||||
"hmac": "49ae99880405c6e3f0b4948b07d398980e223a91d33dc7d0c7f0f9254463fa92"
|
||||
"hmac": "b472bf9b0874ff3f6f460cace7965d53cdfce823ee6f2d1f91914e43f003e92b"
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user