kuhyx/wake-alarm
1
0
Fork 0
mirror of https://github.com/kuhyx/wake-alarm.git synced 2026-11-10 18:07:05 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Fire the alarm at the phone's time via a per-minute timer
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Details
Tests / test (3.10) (push) Waiting to run
Details
Tests / test (3.11) (push) Waiting to run
Details
Tests / test (3.12) (push) Waiting to run
Details

Browse Source 
The PC is no longer shut down at night, which broke the scheduling model
outright. The wake time used to be decided once, at hibernate, and baked
into the RTC by shutdown-wrapper.sh. With no shutdown there is no resume,
so sleep-hook.sh never ran and nothing fired the alarm at all.

Worse, wake-alarm.service was WantedBy=graphical-session.target and
enabled by install.sh, while _should_run_alarm() gates on day / dismissed
/ workout-logged but has no *time* gate. On an always-on machine that
target is reached on every login, so the alarm fired at login time on any
alarm day — which is every day.

Replace both with a systemd user timer ticking once a minute:

- _sync.py reads the alarm the phone pushed to kuhyx/syncs, mirroring
  shutdown-wrapper.sh's jq parse so the two readers cannot drift. Caches
  each read, then degrades fresh cache -> fetch -> stale cache ->
  FALLBACK_ALARM_HOUR, so a network blip can never cost a wake-up.
  stdlib-only: the units run /usr/bin/python, so a dependency here is a
  production install-path hazard.
- _trigger.py fires morning-routine.service (never wake-alarm.service —
  that skips the workout lock) when local time is inside
  [alarm, alarm + TRIGGER_CATCHUP_MINUTES). The catch-up window covers a
  suspended or just-booted machine; starting an already-active unit is a
  no-op, so an alarm on screen needs no extra dedupe state.
- wake-alarm.service loses its [Install]; install.sh disables it.

A timer rather than a resident scheduler: each tick is a fresh process,
so there is no daemon to die silently and no wall-clock jump to handle
across suspend/resume. Ticks are per-minute for exact-minute firing, but
only hit the network when the cache is older than 5 minutes — ~288
GitHub reads/day, and phone changes land within ~5 minutes.

Degraded reads log at INFO, not WARNING: at one tick a minute a
persistent condition would be 1440 identical journal lines a day. The one
WARNING kept is _trigger firing on a non-phone alarm time.

Token resolution gains a fallback to the sibling apps' repo-scoped
sync_token, in both readers, so one PAT covers every app on kuhyx/syncs.

Note this surfaced that wake-alarm-sync/ does not exist in kuhyx/syncs at
all: the phone has never completed a push since the crdt_sync migration,
so the hibernate path had been silently using the +8h fallback. Fixing
that is a phone-side action (reconnect GitHub with repo scope, then one
"Set Alarm & Sync"). Alarms edited directly in the Clock app still do not
reach the sync — that needs an ACTION_NEXT_ALARM_CLOCK_CHANGED receiver.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-07-19 20:21:17 +02:00
parent 64c25e67e9
commit 87d2629e88
14 changed files with 1541 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
CLAUDE.md
View File

@ -10,27 +10,82 @@ fullscreen Tk window that must be dismissed (with a typed challenge — see
through whatever sink comes up after the monitor wakes, and optionally
toggles a TP-Link Tapo P110 smart plug via `python-kasa`.
## Scheduling — hibernate-based, not a systemd timer
## Scheduling — a per-minute timer, with hibernate as the fallback
There is **no systemd timer** for this package. The wake mechanism is:
The PC is **always on**. The alarm time comes from the phone, which can change
it at any hour, so nothing about the schedule may be baked in ahead of time.
1. `shutdown-wrapper.sh`, installed to `/usr/local/bin/shutdown` (shadowing
`/usr/bin/shutdown` via PATH order), intercepts shutdown/poweroff calls on
alarm nights and calls `rtcwake -m disk` instead — hibernating with the
RTC alarm set to wake the machine at `WAKE_AFTER_HOURS` (`_constants.py`)
from now.
2. `sleep-hook.sh`, installed to `/usr/lib/systemd/system-sleep/`, fires on
resume (`$1 == post`) and starts `morning-routine.service` for every
logged-in session. That orchestrator (lives in testsAndMisc, not here —
see below) runs the alarm first, then the workout screen lock, so the two
never fight for the fullscreen.
3. `wake-alarm.service` itself is `Type=simple`, started either directly or
by the orchestrator, and exits once the alarm is dismissed.
**Primary path — `wake-alarm-trigger.timer`:**
1. The timer ticks every minute (`OnCalendar=*:*:00`, `Persistent=true`) and
runs `python -m wake_alarm._trigger` — a short, stateless check-and-fire.
2. `_trigger` resolves the alarm via `_sync.get_alarm()`, builds today's fire
instant in **local** time, and fires iff `now` is in
`[fire_time, fire_time + TRIGGER_CATCHUP_MINUTES)`. The catch-up window is
what makes a suspended or just-booted machine still fire.
3. Firing = `systemctl --user start morning-routine.service`. **Never**
`wake-alarm.service` directly — see the orchestrator section below.
Starting an already-active unit is a no-op, so an alarm on screen needs no
extra dedupe state; the `was_alarm_dismissed_today()` /
`was_workout_logged_today()` gates cover the rest.
A timer, not a resident daemon: each tick is a fresh process, so there is no
long-lived scheduler to die silently and no wall-clock jump to handle across
suspend/resume.
**Fallback path — hibernate (only when the PC really is powered down):**
`shutdown-wrapper.sh` (installed to `/usr/local/bin/shutdown`, shadowing
`/usr/bin/shutdown` via PATH order) still intercepts shutdown/poweroff, reads
the same synced alarm, and sets `rtcwake`. `sleep-hook.sh` (in
`/usr/lib/systemd/system-sleep/`) fires on resume (`$1 == post`) and starts
`morning-routine.service`. This is now a backstop, not the main mechanism.
If you change `WAKE_AFTER_HOURS` in `_constants.py`, you must also update the
duplicate constant in `shutdown-wrapper.sh` (`WAKE_AFTER_HOURS=8`) — they are
not wired together, by design (the shell wrapper has no Python runtime
available at the point it intercepts `shutdown`).
available at the point it intercepts `shutdown`). The same duplication applies
to the token search order: `SYNC_CONFIG_FILE` / `SYNC_TOKEN_FALLBACK_FILES` in
`_constants.py` are mirrored by `read_sync_token()` in the wrapper.
## `wake-alarm.service` must never auto-start
`wake-alarm.service` deliberately has **no `[Install]` section** and nothing
starts it automatically. `_should_run_alarm()` gates on day / dismissed /
workout-logged but has **no time gate**, so a
`WantedBy=graphical-session.target` there fires the alarm at *login* time on
any alarm day — which is every day now that the machine no longer reboots into
the alarm. `install.sh` explicitly runs `systemctl --user disable
wake-alarm.service` to clear the stale symlink from older installs.
The unit survives only as a manual test handle. Do not "fix" its missing
auto-start.
## Reading the phone's alarm
`_sync.py` is the Python reader; `shutdown-wrapper.sh` is a second, jq-based
reader of the **same file**. Keep their parse and token order in step.
- Source of truth: `wake-alarm-sync/devices/phone/alarm.json` in the private
`kuhyx/syncs` repo, written by `phone_app/` as a `crdt_sync` log. The value
lives at `.alarm.fields.alarm_hour[0]` / `.alarm_minute[0]`.
- Token: `~/.config/wake_alarm/sync.json` key `token`, else a sibling app's
raw `~/.config/{diet_guard,screen_locker}/sync_token`. Needs `repo` scope —
the repo is private, and a token without access returns **404, not 403**.
- `_sync` is **stdlib-only** (`urllib.request`) on purpose: the service runs
`/usr/bin/python`, so every dependency is a production install-path hazard.
- Resolution chain in `get_alarm()`: fresh cache (< 5 min) → live fetch →
stale cache → `FALLBACK_ALARM_HOUR`. The cache
(`~/.config/wake_alarm/alarm_cache.json`) is what bounds GitHub reads to
~288/day despite the per-minute tick, and what keeps a network blip from
costing a wake-up.
- **Degraded reads log at INFO, not WARNING.** `get_alarm()` runs every minute,
so a persistent condition logged loudly is 1440 identical journal lines a
day. `_trigger` emits the single WARNING that matters: it is firing *now* on
a non-phone alarm time.
Check what the PC currently resolves: `python -m wake_alarm._sync --show`.
Check what it would do right now: `python -m wake_alarm._trigger --dry-run`.
## Cross-repo coupling — not a bug
@ -110,12 +165,26 @@ considering a dependency change done.
- Run tests: `python -m pytest wake_alarm/tests/ --cov=wake_alarm --cov-branch --cov-fail-under=100`
- Lint: `pre-commit run --all-files`
- Test the lock manually (safe, closeable): `python -m wake_alarm._alarm --demo`
- Show the resolved alarm: `python -m wake_alarm._sync --show` (add
`--no-cache` to force a live GitHub read)
- Show what the next tick would do: `python -m wake_alarm._trigger --dry-run`
- Watch it live: `journalctl --user -fu wake-alarm-trigger.service`
- Install for production: `bash install.sh`
Note the dev venv lacks `gatelock`, so anything importing `_state` (including
`_trigger`) must be run with `/usr/bin/python3`, not `.venv/bin/python`.
## Do NOT
- Don't add a dependency without doing the production install-path check
above.
above. `_sync.py` is stdlib-only for exactly this reason — don't "improve"
it to `requests`.
- Don't forget the orchestrator when changing this package's module path or
invocation — see "The morning_routine orchestrator lives elsewhere" above.
- Don't give `wake-alarm.service` an `[Install]` section or enable it. It
fires at login time, not alarm time — see the section above.
- Don't make the trigger start `wake-alarm.service`; it must start
`morning-routine.service` so the workout lock still runs.
- Don't raise the log level of the routine degraded-read paths in `_sync` —
the per-minute tick turns any WARNING there into journal spam.
- Don't commit `wake_state.json`.

48
MIGRATION.md
View File

@ -37,3 +37,51 @@ sync.
`LogStore` is always the source of truth.
- Any parse miss still falls back to `WAKE_AFTER_HOURS`, so a bad read can
never brick the morning wake.
# Migration: hibernate-scheduled → per-minute trigger timer
The PC is no longer shut down at night, which broke the whole scheduling
model: the wake time used to be decided once, at hibernate, and baked into the
RTC. With no shutdown there is no resume, so nothing fired the alarm at all —
and `wake-alarm.service` was `WantedBy=graphical-session.target`, so on an
always-on machine it fired at *login* time instead (`_should_run_alarm()` has
no time gate).
## What changed
- **New `wake-alarm-trigger.timer`** ticks every minute and runs
`wake_alarm._trigger`, which fires `morning-routine.service` when the
current local time is inside
`[phone alarm, phone alarm + TRIGGER_CATCHUP_MINUTES)`. This is now the
primary path; the phone can move the alarm at any hour and the PC follows.
- **New `wake_alarm/_sync.py`** — a Python reader for the same alarm file
`shutdown-wrapper.sh` parses with `jq`, plus an on-disk cache
(`~/.config/wake_alarm/alarm_cache.json`).
- **`wake-alarm.service` lost its `[Install]` section** and is no longer
enabled; `install.sh` disables it to clear stale symlinks.
- **Token resolution gained a fallback** to the sibling apps'
`~/.config/{diet_guard,screen_locker}/sync_token`, in both readers.
- `shutdown-wrapper.sh` and `sleep-hook.sh` keep working unchanged, now as the
PC-is-powered-down fallback rather than the main mechanism.
## Surprises / non-obvious decisions
- **A timer, not a resident scheduler.** Each tick is a fresh process, so
there is no daemon to die silently and no wall-clock jump to reason about
across suspend/resume. `Persistent=true` plus the catch-up window is what
makes a suspended PC still fire on resume.
- **Per-minute ticks, but a 5-minute cache.** Ticking every 5 minutes would
also quantise *firing* to 5 minutes (07:03 → 07:05). Ticking every minute
and only hitting the network when the cache is stale gives exact-minute
firing at ~288 GitHub reads/day.
- **Degraded reads log at INFO.** At one tick a minute, a WARNING on a
persistent condition is 1440 journal lines a day. The single WARNING that
survives is `_trigger`'s "firing now on a degraded source".
- **The sync had been dead since the crdt_sync migration.** `wake-alarm-sync/`
did not exist in `kuhyx/syncs` at all — the phone had never successfully
pushed, and `~/.config/wake_alarm/sync.json` was never created, so the
hibernate path had been silently using the `+8h` fallback. The phone app
needs one successful "Set Alarm & Sync" to create that path.
- **Still open:** alarms edited directly in the Android Clock app do not reach
the sync. Closing that needs an `ACTION_NEXT_ALARM_CLOCK_CHANGED` receiver
in the phone app — planned, not yet built.

37
README.md
View File

@ -1,9 +1,14 @@
# wake_alarm
A hibernate-scheduled weekend wake alarm: the machine hibernates overnight
on alarm days and wakes itself via `rtcwake`, then shows a fullscreen,
challenge-dismissed alarm with fan ramp and optional TP-Link Tapo P110
smart-plug control.
A phone-driven wake alarm for an always-on PC. The alarm time is whatever the
companion phone app last synced (via the shared `crdt_sync` transport over a
private GitHub repo); a per-minute systemd user timer fires at exactly that
time, showing a fullscreen, challenge-dismissed alarm with fan ramp and
optional TP-Link Tapo P110 smart-plug control.
Change the alarm on the phone at any hour and the PC follows within ~5
minutes. If the PC does happen to be powered down, a `shutdown` wrapper
hibernates with `rtcwake` set to the same synced time, as a fallback.
## Install
@ -12,20 +17,28 @@ bash install.sh
```
This installs the package + dependencies into system Python's user
site-packages (the systemd service runs `/usr/bin/python` directly, not a
venv — see `CLAUDE.md`), installs the systemd user service, the
systemd-sleep resume hook, the `shutdown` wrapper that triggers hibernate on
alarm nights, the fan-ramp script, and (optionally) `python-kasa` for
smart-plug control.
site-packages (the systemd units run `/usr/bin/python` directly, not a venv —
see `CLAUDE.md`), installs and enables the per-minute trigger timer, the
systemd-sleep resume hook, the `shutdown` wrapper that hibernates with
`rtcwake` as a fallback, the fan-ramp script, and (optionally) `python-kasa`
for smart-plug control.
Reading the phone's alarm needs a GitHub token with `repo` scope, at
`~/.config/wake_alarm/sync.json` as `{"token": "..."}`. If that file is
absent, a sibling app's `~/.config/{diet_guard,screen_locker}/sync_token` is
used instead, so one token covers every app syncing through `kuhyx/syncs`.
## Usage
```bash
python -m wake_alarm._alarm --demo # test the alarm window (safe, closeable)
python -m wake_alarm._alarm --demo # test the alarm window (safe, closeable)
python -m wake_alarm._sync --show # what alarm time the PC resolves
python -m wake_alarm._trigger --dry-run # what the next tick would do
journalctl --user -fu wake-alarm-trigger.service # watch it live
```
The alarm fires automatically via the hibernate/wake cycle once installed;
no manual invocation is needed in normal operation.
The alarm fires automatically once installed; no manual invocation is needed
in normal operation.
## Development

35
install.sh
View File

@ -6,7 +6,8 @@
# What it does:
# 1. Installs wake_alarm + dependencies for /usr/bin/python
# 2. Installs system dependencies (alsa-utils, ddcutil)
# 3. Copies wake-alarm.service to ~/.config/systemd/user/ and enables it
# 3. Installs the systemd user units and enables the per-minute trigger timer
# (which fires the alarm at the time the phone synced)
# 4. Installs the systemd-sleep hook (restarts alarm after hibernate resume)
# 5. Adds a sudoers entry for passwordless rtcwake
# 6. Installs shutdown wrapper so "shutdown now" also hibernates on alarm nights
@ -21,6 +22,8 @@ SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
readonly SCRIPT_DIR
readonly REPO_DIR="$SCRIPT_DIR"
SERVICE_FILE="$SCRIPT_DIR/wake-alarm.service"
TRIGGER_SERVICE_SRC="$SCRIPT_DIR/wake-alarm-trigger.service"
TRIGGER_TIMER_SRC="$SCRIPT_DIR/wake-alarm-trigger.timer"
SLEEP_HOOK_SRC="$SCRIPT_DIR/sleep-hook.sh"
SHUTDOWN_WRAPPER_SRC="$SCRIPT_DIR/shutdown-wrapper.sh"
FANS_SCRIPT_SRC="$SCRIPT_DIR/wake-alarm-fans.sh"
@ -49,14 +52,26 @@ else
echo " alsa-utils already installed"
fi
# 3. Install systemd user service
echo "[3/9] Installing systemd user service..."
# 3. Install systemd user units (alarm + the trigger timer that fires it)
echo "[3/9] Installing systemd user units..."
mkdir -p "$SYSTEMD_USER_DIR"
cp "$SERVICE_FILE" "$SYSTEMD_USER_DIR/wake-alarm.service"
cp "$TRIGGER_SERVICE_SRC" "$SYSTEMD_USER_DIR/wake-alarm-trigger.service"
cp "$TRIGGER_TIMER_SRC" "$SYSTEMD_USER_DIR/wake-alarm-trigger.timer"
systemctl --user daemon-reload
echo " Installed to $SYSTEMD_USER_DIR/wake-alarm.service"
systemctl --user enable wake-alarm.service
echo " Service enabled (will start on next boot)"
echo " Installed wake-alarm.service, wake-alarm-trigger.{service,timer}"
# wake-alarm.service must NOT auto-start: the PC is always on, so
# graphical-session.target is reached on every login and the alarm has no time
# gate — it would fire at login time. The timer owns firing (via
# morning-routine.service). Both guards matter under `set -euo pipefail`:
# a unit with no [Install] can't be enabled, and disabling a never-enabled unit
# can also return non-zero.
systemctl --user disable wake-alarm.service 2>/dev/null || true
echo " wake-alarm.service left disabled (started only by the trigger)"
systemctl --user enable --now wake-alarm-trigger.timer
echo " Trigger timer enabled and started (ticks once a minute)"
# 4. Install systemd-sleep hook (restarts alarm after hibernate resume)
echo "[4/9] Installing systemd-sleep hook..."
@ -134,7 +149,11 @@ else
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."
echo "The trigger timer ticks once a minute and fires the alarm at the time"
echo "the phone last synced (needs ~/.config/wake_alarm/sync.json with a"
echo "repo-scoped GitHub token — see MIGRATION.md)."
echo "Check what it resolves right now: python -m wake_alarm._sync --show"
echo "If the PC is powered down instead, the shutdown wrapper hibernates with"
echo "rtcwake set to the same alarm, and the sleep hook runs it on resume."
echo "Fans will ramp to 100% while the alarm is active, then restore automatically."
echo "To test now: python -m wake_alarm._alarm --demo"

6
pyproject.toml
View File

@ -21,6 +21,12 @@ ignore = [
"COM812",
"ISC001",
"S603",
# urllib "audit URL open for permitted schemes". _sync.py builds its one
# URL from module constants (never from remote input) and asserts
# request.type == "https" before sending the token. Ignored repo-wide
# rather than per-line because inline `# noqa` is blocked by a
# pre-commit hook; same rationale as S603 above.
"S310",
]
fixable = ["ALL"]
unfixable = []

49
shutdown-wrapper.sh
View File

@ -7,11 +7,17 @@
# Install to /usr/local/bin/shutdown (takes priority over /usr/bin/shutdown
# because /usr/local/bin appears first in PATH).
#
# This is the FALLBACK path, for the rare night the PC is actually powered
# down. The primary path is wake-alarm-trigger.timer, which ticks once a
# minute on the always-on machine and fires at the phone's alarm time without
# any rtcwake involvement (see wake_alarm/_trigger.py).
#
# Wake time is taken from the alarm the phone app pushes to the private
# kuhyx/syncs repo via the shared crdt_sync transport (see phone_app/). Falls
# back to WAKE_AFTER_HOURS if sync is not configured or unreachable. Config:
# ~/.config/wake_alarm/sync.json with key "token" (GitHub PAT/token with
# "repo" scope, so the Contents API can read the private repo).
# back to WAKE_AFTER_HOURS if sync is not configured or unreachable. Token:
# ~/.config/wake_alarm/sync.json key "token", else a sibling app's
# ~/.config/{diet_guard,screen_locker}/sync_token. Needs "repo" scope, since
# the Contents API must read a private repo.
set -euo pipefail
@ -30,6 +36,37 @@ for arg in "$@"; do
esac
done
# Resolve the GitHub token, preferring wake_alarm's own JSON config and falling
# back to the sibling companion apps' raw single-line token files (one
# repo-scoped PAT covers every app syncing through kuhyx/syncs).
# Mirrors SYNC_CONFIG_FILE / SYNC_TOKEN_FALLBACK_FILES in
# wake_alarm/_constants.py — keep the two readers in step.
read_sync_token() {
local cfg="$HOME/.config/wake_alarm/sync.json"
local token fallback
if [[ -f "$cfg" ]]; then
token=$(jq -r '.token // empty' "$cfg" 2>/dev/null) || token=""
if [[ -n "$token" ]]; then
printf '%s' "$token"
return 0
fi
fi
for fallback in \
"$HOME/.config/diet_guard/sync_token" \
"$HOME/.config/screen_locker/sync_token"; do
[[ -f "$fallback" ]] || continue
token=$(tr -d '[:space:]' < "$fallback") || continue
if [[ -n "$token" ]]; then
printf '%s' "$token"
return 0
fi
done
return 1
}
# Read the alarm time the phone pushed to kuhyx/syncs via crdt_sync.
# Returns the Unix epoch (seconds) of the next occurrence of the stored H:M.
# Exits non-zero if sync is not configured, unreachable, or malformed.
@ -41,12 +78,8 @@ done
# [H, "<hlc>"], "alarm_minute": [M, "<hlc>"]}, ...}}. The raw media type makes
# the Contents API return that JSON directly, so no base64 decode is needed.
get_synced_alarm_epoch() {
local cfg="$HOME/.config/wake_alarm/sync.json"
[[ -f "$cfg" ]] || return 1
local token log_json hour minute target_epoch now_epoch
token=$(jq -r .token "$cfg" 2>/dev/null) || return 1
[[ -n "$token" && "$token" != "null" ]] || return 1
token=$(read_sync_token) || return 1
local url="https://api.github.com/repos/kuhyx/syncs/contents"
url="$url/wake-alarm-sync/devices/phone/alarm.json"

12
wake-alarm-trigger.service Normal file
View File

@ -0,0 +1,12 @@
[Unit]
Description=Wake Alarm trigger tick (fire at the phone's alarm time)
Documentation=https://github.com/kuhyx/wake-alarm
[Service]
Type=oneshot
# One short check-and-fire tick, run once a minute by wake-alarm-trigger.timer.
# It reads the alarm the phone pushed and, on the alarm minute, starts
# morning-routine.service. No DISPLAY needed: this process never opens a
# window — the unit it starts does.
# No PYTHONPATH needed: wake_alarm is pip-installed (see install.sh).
ExecStart=/usr/bin/python -m wake_alarm._trigger

17
wake-alarm-trigger.timer Normal file
View File

@ -0,0 +1,17 @@
[Unit]
Description=Wake Alarm trigger tick, once a minute
Documentation=https://github.com/kuhyx/wake-alarm
[Timer]
# Tick every minute on the minute so the alarm fires on its exact minute.
# The tick itself only hits the network when its cache is older than
# ALARM_CACHE_MAX_AGE_SECONDS (5 min), so this costs ~288 GitHub reads/day,
# not 1440 - see wake_alarm/_sync.py.
OnCalendar=*:*:00
AccuracySec=1s
# Run a missed tick immediately after resume/boot. Combined with the trigger's
# catch-up window this is what makes a suspended PC still fire on wake.
Persistent=true
[Install]
WantedBy=timers.target

15
wake-alarm.service
View File

@ -16,5 +16,16 @@ ExecStart=/usr/bin/python -m wake_alarm._alarm --production
Restart=on-failure
RestartSec=10
[Install]
WantedBy=graphical-session.target
# There is intentionally NO [Install] section, and nothing auto-starts this
# unit. The PC is always on now, so graphical-session.target is reached on
# every login — and _should_run_alarm() has no time gate, so a
# WantedBy=graphical-session.target here fires the alarm at *login time* on
# any alarm day (i.e. every day). That was the bug this design removed.
#
# What actually runs the alarm: wake-alarm-trigger.timer ticks once a minute
# and starts morning-routine.service at the phone's alarm time; that
# orchestrator runs `python -m wake_alarm._alarm` as a subprocess (the alarm
# first, the workout lock after). This unit survives only as a manual test
# handle: `systemctl --user start wake-alarm.service`.
#
# Do not "fix" the missing auto-start.

64
wake_alarm/_constants.py
View File

@ -8,13 +8,17 @@ from pathlib import Path
# Every day of the week.
ALARM_DAYS: frozenset[int] = frozenset({0, 1, 2, 3, 4, 5, 6})
# How many hours after shutdown the PC should wake
# How many hours after shutdown the PC should wake.
# Only used by the hibernate fallback path (shutdown-wrapper.sh) for the rare
# night the PC is actually powered down; the primary path is the per-minute
# trigger timer, which reads the phone's alarm directly (see _trigger.py).
WAKE_AFTER_HOURS: int = 8
# Minutes after alarm starts within which you must dismiss to earn skip
DISMISS_WINDOW_MINUTES: int = 30
# Hour at which the second (fallback) alarm fires if the first was missed
# Hour at which the alarm fires when the phone's alarm cannot be resolved at
# all — no network *and* no cached read ever written. Minute is implicitly 0.
FALLBACK_ALARM_HOUR: int = 13
# Alarm escalation phase boundaries (minutes from alarm start)
@ -58,6 +62,62 @@ WORKOUT_LOG_FILE: Path = (
Path.home() / "screen-locker" / "screen_locker" / "workout_log.json"
)
# --- Phone alarm sync (crdt_sync over GitHub) -----------------------------
# The phone app pushes its alarm to a private repo as a crdt_sync log; the PC
# reads it back. shutdown-wrapper.sh reads the SAME file with jq for the
# hibernate fallback — keep the two readers in step if any of this changes.
# GitHub token config, key "token" (PAT with "repo" scope: the repo is private).
SYNC_CONFIG_FILE: Path = Path.home() / ".config" / "wake_alarm" / "sync.json"
# Fallback token locations, tried in order when SYNC_CONFIG_FILE is absent.
# The sibling companion apps already store a repo-scoped PAT for this same
# kuhyx/syncs repo as a raw single-line file, so reusing one means a single
# token to rotate instead of three copies drifting out of sync.
# shutdown-wrapper.sh reads the same list — keep the two in step.
SYNC_TOKEN_FALLBACK_FILES: tuple[Path, ...] = (
Path.home() / ".config" / "diet_guard" / "sync_token",
Path.home() / ".config" / "screen_locker" / "sync_token",
)
# Where a successful read is cached, so a network blip can't cost a wake-up.
ALARM_CACHE_FILE: Path = Path.home() / ".config" / "wake_alarm" / "alarm_cache.json"
# The shared private repo every companion app syncs through, and the phone's
# path within it. The phone is the sole writer under a fixed device id, so one
# read of its alarm.json is enough — no device discovery or HLC ordering here.
SYNC_API_BASE: str = "https://api.github.com"
SYNC_REPO_OWNER: str = "kuhyx"
SYNC_REPO_NAME: str = "syncs"
SYNC_ALARM_PATH: str = "wake-alarm-sync/devices/phone/alarm.json"
# Valid ranges for a wall-clock alarm, used to reject a malformed sync payload
# or a corrupted cache before it can become a fire time.
MIN_HOUR: int = 0
MAX_HOUR: int = 23
MIN_MINUTE: int = 0
MAX_MINUTE: int = 59
# Timeout for the Contents API read. Short: a slow read must never delay the
# per-minute trigger tick, and a stale cache is a perfectly good answer.
SYNC_HTTP_TIMEOUT_SECONDS: float = 10.0
# How long a cached alarm is served without re-fetching. The trigger ticks once
# a minute (so firing lands on the exact minute), but only touches the network
# when the cache is older than this — bounding phone->PC latency at ~5 minutes
# while keeping GitHub reads to ~288/day.
ALARM_CACHE_MAX_AGE_SECONDS: float = 300.0
# How long after the alarm instant the trigger will still fire. Covers a
# suspended PC, a missed timer tick, and a machine booted mid-window. Also the
# reason a phone alarm moved *earlier* mid-window still fires promptly.
TRIGGER_CATCHUP_MINUTES: int = 60
# The unit the trigger starts. Deliberately the orchestrator, never
# wake-alarm.service directly: the orchestrator runs the alarm first and the
# workout lock after, so the two never fight for the fullscreen.
TRIGGER_TARGET_UNIT: str = "morning-routine.service"
# TP-Link Tapo P110 smart-plug config file (JSON).
# Create with mode 0600 and these keys: host, email, password.
# Example contents: a JSON object mapping host -> "192.168.x.x", email ->

368
wake_alarm/_sync.py Normal file
View File

@ -0,0 +1,368 @@
"""Read the alarm time the phone pushed, with an on-disk cache and fallbacks.
The phone app (``phone_app/``) is the sole writer: it stores the chosen alarm
in a ``crdt_sync`` ``LogStore`` and pushes that log to a fixed path in the
private ``kuhyx/syncs`` repo. This module is the PC-side reader used by the
per-minute trigger (see ``_trigger.py``).
``shutdown-wrapper.sh`` reads the **same file** with ``jq`` for the hibernate
fallback path. The parse here mirrors it exactly (``.alarm.fields.alarm_hour``
and ``.alarm.fields.alarm_minute``, first tuple element) so the two readers
cannot drift.
Deliberately stdlib-only (``urllib.request`` + ``json``): the alarm runs under
``/usr/bin/python``, not a venv, so every added dependency is a production
install-path hazard (see ``CLAUDE.md``). A read is not worth that risk.
"""
from __future__ import annotations
import argparse
from dataclasses import dataclass
import json
import logging
import sys
import time
from typing import Any
import urllib.error
import urllib.request
from wake_alarm._constants import (
ALARM_CACHE_FILE,
ALARM_CACHE_MAX_AGE_SECONDS,
FALLBACK_ALARM_HOUR,
MAX_HOUR,
MAX_MINUTE,
MIN_HOUR,
MIN_MINUTE,
SYNC_ALARM_PATH,
SYNC_API_BASE,
SYNC_CONFIG_FILE,
SYNC_HTTP_TIMEOUT_SECONDS,
SYNC_REPO_NAME,
SYNC_REPO_OWNER,
SYNC_TOKEN_FALLBACK_FILES,
)
from wake_alarm._logging_setup import configure_logging
_logger = logging.getLogger(__name__)
# Keys inside the crdt_sync log the phone writes.
_RECORD_ID = "alarm"
_HOUR_FIELD = "alarm_hour"
_MINUTE_FIELD = "alarm_minute"
@dataclass(frozen=True)
class Alarm:
"""A wall-clock alarm time, in the phone's (and the PC's) local timezone.
Attributes:
hour: Hour of day, 0-23.
minute: Minute of hour, 0-59.
source: Where the value came from ("remote", "cache", "stale-cache"
or "fallback"). Purely informational — used for logging and the
``--show`` output so a wrong alarm is diagnosable at a glance.
"""
hour: int
minute: int
source: str
def __str__(self) -> str:
"""Render as ``HH:MM (source)``."""
return f"{self.hour:02d}:{self.minute:02d} ({self.source})"
def _in_range(hour: int, minute: int) -> bool:
"""Check that an (hour, minute) pair is a real wall-clock time.
Guards both readers: a malformed sync payload and a corrupted cache must
never become a fire time.
Args:
hour: Candidate hour.
minute: Candidate minute.
Returns:
True if both values are within a 24-hour clock.
"""
return MIN_HOUR <= hour <= MAX_HOUR and MIN_MINUTE <= minute <= MAX_MINUTE
def _read_token_config() -> str | None:
"""Read the token from this package's own JSON config.
Returns:
The token, or None if the file is missing, unreadable, malformed, or
holds no usable token.
"""
try:
with SYNC_CONFIG_FILE.open() as handle:
config = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
if not isinstance(config, dict):
return None
token = config.get("token")
if not isinstance(token, str) or not token:
return None
return token
def _read_token_fallback() -> str | None:
"""Read the token from a sibling app's raw single-line token file.
Returns:
The first usable token found, or None if none of the fallback files
exists or holds one.
"""
for path in SYNC_TOKEN_FALLBACK_FILES:
try:
token = path.read_text().strip()
except OSError:
continue
if token:
_logger.debug("Using sync token from %s", path)
return token
return None
def _read_token() -> str | None:
"""Read the GitHub token, preferring this package's own config.
Falls back to the sibling companion apps' token files so one repo-scoped
PAT covers every app syncing through ``kuhyx/syncs``.
Returns:
The token, or None if no configured source holds one.
"""
# No warning on a miss here: the caller reports "no token" once. Logging
# twice for one unconfigured file makes every tick noisy in the journal.
return _read_token_config() or _read_token_fallback()
def _extract_field(log: dict[str, Any], field: str) -> int | None:
"""Pull one integer field out of the phone's crdt_sync log.
The stored shape is ``{"alarm": {"fields": {"<field>": [value, "<hlc>"]}}}``
— each field is a (value, HLC) pair, so the value is element 0.
Args:
log: The parsed log JSON.
field: Field name to extract (e.g. ``alarm_hour``).
Returns:
The integer value, or None if anything about the shape is unexpected.
"""
record = log.get(_RECORD_ID)
if not isinstance(record, dict):
return None
fields = record.get("fields")
if not isinstance(fields, dict):
return None
pair = fields.get(field)
# bool is an int subclass; an alarm hour of True would be nonsense.
if not isinstance(pair, list) or not pair:
return None
value = pair[0]
if isinstance(value, bool) or not isinstance(value, int):
return None
return value
def _parse_log(raw: str) -> tuple[int, int] | None:
"""Parse the phone's log into an (hour, minute) pair.
Args:
raw: The raw JSON text returned by the Contents API.
Returns:
The (hour, minute) pair, or None if the payload is malformed or the
values are out of range.
"""
try:
log = json.loads(raw)
except json.JSONDecodeError:
_logger.warning("Synced alarm is not valid JSON")
return None
if not isinstance(log, dict):
return None
hour = _extract_field(log, _HOUR_FIELD)
minute = _extract_field(log, _MINUTE_FIELD)
if hour is None or minute is None:
_logger.warning("Synced alarm is missing hour/minute fields")
return None
if not _in_range(hour, minute):
_logger.warning("Synced alarm out of range: %d:%d", hour, minute)
return None
return hour, minute
def fetch_remote_alarm() -> tuple[int, int] | None:
"""Fetch the alarm the phone last pushed to GitHub.
Uses the raw media type so the Contents API returns the file body directly
rather than base64-wrapped JSON.
Returns:
The (hour, minute) pair, or None if unconfigured, unreachable, or
malformed. Never raises — a failed read must degrade to the cache,
not crash the trigger.
"""
token = _read_token()
if token is None:
# INFO, not WARNING: get_alarm() runs on every one-minute trigger tick,
# so a persistent condition logged loudly here is 1440 identical
# journal lines a day. The trigger warns once, at the moment it
# actually fires on a degraded source — that is the actionable event.
_logger.info("No sync token configured at %s", SYNC_CONFIG_FILE)
return None
url = (
f"{SYNC_API_BASE}/repos/{SYNC_REPO_OWNER}/{SYNC_REPO_NAME}"
f"/contents/{SYNC_ALARM_PATH}"
)
request = urllib.request.Request(
url,
headers={
"Authorization": f"token {token}",
"Accept": "application/vnd.github.raw",
},
)
# The URL is built from module constants, never from remote input, but
# assert the scheme anyway: this call carries a credential, and an http://
# or file:// target would leak it. Belt and braces for a bad edit above.
if request.type != "https":
_logger.error("Refusing to send the sync token over %s", request.type)
return None
try:
with urllib.request.urlopen(
request, timeout=SYNC_HTTP_TIMEOUT_SECONDS
) as response:
raw = response.read().decode("utf-8")
except (urllib.error.URLError, OSError, UnicodeDecodeError) as exc:
# INFO for the same reason as the no-token path: a persistent failure
# (404 because the phone never pushed, an expired token, no network)
# would otherwise be one WARNING per cache expiry, forever.
_logger.info("Cannot fetch synced alarm: %s", exc)
return None
return _parse_log(raw)
def _read_cache() -> tuple[int, int, float] | None:
"""Read the cached alarm.
Returns:
An (hour, minute, fetched_at) triple, or None if there is no usable
cache.
"""
try:
with ALARM_CACHE_FILE.open() as handle:
cached = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
if not isinstance(cached, dict):
return None
hour = cached.get("hour")
minute = cached.get("minute")
fetched_at = cached.get("fetched_at")
if not isinstance(hour, int) or not isinstance(minute, int):
return None
if not isinstance(fetched_at, (int, float)):
return None
if not _in_range(hour, minute):
return None
return hour, minute, float(fetched_at)
def _write_cache(hour: int, minute: int) -> None:
"""Persist a freshly fetched alarm so later reads survive a network loss.
Failures are logged and swallowed: an unwritable cache degrades the next
read to a live fetch, which is not worth failing a trigger tick over.
Args:
hour: Hour of day, 0-23.
minute: Minute of hour, 0-59.
"""
entry = {"hour": hour, "minute": minute, "fetched_at": time.time()}
try:
ALARM_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
with ALARM_CACHE_FILE.open("w") as handle:
json.dump(entry, handle, indent=2)
except OSError as exc:
_logger.warning("Cannot write alarm cache: %s", exc)
def get_alarm(*, max_cache_age_seconds: float = ALARM_CACHE_MAX_AGE_SECONDS) -> Alarm:
"""Resolve the alarm to fire today, degrading gracefully at every step.
Resolution order:
1. A cache entry younger than *max_cache_age_seconds* (no network).
2. A live fetch from GitHub (cached on success).
3. Any cache entry, however stale — better an old alarm than none.
4. ``FALLBACK_ALARM_HOUR:00``, if the alarm has never been read at all.
Args:
max_cache_age_seconds: How long a cached read is served without
re-fetching. Injectable so tests and ``--show`` can force a live
read.
Returns:
The resolved alarm. Always succeeds.
"""
cached = _read_cache()
if cached is not None:
hour, minute, fetched_at = cached
if time.time() - fetched_at < max_cache_age_seconds:
return Alarm(hour, minute, "cache")
remote = fetch_remote_alarm()
if remote is not None:
_write_cache(*remote)
return Alarm(remote[0], remote[1], "remote")
# Both degraded paths log at INFO for the same per-minute-tick reason as
# _read_token above; _trigger warns when it fires on a degraded source.
if cached is not None:
_logger.info("Using stale cached alarm — sync read failed")
return Alarm(cached[0], cached[1], "stale-cache")
_logger.info(
"No synced or cached alarm — falling back to %02d:00",
FALLBACK_ALARM_HOUR,
)
return Alarm(FALLBACK_ALARM_HOUR, 0, "fallback")
def _parse_args(argv: list[str]) -> argparse.Namespace:
"""Parse CLI arguments for the sync reader."""
parser = argparse.ArgumentParser(
description="Show the alarm time the phone last synced."
)
parser.add_argument(
"--show",
action="store_true",
help="Print the resolved alarm and where it came from.",
)
parser.add_argument(
"--no-cache",
action="store_true",
help="Ignore cache freshness and force a live fetch.",
)
return parser.parse_args(argv)
def main() -> None:
"""Entry point: print the resolved alarm (``--show`` is the only mode)."""
configure_logging()
args = _parse_args(sys.argv[1:])
max_age = 0.0 if args.no_cache else ALARM_CACHE_MAX_AGE_SECONDS
sys.stdout.write(f"{get_alarm(max_cache_age_seconds=max_age)}\n")
if __name__ == "__main__":
main()

198
wake_alarm/_trigger.py Normal file
View File

@ -0,0 +1,198 @@
"""Per-minute tick that fires the morning routine at the phone's alarm time.
Run by ``wake-alarm-trigger.timer`` once a minute. Exits in milliseconds on an
ordinary tick; on the alarm minute it starts ``morning-routine.service`` and
exits.
Why a timer rather than a resident scheduler: each tick is a fresh, stateless
process, so there is no long-lived daemon to die silently and no wall-clock
jump to reason about when the machine suspends and resumes.
Why it starts the *orchestrator* and not ``wake-alarm.service``: the
orchestrator runs the alarm first and the workout lock afterwards, so the two
never fight for the fullscreen. Bypassing it is the exact failure ``CLAUDE.md``
documents. Starting an already-running unit is a no-op, which also means an
alarm currently on screen needs no extra dedupe bookkeeping.
All times are **local**. The phone stores a wall-clock H:M, so the fire instant
must be built in the same timezone the user reads off their phone.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timedelta, timezone
import logging
import subprocess
import sys
from wake_alarm._constants import (
ALARM_DAYS,
TRIGGER_CATCHUP_MINUTES,
TRIGGER_TARGET_UNIT,
)
from wake_alarm._logging_setup import configure_logging
from wake_alarm._state import (
was_alarm_dismissed_today,
was_workout_logged_today,
)
from wake_alarm._sync import Alarm, get_alarm
_logger = logging.getLogger(__name__)
def _now() -> datetime:
"""Return the current time as an aware datetime in the system's timezone.
Aware-and-local, not naive: the phone stores a wall-clock H:M, so the fire
instant has to be built in the timezone the user reads off their phone —
but carrying the offset explicitly keeps the comparisons unambiguous.
Returns:
The current local time, with tzinfo set. Also the seam tests patch.
"""
return datetime.now(tz=timezone.utc).astimezone()
def _fire_window(alarm: Alarm, now: datetime) -> tuple[datetime, datetime]:
"""Compute today's fire instant and the end of its catch-up window.
Args:
alarm: The resolved alarm time.
now: Current local time, used only for its date.
Returns:
A (fire_at, window_end) pair in local time.
"""
# Drop to a naive wall-clock time before substituting H:M, then let
# astimezone() re-derive the offset. Calling replace() straight on an
# aware datetime would keep *now*'s UTC offset, which is wrong across a
# DST boundary — e.g. deciding a 06:40 alarm from a 03:00 tick on the
# night the clocks move.
fire_at = (
now.replace(tzinfo=None)
.replace(hour=alarm.hour, minute=alarm.minute, second=0, microsecond=0)
.astimezone()
)
return fire_at, fire_at + timedelta(minutes=TRIGGER_CATCHUP_MINUTES)
def _skip_reason(now: datetime) -> str | None:
"""Return why the alarm must not fire right now, or None if it may.
Checks only the gates that are independent of the clock: the day gate and
the two "already handled today" gates shared with ``_alarm.py``.
Args:
now: Current local time.
Returns:
A human-readable reason, or None when nothing blocks firing.
"""
if now.weekday() not in ALARM_DAYS:
return "not an alarm day"
if was_alarm_dismissed_today():
return "alarm already dismissed today"
if was_workout_logged_today():
return "workout already logged today"
return None
def _start_target_unit() -> bool:
"""Start the morning routine unit.
Returns:
True if systemd accepted the start, False otherwise. A failure is
logged and swallowed: the next tick retries, and the catch-up window
means a transient failure does not cost the whole wake-up.
"""
cmd = ["systemctl", "--user", "start", TRIGGER_TARGET_UNIT]
try:
result = subprocess.run(cmd, check=False, capture_output=True)
except OSError:
_logger.warning("Failed to run %s", " ".join(cmd), exc_info=True)
return False
if result.returncode != 0:
_logger.warning(
"%s failed (rc=%d): %s",
" ".join(cmd),
result.returncode,
result.stderr.decode("utf-8", "replace").strip(),
)
return False
return True
def _decide(now: datetime, alarm: Alarm) -> tuple[bool, str]:
"""Decide whether this tick should fire the alarm.
Args:
now: Current local time.
alarm: The resolved alarm time.
Returns:
A (should_fire, reason) pair; *reason* explains the decision either
way and is what ``--dry-run`` prints.
"""
skip = _skip_reason(now)
if skip is not None:
return False, skip
fire_at, window_end = _fire_window(alarm, now)
if now < fire_at:
return False, f"alarm {alarm} is still ahead"
if now >= window_end:
return False, (
f"alarm {alarm} passed over "
f"{TRIGGER_CATCHUP_MINUTES} minutes ago — window closed"
)
return True, f"alarm {alarm} is due (since {fire_at:%H:%M})"
def _parse_args(argv: list[str]) -> argparse.Namespace:
"""Parse CLI arguments for the trigger."""
parser = argparse.ArgumentParser(
description="Fire the morning routine at the phone's alarm time."
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print the decision without starting anything.",
)
return parser.parse_args(argv)
def main() -> None:
"""Entry point for one timer tick."""
configure_logging()
args = _parse_args(sys.argv[1:])
now = _now()
alarm = get_alarm()
should_fire, reason = _decide(now, alarm)
if args.dry_run:
verdict = "WOULD FIRE" if should_fire else "not firing"
sys.stdout.write(f"{now:%Y-%m-%d %H:%M} — {verdict}: {reason}\n")
return
if not should_fire:
_logger.debug("Not firing: %s", reason)
return
# The one moment a degraded read is worth shouting about: we are firing
# now, and the time we are firing at did not come from the phone.
if alarm.source not in {"remote", "cache"}:
_logger.warning(
"Firing on a degraded alarm source (%s) — the phone's alarm "
"could not be read; check ~/.config/wake_alarm/sync.json",
alarm.source,
)
_logger.warning("Firing %s: %s", TRIGGER_TARGET_UNIT, reason)
_start_target_unit()
if __name__ == "__main__":
main()

387
wake_alarm/tests/test_sync.py Normal file
View File

@ -0,0 +1,387 @@
"""Tests for _sync.py — reading the alarm the phone pushed, with fallbacks."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
import urllib.error
import pytest
from wake_alarm._sync import (
Alarm,
_extract_field,
_parse_args,
_parse_log,
_read_cache,
_read_token,
_read_token_fallback,
_write_cache,
fetch_remote_alarm,
get_alarm,
main,
)
if TYPE_CHECKING:
from pathlib import Path
def _log(hour: object = 7, minute: object = 30) -> str:
"""Build a crdt_sync log payload shaped like the phone's."""
return json.dumps(
{
"alarm": {
"fields": {
"alarm_hour": [hour, "hlc-1"],
"alarm_minute": [minute, "hlc-1"],
}
}
}
)
@pytest.fixture
def config_file(tmp_path: Path) -> Path:
"""Provide a temporary sync config path."""
return tmp_path / "sync.json"
@pytest.fixture
def cache_file(tmp_path: Path) -> Path:
"""Provide a temporary alarm cache path."""
return tmp_path / "cache" / "alarm_cache.json"
@pytest.fixture
def fallback_files(tmp_path: Path) -> tuple[Path, Path]:
"""Provide two temporary sibling-app token paths."""
return tmp_path / "diet_guard_token", tmp_path / "screen_locker_token"
@pytest.fixture(autouse=True)
def _patch_paths(
config_file: Path, cache_file: Path, fallback_files: tuple[Path, Path]
) -> None:
"""Redirect every on-disk path into tmp_path for every test."""
with (
patch("wake_alarm._sync.SYNC_CONFIG_FILE", config_file),
patch("wake_alarm._sync.ALARM_CACHE_FILE", cache_file),
patch("wake_alarm._sync.SYNC_TOKEN_FALLBACK_FILES", fallback_files),
):
yield
class TestAlarm:
"""Tests for the Alarm value object."""
def test_str_pads_and_names_source(self) -> None:
"""Render zero-padded HH:MM with the source in parentheses."""
assert str(Alarm(7, 5, "remote")) == "07:05 (remote)"
class TestReadToken:
"""Tests for _read_token."""
def test_missing_file(self) -> None:
"""A missing config file yields no token."""
assert _read_token() is None
def test_malformed_json(self, config_file: Path) -> None:
"""Unparsable JSON yields no token."""
config_file.write_text("{not json")
assert _read_token() is None
def test_not_a_dict(self, config_file: Path) -> None:
"""A JSON array is not a config object."""
config_file.write_text("[]")
assert _read_token() is None
@pytest.mark.parametrize("value", [None, "", 42])
def test_unusable_token_values(self, config_file: Path, value: object) -> None:
"""Absent, empty and non-string tokens are all rejected."""
config_file.write_text(json.dumps({"token": value}))
assert _read_token() is None
def test_valid_token(self, config_file: Path) -> None:
"""A non-empty string token is returned."""
config_file.write_text(json.dumps({"token": "ghp_x"}))
assert _read_token() == "ghp_x"
class TestTokenFallback:
"""Tests for reusing a sibling companion app's token file."""
def test_no_fallback_files(self) -> None:
"""With nothing on disk anywhere there is no token."""
assert _read_token_fallback() is None
assert _read_token() is None
def test_reads_raw_single_line_token(
self, fallback_files: tuple[Path, Path]
) -> None:
"""A sibling token file is a raw PAT, not JSON — and is stripped."""
fallback_files[0].write_text(" ghp_sibling\n")
assert _read_token_fallback() == "ghp_sibling"
def test_skips_empty_file(self, fallback_files: tuple[Path, Path]) -> None:
"""An empty first file falls through to the second."""
fallback_files[0].write_text(" \n")
fallback_files[1].write_text("ghp_second")
assert _read_token_fallback() == "ghp_second"
def test_skips_unreadable_file(self, fallback_files: tuple[Path, Path]) -> None:
"""An unreadable file is skipped rather than fatal."""
fallback_files[1].write_text("ghp_second")
with patch(
"pathlib.Path.read_text",
side_effect=[OSError("denied"), "ghp_second"],
):
assert _read_token_fallback() == "ghp_second"
def test_own_config_wins(
self, config_file: Path, fallback_files: tuple[Path, Path]
) -> None:
"""wake_alarm's own config takes precedence over a sibling's."""
config_file.write_text(json.dumps({"token": "ghp_own"}))
fallback_files[0].write_text("ghp_sibling")
assert _read_token() == "ghp_own"
def test_falls_back_when_config_absent(
self, fallback_files: tuple[Path, Path]
) -> None:
"""With no own config, the sibling token is used."""
fallback_files[0].write_text("ghp_sibling")
assert _read_token() == "ghp_sibling"
class TestExtractField:
"""Tests for _extract_field's tolerance of malformed logs."""
@pytest.mark.parametrize(
"log",
[
{},
{"alarm": "not a dict"},
{"alarm": {}},
{"alarm": {"fields": "not a dict"}},
{"alarm": {"fields": {}}},
{"alarm": {"fields": {"alarm_hour": "not a list"}}},
{"alarm": {"fields": {"alarm_hour": []}}},
{"alarm": {"fields": {"alarm_hour": ["7", "hlc"]}}},
{"alarm": {"fields": {"alarm_hour": [True, "hlc"]}}},
],
)
def test_malformed_shapes_yield_none(self, log: dict[str, Any]) -> None:
"""Every unexpected shape degrades to None rather than raising."""
assert _extract_field(log, "alarm_hour") is None
def test_extracts_value(self) -> None:
"""The value is element 0 of the (value, HLC) pair."""
log = {"alarm": {"fields": {"alarm_hour": [6, "hlc"]}}}
assert _extract_field(log, "alarm_hour") == 6
class TestParseLog:
"""Tests for _parse_log."""
def test_valid(self) -> None:
"""A well-formed log parses to an (hour, minute) pair."""
assert _parse_log(_log(7, 30)) == (7, 30)
def test_invalid_json(self) -> None:
"""Unparsable JSON yields None."""
assert _parse_log("{nope") is None
def test_not_a_dict(self) -> None:
"""A JSON array is not a log."""
assert _parse_log("[]") is None
def test_missing_fields(self) -> None:
"""A log without the alarm fields yields None."""
assert _parse_log(json.dumps({"alarm": {"fields": {}}})) is None
@pytest.mark.parametrize(("hour", "minute"), [(24, 0), (-1, 0), (7, 60), (7, -1)])
def test_out_of_range(self, hour: int, minute: int) -> None:
"""Values outside 0-23 / 0-59 are rejected."""
assert _parse_log(_log(hour, minute)) is None
class TestFetchRemoteAlarm:
"""Tests for fetch_remote_alarm."""
def test_no_token(self) -> None:
"""Without a token there is nothing to fetch."""
assert fetch_remote_alarm() is None
def _with_token(self, config_file: Path) -> None:
config_file.write_text(json.dumps({"token": "ghp_x"}))
def test_success(self, config_file: Path) -> None:
"""A 200 with a valid body yields the pair."""
self._with_token(config_file)
response = MagicMock()
response.read.return_value = _log(6, 15).encode()
response.__enter__.return_value = response
with patch("urllib.request.urlopen", return_value=response):
assert fetch_remote_alarm() == (6, 15)
def test_sends_raw_media_type_and_auth(self, config_file: Path) -> None:
"""The request carries the token and the raw Accept header."""
self._with_token(config_file)
response = MagicMock()
response.read.return_value = _log().encode()
response.__enter__.return_value = response
with patch("urllib.request.urlopen", return_value=response) as urlopen:
fetch_remote_alarm()
request = urlopen.call_args[0][0]
assert request.get_header("Authorization") == "token ghp_x"
assert request.get_header("Accept") == "application/vnd.github.raw"
assert "kuhyx/syncs" in request.full_url
def test_refuses_a_non_https_scheme(self, config_file: Path) -> None:
"""The token is never sent over a plaintext or file:// URL."""
self._with_token(config_file)
with (
patch("wake_alarm._sync.SYNC_API_BASE", "http://api.github.com"),
patch("urllib.request.urlopen") as urlopen,
):
assert fetch_remote_alarm() is None
urlopen.assert_not_called()
@pytest.mark.parametrize(
"exc",
[
urllib.error.URLError("down"),
OSError("timeout"),
UnicodeDecodeError("utf-8", b"\xff", 0, 1, "bad"),
],
)
def test_network_failures_yield_none(
self, config_file: Path, exc: Exception
) -> None:
"""Every transport failure degrades to None, never raises."""
self._with_token(config_file)
with patch("urllib.request.urlopen", side_effect=exc):
assert fetch_remote_alarm() is None
class TestCache:
"""Tests for the on-disk cache helpers."""
def test_roundtrip(self, cache_file: Path) -> None:
"""A written cache reads back with the same values."""
_write_cache(6, 45)
cached = _read_cache()
assert cached is not None
assert cached[0] == 6
assert cached[1] == 45
def test_write_creates_parent_dir(self, cache_file: Path) -> None:
"""The cache directory is created on demand."""
_write_cache(6, 45)
assert cache_file.exists()
def test_write_failure_is_swallowed(self) -> None:
"""An unwritable cache logs but does not raise."""
with patch("pathlib.Path.mkdir", side_effect=OSError("read-only")):
_write_cache(6, 45)
def test_missing_file(self) -> None:
"""No cache file yields None."""
assert _read_cache() is None
@pytest.mark.parametrize(
"content",
[
"{not json",
"[]",
json.dumps({"hour": "6", "minute": 45, "fetched_at": 1.0}),
json.dumps({"hour": 6, "minute": "45", "fetched_at": 1.0}),
json.dumps({"hour": 6, "minute": 45, "fetched_at": "soon"}),
json.dumps({"hour": 24, "minute": 45, "fetched_at": 1.0}),
json.dumps({"hour": 6, "minute": 60, "fetched_at": 1.0}),
],
)
def test_malformed_cache_yields_none(self, cache_file: Path, content: str) -> None:
"""Any unexpected cache shape is treated as no cache."""
cache_file.parent.mkdir(parents=True, exist_ok=True)
cache_file.write_text(content)
assert _read_cache() is None
class TestGetAlarm:
"""Tests for the get_alarm resolution chain."""
def test_fresh_cache_skips_network(self, cache_file: Path) -> None:
"""A young cache entry is served without any fetch."""
_write_cache(6, 45)
with patch("wake_alarm._sync.fetch_remote_alarm") as fetch:
alarm = get_alarm()
fetch.assert_not_called()
assert (alarm.hour, alarm.minute, alarm.source) == (6, 45, "cache")
def test_stale_cache_triggers_fetch(self, cache_file: Path) -> None:
"""An old cache entry is refreshed from the network."""
_write_cache(6, 45)
with patch("wake_alarm._sync.fetch_remote_alarm", return_value=(7, 0)):
alarm = get_alarm(max_cache_age_seconds=0)
assert (alarm.hour, alarm.minute, alarm.source) == (7, 0, "remote")
def test_fetch_result_is_cached(self, cache_file: Path) -> None:
"""A successful fetch is written to the cache for next time."""
with patch("wake_alarm._sync.fetch_remote_alarm", return_value=(7, 0)):
get_alarm()
assert json.loads(cache_file.read_text())["hour"] == 7
def test_falls_back_to_stale_cache(self, cache_file: Path) -> None:
"""When the fetch fails, an old cache entry is still used."""
_write_cache(6, 45)
with patch("wake_alarm._sync.fetch_remote_alarm", return_value=None):
alarm = get_alarm(max_cache_age_seconds=0)
assert (alarm.hour, alarm.minute, alarm.source) == (
6,
45,
"stale-cache",
)
def test_falls_back_to_constant(self) -> None:
"""With no cache and no network, the fixed fallback hour is used."""
with patch("wake_alarm._sync.fetch_remote_alarm", return_value=None):
alarm = get_alarm()
assert (alarm.minute, alarm.source) == (0, "fallback")
assert alarm.hour == 13
class TestCli:
"""Tests for the --show entry point."""
def test_parse_args_defaults(self) -> None:
"""Both flags default to off."""
args = _parse_args([])
assert not args.show
assert not args.no_cache
def test_main_prints_alarm(self, capsys: pytest.CaptureFixture[str]) -> None:
"""main prints the resolved alarm."""
with (
patch("sys.argv", ["_sync", "--show"]),
patch(
"wake_alarm._sync.get_alarm",
return_value=Alarm(7, 30, "remote"),
),
):
main()
assert "07:30 (remote)" in capsys.readouterr().out
def test_no_cache_forces_live_read(self) -> None:
"""--no-cache asks get_alarm for a zero-age cache."""
with (
patch("sys.argv", ["_sync", "--no-cache"]),
patch(
"wake_alarm._sync.get_alarm",
return_value=Alarm(7, 30, "remote"),
) as get,
):
main()
assert get.call_args.kwargs["max_cache_age_seconds"] == 0.0

252
wake_alarm/tests/test_trigger.py Normal file
View File

@ -0,0 +1,252 @@
"""Tests for _trigger.py — the per-minute check-and-fire tick."""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch
import pytest
from wake_alarm._sync import Alarm
from wake_alarm._trigger import (
_decide,
_fire_window,
_now,
_parse_args,
_skip_reason,
_start_target_unit,
main,
)
def _at(hour: int, minute: int) -> datetime:
"""Return a Sunday at the given local wall-clock time, timezone-aware.
Built the same way production does — naive wall clock, then astimezone()
to attach the system offset — so the tests exercise the real comparison
semantics rather than a hand-picked fixed offset.
"""
return datetime(2026, 7, 19, hour, minute, tzinfo=None).astimezone()
@pytest.fixture(autouse=True)
def _ungated() -> None:
"""Open both "already handled today" gates unless a test closes them."""
with (
patch("wake_alarm._trigger.was_alarm_dismissed_today", return_value=False),
patch("wake_alarm._trigger.was_workout_logged_today", return_value=False),
):
yield
class TestNow:
"""Tests for the _now seam."""
def test_returns_aware_local_time(self) -> None:
"""The clock is timezone-aware and in local time, not UTC."""
now = _now()
assert now.tzinfo is not None
# Local, not UTC: the wall-clock hour must match the system clock's.
assert now.hour == datetime.now(tz=timezone.utc).astimezone().hour
class TestFireWindow:
"""Tests for _fire_window."""
def test_window_starts_at_the_alarm_minute(self) -> None:
"""The fire instant is today at the alarm's H:M, seconds zeroed."""
fire_at, _ = _fire_window(Alarm(6, 45, "remote"), _at(3, 12))
assert fire_at == _at(6, 45)
assert fire_at.second == 0
def test_window_is_the_catchup_length(self) -> None:
"""The window ends TRIGGER_CATCHUP_MINUTES after the alarm."""
fire_at, end = _fire_window(Alarm(6, 45, "remote"), _at(3, 12))
assert (end - fire_at).total_seconds() == 60 * 60
def test_fire_instant_is_aware(self) -> None:
"""The fire instant keeps a timezone, so comparisons are unambiguous."""
fire_at, end = _fire_window(Alarm(6, 45, "remote"), _at(3, 12))
assert fire_at.tzinfo is not None
assert end.tzinfo is not None
def test_offset_is_derived_from_the_target_wall_time(self) -> None:
"""The offset is re-derived, not inherited from *now*.
Guards the DST case: deciding a morning alarm from a tick taken on the
other side of a clock change must not carry *now*'s stale UTC offset.
"""
fire_at, _ = _fire_window(Alarm(6, 45, "remote"), _at(3, 12))
expected = datetime(2026, 7, 19, 6, 45, tzinfo=None).astimezone()
assert fire_at.utcoffset() == expected.utcoffset()
assert fire_at == expected
class TestSkipReason:
"""Tests for the clock-independent gates."""
def test_open_when_nothing_blocks(self) -> None:
"""With every gate open there is no reason to skip."""
assert _skip_reason(_at(7, 0)) is None
def test_not_an_alarm_day(self) -> None:
"""A weekday outside ALARM_DAYS blocks firing."""
with patch("wake_alarm._trigger.ALARM_DAYS", frozenset()):
assert _skip_reason(_at(7, 0)) == "not an alarm day"
def test_already_dismissed(self) -> None:
"""A dismissal recorded today blocks a re-fire."""
with patch(
"wake_alarm._trigger.was_alarm_dismissed_today",
return_value=True,
):
assert _skip_reason(_at(7, 0)) == "alarm already dismissed today"
def test_workout_already_logged(self) -> None:
"""A logged workout means the alarm is no longer needed."""
with patch("wake_alarm._trigger.was_workout_logged_today", return_value=True):
assert _skip_reason(_at(7, 0)) == "workout already logged today"
class TestDecide:
"""Tests for the fire/don't-fire decision."""
def test_before_the_alarm(self) -> None:
"""A tick before the alarm minute does not fire."""
fire, reason = _decide(_at(6, 44), Alarm(6, 45, "remote"))
assert not fire
assert "still ahead" in reason
def test_exactly_on_the_alarm_minute(self) -> None:
"""The alarm minute itself fires."""
fire, _ = _decide(_at(6, 45), Alarm(6, 45, "remote"))
assert fire
def test_within_the_catchup_window(self) -> None:
"""A tick inside the catch-up window still fires (e.g. after resume)."""
fire, _ = _decide(_at(7, 30), Alarm(6, 45, "remote"))
assert fire
def test_at_the_window_edge(self) -> None:
"""The window is half-open: its end instant no longer fires."""
fire, reason = _decide(_at(7, 45), Alarm(6, 45, "remote"))
assert not fire
assert "window closed" in reason
def test_long_after_the_alarm(self) -> None:
"""Hours later, the window is closed."""
fire, reason = _decide(_at(20, 0), Alarm(6, 45, "remote"))
assert not fire
assert "window closed" in reason
def test_gate_beats_the_clock(self) -> None:
"""A closed gate wins even at the exact alarm minute."""
with patch(
"wake_alarm._trigger.was_alarm_dismissed_today",
return_value=True,
):
fire, reason = _decide(_at(6, 45), Alarm(6, 45, "remote"))
assert not fire
assert reason == "alarm already dismissed today"
def test_alarm_moved_earlier_fires_immediately(self) -> None:
"""A phone alarm moved back before now fires on the very next tick."""
fire, _ = _decide(_at(6, 50), Alarm(6, 40, "remote"))
assert fire
class TestStartTargetUnit:
"""Tests for the systemctl invocation."""
def test_starts_the_orchestrator_not_the_alarm(self) -> None:
"""The trigger must start morning-routine, never wake-alarm."""
completed = MagicMock(returncode=0)
with patch("subprocess.run", return_value=completed) as run:
assert _start_target_unit()
assert run.call_args[0][0] == [
"systemctl",
"--user",
"start",
"morning-routine.service",
]
def test_nonzero_exit_is_reported(self) -> None:
"""A systemctl failure returns False rather than raising."""
completed = MagicMock(returncode=1, stderr=b"no such unit")
with patch("subprocess.run", return_value=completed):
assert not _start_target_unit()
def test_missing_systemctl_is_swallowed(self) -> None:
"""An OSError (no systemctl) returns False rather than raising."""
with patch("subprocess.run", side_effect=OSError("not found")):
assert not _start_target_unit()
def test_does_not_raise_on_failure(self) -> None:
"""check=False: a failing start never raises CalledProcessError."""
completed = MagicMock(returncode=1, stderr=b"")
with patch("subprocess.run", return_value=completed) as run:
_start_target_unit()
assert run.call_args.kwargs["check"] is False
class TestMain:
"""Tests for the tick entry point."""
def test_parse_args_default(self) -> None:
"""--dry-run defaults to off."""
assert not _parse_args([]).dry_run
def _run_main(self, argv: list[str], now: datetime, alarm: Alarm) -> MagicMock:
with (
patch("sys.argv", ["_trigger", *argv]),
patch("wake_alarm._trigger._now", return_value=now),
patch("wake_alarm._trigger.get_alarm", return_value=alarm),
patch("wake_alarm._trigger._start_target_unit") as start,
):
main()
return start
def test_fires_at_the_alarm_minute(self) -> None:
"""On the alarm minute the target unit is started."""
start = self._run_main([], _at(6, 45), Alarm(6, 45, "remote"))
start.assert_called_once()
def test_does_not_fire_early(self) -> None:
"""Before the alarm nothing is started."""
start = self._run_main([], _at(5, 0), Alarm(6, 45, "remote"))
start.assert_not_called()
def test_dry_run_never_starts_anything(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""--dry-run reports WOULD FIRE without touching systemd."""
start = self._run_main(["--dry-run"], _at(6, 45), Alarm(6, 45, "remote"))
start.assert_not_called()
assert "WOULD FIRE" in capsys.readouterr().out
def test_dry_run_reports_not_firing(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""--dry-run explains why it is not firing."""
self._run_main(["--dry-run"], _at(5, 0), Alarm(6, 45, "remote"))
out = capsys.readouterr().out
assert "not firing" in out
assert "still ahead" in out
@pytest.mark.parametrize("source", ["fallback", "stale-cache"])
def test_warns_when_firing_on_a_degraded_source(
self, caplog: pytest.LogCaptureFixture, source: str
) -> None:
"""Firing on a non-phone alarm time is logged loudly."""
with caplog.at_level("WARNING"):
self._run_main([], _at(6, 45), Alarm(6, 45, source))
assert "degraded alarm source" in caplog.text
@pytest.mark.parametrize("source", ["remote", "cache"])
def test_quiet_when_firing_on_the_phone_alarm(
self, caplog: pytest.LogCaptureFixture, source: str
) -> None:
"""A normal fire does not emit the degraded-source warning."""
with caplog.at_level("WARNING"):
self._run_main([], _at(6, 45), Alarm(6, 45, source))
assert "degraded alarm source" not in caplog.text