From 373a0756067e3868471f1528c2e6010b81f0ce1e Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 9 Oct 2026 10:43:45 +0200 Subject: [PATCH] Shutdown: walks as workout, rest days, grace floor, workout ledger (earned-time 0.4.0) - A day's RunnerUp walks (_Walking.tcx) count once their moving time sums to 40 min - Rest days: declared ahead, 2 per ISO week, HMAC-signed; pay the first workout unit - Grace floor: shutdown >= first task + 60 min (capped); skipped on a sick day - Workout ledger: one signed credit row per log credit slot and rest day - Day-aware ladder amounts from earned_time 0.4.0; preview_shutdown dry-run script Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QrjUTo1yhgAsUUasdw4bpG Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 4 + pyproject.toml | 2 +- requirements.txt | 2 +- screen_locker/_cli.py | 12 + screen_locker/_constants.py | 21 +- screen_locker/_earned.py | 48 +++- screen_locker/_grace_floor.py | 204 +++++++++++++++ screen_locker/_log_mixin.py | 5 + screen_locker/_rest_day.py | 193 ++++++++++++++ screen_locker/_rest_day_cli.py | 81 ++++++ screen_locker/_runnerup_backfill.py | 70 +++++- screen_locker/_runnerup_tcx.py | 16 +- screen_locker/_runnerup_verification.py | 11 + screen_locker/_shutdown.py | 35 ++- screen_locker/_shutdown_base.py | 27 +- screen_locker/_shutdown_target.py | 133 ++++++++++ screen_locker/_status_view_verify.py | 11 +- screen_locker/_walk_tcx.py | 144 +++++++++++ screen_locker/_workout_credit.py | 24 +- screen_locker/_workout_ledger.py | 238 ++++++++++++++++++ screen_locker/_workout_ledger_cli.py | 86 +++++++ screen_locker/tests/_isolated_state.py | 11 + .../tests/test_daily_limits_wiring.py | 180 +++++++++++++ screen_locker/tests/test_earned.py | 5 +- screen_locker/tests/test_earner_units.py | 2 +- screen_locker/tests/test_grace_floor.py | 207 +++++++++++++++ screen_locker/tests/test_rest_day.py | 219 ++++++++++++++++ screen_locker/tests/test_runnerup_walks.py | 177 +++++++++++++ .../tests/test_shutdown_flat_earners.py | 16 +- screen_locker/tests/test_shutdown_reading.py | 22 +- screen_locker/tests/test_shutdown_target.py | 77 ++++++ screen_locker/tests/test_walk_tcx.py | 134 ++++++++++ screen_locker/tests/test_workout_ledger.py | 197 +++++++++++++++ .../tests/test_workout_ledger_cli.py | 95 +++++++ scripts/_preview_sim.py | 105 ++++++++ scripts/preview_shutdown.py | 187 ++++++++++++++ 36 files changed, 2940 insertions(+), 61 deletions(-) create mode 100644 screen_locker/_grace_floor.py create mode 100644 screen_locker/_rest_day.py create mode 100644 screen_locker/_rest_day_cli.py create mode 100644 screen_locker/_shutdown_target.py create mode 100644 screen_locker/_walk_tcx.py create mode 100644 screen_locker/_workout_ledger.py create mode 100644 screen_locker/_workout_ledger_cli.py create mode 100644 screen_locker/tests/test_daily_limits_wiring.py create mode 100644 screen_locker/tests/test_grace_floor.py create mode 100644 screen_locker/tests/test_rest_day.py create mode 100644 screen_locker/tests/test_runnerup_walks.py create mode 100644 screen_locker/tests/test_shutdown_target.py create mode 100644 screen_locker/tests/test_walk_tcx.py create mode 100644 screen_locker/tests/test_workout_ledger.py create mode 100644 screen_locker/tests/test_workout_ledger_cli.py create mode 100644 scripts/_preview_sim.py create mode 100755 scripts/preview_shutdown.py diff --git a/.gitignore b/.gitignore index 38d2c3c..6001293 100644 --- a/.gitignore +++ b/.gitignore @@ -41,3 +41,7 @@ web/dist/ web/coverage/ web/node_modules web/dist +screen_locker/rest_days.json +screen_locker/grace_floor.json +screen_locker/grace_floor.lock +screen_locker/rest_days.lock diff --git a/pyproject.toml b/pyproject.toml index 3e02e6b..6185fa3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ dependencies = [ "gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock", "crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync", "freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays", - "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.3.0#subdirectory=earned_time", + "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.4.0#subdirectory=earned_time", ] [project.scripts] diff --git a/requirements.txt b/requirements.txt index f5bfabc..be649e8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,7 +4,7 @@ bandit==1.9.4 codespell==2.4.3 coverage==7.16.2 crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.1#subdirectory=crdt-sync -earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.3.0#subdirectory=earned_time +earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.4.0#subdirectory=earned_time freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.0#subdirectory=freedays gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.9.0#subdirectory=gatelock mcp==2.3.0 diff --git a/screen_locker/_cli.py b/screen_locker/_cli.py index 86a7da0..b284f99 100644 --- a/screen_locker/_cli.py +++ b/screen_locker/_cli.py @@ -16,8 +16,10 @@ from typing import TYPE_CHECKING from screen_locker._constants import SHUTDOWN_BASE_FILE from screen_locker._decision_log import record_no_decision from screen_locker._manual_cli import run_manual_log +from screen_locker._rest_day_cli import run_declare_rest_day from screen_locker._shutdown_base import apply_flat_bonuses_if_new from screen_locker._status import run_status +from screen_locker._workout_ledger_cli import run_backfill if TYPE_CHECKING: from screen_locker.screen_lock import ScreenLocker @@ -28,6 +30,8 @@ _logger = logging.getLogger(__name__) _LOG_FILE_NAME = "log.json" _LOG_MANUAL_FLAG = "--log-manual-workout" +_REST_DAY_FLAG = "--declare-rest-day" +_BACKFILL_FLAG = "--backfill-workout-ledger" def _headless_locker(locker_cls: type[ScreenLocker]) -> ScreenLocker: @@ -65,6 +69,14 @@ def main(locker_cls: type[ScreenLocker], argv: list[str]) -> None: rest = argv[argv.index(_LOG_MANUAL_FLAG) + 1 :] sys.exit(run_manual_log(_headless_locker(locker_cls).log_file, rest)) + if _REST_DAY_FLAG in argv: + # Headless, no Tk: sign a FUTURE rest day (see screen_locker._rest_day). + sys.exit(run_declare_rest_day(argv[argv.index(_REST_DAY_FLAG) + 1 :])) + + if _BACKFILL_FLAG in argv: + rest = argv[argv.index(_BACKFILL_FLAG) + 1 :] + sys.exit(run_backfill(_headless_locker(locker_cls).log_file, rest)) + if "--sync-only" in argv: # Headless sync for the timer unit: pull other devices' workouts and # apply their credit, with no Tk and no lock screen. Sync used to run diff --git a/screen_locker/_constants.py b/screen_locker/_constants.py index 4029ec1..4134ac4 100644 --- a/screen_locker/_constants.py +++ b/screen_locker/_constants.py @@ -12,8 +12,16 @@ from gatelock.log_integrity import DEFAULT_HMAC_KEY_FILE # log.json/sick_history.json below. INSTANCE_LOCK_FILE = default_runtime_dir() / "screen_locker-instance.lock" -# Single-sourced from gatelock so the literal key path can't drift. +# Single-sourced from gatelock so the literal key path can't drift. The key is +# root:root 0644 ON PURPOSE, never 0600: leetcode-guard, book-guard and +# anki-guard run as user timers and sign their ledgers with it, and this +# locker signs log.json, rest_days.json and the workout ledger as the user. +# Nothing in this repo may chmod it; root ownership keeps it from being +# replaced, which is what the signatures rely on. HMAC_KEY_FILE = DEFAULT_HMAC_KEY_FILE +# The workout ledger (earned_time.WORKOUT.ledger): one signed credit row per +# verified RunnerUp TCX credit and per rest day, resolved at call time. +WORKOUT_LEDGER_RELATIVE = Path(".local/share/workout_locker/ledger.json") SICK_LOCKOUT_SECONDS = 120 # base 2 minutes wait when sick (escalates with usage) PHONE_PENALTY_DELAY_DEMO = 10 PHONE_PENALTY_DELAY_PRODUCTION = 100 @@ -120,6 +128,17 @@ SCHEDULED_SKIPS_FILE = Path(__file__).resolve().parent / "scheduled_skips.json" EXTRA_BENEFITS_FILE = Path(__file__).resolve().parent / "extra_benefits_state.json" # State file storing the base (pre-bonus) shutdown hours and last reset date. SHUTDOWN_BASE_FILE = Path(__file__).resolve().parent / "shutdown_base.json" +# HMAC-signed rest days, declared ahead (see _rest_day), and how far the grace +# floor has lifted today's shutdown above the earned time (see _grace_floor). +REST_DAY_FILE = Path(__file__).resolve().parent / "rest_days.json" +REST_DAYS_PER_ISO_WEEK = 2 +GRACE_STATE_FILE = Path(__file__).resolve().parent / "grace_floor.json" +GRACE_MINUTES = 60 +# A day's RunnerUp walks (commute legs) count as the workout once their +# MOVING time sums to this; moving = trackpoint pairs at walking speed. +WALK_MIN_MOVING_MINUTES = 40 +WALK_SPEED_MPS = (0.5, 3.0) +WALK_MAX_GAP_SECONDS = 30 # Workout records this PC must keep deleted in the sync store, pushed as # tombstones on every tick (see _sync_tombstones for why once is not enough). SYNC_TOMBSTONES_FILE = Path(__file__).resolve().parent / "sync_tombstones.json" diff --git a/screen_locker/_earned.py b/screen_locker/_earned.py index cffcf1e..06cefb7 100644 --- a/screen_locker/_earned.py +++ b/screen_locker/_earned.py @@ -16,6 +16,7 @@ midnight, so a 30-minute earner is applied exactly. from __future__ import annotations +import functools import logging from pathlib import Path from typing import TYPE_CHECKING, Final @@ -25,7 +26,7 @@ import earned_time from screen_locker._constants import HMAC_KEY_FILE if TYPE_CHECKING: - from datetime import datetime + from datetime import date, datetime _logger: Final = logging.getLogger(__name__) @@ -86,3 +87,48 @@ def flat_answers() -> dict[str, bool | None]: ) answers[earner.name] = answer return answers + + +# The day-aware ladder API (earned_time >= 0.4, from 2026-10-10). Looked up at +# call time so the installed 0.3.0 -- which has none of it -- still runs, on +# the raw fields it always used. +def first_minutes(earner: earned_time.Earner, day: date) -> int: + """What ``earner``'s first unit pushes shutdown later by on ``day``.""" + fn = getattr(earned_time, "shutdown_minutes_for", None) + return earner.shutdown_minutes if fn is None else int(fn(earner, day)) + + +def extra_minutes(earner: earned_time.Earner, day: date) -> int: + """What each further unit of a counted earner earns on ``day``.""" + fn = getattr(earned_time, "extra_shutdown_minutes_for", None) + return earner.extra_shutdown_minutes if fn is None else int(fn(earner, day)) + + +def ceiling(day: date) -> int: + """The latest shutdown ``day`` can earn, minutes after its midnight.""" + fn = getattr(earned_time, "shutdown_ceiling_for", None) + return earned_time.SHUTDOWN_CEILING_MINUTES if fn is None else int(fn(day)) + + +def first_credit_time(earner: earned_time.Earner, day: date) -> float | None: + """Unix time of ``earner``'s first verified credit counting for ``day``. + + ``None`` when there is none, when it cannot be checked, and on an + ``earned_time`` without ``first_credit_at`` (0.3.0) -- logged, because + then the grace floor sees no ledger earner at all. + """ + fn = getattr(earned_time, "first_credit_at", None) + if fn is None: + _warn_no_first_credit() + return None + stamp: float | None = fn(earner, ledger_file(earner), HMAC_KEY_FILE, day) + return stamp + + +@functools.cache +def _warn_no_first_credit() -> None: + """Once per process: the installed earned_time predates first_credit_at.""" + _logger.warning( + "earned_time has no first_credit_at (needs >= 0.4); the grace floor " + "sees no ledger earner, only the workout log" + ) diff --git a/screen_locker/_grace_floor.py b/screen_locker/_grace_floor.py new file mode 100644 index 0000000..7f4b877 --- /dev/null +++ b/screen_locker/_grace_floor.py @@ -0,0 +1,204 @@ +"""Grace floor: shutdown is never earlier than the day's first task + 60 min. + +Someone who gets home at 18:10 and finishes their first task at 18:55 is not +cut off at 19:00. The floor is ``first done + GRACE_MINUTES``, capped at the +day's ceiling. "First done" is the earliest of every ledger earner's first +verified credit (``earned_time.first_credit_at``) and the workout's own +completion time -- the end of the run or walk, never the moment a sync +happened to log it, so a late upload does not move the floor later. + +**A floor, not a bonus.** The config is otherwise read-add-write: every +earner adds its minutes to whatever is there. Lifting it to the floor and +then letting the next earner add on top would pay the lift twice (anki at +21:00 lifts to 22:00; reading then makes 22:30 where max(earned, floor) is +22:00). So the lift is remembered for the day in ``grace_floor.json`` and +later additions are paid out of it first: with ``E`` earned, floor ``G`` and +lift ``L = max(0, G - E)``, adding ``m`` moves the config by +``max(0, m - L)`` and leaves ``L' = max(0, L - m)`` -- which keeps the config +at ``max(E + m, G)`` without ever needing to know ``E``. +""" + +from __future__ import annotations + +from datetime import date, datetime +import json +import logging +from typing import TYPE_CHECKING + +import earned_time + +from screen_locker._bonus_lock import bonus_lock +from screen_locker._constants import GRACE_MINUTES, GRACE_STATE_FILE +from screen_locker._day import today_str +from screen_locker._earned import ceiling, first_credit_time, flat_earners, hhmm +from screen_locker._log_io import load_workout_log +from screen_locker._rest_day import is_rest_day +from screen_locker._sick_tracker import is_sick_day, load_history +from screen_locker._weekly_check import credit_key, day_workout_index +from screen_locker._workout_ledger import entry_done_at + +if TYPE_CHECKING: + from pathlib import Path + +_logger = logging.getLogger(__name__) + +_MON_WED = frozenset({0, 1, 2}) + + +def workout_done_at(log_file: Path, day: date) -> float | None: + """The completion time of ``day``'s first counted workout, if any.""" + iso = day.isoformat() + entries = [ + e for e in load_workout_log(log_file).get(iso, []) if isinstance(e, dict) + ] + siblings = day_workout_index(entries) + times = [ + stamp + for index, entry in enumerate(entries) + if credit_key(iso, index, entry, siblings) is not None + and (stamp := entry_done_at(entry, day)) is not None + ] + return min(times, default=None) + + +def _workout_time(day: date, log_file: Path | None, *, rest_day: bool) -> float | None: + """When ``day``'s first workout was done: the ledger's and the log's view. + + The workout ledger (``earned_time`` >= 0.4) holds only RunnerUp and rest + days, so the log still answers for manual and StrongLifts workouts. On a + rest day the ledger is skipped: its rest row is stamped at the day's + midnight, which is not a completion and would otherwise erase the floor. + """ + times: list[float | None] = [] + if log_file is not None: + times.append(workout_done_at(log_file, day)) + if not rest_day and earned_time.WORKOUT.ledger is not None: + times.append(first_credit_time(earned_time.WORKOUT, day)) + return min((t for t in times if t is not None), default=None) + + +def first_done_at( + day: date, log_file: Path | None, *, rest_day: bool = False +) -> float | None: + """The earliest moment any earner was done on ``day``; ``None`` if none.""" + times = [first_credit_time(e, day) for e in flat_earners()] + times.append(_workout_time(day, log_file, rest_day=rest_day)) + on_day = [ + t + for t in times + if t is not None and datetime.fromtimestamp(t).astimezone().date() == day + ] + return min(on_day, default=None) + + +def sick_on(day: date) -> bool: + """Whether ``day`` is a sick day, from the sick-day history itself. + + On a sick day the sick-day shutdown wins and there is no grace floor (the + user's call, 2026-10-09): the floor would otherwise lift the sick-day + reduction straight back up. + """ + return is_sick_day(load_history(), today=day.isoformat()) + + +def grace_for(first_done: float | None, day: date) -> int | None: + """The floor for ``day`` in minutes after midnight, or ``None`` if none.""" + if first_done is None: + return None + moment = datetime.fromtimestamp(first_done).astimezone() + if moment.date() != day: + return None + done = moment.hour * 60 + moment.minute + return min(ceiling(day), done + GRACE_MINUTES) + + +def _load_lift(state_file: Path, today: str) -> int: + if not state_file.exists(): + return 0 # no lift recorded yet: nothing to absorb + try: + raw = json.loads(state_file.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + _logger.warning( + "Grace state %s unreadable (%s); lift taken as 0", state_file, exc + ) + return 0 + if not isinstance(raw, dict) or raw.get("date") != today: + return 0 + return max(0, int(raw.get("lift", 0))) + + +def save_lift(lift: int, state_file: Path | None = None) -> None: + """Record how far the floor lifts today's shutdown above the earned time.""" + state_file = state_file or GRACE_STATE_FILE + payload = {"date": today_str(), "lift": max(0, lift)} + try: + state_file.write_text(json.dumps(payload), encoding="utf-8") + except OSError as exc: + _logger.warning("Could not record the grace lift in %s (%s)", state_file, exc) + + +def plan_absorb(minutes: int, state_file: Path | None = None) -> tuple[int, int]: + """How much of an ``minutes`` addition moves the config, and the lift left. + + Commit the lift with :func:`save_lift` only once the write succeeded. + """ + lift = _load_lift(state_file or GRACE_STATE_FILE, today_str()) + return max(0, minutes - lift), max(0, lift - minutes) + + +def _tonight(config: tuple[int, int, int], day: date) -> int: + mon_wed, thu_sun, _ = config + return mon_wed if day.weekday() in _MON_WED else thu_sun + + +class GraceFloorMixin: + """Lifts tonight's shutdown to the grace floor; composed into ShutdownMixin.""" + + if TYPE_CHECKING: + + def _read_shutdown_config(self) -> tuple[int, int, int] | None: ... + + def _write_shutdown_config( + self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = ... + ) -> bool: ... + + def _apply_grace_floor( + self, log_file: Path | None, *, first_done: float | None = None + ) -> bool: + """Lift tonight's shutdown to the grace floor if it sits below it. + + ``first_done`` overrides the ledgers' answer (the preview script's + simulations). Returns True when the config was raised on this call. + """ + day = date.fromisoformat(today_str()) + if sick_on(day): + _logger.info("Grace floor: skipped, today is a sick day") + return False + if first_done is None: + first_done = first_done_at(day, log_file, rest_day=is_rest_day(day)) + floor = grace_for(first_done, day) + if floor is None: + return False + with bonus_lock(GRACE_STATE_FILE): + config = self._read_shutdown_config() + if config is None: + _logger.warning("Grace floor: shutdown config unreadable; not applied") + return False + current = _tonight(config, day) + if current >= floor: + return False + mon_wed, thu_sun, morning = config + if not self._write_shutdown_config( + max(mon_wed, floor), max(thu_sun, floor), morning, restore=True + ): + _logger.warning("Grace floor: failed to write shutdown config") + return False + lift = _load_lift(GRACE_STATE_FILE, today_str()) + floor - current + save_lift(lift, GRACE_STATE_FILE) + _logger.info( + "Grace floor: shutdown %s -> %s (first task done + %d min).", + hhmm(current), + hhmm(floor), + GRACE_MINUTES, + ) + return True diff --git a/screen_locker/_log_mixin.py b/screen_locker/_log_mixin.py index 0c2419e..ad788a1 100644 --- a/screen_locker/_log_mixin.py +++ b/screen_locker/_log_mixin.py @@ -16,6 +16,7 @@ from screen_locker._bonus_lock import bonus_lock from screen_locker._day import today_str from screen_locker._log_io import load_workout_log from screen_locker._manual_workout import MANUAL_WORKOUT_TYPE, manual_sync_record_id +from screen_locker._workout_ledger import record_log_entry if TYPE_CHECKING: from collections.abc import Mapping @@ -162,6 +163,10 @@ def _write_signed_entry_locked( json.dump(logs, f, indent=2) except OSError as e: _logger.warning("Could not save workout log: %s", e) + else: + # The one chokepoint every credit passes, so the signed workout + # ledger (earned_time's view of the workout) can never miss one. + record_log_entry(date, entries) return RecordResult(appended=True, prior_entries=prior) diff --git a/screen_locker/_rest_day.py b/screen_locker/_rest_day.py new file mode 100644 index 0000000..84b4cf6 --- /dev/null +++ b/screen_locker/_rest_day.py @@ -0,0 +1,193 @@ +"""Rest days: declared AHEAD of time, at most two per ISO week, HMAC-signed. + +A declared rest day earns the workout's first-unit shutdown bonus, so the day +resolves as if its workout were done. Two rules keep it from being an escape: + +* **Ahead only.** A rest day must be declared on an earlier local day than + the one it names -- never "today", so it cannot be reached for at the + moment the shutdown bites. Checked when declaring AND again when reading + (``declared_on < day``), so a row written by any other route still fails. + The CLI refuses to run on a skewed clock (``check_clock_skew``), the one + way to make tomorrow look like today. +* **Two per ISO week**, counted by the rest day's own week over the verified + rows, oldest declaration first; a third row in a week is ignored on read. + +Each declaration is mirrored as a ``rest_day`` credit row in the workout +ledger (:mod:`screen_locker._workout_ledger`), which ``earned_time`` reads. + +Rows live in ``rest_days.json`` as ``{"entries": [...]}``, each signed with +``earned_time.entry_signature`` and the shared key, the same as every gate's +ledger: editing the file without the key breaks the row, and a broken row +earns nothing. + +Unlike a sick day -- decided on the day, at the lock screen, behind a +countdown, rate-limited over 7/30/90 days, and paid back as workout debt -- a +rest day is planned, costs no debt, and only touches shutdown time. It does +not stand down the screen lock or the weekly minimum (``freedays`` does that). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, date, datetime +import json +import logging +from typing import TYPE_CHECKING, Any + +import earned_time + +from screen_locker import _earned +from screen_locker._constants import REST_DAY_FILE, REST_DAYS_PER_ISO_WEEK +from screen_locker._workout_ledger import append_rows, rest_row + +if TYPE_CHECKING: + from pathlib import Path + +_logger = logging.getLogger(__name__) + +_KIND = "rest_day" + + +@dataclass(frozen=True) +class Declared: + """Outcome of :func:`declare`: ``ok`` and a sentence a human can act on.""" + + ok: bool + reason: str + + +def _key(key_file: Path) -> bytes | None: + try: + key = key_file.read_bytes().strip() + except OSError as exc: + _logger.warning("Cannot read the rest-day signing key %s (%s)", key_file, exc) + return None + return key or None + + +def _rows(path: Path) -> list[Any] | None: + """The file's ``entries``; ``[]`` if it does not exist, ``None`` if unreadable.""" + if not path.exists(): + return [] + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + _logger.warning("Rest-day file %s is unreadable (%s)", path, exc) + return None + rows = raw.get("entries") if isinstance(raw, dict) else None + if not isinstance(rows, list): + _logger.warning("Rest-day file %s has no entries array", path) + return None + return rows + + +def _iso_week(day: date) -> tuple[int, int]: + year, week, _ = day.isocalendar() + return year, week + + +def _valid(row: object, key: bytes) -> bool: + """A signed rest-day row, declared on an earlier day than it names.""" + if not isinstance(row, dict) or row.get("kind") != _KIND: + return False + if not earned_time.verified(row, key): + _logger.warning("Rest-day row %r fails its signature; ignored", row.get("day")) + return False + try: + return date.fromisoformat(str(row["declared_on"])) < date.fromisoformat( + str(row["day"]) + ) + except (KeyError, ValueError) as exc: + _logger.warning("Rest-day row %r is malformed (%s); ignored", row, exc) + return False + + +def rest_day_declarations( + path: Path | None = None, key_file: Path | None = None +) -> dict[date, float]: + """Every rest day that counts, with the unix time it was declared. + + Counted: verified, declared ahead, within the weekly cap (oldest first). + """ + key = _key(key_file or _earned.HMAC_KEY_FILE) + rows = _rows(path or REST_DAY_FILE) + if key is None or rows is None: + return {} + good = sorted( + (r for r in rows if _valid(r, key)), key=lambda r: float(r["declared_at"]) + ) + counted: dict[date, float] = {} + for row in good: + day = date.fromisoformat(str(row["day"])) + same_week = sum(1 for d in counted if _iso_week(d) == _iso_week(day)) + if day not in counted and same_week < REST_DAYS_PER_ISO_WEEK: + counted[day] = float(row["declared_at"]) + return counted + + +def rest_days(path: Path | None = None, key_file: Path | None = None) -> set[date]: + """Every rest day that counts (see :func:`rest_day_declarations`).""" + return set(rest_day_declarations(path, key_file)) + + +def is_rest_day( + day: date, path: Path | None = None, key_file: Path | None = None +) -> bool: + """Whether ``day`` is a declared rest day (``False`` when unverifiable).""" + return day in rest_days(path, key_file) + + +def declare( + day: date, + *, + now: datetime | None = None, + path: Path | None = None, + key_file: Path | None = None, + ledger: Path | None = None, +) -> Declared: + """Sign and append a rest day for ``day``, if the rules allow it. + + Paths default at call time to ``REST_DAY_FILE`` and the key ``_earned`` + verifies every ledger with, so the test suite's redirects apply. + """ + path = path or REST_DAY_FILE + key_file = key_file or _earned.HMAC_KEY_FILE + moment = (now or datetime.now(tz=UTC)).astimezone() + if day <= moment.date(): + return Declared( + ok=False, + reason=f"{day} is not in the future: rest days are declared ahead", + ) + existing = rest_days(path, key_file) + if day in existing: + return Declared(ok=True, reason=f"{day} is already a rest day") + used = sum(1 for d in existing if _iso_week(d) == _iso_week(day)) + if used >= REST_DAYS_PER_ISO_WEEK: + return Declared( + ok=False, + reason=f"ISO week {_iso_week(day)[1]} already has " + f"{REST_DAYS_PER_ISO_WEEK} rest days", + ) + key = _key(key_file) + rows = _rows(path) + if key is None or rows is None: + return Declared(ok=False, reason="rest-day file or signing key unreadable") + row: dict[str, object] = { + "kind": _KIND, + "day": day.isoformat(), + "declared_on": moment.date().isoformat(), + "declared_at": moment.timestamp(), + } + row["hmac"] = earned_time.entry_signature(row, key) + rows.append(row) + try: + path.write_text(json.dumps({"entries": rows}, indent=2), encoding="utf-8") + except OSError as exc: + _logger.warning("Could not write rest day to %s (%s)", path, exc) + return Declared(ok=False, reason=f"could not write {path}: {exc}") + append_rows([rest_row(day, moment.timestamp())], ledger, key_file) + left = REST_DAYS_PER_ISO_WEEK - used - 1 + return Declared( + ok=True, + reason=f"{day} is a rest day ({left} left in ISO week {_iso_week(day)[1]})", + ) diff --git a/screen_locker/_rest_day_cli.py b/screen_locker/_rest_day_cli.py new file mode 100644 index 0000000..4430fd9 --- /dev/null +++ b/screen_locker/_rest_day_cli.py @@ -0,0 +1,81 @@ +"""``--declare-rest-day``: plan a rest day from the command line. + + python -m screen_locker.screen_lock --declare-rest-day 2026-10-12 + python -m screen_locker.screen_lock --declare-rest-day tomorrow + python -m screen_locker.screen_lock --declare-rest-day --list + +The clock is checked against NTP first and an unconfirmable clock is a +refusal, not a pass: unlike the lock's own check (which fails open when NTP +is unreachable), a wound-back clock with the network off is exactly how +"today" would be passed off as tomorrow. +""" + +from __future__ import annotations + +import argparse +from datetime import date, timedelta +import logging +import sys +from typing import TYPE_CHECKING + +from screen_locker._constants import MAX_CLOCK_SKEW_SECONDS +from screen_locker._day import today_str +from screen_locker._rest_day import declare, rest_days +from screen_locker._time_check import _query_ntp_offset + +if TYPE_CHECKING: + from collections.abc import Sequence + +_logger = logging.getLogger(__name__) + + +def _say(message: str) -> None: + sys.stdout.write(f"{message}\n") + + +def _parse_day(raw: str) -> date: + today = date.fromisoformat(today_str()) + if raw == "tomorrow": + return today + timedelta(days=1) + return date.fromisoformat(raw) + + +def _clock_confirmed() -> str | None: + """``None`` if NTP confirms the clock, else why the declaration is refused.""" + offset = _query_ntp_offset() + if offset is None: + return "NTP is unreachable, so the clock cannot be confirmed; try online" + if abs(offset) > MAX_CLOCK_SKEW_SECONDS: + return f"the clock is off by {abs(offset):.0f}s from NTP" + return None + + +def run_declare_rest_day(argv: Sequence[str]) -> int: + """Declare (or list) rest days. Returns a process exit code.""" + parser = argparse.ArgumentParser( + prog="screen-locker --declare-rest-day", + description="Declare a FUTURE rest day (max 2 per ISO week).", + ) + parser.add_argument("day", nargs="?", help="YYYY-MM-DD or 'tomorrow'") + parser.add_argument("--list", action="store_true", help="list rest days") + args = parser.parse_args(list(argv)) + if args.list or args.day is None: + days = sorted(rest_days()) + _say("\n".join(d.isoformat() for d in days) or "no rest days declared") + return 0 + try: + day = _parse_day(args.day) + except ValueError as exc: + _logger.warning("Unparsable rest day %r: %s", args.day, exc) + _say(f"error: {exc}") + return 2 + refusal = _clock_confirmed() + if refusal is not None: + _logger.warning("Rest day refused: %s", refusal) + _say(f"refused: {refusal}") + return 1 + result = declare(day) + if not result.ok: + _logger.warning("Rest day refused: %s", result.reason) + _say(result.reason if result.ok else f"refused: {result.reason}") + return 0 if result.ok else 1 diff --git a/screen_locker/_runnerup_backfill.py b/screen_locker/_runnerup_backfill.py index 738f187..2b546cf 100644 --- a/screen_locker/_runnerup_backfill.py +++ b/screen_locker/_runnerup_backfill.py @@ -12,9 +12,10 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from screen_locker._log_mixin import write_signed_entry +from screen_locker._walk_tcx import walk_day, walk_shortfall if TYPE_CHECKING: from collections.abc import Callable @@ -27,6 +28,24 @@ if TYPE_CHECKING: _logger = logging.getLogger(__name__) +def _record( + log_file: Path, + date_str: str, + workout_data: dict[str, Any], + on_ingested: OnIngestedCallback | None, +) -> bool: + """Append one verified RunnerUp entry and credit it; False if a duplicate.""" + result = write_signed_entry(log_file, date_str, workout_data) + if not result.appended: + return False + _logger.info( + "Auto-filled RunnerUp entry for %s: %s", date_str, workout_data["source"] + ) + if on_ingested is not None: + on_ingested(workout_data, result.prior_entries) + return True + + class RunnerUpBackfillMixin: """Fills the log from RunnerUp exports for a date or a whole week.""" @@ -53,10 +72,15 @@ class RunnerUpBackfillMixin: Returns True if a new verified entry was appended for ``date_str``. """ + walks: list[dict[str, Any]] = [] for remote in self._find_runnerup_exports_for_date(date_str): data = self._pull_and_parse_tcx(remote) if data is None: continue + if data.get("walking"): + # Never one walk at a time: a day's walks are summed below. + walks.append(data) + continue status, msg = self._validate_runnerup_data(data) if status != "verified": _logger.warning( @@ -72,15 +96,43 @@ class RunnerUpBackfillMixin: "source": f"Auto-scanned: {msg}", "distance_km": round(data["distance_m"] / 1000, 2), "duration_minutes": round(data["duration_seconds"] / 60, 1), + "completed_at": data.get("ended_at"), } - result = write_signed_entry(log_file, date_str, workout_data) - if not result.appended: - return False - _logger.info("Auto-filled RunnerUp entry for %s: %s", date_str, msg) - if on_ingested is not None: - on_ingested(workout_data, result.prior_entries) - return True - return False + return _record(log_file, date_str, workout_data, on_ingested) + return self._try_fill_walks(date_str, walks, log_file, on_ingested) + + @staticmethod + def _try_fill_walks( + date_str: str, + walks: list[dict[str, Any]], + log_file: Path, + on_ingested: OnIngestedCallback | None, + ) -> bool: + """Append the day's walking as its RunnerUp workout, once it adds up. + + Same type and so the same ``runnerup_verified:{date}`` slot as a run: + a walk and a run on one day are one RunnerUp credit, exactly as two + runs already were. + """ + if not walks: + return False + day = walk_day(walks) + if not day.qualifies: + _logger.warning( + "RunnerUp walks for %s do not count yet: %s", + date_str, + walk_shortfall(day), + ) + return False + workout_data = { + "type": "runnerup_verified", + "activity": "walking", + "source": f"Auto-scanned: {day.message()}", + "distance_km": round(day.distance_km, 2), + "duration_minutes": round(day.moving_minutes, 1), + "completed_at": day.completed_at, + } + return _record(log_file, date_str, workout_data, on_ingested) def _scan_and_fill_week_runnerup( self, log_file: Path, on_ingested: OnIngestedCallback | None = None diff --git a/screen_locker/_runnerup_tcx.py b/screen_locker/_runnerup_tcx.py index 504a8a9..0727478 100644 --- a/screen_locker/_runnerup_tcx.py +++ b/screen_locker/_runnerup_tcx.py @@ -17,6 +17,8 @@ import tempfile from typing import Any import xml.etree.ElementTree as ET +from screen_locker._walk_tcx import is_walking_export, track_summary + # TCX XML namespace used by Garmin/RunnerUp. _TCX_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" @@ -48,8 +50,9 @@ class RunnerUpTcxMixin: phone path and goes through ``adb pull``. Returns the activity dict or None. """ + name = Path(remote_path).name if Path(remote_path).is_file(): - return self._parse_tcx(remote_path) + return _mark_walk(self._parse_tcx(remote_path), name) tmp_dir = tempfile.mkdtemp(prefix="runnerup_tcx_") local_path = str(Path(tmp_dir) / "activity.tcx") try: @@ -57,7 +60,7 @@ class RunnerUpTcxMixin: if not ok or not Path(local_path).exists(): _logger.info("Failed to pull TCX file: %s", remote_path) return None - return self._parse_tcx(local_path) + return _mark_walk(self._parse_tcx(local_path), name) finally: shutil.rmtree(tmp_dir, ignore_errors=True) @@ -101,4 +104,13 @@ class RunnerUpTcxMixin: "sport": sport_int, "duration_seconds": int(total_seconds), "distance_m": total_distance, + "walking": is_walking_export(Path(tcx_path).name, sport_str), + **track_summary(activity), } + + +def _mark_walk(data: dict[str, Any] | None, name: str) -> dict[str, Any] | None: + """Flag a walk by its ORIGINAL file name: an adb pull renames the copy.""" + if data is not None: + data["walking"] = data["walking"] or is_walking_export(name, "") + return data diff --git a/screen_locker/_runnerup_verification.py b/screen_locker/_runnerup_verification.py index 7549973..b60d153 100644 --- a/screen_locker/_runnerup_verification.py +++ b/screen_locker/_runnerup_verification.py @@ -23,6 +23,7 @@ from screen_locker._runnerup_backfill import RunnerUpBackfillMixin from screen_locker._runnerup_db import RunnerUpDbMixin from screen_locker._runnerup_tcx import RunnerUpTcxMixin from screen_locker._time_check import check_clock_skew +from screen_locker._walk_tcx import walk_day, walk_shortfall _logger = logging.getLogger(__name__) @@ -121,16 +122,26 @@ class RunnerUpVerificationMixin( return None # Try each file; return the best result (verified > validation error). + # Walks never qualify one file at a time: they are summed below. best: tuple[str, str] | None = None + walks: list[dict[str, Any]] = [] for remote in exports: data = self._pull_and_parse_tcx(remote) if data is None: continue + if data.get("walking"): + walks.append(data) + continue status, msg = self._validate_runnerup_data(data) if status == "verified": return status, msg if best is None: best = (status, msg) + if walks: + day = walk_day(walks) + if day.qualifies: + return "verified", day.message() + best = best or ("too_short", walk_shortfall(day)) # All files found but none passed validation. return best or ( diff --git a/screen_locker/_shutdown.py b/screen_locker/_shutdown.py index ae7b625..76d7c65 100644 --- a/screen_locker/_shutdown.py +++ b/screen_locker/_shutdown.py @@ -2,18 +2,21 @@ from __future__ import annotations +from datetime import date import logging import subprocess from typing import TYPE_CHECKING +import earned_time + from screen_locker._constants import ( ADJUST_SHUTDOWN_SCRIPT, SHUTDOWN_CONFIG_FILE, ) from screen_locker._day import today_str -from screen_locker._earned import hhmm, span +from screen_locker._earned import ceiling, first_minutes, hhmm, span +from screen_locker._grace_floor import GraceFloorMixin, plan_absorb, save_lift from screen_locker._shutdown_sick_state import SickDayStateMixin -from screen_locker._workout_credit import FIRST_WORKOUT_BONUS_MINUTES if TYPE_CHECKING: from pathlib import Path @@ -29,8 +32,6 @@ _MINUTES_PER_HOUR = 60 # A sick day moves shutdown an hour earlier, but never before 18:00. _SICK_DAY_STEP = 60 _SICK_DAY_FLOOR = 18 * _MINUTES_PER_HOUR -# The workout reward stops at 23:00 (adjust_shutdown_schedule.sh's ceiling). -_WORKOUT_CEILING = 23 * _MINUTES_PER_HOUR # Extra bonuses may name midnight; the helper still clamps them to 23:00. _MIDNIGHT = 24 * _MINUTES_PER_HOUR @@ -69,7 +70,7 @@ def read_shutdown_config(path: Path) -> tuple[int, int, int] | None: return mon_wed, thu_sun, morning_end -class ShutdownMixin(SickDayStateMixin): +class ShutdownMixin(GraceFloorMixin, SickDayStateMixin): """Mixin providing shutdown schedule adjustment functionality.""" def _apply_earlier_shutdown(self, today: str) -> bool: @@ -107,24 +108,29 @@ class ShutdownMixin(SickDayStateMixin): return False def _adjust_shutdown_time_later(self) -> bool: - """Push shutdown later by the first workout's reward, capped at 23:00. + """Push shutdown later by the first workout's reward, capped at the ceiling. Returns True if successful, False otherwise. """ + day = date.fromisoformat(today_str()) + bonus, lift = plan_absorb(first_minutes(earned_time.WORKOUT, day)) try: config_values = self._read_shutdown_config() if config_values is None: return False mon_wed, thu_sun, morning_end = config_values - return self._write_shutdown_config( - min(_WORKOUT_CEILING, mon_wed + FIRST_WORKOUT_BONUS_MINUTES), - min(_WORKOUT_CEILING, thu_sun + FIRST_WORKOUT_BONUS_MINUTES), + ok = self._write_shutdown_config( + min(ceiling(day), mon_wed + bonus), + min(ceiling(day), thu_sun + bonus), morning_end, restore=True, ) except (OSError, ValueError) as e: _logger.warning("Failed to adjust shutdown time for workout: %s", e) return False + if ok: + save_lift(lift) + return ok def _adjust_shutdown_time_by(self, extra_minutes: int) -> bool: """Push shutdown later by *extra_minutes*, capped at 24:00 (midnight). @@ -135,14 +141,16 @@ class ShutdownMixin(SickDayStateMixin): Returns True if successful, False otherwise. """ + # Paid out of the grace lift first, so the floor is never stacked on. + moved, lift = plan_absorb(extra_minutes) try: config_values = self._read_shutdown_config() if config_values is None: return False mw, ts, morning = config_values - return self._write_shutdown_config( - min(_MIDNIGHT, mw + extra_minutes), - min(_MIDNIGHT, ts + extra_minutes), + ok = self._write_shutdown_config( + min(_MIDNIGHT, mw + moved), + min(_MIDNIGHT, ts + moved), morning, restore=True, ) @@ -151,6 +159,9 @@ class ShutdownMixin(SickDayStateMixin): "Failed to adjust shutdown time by %s: %s", span(extra_minutes), e ) return False + if ok: + save_lift(lift) + return ok def _read_shutdown_config(self) -> tuple[int, int, int] | None: """Read shutdown config as (mon_wed, thu_sun, morning_end) minutes, or None.""" diff --git a/screen_locker/_shutdown_base.py b/screen_locker/_shutdown_base.py index 24ab951..e35fa32 100644 --- a/screen_locker/_shutdown_base.py +++ b/screen_locker/_shutdown_base.py @@ -37,12 +37,14 @@ from screen_locker._bonus_lock import bonus_lock from screen_locker._day import today_str from screen_locker._earned import ( earned_today, - flat_answers, + first_minutes, flat_earners, hhmm, span, ) +from screen_locker._grace_floor import save_lift from screen_locker._log_io import load_workout_log +from screen_locker._shutdown_target import gather from screen_locker._weekly_check import count_day_credits if TYPE_CHECKING: @@ -153,16 +155,9 @@ def _reset( if _load_state(state_file).get("last_reset_date") == today: return False - answers: dict[str, int | bool | None] = dict(flat_answers()) - answers[earned_time.WORKOUT.name] = ( - today_credit_count(log_file, today) if log_file is not None else 0 - ) - # The registry is passed explicitly: flat_answers iterated this same - # tuple, so the earners asked and the earners summed can never differ. - day = earned_time.resolve( - answers, datetime.now().astimezone().date(), earned_time.EARNERS - ) - target = day.shutdown_minutes + derived = gather(date.fromisoformat(today), log_file) + day = derived.resolution + target = derived.minutes # Preserve the morning end from the live config. config = mixin._read_shutdown_config() @@ -174,6 +169,8 @@ def _reset( return False _clear_sick_day_state(sick_day_state_file) + # Later additions are paid out of the grace lift first (see _grace_floor). + save_lift(derived.lift) # The flat bonuses are stamped here too: the reset already included them, # so the live pass below must not add them a second time today. @@ -187,10 +184,12 @@ def _reset( f"{span(t.shutdown_minutes)} {t.earner.label}" for t in day.terms ) _logger.info( - "Daily base reset: %s (base %s + %s already earned today).", + "Daily base reset: %s (base %s + %s already earned today%s%s).", hhmm(target), hhmm(day.base.shutdown_minutes), earned, + "; rest day" if derived.rest_day else "", + f"; grace floor {hhmm(derived.grace)}" if derived.grace is not None else "", ) return True @@ -223,7 +222,7 @@ def _apply_flat_bonus( return False if not answer: return False - minutes = earner.shutdown_minutes + minutes = first_minutes(earner, date.fromisoformat(today)) if not mixin._adjust_shutdown_time_by(minutes): _logger.warning("%s bonus: failed to write shutdown config.", earner.label) return False @@ -242,3 +241,5 @@ def apply_flat_bonuses_if_new(state_file: Path, mixin: object) -> None: with bonus_lock(state_file): for earner in flat_earners(): _apply_flat_bonus(state_file, mixin, earner) + # Last, so the floor compares against everything earned so far. + mixin._apply_grace_floor(getattr(mixin, "log_file", None)) diff --git a/screen_locker/_shutdown_target.py b/screen_locker/_shutdown_target.py new file mode 100644 index 0000000..b7f7ecd --- /dev/null +++ b/screen_locker/_shutdown_target.py @@ -0,0 +1,133 @@ +"""Tonight's shutdown, derived from scratch: earned time, rest day, grace floor. + +:func:`derive` is pure, and is the one formula behind both the daily reset +(:mod:`screen_locker._shutdown_base`) and the read-only preview +(``scripts/preview_shutdown.py``), so a dry run exercises the real sum:: + + earned = earned_time.resolve(answers) # base + every earner, capped + target = min(ceiling, max(earned, grace floor)) + +A declared rest day answers the workout with at least one unit, i.e. its +first-unit bonus; a real workout on top of it earns only what a further unit +earns. :func:`gather` reads today's real inputs; it writes nothing. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import earned_time + +from screen_locker._earned import ceiling, flat_answers +from screen_locker._grace_floor import first_done_at, grace_for, sick_on +from screen_locker._log_io import load_workout_log +from screen_locker._rest_day import is_rest_day +from screen_locker._weekly_check import count_day_credits + +if TYPE_CHECKING: + from collections.abc import Mapping + from datetime import date + from pathlib import Path + + +@dataclass(frozen=True) +class Target: + """One day's derived shutdown and what made it. + + Attributes: + resolution: ``earned_time``'s sum (base + terms, capped). + rest_day: Whether the day is a declared rest day. + first_done_at: Unix time the day's first earner was done, if any. + grace: The grace floor in minutes, or ``None``. + minutes: The shutdown to write, minutes after midnight. + """ + + resolution: earned_time.Resolution + rest_day: bool + first_done_at: float | None + grace: int | None + minutes: int + + @property + def earned(self) -> int: + """The earned shutdown, before the grace floor.""" + return self.resolution.shutdown_minutes + + @property + def lift(self) -> int: + """How far the grace floor lifts the earned shutdown.""" + return self.minutes - self.earned + + +def workout_units(count: int, *, rest_day: bool) -> int: + """A rest day stands in for the first workout, never for a further one.""" + return max(count, 1) if rest_day else count + + +@dataclass(frozen=True) +class DayInputs: + """Everything one day's shutdown is derived from. + + Attributes: + day: The day. + flat: Each flat earner's answer (``None`` = could not check). + workout_credits: The log's distinct workout credits. + rest_day: Declared rest day. + first_done: Unix time the day's first earner was done, if any. + sick_day: Sick day: the sick-day shutdown wins, no grace floor. + """ + + day: date + flat: Mapping[str, bool | None] + workout_credits: int = 0 + rest_day: bool = False + first_done: float | None = None + sick_day: bool = False + + +def derive(inputs: DayInputs) -> Target: + """The pure formula: earned time, then the grace floor, then the ceiling.""" + day = inputs.day + answers: dict[str, int | bool | None] = dict(inputs.flat) + answers[earned_time.WORKOUT.name] = workout_units( + inputs.workout_credits, rest_day=inputs.rest_day + ) + # The registry is passed explicitly: flat_answers iterated this same + # tuple, so the earners asked and the earners summed can never differ. + resolution = earned_time.resolve(answers, day, earned_time.EARNERS) + grace = None if inputs.sick_day else grace_for(inputs.first_done, day) + minutes = resolution.shutdown_minutes + if grace is not None: + minutes = min(ceiling(day), max(minutes, grace)) + return Target( + resolution=resolution, + rest_day=inputs.rest_day, + first_done_at=inputs.first_done, + grace=grace, + minutes=minutes, + ) + + +def day_credit_count(log_file: Path | None, day: date) -> int: + """How many distinct workouts ``day``'s log counts (0 without a log).""" + if log_file is None: + return 0 + iso = day.isoformat() + entries = load_workout_log(log_file).get(iso, []) + return count_day_credits(iso, [e for e in entries if isinstance(e, dict)]) + + +def gather(day: date, log_file: Path | None) -> Target: + """Derive ``day``'s target from the real ledgers, log and rest days.""" + rest_day = is_rest_day(day) + return derive( + DayInputs( + day=day, + flat=flat_answers(), + workout_credits=day_credit_count(log_file, day), + rest_day=rest_day, + first_done=first_done_at(day, log_file, rest_day=rest_day), + sick_day=sick_on(day), + ) + ) diff --git a/screen_locker/_status_view_verify.py b/screen_locker/_status_view_verify.py index ecd90b0..173c3af 100644 --- a/screen_locker/_status_view_verify.py +++ b/screen_locker/_status_view_verify.py @@ -10,12 +10,15 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module from dataclasses import dataclass +from datetime import date from typing import TYPE_CHECKING -from screen_locker._earned import span +import earned_time + +from screen_locker._day import today_str +from screen_locker._earned import extra_minutes, span from screen_locker._status_data import gather_status from screen_locker._weekly_check import WEEKLY_WORKOUT_MINIMUM, count_weekly_workouts -from screen_locker._workout_credit import EXTRA_WORKOUT_BONUS_MINUTES if TYPE_CHECKING: from collections.abc import Callable @@ -84,7 +87,9 @@ def _backfill_week_and_apply_bonus(verifier: ScreenLocker) -> str | None: if sl_filled: new_count = count_weekly_workouts(verifier.log_file) extra = max(0, new_count - max(WEEKLY_WORKOUT_MINIMUM, prev_count + filled)) - bonus = extra * EXTRA_WORKOUT_BONUS_MINUTES + # Day-aware: 0 per further unit on earned_time's ladder (2026-10-10). + today = date.fromisoformat(today_str()) + bonus = extra * extra_minutes(earned_time.WORKOUT, today) if bonus > 0: verifier._adjust_shutdown_time_by(bonus) filled += sl_filled diff --git a/screen_locker/_walk_tcx.py b/screen_locker/_walk_tcx.py new file mode 100644 index 0000000..61a25c5 --- /dev/null +++ b/screen_locker/_walk_tcx.py @@ -0,0 +1,144 @@ +"""Walking in RunnerUp TCX exports: which files are walks, and their moving time. + +A day's walking counts as the workout once its walks -- usually separate +commute legs, one file each -- add up to ``WALK_MIN_MOVING_MINUTES`` of +*moving* time. + +**Which files are walks.** RunnerUp's WebDAV and File synchronisers both +export with ``ExportOptions.getDefault()``, whose TCX writer knows only +Running/Biking and writes ``Sport="Other"`` for a walk. The sport survives +only in the file name (``Sport.TapiriikType()``): ``..._Walking.tcx``. The +Strava export variant writes ``Sport="Walking"``, so that counts too. +``Other`` alone is not a walk: it is every sport RunnerUp cannot name. + +**Moving time, not elapsed.** A lap's ``TotalTimeSeconds`` is wall-clock: one +real lap reads 101 min for 1 km of standing about. Moving time is summed over +consecutive trackpoint pairs whose speed is walking speed +(``WALK_SPEED_MPS``): standing still is too slow, a tram or a bus too fast. +A pair further apart than ``WALK_MAX_GAP_SECONDS`` is a GPS dropout whose +motion is unknown and earns nothing -- the walk is never credited for time +nobody saw. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +import itertools +import logging +from typing import TYPE_CHECKING, Any + +from screen_locker._constants import ( + WALK_MAX_GAP_SECONDS, + WALK_MIN_MOVING_MINUTES, + WALK_SPEED_MPS, +) + +if TYPE_CHECKING: + from collections.abc import Iterable + import xml.etree.ElementTree as ET + +_logger = logging.getLogger(__name__) + +_TCX_NS = "{http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2}" +_WALKING = "Walking" + + +def is_walking_export(name: str, sport: str) -> bool: + """Whether a TCX export is a walk, from its file name or its Sport tag.""" + return name.endswith(f"_{_WALKING}.tcx") or sport == _WALKING + + +def _stamp(raw: str | None) -> float | None: + if not raw: + return None + try: + return datetime.fromisoformat(raw).timestamp() + except ValueError: + _logger.warning("TCX trackpoint has an unparsable time %r; skipped", raw) + return None + + +def _trackpoints(activity: ET.Element) -> list[tuple[float, float]]: + """``(unix time, cumulative metres)`` of every usable trackpoint, in order.""" + points: list[tuple[float, float]] = [] + for point in activity.iter(f"{_TCX_NS}Trackpoint"): + when = _stamp(point.findtext(f"{_TCX_NS}Time")) + metres = point.findtext(f"{_TCX_NS}DistanceMeters") + if when is None or not metres: + continue + points.append((when, float(metres))) + points.sort() + return points + + +def track_summary(activity: ET.Element) -> dict[str, Any]: + """Moving seconds and end time of one activity, from its trackpoints.""" + points = _trackpoints(activity) + slow, fast = WALK_SPEED_MPS + moving = 0.0 + for (t0, d0), (t1, d1) in itertools.pairwise(points): + gap = t1 - t0 + if 0 < gap <= WALK_MAX_GAP_SECONDS and slow <= (d1 - d0) / gap <= fast: + moving += gap + return { + "activity_id": activity.findtext(f"{_TCX_NS}Id") or "", + "moving_seconds": moving, + "ended_at": points[-1][0] if points else None, + } + + +@dataclass(frozen=True) +class WalkDay: + """A day's walks, summed. + + Attributes: + legs: Distinct walks (deduplicated by the activity's ``Id``: the + WebDAV copy and the phone's copy are the same walk). + moving_minutes: Their summed moving time. + distance_km: Their summed distance. + completed_at: Unix time the walk that crossed the bar ended -- when + the workout was actually done -- or ``None`` if it never did. + """ + + legs: int + moving_minutes: float + distance_km: float + completed_at: float | None + + @property + def qualifies(self) -> bool: + """Whether the day's walking counts as its workout.""" + return self.moving_minutes >= WALK_MIN_MOVING_MINUTES + + def message(self) -> str: + """The verification line, in the RunnerUp verifier's own style.""" + return ( + f"Walking: {self.legs} walk(s), {self.distance_km:.1f} km, " + f"{self.moving_minutes:.0f} min moving" + ) + + +def walk_shortfall(day: WalkDay) -> str: + """Why a day's walking did not count yet.""" + return f"{day.message()} -- need {WALK_MIN_MOVING_MINUTES}+ min moving" + + +def walk_day(walks: Iterable[dict[str, Any]]) -> WalkDay: + """Sum parsed walk exports (``_parse_tcx`` dicts) into one :class:`WalkDay`.""" + unique: dict[str, dict[str, Any]] = {} + for walk in walks: + unique.setdefault(str(walk.get("activity_id") or id(walk)), walk) + ordered = sorted(unique.values(), key=lambda w: w.get("ended_at") or 0.0) + moving = 0.0 + completed_at: float | None = None + for walk in ordered: + moving += float(walk.get("moving_seconds", 0.0)) + if completed_at is None and moving / 60 >= WALK_MIN_MOVING_MINUTES: + completed_at = walk.get("ended_at") + return WalkDay( + legs=len(ordered), + moving_minutes=moving / 60, + distance_km=sum(float(w.get("distance_m", 0.0)) for w in ordered) / 1000, + completed_at=completed_at, + ) diff --git a/screen_locker/_workout_credit.py b/screen_locker/_workout_credit.py index 7309243..05c8863 100644 --- a/screen_locker/_workout_credit.py +++ b/screen_locker/_workout_credit.py @@ -10,12 +10,16 @@ miss (see ``project-lock-disabled-pending-manual-log`` memory). from __future__ import annotations from dataclasses import dataclass +from datetime import date import logging from typing import TYPE_CHECKING import earned_time from screen_locker import _sick_tracker +from screen_locker._day import today_str +from screen_locker._earned import extra_minutes +from screen_locker._rest_day import is_rest_day from screen_locker._weekly_check import ( COUNTED_WORKOUT_TYPES, PC_WORKOUT_TYPE, @@ -30,6 +34,8 @@ _logger = logging.getLogger(__name__) # The shutdown reward for a day's workouts, from the shared registry # (``earned_time.WORKOUT``): the first counted workout pushes shutdown later by # FIRST_WORKOUT_BONUS_MINUTES, every further one by EXTRA_WORKOUT_BONUS_MINUTES. +# Those are the pre-ladder fields; from earned_time's ladder (2026-10-10) the +# live paths ask the day-aware _earned.first_minutes / extra_minutes instead. # The live credit paths below and the daily base reset (``_shutdown_base``) # both derive from the registry, so a day's earned time can be recomputed # from the log instead of being lost when the credit landed before the reset. @@ -124,11 +130,23 @@ class WorkoutCreditMixin: shutdown_adjusted = False extra_bonus_delta = 0 - if first_counted_today: + day = date.fromisoformat(today_str()) + # A rest day already paid the first unit (the daily reset counts it): + # the first real workout on it earns nothing, a second one what a + # further unit earns -- the reset's own max(credits, 1) rule. + rest_paid = first_counted_today and is_rest_day(day) + if rest_paid: + _logger.info("Rest day: its first-unit bonus is already in tonight's time") + extra = extra_minutes(earned_time.WORKOUT, day) + if first_counted_today and not rest_paid: shutdown_adjusted = self._try_adjust_shutdown_for_workout() - elif self.workout_data.get("type") in COUNTED_WORKOUT_TYPES: + elif ( + not first_counted_today + and extra + and self.workout_data.get("type") in COUNTED_WORKOUT_TYPES + ): old_cfg = self._read_shutdown_config() - if old_cfg and self._adjust_shutdown_time_by(EXTRA_WORKOUT_BONUS_MINUTES): + if old_cfg and self._adjust_shutdown_time_by(extra): new_cfg = self._read_shutdown_config() if new_cfg: extra_bonus_delta = new_cfg[1] - old_cfg[1] diff --git a/screen_locker/_workout_ledger.py b/screen_locker/_workout_ledger.py new file mode 100644 index 0000000..411f875 --- /dev/null +++ b/screen_locker/_workout_ledger.py @@ -0,0 +1,238 @@ +"""The workout ledger: signed credit rows, the workout's side of ``earned_time``. + +Every other earner publishes its credits as HMAC-signed rows in a ledger that +``earned_time`` reads (``first_credit_at``, ``credit_units``). This is the +workout's: ``~/.local/share/workout_locker/ledger.json``, one row per + +* workout credit slot in ``log.json`` -- RunnerUp (a run, or the day's walks + summed), StrongLifts (phone or PC session), manual, any lock-screen or + verify path -- appended by the one log write chokepoint + (``_log_mixin.write_signed_entry``), so no path can credit without it; and +* declared rest day -- appended when it is declared. + +Row:: + + {"kind": "credit", "entry_id": ..., "day": "YYYY-MM-DD", + "detail": {"completed_at": "", "source": "runnerup_tcx" | + "stronglifts" | "manual" | "rest_day", ["declared_at": ...]}, + "hmac": ...} + +``credit_units(WORKOUT)`` therefore equals the log's own credit count: one row +per ``credit_key`` slot, idempotent on ``entry_id`` (the slot's first entry's +``workout_id``), so a re-scan or a second ingestion path writes nothing. +``completed_at`` is the activity's real end where the log knows it +(:func:`entry_done_at`). A rest day's is local midnight at the start of that +day, and its ``declared_at`` (earlier, by construction) proves it was planned. +""" + +from __future__ import annotations + +from datetime import date, datetime, time +import json +import logging +import os +from pathlib import Path +import tempfile +from typing import TYPE_CHECKING, Any + +import earned_time + +from screen_locker import _earned +from screen_locker._bonus_lock import bonus_lock +from screen_locker._constants import WORKOUT_LEDGER_RELATIVE +from screen_locker._weekly_check import credit_key, day_workout_index + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping, Sequence + +_logger = logging.getLogger(__name__) + +# The ledger's ``detail.source`` per log type; anything else keeps its type. +_SOURCES = { + "runnerup_verified": "runnerup_tcx", + "phone_verified": "stronglifts", + "pc_workout_verified": "stronglifts", + "manual_workout": "manual", +} +_HHMM = len("HH:MM") + + +def ledger_path() -> Path: + """Where the workout ledger lives (redirected with ``_earned.LEDGER_HOME``).""" + return _earned.LEDGER_HOME / WORKOUT_LEDGER_RELATIVE + + +def _as_stamp(raw: object, day: date) -> float | None: + """A unix time from a number, a numeric string, ISO, or ``HH:MM`` on ``day``.""" + if isinstance(raw, int | float): + return float(raw) + text = str(raw or "").strip() + if not text: + return None + try: + if text.replace(".", "", 1).isdigit(): + return float(text) + if len(text) <= _HHMM: + hours, _, minutes = text.partition(":") + clock = time(int(hours), int(minutes)) + return datetime.combine(day, clock).astimezone().timestamp() + return datetime.fromisoformat(text).astimezone().timestamp() + except ValueError: + _logger.warning("Unusable workout time %r; ignored", raw) + return None + + +def entry_done_at(entry: Mapping[str, Any], day: date) -> float | None: + """When a logged workout ended; the moment it was logged as a last resort.""" + real = _real_end(entry, day) + return real if real is not None else _as_stamp(entry.get("timestamp"), day) + + +def _real_end(entry: Mapping[str, Any], day: date) -> float | None: + """When a logged workout actually ended, as unix seconds. + + In order: the RunnerUp TCX end (``completed_at``), a manual workout's + ``end_time``, a StrongLifts session's start (``sync_record_id``) plus its + ``duration_minutes``; ``None`` when the entry records no end of its own. + """ + data = entry.get("workout_data") + data = data if isinstance(data, dict) else {} + for raw in (data.get("completed_at"), data.get("end_time")): + stamp = _as_stamp(raw, day) + if stamp is not None: + return stamp + if data.get("type") == "pc_workout_verified": + start = _as_stamp(data.get("sync_record_id"), day) + try: + minutes = float(str(data.get("duration_minutes", ""))) + except ValueError: + _logger.warning("Session on %s has no usable duration", day) + minutes = None + if start is not None and minutes is not None: + return start + minutes * 60 + return None + + +def credit_row( + day: str, entries: Sequence[Mapping[str, Any]], index: int +) -> dict[str, Any] | None: + """The unsigned row for ``entries[index]``; ``None`` if it earns no credit. + + One row per CREDIT SLOT (``_weekly_check.credit_key``), not per entry: the + two StrongLifts ingestion paths and a synced copy of an entry share their + slot, and so share the ``entry_id`` -- the slot's first entry's own + ``workout_id`` -- and the second never writes a row. + """ + siblings = day_workout_index(entries) + slot = credit_key(day, index, entries[index], siblings) + if slot is None: + return None + members = [ + e for i, e in enumerate(entries) if credit_key(day, i, e, siblings) == slot + ] + owner = members[0] + data = owner.get("workout_data") + wtype = str(data.get("type")) if isinstance(data, dict) else "" + # The slot's real end, from whichever member recorded one (the PC session + # of a StrongLifts slot knows its duration, the phone copy does not). + when = date.fromisoformat(day) + ends = [t for e in members if (t := _real_end(e, when)) is not None] + done = min(ends) if ends else entry_done_at(owner, when) + if done is None: + _logger.warning("Workout on %s has no usable time; no ledger row", day) + return None + entry_id = owner.get("workout_id") or f"{slot[0]}:{slot[1]}" + return { + "kind": "credit", + "entry_id": str(entry_id), + "day": day, + "detail": {"completed_at": str(done), "source": _SOURCES.get(wtype, wtype)}, + } + + +def rest_row(day: date, declared_at: float) -> dict[str, Any]: + """The unsigned row for a declared rest day, stamped at its own midnight.""" + midnight = datetime.combine(day, time.min).astimezone().timestamp() + return { + "kind": "credit", + "entry_id": f"rest:{day.isoformat()}", + "day": day.isoformat(), + "detail": { + "completed_at": str(midnight), + "declared_at": str(declared_at), + "source": "rest_day", + }, + } + + +def existing_rows(path: Path) -> list[Any] | None: + """The ledger's rows; ``[]`` if absent, ``None`` (logged) if malformed.""" + if not path.exists(): + return [] + raw = json.loads(path.read_text(encoding="utf-8")) + rows = raw.get("entries") if isinstance(raw, dict) else None + if not isinstance(rows, list): + _logger.warning("Workout ledger %s has no entries array", path) + return None + return rows + + +def _atomic_write(path: Path, rows: list[Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".ledger-", suffix=".tmp") + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump({"entries": rows}, handle, indent=2) + Path(tmp).replace(path) + + +def new_rows( + rows: Iterable[dict[str, Any]], existing: list[Any] +) -> list[dict[str, Any]]: + """The rows whose ``entry_id`` is not among ``existing`` yet.""" + seen = {r.get("entry_id") for r in existing if isinstance(r, dict)} + fresh: list[dict[str, Any]] = [] + for row in rows: + if row["entry_id"] not in seen: + seen.add(row["entry_id"]) + fresh.append(row) + return fresh + + +def append_rows( + rows: Iterable[dict[str, Any]], + path: Path | None = None, + key_file: Path | None = None, +) -> int: + """Sign and append every row not already there. Returns how many were new. + + Never raises: a ledger that cannot be written is logged loudly, and the + credit it mirrors (log.json) stands regardless. + """ + target = path or ledger_path() + # The same key binding ``_earned`` verifies the gates' ledgers with. + key_path = key_file or _earned.HMAC_KEY_FILE + try: + key = key_path.read_bytes().strip() + target.parent.mkdir(parents=True, exist_ok=True) + with bonus_lock(target): + existing = existing_rows(target) + if existing is None: + return 0 + fresh = new_rows(rows, existing) + if not fresh: + return 0 + for row in fresh: + row["hmac"] = earned_time.entry_signature(row, key) + _atomic_write(target, [*existing, *fresh]) + except (OSError, ValueError) as exc: + _logger.warning("Workout ledger %s NOT updated: %s", target, exc) + return 0 + _logger.info("Workout ledger: %d new credit row(s) in %s", len(fresh), target) + return len(fresh) + + +def record_log_entry(day: str, entries: Sequence[Mapping[str, Any]]) -> None: + """Mirror the day's newest log entry into the ledger, if it earns a credit.""" + row = credit_row(day, entries, len(entries) - 1) + if row is not None: + append_rows([row]) diff --git a/screen_locker/_workout_ledger_cli.py b/screen_locker/_workout_ledger_cli.py new file mode 100644 index 0000000..23af477 --- /dev/null +++ b/screen_locker/_workout_ledger_cli.py @@ -0,0 +1,86 @@ +"""``--backfill-workout-ledger``: one-shot, idempotent ledger backfill. + + python -m screen_locker.screen_lock --backfill-workout-ledger --dry-run + python -m screen_locker.screen_lock --backfill-workout-ledger + --from 2026-10-01 --to 2026-10-09 [--dry-run] + +Writes a signed row for every workout credit slot already in ``log.json`` for +the days (default: today) and for every declared rest day in that range -- +the credits made before the ledger existed. Rows already present (same +``entry_id``) are skipped, so running it twice writes nothing the second +time. ``--dry-run`` prints the rows it would add, unsigned, with a count per +source, and writes nothing. +""" + +from __future__ import annotations + +import argparse +from collections import Counter +from datetime import date, timedelta +import json +import sys +from typing import TYPE_CHECKING, Any + +from screen_locker._day import today_str +from screen_locker._log_io import load_workout_log +from screen_locker._rest_day import rest_day_declarations +from screen_locker._workout_ledger import ( + append_rows, + credit_row, + existing_rows, + ledger_path, + new_rows, + rest_row, +) + +if TYPE_CHECKING: + from collections.abc import Sequence + from pathlib import Path + + +def backfill_rows(log_file: Path, first: date, last: date) -> list[dict[str, Any]]: + """Unsigned rows for every credit slot and rest day from ``first`` to ``last``.""" + logs = load_workout_log(log_file) + declared = rest_day_declarations() + rows: list[dict[str, Any]] = [] + day = first + while day <= last: + iso = day.isoformat() + entries = [e for e in logs.get(iso, []) if isinstance(e, dict)] + for index in range(len(entries)): + row = credit_row(iso, entries, index) + if row is not None and row["entry_id"] not in {r["entry_id"] for r in rows}: + rows.append(row) + if day in declared: + rows.append(rest_row(day, declared[day])) + day += timedelta(days=1) + return rows + + +def run_backfill(log_file: Path, argv: Sequence[str]) -> int: + """Backfill (or preview) the workout ledger. Returns a process exit code.""" + parser = argparse.ArgumentParser(prog="screen-locker --backfill-workout-ledger") + parser.add_argument("--from", dest="first", default=None, help="YYYY-MM-DD") + parser.add_argument("--to", dest="last", default=None, help="YYYY-MM-DD") + parser.add_argument("--dry-run", action="store_true", help="print, write nothing") + args = parser.parse_args(list(argv)) + today = date.fromisoformat(today_str()) + first = date.fromisoformat(args.first) if args.first else today + last = date.fromisoformat(args.last) if args.last else first + rows = backfill_rows(log_file, first, last) + path = ledger_path() + if args.dry_run: + existing = existing_rows(path) + fresh = new_rows(rows, existing or []) + for row in fresh: + sys.stdout.write(json.dumps(row) + "\n") + per_source = Counter(r["detail"]["source"] for r in fresh) + sys.stdout.write( + f"ledger {path} ({'present' if path.exists() else 'missing'}), " + f"{first}..{last}: {len(rows)} candidate row(s), {len(fresh)} new; " + f"per source: {dict(sorted(per_source.items()))}\n" + ) + return 0 + added = append_rows(rows, path) + sys.stdout.write(f"{added} new row(s) appended to {path}\n") + return 0 diff --git a/screen_locker/tests/_isolated_state.py b/screen_locker/tests/_isolated_state.py index 7ebb676..24a80b3 100644 --- a/screen_locker/tests/_isolated_state.py +++ b/screen_locker/tests/_isolated_state.py @@ -87,6 +87,17 @@ ISOLATED_STATE: tuple[tuple[str, tuple[str, ...]], ...] = ( "earned_hmac.key", ("_earned.HMAC_KEY_FILE",), ), + # Declared rest days: read by every workout credit and every reset, so an + # unredirected test would see the user's real rest days. + ( + "rest_days.json", + ("_constants.REST_DAY_FILE", "_rest_day.REST_DAY_FILE"), + ), + # Today's grace-floor lift; every shutdown add reads and writes it. + ( + "grace_floor.json", + ("_constants.GRACE_STATE_FILE", "_grace_floor.GRACE_STATE_FILE"), + ), # wake-alarm's workday-stick file. A test that forgot this would read # the developer's real penalty state. ( diff --git a/screen_locker/tests/test_daily_limits_wiring.py b/screen_locker/tests/test_daily_limits_wiring.py new file mode 100644 index 0000000..22cfc2b --- /dev/null +++ b/screen_locker/tests/test_daily_limits_wiring.py @@ -0,0 +1,180 @@ +"""The wiring around rest days, the grace lift and the ladder accessors. + +Live workout credit on a rest day, lift absorption in the shutdown adds, +the day-aware ``_earned`` accessors on an ``earned_time`` without the ladder +API (0.3.0), and the new CLI flags. +""" + +from __future__ import annotations + +from datetime import date +import logging +from typing import TYPE_CHECKING, Any +from unittest.mock import patch + +import earned_time +import pytest + +from screen_locker import _cli, _earned, _grace_floor, _runnerup_tcx, _workout_credit +from screen_locker._shutdown import ShutdownMixin +from screen_locker._workout_credit import WorkoutCreditMixin + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_ISO = "2026-10-12" # a Monday + + +class Host(WorkoutCreditMixin, ShutdownMixin): + """The real credit and add paths over an in-memory config.""" + + def __init__(self, tmp_path: Path, minutes: int, *, ok: bool = True) -> None: + self.config = (minutes, minutes, 300) + self.ok = ok + self.log_file = tmp_path / "log.json" + self.workout_data = {"type": "runnerup_verified"} + + def _read_shutdown_config(self) -> tuple[int, int, int] | None: + return self.config + + def _write_shutdown_config( + self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = False + ) -> bool: + del restore + if self.ok: + self.config = (min(mon_wed, 1380), min(thu_sun, 1380), morning_end) + return self.ok + + +@pytest.fixture +def monday() -> Iterator[None]: + with ( + patch("screen_locker._workout_credit.today_str", return_value=_ISO), + patch("screen_locker._shutdown.today_str", return_value=_ISO), + patch.object(_grace_floor, "today_str", return_value=_ISO), + ): + yield + + +_COUNTED: list[dict[str, Any]] = [{"workout_data": {"type": "runnerup_verified"}}] + + +@pytest.mark.usefixtures("monday") +class TestRestDayCredit: + def test_the_first_real_workout_earns_nothing_more( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + host = Host(tmp_path, 20 * 60) + with ( + patch.object(_workout_credit, "is_rest_day", return_value=True), + caplog.at_level(logging.INFO), + ): + result = host._apply_credit_for_written_entry([]) + assert host.config[0] == 20 * 60 + assert not result.shutdown_adjusted + assert "already in tonight's time" in caplog.text + + def test_a_second_one_earns_a_further_unit(self, tmp_path: Path) -> None: + host = Host(tmp_path, 20 * 60) + extra = _earned.extra_minutes(earned_time.WORKOUT, _grace_date()) + with patch.object(_workout_credit, "is_rest_day", return_value=True): + result = host._apply_credit_for_written_entry(_COUNTED) + assert host.config[0] == 20 * 60 + extra + assert result.extra_bonus_delta == extra + + +def _grace_date() -> date: + return date.fromisoformat(_ISO) + + +@pytest.mark.usefixtures("monday") +class TestLiftAbsorption: + def test_adds_pay_from_the_lift_first(self, tmp_path: Path) -> None: + _grace_floor.save_lift(30) + host = Host(tmp_path, 19 * 60 + 55) + assert host._adjust_shutdown_time_by(25) + assert host.config[0] == 19 * 60 + 55 + assert host._adjust_shutdown_time_by(25) + assert host.config[0] == 20 * 60 + 15 + + def test_a_failed_write_keeps_the_lift(self, tmp_path: Path) -> None: + _grace_floor.save_lift(30) + host = Host(tmp_path, 1195, ok=False) + assert not host._adjust_shutdown_time_by(25) + assert not host._adjust_shutdown_time_later() + assert _grace_floor.plan_absorb(0) == (0, 30) + + def test_the_first_workout_is_day_aware(self, tmp_path: Path) -> None: + host = Host(tmp_path, 1140) + assert host._adjust_shutdown_time_later() + first = _earned.first_minutes(earned_time.WORKOUT, _grace_date()) + assert host.config[0] == 1140 + first + + +class TestPreLadderEarnedTime: + """On 0.3.0 the accessors fall back to the raw fields, and say so once.""" + + def test_raw_fields_without_the_ladder_api( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + for name in ( + "shutdown_minutes_for", + "extra_shutdown_minutes_for", + "shutdown_ceiling_for", + "first_credit_at", + ): + monkeypatch.delattr(earned_time, name, raising=False) + day = _grace_date() + workout = earned_time.WORKOUT + assert _earned.first_minutes(workout, day) == workout.shutdown_minutes + assert _earned.extra_minutes(workout, day) == workout.extra_shutdown_minutes + assert _earned.ceiling(day) == earned_time.SHUTDOWN_CEILING_MINUTES + _earned._warn_no_first_credit.cache_clear() + with caplog.at_level(logging.WARNING): + assert _earned.first_credit_time(earned_time.ANKI, day) is None + assert _earned.first_credit_time(earned_time.ANKI, day) is None + assert caplog.text.count("has no first_credit_at") == 1 + + def test_the_ladder_api_reads_the_ledger(self, tmp_path: Path) -> None: + if not hasattr(earned_time, "first_credit_at"): + pytest.skip("installed earned_time predates first_credit_at") + assert _earned.first_credit_time(earned_time.ANKI, _grace_date()) is None + + +def test_a_failed_tcx_parse_is_not_marked() -> None: + assert _runnerup_tcx._mark_walk(None, "x_Walking.tcx") is None + + +class _Bare: + """What ``_headless_locker`` builds on: any class with a ``__dict__``.""" + + log_file: Path + workout_data: dict[str, str] + + +class TestCliFlags: + def test_declare_rest_day_dispatches_the_rest_of_argv(self) -> None: + with ( + patch.object(_cli.logging, "basicConfig"), + patch.object(_cli, "run_declare_rest_day", return_value=0) as run, + patch.object(_cli.sys, "exit", side_effect=SystemExit) as sys_exit, + pytest.raises(SystemExit), + ): + _cli.main(_Bare, ["screen_lock.py", "--declare-rest-day", "tomorrow"]) + run.assert_called_once_with(["tomorrow"]) + sys_exit.assert_called_once_with(0) + + def test_backfill_dispatches_with_the_log(self) -> None: + with ( + patch.object(_cli.logging, "basicConfig"), + patch.object(_cli, "run_backfill", return_value=0) as run, + patch.object(_cli.sys, "exit", side_effect=SystemExit), + pytest.raises(SystemExit), + ): + _cli.main( + _Bare, ["screen_lock.py", "--backfill-workout-ledger", "--dry-run"] + ) + log_file, argv = run.call_args.args + assert log_file.name == "log.json" + assert argv == ["--dry-run"] diff --git a/screen_locker/tests/test_earned.py b/screen_locker/tests/test_earned.py index 36772e3..0b16aef 100644 --- a/screen_locker/tests/test_earned.py +++ b/screen_locker/tests/test_earned.py @@ -95,8 +95,11 @@ class TestLedgerFile: assert not any(ledger_file(e).exists() for e in flat_earners()) def test_earner_without_a_ledger_raises(self) -> None: + bare = earned_time.Earner( + name="bare", label="Bare", gaming_minutes=0, shutdown_minutes=0 + ) with pytest.raises(ValueError, match="has no ledger"): - ledger_file(earned_time.WORKOUT) + ledger_file(bare) class TestFlatEarners: diff --git a/screen_locker/tests/test_earner_units.py b/screen_locker/tests/test_earner_units.py index 90fdbe7..eec5cdb 100644 --- a/screen_locker/tests/test_earner_units.py +++ b/screen_locker/tests/test_earner_units.py @@ -137,7 +137,7 @@ class TestBonusLock: ) as lock, patch("screen_locker._shutdown_base.flat_earners", return_value=()), ): - apply_flat_bonuses_if_new(state, object()) + apply_flat_bonuses_if_new(state, MagicMock()) lock.assert_called_once_with(state) assert seen == [True] diff --git a/screen_locker/tests/test_grace_floor.py b/screen_locker/tests/test_grace_floor.py new file mode 100644 index 0000000..ab83e75 --- /dev/null +++ b/screen_locker/tests/test_grace_floor.py @@ -0,0 +1,207 @@ +"""Tests for the grace floor, its lift bookkeeping and the derived target.""" + +from __future__ import annotations + +import dataclasses +from datetime import date, datetime, time, timedelta +import json +import logging +from typing import TYPE_CHECKING, Any +from unittest.mock import patch + +import earned_time +import pytest + +from screen_locker import _earned, _grace_floor, _sick_tracker +from screen_locker._grace_floor import ( + GraceFloorMixin, + first_done_at, + grace_for, + plan_absorb, + save_lift, + sick_on, + workout_done_at, +) + +if TYPE_CHECKING: + from pathlib import Path + +_DAY = date(2026, 10, 10) +_ISO = _DAY.isoformat() +_FLAT = {"leetcode": False, "reading": False, "anki": False, "automation": False} + + +def at(hhmm: str, day: date = _DAY) -> float: + hours, minutes = map(int, hhmm.split(":")) + return datetime.combine(day, time(hours, minutes)).astimezone().timestamp() + + +def log_with(tmp_path: Path, entries: list[dict[str, Any]]) -> Path: + log = tmp_path / "log.json" + log.write_text(json.dumps({_ISO: entries})) + return log + + +class TestWhenTheDayStarted: + def test_the_first_counted_workout_by_its_own_end(self, tmp_path: Path) -> None: + log = log_with( + tmp_path, + [ + {"workout_data": {"type": "relaxed_day_skip"}, "timestamp": "x"}, + {"workout_data": {"type": "manual_workout", "end_time": "18:40"}}, + { + "workout_data": { + "type": "runnerup_verified", + "completed_at": at("17:00"), + } + }, + ], + ) + assert workout_done_at(log, _DAY) == at("17:00") + + def test_no_workout_is_none(self, tmp_path: Path) -> None: + assert workout_done_at(log_with(tmp_path, []), _DAY) is None + + def test_the_earliest_earner_on_the_day_wins(self, tmp_path: Path) -> None: + log = log_with( + tmp_path, + [{"workout_data": {"type": "manual_workout", "end_time": "19:30"}}], + ) + times = {"anki": at("18:55"), "leetcode": at("23:00", _DAY - timedelta(1))} + with patch.object( + _grace_floor, + "first_credit_time", + side_effect=lambda e, _d: times.get(e.name), + ): + assert first_done_at(_DAY, log) == at("18:55") + assert first_done_at(_DAY, None) == at("18:55") + + def test_a_rest_day_skips_the_workout_ledger(self) -> None: + """Its rest row is stamped at midnight, which is not a completion.""" + ledgered = dataclasses.replace(earned_time.WORKOUT, ledger="w.json") + with ( + patch.object(earned_time, "WORKOUT", ledgered), + patch.object(_grace_floor, "first_credit_time", return_value=at("00:00")), + ): + assert first_done_at(_DAY, None) == at("00:00") + with patch.object(_grace_floor, "flat_earners", return_value=()): + assert first_done_at(_DAY, None, rest_day=True) is None + + def test_without_a_workout_ledger_only_the_log_answers(self) -> None: + bare = earned_time.Earner( + name="workout", label="workout", gaming_minutes=0, shutdown_minutes=0 + ) + with ( + patch.object(earned_time, "WORKOUT", bare), + patch.object(_grace_floor, "first_credit_time", return_value=None), + ): + assert first_done_at(_DAY, None) is None + + +class TestGraceFor: + def test_an_hour_after_the_first_task(self) -> None: + assert grace_for(at("18:55"), _DAY) == 19 * 60 + 55 + + def test_capped_at_the_ceiling(self) -> None: + assert grace_for(at("22:40"), _DAY) == _earned.ceiling(_DAY) + + def test_nothing_done_or_another_day_is_no_floor(self) -> None: + assert grace_for(None, _DAY) is None + assert grace_for(at("12:00", _DAY - timedelta(1)), _DAY) is None + + +class TestLift: + def test_absorb_pays_from_the_lift_first(self) -> None: + with patch.object(_grace_floor, "today_str", return_value=_ISO): + assert plan_absorb(30) == (30, 0) + save_lift(45) + assert plan_absorb(30) == (0, 15) + assert plan_absorb(60) == (15, 0) + + def test_yesterdays_lift_is_spent(self) -> None: + with patch.object(_grace_floor, "today_str", return_value="2026-10-09"): + save_lift(45) + with patch.object(_grace_floor, "today_str", return_value=_ISO): + assert plan_absorb(30) == (30, 0) + + def test_an_unreadable_state_is_no_lift( + self, caplog: pytest.LogCaptureFixture + ) -> None: + _grace_floor.GRACE_STATE_FILE.write_text("{bad") + with caplog.at_level(logging.WARNING): + assert plan_absorb(10) == (10, 0) + assert "unreadable" in caplog.text + + def test_a_failed_save_is_logged( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.WARNING): + save_lift(5, tmp_path / "missing-dir" / "g.json") + assert "Could not record the grace lift" in caplog.text + + +class Host(GraceFloorMixin): + """The mixin over an in-memory config.""" + + def __init__(self, config: tuple[int, int, int] | None, *, ok: bool = True) -> None: + self.config = config + self.ok = ok + + def _read_shutdown_config(self) -> tuple[int, int, int] | None: + return self.config + + def _write_shutdown_config( + self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = False + ) -> bool: + assert restore + if self.ok: + self.config = (mon_wed, thu_sun, morning_end) + return self.ok + + +@pytest.fixture +def saturday() -> Any: + with patch.object(_grace_floor, "today_str", return_value=_ISO): + yield + + +@pytest.mark.usefixtures("saturday") +class TestApply: + def test_lifts_to_the_floor_and_records_the_lift(self) -> None: + host = Host((19 * 60 + 25, 19 * 60 + 25, 300)) + assert host._apply_grace_floor(None, first_done=at("18:55")) is True + assert host.config == (19 * 60 + 55, 19 * 60 + 55, 300) + assert plan_absorb(0) == (0, 30) + + def test_already_above_the_floor(self) -> None: + host = Host((21 * 60, 21 * 60, 300)) + assert host._apply_grace_floor(None, first_done=at("18:55")) is False + + def test_no_first_task_no_floor(self) -> None: + with patch.object(_grace_floor, "first_done_at", return_value=None): + assert Host((1140, 1140, 300))._apply_grace_floor(None) is False + + def test_unreadable_config(self, caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING): + assert Host(None)._apply_grace_floor(None, first_done=at("18:55")) is False + assert "config unreadable" in caplog.text + + def test_failed_write(self, caplog: pytest.LogCaptureFixture) -> None: + host = Host((1140, 1140, 300), ok=False) + with caplog.at_level(logging.WARNING): + assert host._apply_grace_floor(None, first_done=at("18:55")) is False + assert "failed to write" in caplog.text + assert plan_absorb(0) == (0, 0) + + def test_a_sick_day_has_no_floor(self) -> None: + """The user's call: the sick-day shutdown wins.""" + _grace_floor_history([_ISO]) + assert sick_on(_DAY) + host = Host((18 * 60, 18 * 60, 300)) + assert host._apply_grace_floor(None, first_done=at("18:55")) is False + assert host.config == (18 * 60, 18 * 60, 300) + + +def _grace_floor_history(days: list[str]) -> None: + + _sick_tracker.SICK_HISTORY_FILE.write_text(json.dumps({"sick_days": days})) diff --git a/screen_locker/tests/test_rest_day.py b/screen_locker/tests/test_rest_day.py new file mode 100644 index 0000000..4168f4b --- /dev/null +++ b/screen_locker/tests/test_rest_day.py @@ -0,0 +1,219 @@ +"""Tests for ``_rest_day`` and its CLI: declared ahead, two a week, signed.""" + +from __future__ import annotations + +from datetime import date, datetime, time +import json +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +import earned_time +import pytest + +from screen_locker import _earned, _rest_day +from screen_locker._rest_day import declare, is_rest_day, rest_day_declarations +from screen_locker._rest_day_cli import run_declare_rest_day +from screen_locker._workout_ledger import ledger_path +from screen_locker.tests._earned_fixtures import KEY, signing_key + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_EVE = datetime.combine(date(2026, 10, 9), time(18, 55)).astimezone() +_SAT, _SUN, _MON = date(2026, 10, 10), date(2026, 10, 11), date(2026, 10, 12) + + +@pytest.fixture +def key(tmp_path: Path) -> Iterator[Path]: + with signing_key(tmp_path) as path: + yield path + + +def rows() -> list[dict[str, object]]: + return json.loads(_rest_day.REST_DAY_FILE.read_text())["entries"] + + +def write(entries: object) -> None: + _rest_day.REST_DAY_FILE.write_text(json.dumps({"entries": entries})) + + +@pytest.mark.usefixtures("key") +class TestDeclare: + def test_a_future_day_is_signed_and_mirrored_to_the_ledger(self) -> None: + result = declare(_SAT, now=_EVE) + assert result.ok + assert result.reason == "2026-10-10 is a rest day (1 left in ISO week 41)" + (row,) = rows() + assert earned_time.verified(row, KEY) + assert row["declared_on"] == "2026-10-09" + (credit,) = json.loads(ledger_path().read_text())["entries"] + assert credit["entry_id"] == "rest:2026-10-10" + assert credit["detail"]["source"] == "rest_day" + assert float(credit["detail"]["declared_at"]) == _EVE.timestamp() + midnight = datetime.combine(_SAT, time.min).astimezone().timestamp() + assert float(credit["detail"]["completed_at"]) == midnight + assert is_rest_day(_SAT) + + @pytest.mark.parametrize("day", [date(2026, 10, 9), date(2026, 10, 1)]) + def test_today_and_the_past_are_refused(self, day: date) -> None: + result = declare(day, now=_EVE) + assert not result.ok + assert "not in the future" in result.reason + assert not _rest_day.REST_DAY_FILE.exists() + + def test_two_per_iso_week_then_refused(self) -> None: + assert declare(_SAT, now=_EVE).ok + assert declare(_SUN, now=_EVE).reason.endswith("(0 left in ISO week 41)") + assert declare(_SUN, now=_EVE).reason == "2026-10-11 is already a rest day" + third = declare(date(2026, 10, 8 + 1), now=datetime(2026, 10, 8).astimezone()) + assert not third.ok + assert third.reason == "ISO week 41 already has 2 rest days" + assert declare(_MON, now=_EVE).ok # the next week has its own two + + def test_an_unreadable_file_refuses(self, caplog: pytest.LogCaptureFixture) -> None: + _rest_day.REST_DAY_FILE.write_text("{not json") + with caplog.at_level(logging.WARNING): + result = declare(_SAT, now=_EVE) + assert not result.ok + assert result.reason == "rest-day file or signing key unreadable" + assert "unreadable" in caplog.text + + def test_a_failed_write_says_so(self, caplog: pytest.LogCaptureFixture) -> None: + with ( + patch.object( + type(_rest_day.REST_DAY_FILE), "write_text", side_effect=OSError("ro") + ), + caplog.at_level(logging.WARNING), + ): + result = declare(_SAT, now=_EVE) + assert not result.ok + assert "could not write" in result.reason + + +class TestReading: + def test_no_key_counts_nothing( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.WARNING): + assert rest_day_declarations() == {} + assert "signing key" in caplog.text + with signing_key(tmp_path, b" "): + assert rest_day_declarations() == {} + + @pytest.mark.usefixtures("key") + def test_a_hand_edited_row_is_ignored( + self, caplog: pytest.LogCaptureFixture + ) -> None: + declare(_SAT, now=_EVE) + edited = rows() + edited[0]["day"] = "2026-10-14" + write(edited) + with caplog.at_level(logging.WARNING): + assert rest_day_declarations() == {} + assert "fails its signature" in caplog.text + + @pytest.mark.usefixtures("key") + def test_a_signed_same_day_row_is_not_ahead(self) -> None: + """A row from any other route still has to be declared ahead.""" + body = { + "kind": "rest_day", + "day": "2026-10-10", + "declared_on": "2026-10-10", + "declared_at": 1.0, + } + write([{**body, "hmac": earned_time.entry_signature(body, KEY)}]) + assert rest_day_declarations() == {} + + @pytest.mark.usefixtures("key") + def test_malformed_and_foreign_rows_are_ignored( + self, caplog: pytest.LogCaptureFixture + ) -> None: + body = { + "kind": "rest_day", + "day": "soon", + "declared_on": "2026-10-09", + "declared_at": 1.0, + } + write( + [ + "x", + {"kind": "other"}, + {**body, "hmac": earned_time.entry_signature(body, KEY)}, + ] + ) + with caplog.at_level(logging.WARNING): + assert rest_day_declarations() == {} + assert "malformed" in caplog.text + + @pytest.mark.usefixtures("key") + def test_a_file_without_entries_counts_nothing( + self, caplog: pytest.LogCaptureFixture + ) -> None: + _rest_day.REST_DAY_FILE.write_text("[]") + with caplog.at_level(logging.WARNING): + assert rest_day_declarations() == {} + assert "no entries array" in caplog.text + + @pytest.mark.usefixtures("key") + def test_the_cap_holds_on_read(self) -> None: + """Three signed rows in a week: the two declared first count.""" + + def row(day: str, at: float) -> dict[str, object]: + body = { + "kind": "rest_day", + "day": day, + "declared_on": "2026-10-01", + "declared_at": at, + } + return {**body, "hmac": earned_time.entry_signature(body, KEY)} + + write([row("2026-10-07", 3.0), row("2026-10-05", 1.0), row("2026-10-06", 2.0)]) + assert set(rest_day_declarations()) == {date(2026, 10, 5), date(2026, 10, 6)} + + +@pytest.mark.usefixtures("key") +class TestCli: + def _run(self, *argv: str, offset: float | None = 0.0) -> int: + with ( + patch("screen_locker._rest_day_cli._query_ntp_offset", return_value=offset), + patch("screen_locker._rest_day_cli.today_str", return_value="2026-10-09"), + patch("screen_locker._rest_day.datetime") as clock, + ): + clock.now.return_value = _EVE + return run_declare_rest_day(list(argv)) + + def test_declare_tomorrow(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("tomorrow") == 0 + assert "2026-10-10 is a rest day" in capsys.readouterr().out + + def test_a_date_and_then_the_list(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("2026-10-11") == 0 + assert self._run("--list") == 0 + assert capsys.readouterr().out.splitlines()[-1] == "2026-10-11" + + def test_an_empty_list(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run() == 0 + assert capsys.readouterr().out.strip() == "no rest days declared" + + def test_a_bad_date(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("someday") == 2 + assert capsys.readouterr().out.startswith("error:") + + def test_no_ntp_refuses(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("tomorrow", offset=None) == 1 + assert "cannot be confirmed" in capsys.readouterr().out + + def test_a_skewed_clock_refuses(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("tomorrow", offset=86400.0) == 1 + assert "off by 86400s" in capsys.readouterr().out + + def test_today_is_refused(self, capsys: pytest.CaptureFixture[str]) -> None: + assert self._run("2026-10-09") == 1 + assert "refused: 2026-10-09 is not in the future" in capsys.readouterr().out + + +def test_the_key_binding_is_the_one_earned_reads() -> None: + """The suite redirects ``_earned.HMAC_KEY_FILE``; rest days must follow it.""" + assert not _earned.HMAC_KEY_FILE.exists() diff --git a/screen_locker/tests/test_runnerup_walks.py b/screen_locker/tests/test_runnerup_walks.py new file mode 100644 index 0000000..ff579d2 --- /dev/null +++ b/screen_locker/tests/test_runnerup_walks.py @@ -0,0 +1,177 @@ +"""A day's RunnerUp walks are summed into one workout; never one file at a time.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import json +import logging +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from screen_locker._runnerup_verification import RunnerUpVerificationMixin + +if TYPE_CHECKING: + from pathlib import Path + +_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" +_DAY = "2026-10-10" + + +def tcx(sport: str, minutes: int, start: str, speed: float = 1.3) -> str: + """A RunnerUp-shaped export: trackpoints every 5 s at ``speed`` m/s.""" + begin = datetime.fromisoformat(f"{_DAY}T{start}:00+00:00") + points = "".join( + f"" + f"{s * speed:.1f}" + for s in range(0, minutes * 60 + 1, 5) + ) + return ( + f'' + f'{_DAY}T{start}' + f"{minutes * 60}" + f"{minutes * 60 * speed}" + f"{points}" + "" + ) + + +class Verifier(RunnerUpVerificationMixin): + """The real mixin, reading one temporary drop directory and no phone.""" + + def __init__(self, folder: Path) -> None: + self.folder = folder + self.adb = MagicMock(return_value=(False, "")) + + def _has_adb_device(self) -> bool: + return False + + def _run_adb(self, args: list[str]) -> tuple[bool, str]: + return self.adb(args) + + def _find_runnerup_exports_for_date(self, date_str: str) -> list[str]: + return sorted(str(p) for p in self.folder.glob(f"*{date_str}*.tcx")) + + +def drop(folder: Path, name: str, body: str) -> None: + (folder / f"RunnerUp_Pixel_6a_{_DAY}-{name}.tcx").write_text(body) + + +@pytest.fixture +def verifier(tmp_path: Path) -> Verifier: + return Verifier(tmp_path) + + +class TestParse: + def test_a_walk_is_flagged_by_its_file_name(self, verifier: Verifier) -> None: + drop(verifier.folder, "08-00-00_Walking", tcx("Other", 10, "08:00")) + (path,) = verifier._find_runnerup_exports_for_date(_DAY) + data = verifier._pull_and_parse_tcx(path) + assert data is not None + assert data["walking"] is True + assert data["moving_seconds"] == pytest.approx(600) + + def test_a_pulled_copy_keeps_the_phone_name(self, verifier: Verifier) -> None: + """adb pulls into ``activity.tcx``; the remote name still decides.""" + body = tcx("Other", 10, "08:00") + + def pull(args: list[str]) -> tuple[bool, str]: + from pathlib import Path + + Path(args[2]).write_text(body) + return True, "" + + verifier.adb.side_effect = pull + data = verifier._pull_and_parse_tcx("/sdcard/RunnerUp_x_Walking.tcx") + assert data is not None + assert data["walking"] is True + + def test_a_failed_pull_is_none(self, verifier: Verifier) -> None: + assert verifier._pull_and_parse_tcx("/sdcard/RunnerUp_x_Walking.tcx") is None + + +class TestVerifyToday: + def _verify(self, verifier: Verifier) -> tuple[str, str] | None: + with patch("screen_locker._runnerup_verification.today_str", return_value=_DAY): + return verifier._verify_runnerup_via_files() + + def test_walks_summing_to_the_bar_verify(self, verifier: Verifier) -> None: + drop(verifier.folder, "08-00-00_Walking", tcx("Other", 25, "08:00")) + drop(verifier.folder, "17-00-00_Walking", tcx("Other", 20, "17:00")) + result = self._verify(verifier) + assert result is not None + assert result[0] == "verified" + assert result[1].startswith("Walking: 2 walk(s)") + + def test_one_long_elapsed_walk_is_judged_by_moving_time( + self, verifier: Verifier + ) -> None: + """60 min on the clock, standing still: not a workout.""" + drop(verifier.folder, "08-00-00_Walking", tcx("Other", 60, "08:00", 0.1)) + assert self._verify(verifier) == ( + "too_short", + "Walking: 1 walk(s), 0.4 km, 0 min moving -- need 40+ min moving", + ) + + def test_a_run_still_verifies_beside_short_walks(self, verifier: Verifier) -> None: + drop(verifier.folder, "07-00-00_Walking", tcx("Other", 10, "07:00")) + drop(verifier.folder, "18-00-00_Running", tcx("Running", 45, "18:00", 2.5)) + result = self._verify(verifier) + assert result is not None + assert result[0] == "verified" + assert result[1].startswith("Running") + + def test_a_short_run_outranks_the_walk_shortfall(self, verifier: Verifier) -> None: + drop(verifier.folder, "07-00-00_Walking", tcx("Other", 10, "07:00")) + drop(verifier.folder, "18-00-00_Running", tcx("Running", 5, "18:00", 2.5)) + result = self._verify(verifier) + assert result is not None + assert result[0] == "too_short" + assert result[1].startswith("Run was") + + +class TestBackfill: + def test_walks_are_logged_as_one_workout( + self, verifier: Verifier, tmp_path: Path + ) -> None: + drop(verifier.folder, "08-00-00_Walking", tcx("Other", 25, "08:00")) + drop(verifier.folder, "17-00-00_Walking", tcx("Other", 20, "17:00")) + log = tmp_path / "log.json" + ingested = MagicMock() + assert verifier._try_fill_runnerup_for_date(_DAY, log, ingested) is True + (entry,) = json.loads(log.read_text())[_DAY] + data = entry["workout_data"] + assert data["type"] == "runnerup_verified" + assert data["activity"] == "walking" + assert data["duration_minutes"] == pytest.approx(45) + assert ( + data["completed_at"] + == datetime(2026, 10, 10, 17, 20, tzinfo=UTC).timestamp() + ) + ingested.assert_called_once() + # The same day again is the same slot: nothing new. + assert verifier._try_fill_runnerup_for_date(_DAY, log, ingested) is False + + def test_walks_short_of_the_bar_log_nothing( + self, + verifier: Verifier, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + drop(verifier.folder, "08-00-00_Walking", tcx("Other", 30, "08:00")) + log = tmp_path / "log.json" + with caplog.at_level(logging.WARNING): + assert verifier._try_fill_runnerup_for_date(_DAY, log) is False + assert "do not count yet" in caplog.text + assert not log.exists() + + def test_a_run_records_its_end_time( + self, verifier: Verifier, tmp_path: Path + ) -> None: + drop(verifier.folder, "18-00-00_Running", tcx("Running", 45, "18:00", 2.5)) + log = tmp_path / "log.json" + assert verifier._try_fill_runnerup_for_date(_DAY, log) is True + (entry,) = json.loads(log.read_text())[_DAY] + end = datetime(2026, 10, 10, 18, 45, tzinfo=UTC).timestamp() + assert entry["workout_data"]["completed_at"] == end diff --git a/screen_locker/tests/test_shutdown_flat_earners.py b/screen_locker/tests/test_shutdown_flat_earners.py index d5624dd..25e586e 100644 --- a/screen_locker/tests/test_shutdown_flat_earners.py +++ b/screen_locker/tests/test_shutdown_flat_earners.py @@ -186,5 +186,17 @@ class TestProjectionFollowsTheCut: lowered = 19 * 60 - sum( e.shutdown_minutes for e in earned_time.EARNERS if e.penalty_from == tuesday ) - assert [d.minutes for d in shutdown.rest_of_week] == [19 * 60] + [lowered] * 6 - assert [d.minutes for d in shutdown.next_week_preview] == [lowered] * 7 + + def base(day: date) -> int: + # From the sleep ladder (Sat 2026-10-10) the registry prices the + # day itself; before it, the cut from Tuesday holds. + return earned_time.base_for(day, earned_time.EARNERS).shutdown_minutes + + week = [date(2026, 10, 5 + i) for i in range(7)] + assert [base(d) for d in week[1:5]] == [lowered] * 4 + assert [d.minutes for d in shutdown.rest_of_week] == [19 * 60] + [ + base(d) for d in week[1:] + ] + assert [d.minutes for d in shutdown.next_week_preview] == [ + base(date(2026, 10, 12 + i)) for i in range(7) + ] diff --git a/screen_locker/tests/test_shutdown_reading.py b/screen_locker/tests/test_shutdown_reading.py index 3523f66..3ac43d3 100644 --- a/screen_locker/tests/test_shutdown_reading.py +++ b/screen_locker/tests/test_shutdown_reading.py @@ -16,7 +16,6 @@ from unittest.mock import MagicMock, patch import earned_time import pytest -from screen_locker import _shutdown_base from screen_locker._day import today_str from screen_locker._earned import flat_earners, hhmm from screen_locker._shutdown_base import ( @@ -65,15 +64,16 @@ class TestBaseHour: assert base_minutes(date(2026, 10, 1)) == 19 * 60 def test_long_after_the_cut_stays_lowered(self) -> None: - """No drift back to 20:00; every later earner's cut holds too.""" - cut = earned_time.READING.penalty_from - assert cut is not None - later_cuts = sum( - e.shutdown_minutes - for e in earned_time.EARNERS - if e.penalty_from is not None and e.penalty_from > cut - ) - assert base_minutes(date(2030, 1, 1)) == 19 * 60 - later_cuts + """No drift back to 20:00: the registry's base for the day holds. + + From earned_time's sleep ladder (2026-10-10) the floor is the ceiling + minus every earner's first unit, not 20:00 minus the cuts, so the day + is priced by the registry and only the "stays lowered" part is ours. + """ + later = date(2030, 1, 1) + expected = earned_time.base_for(later, earned_time.EARNERS).shutdown_minutes + assert base_minutes(later) == expected + assert expected <= 19 * 60 def test_default_is_the_local_today(self) -> None: assert base_minutes() == base_minutes(datetime.now().astimezone().date()) @@ -120,7 +120,7 @@ class TestResetIncludesReading: answers[0]["leetcode"] = True answers[0]["reading"] = True mixin = _mixin() - with patch.object(_shutdown_base, "today_credit_count", return_value=9): + with patch("screen_locker._shutdown_target.day_credit_count", return_value=9): reset_to_base_if_new_day( tmp_path / "state.json", mixin, log_file=tmp_path / "log.json" ) diff --git a/screen_locker/tests/test_shutdown_target.py b/screen_locker/tests/test_shutdown_target.py new file mode 100644 index 0000000..acbbf34 --- /dev/null +++ b/screen_locker/tests/test_shutdown_target.py @@ -0,0 +1,77 @@ +"""Tests for ``_shutdown_target``: the one formula behind reset and preview.""" + +from __future__ import annotations + +from datetime import date, datetime, time +import json +from typing import TYPE_CHECKING +from unittest.mock import patch + +from screen_locker import _grace_floor +from screen_locker._shutdown_target import ( + DayInputs, + day_credit_count, + derive, + gather, + workout_units, +) + +if TYPE_CHECKING: + from pathlib import Path + +_DAY = date(2026, 10, 10) +_ISO = _DAY.isoformat() +_FLAT = {"leetcode": False, "reading": False, "anki": False, "automation": False} + + +def at(hhmm: str) -> float: + hours, minutes = map(int, hhmm.split(":")) + return datetime.combine(_DAY, time(hours, minutes)).astimezone().timestamp() + + +def log_with(tmp_path: Path, entries: list[dict[str, object]]) -> Path: + log = tmp_path / "log.json" + log.write_text(json.dumps({_ISO: entries})) + return log + + +class TestTarget: + def test_a_rest_day_stands_in_for_the_first_workout_only(self) -> None: + assert workout_units(0, rest_day=True) == 1 + assert workout_units(2, rest_day=True) == 2 + assert workout_units(0, rest_day=False) == 0 + + def test_the_floor_lifts_and_the_lift_is_reported(self) -> None: + target = derive( + DayInputs(_DAY, {**_FLAT, "anki": True}, first_done=at("18:55")) + ) + assert target.minutes == max(target.earned, 19 * 60 + 55) + assert target.lift == target.minutes - target.earned + + def test_a_sick_day_drops_the_floor(self) -> None: + target = derive(DayInputs(_DAY, _FLAT, first_done=at("18:55"), sick_day=True)) + assert target.grace is None + assert target.minutes == target.earned + + def test_no_log_counts_no_workouts(self) -> None: + assert day_credit_count(None, _DAY) == 0 + + def test_gather_reads_every_real_source(self, tmp_path: Path) -> None: + log = log_with( + tmp_path, + [ + { + "workout_data": {"type": "manual_workout", "end_time": "18:00"}, + "timestamp": "2026-10-10T20:00:00+00:00", + } + ], + ) + with ( + patch("screen_locker._shutdown_target.flat_answers", return_value=_FLAT), + patch.object(_grace_floor, "first_credit_time", return_value=None), + ): + target = gather(_DAY, log) + assert target.resolution.term("workout").answer == 1 + assert target.first_done_at == at("18:00") + assert target.grace == 19 * 60 + assert not target.rest_day diff --git a/screen_locker/tests/test_walk_tcx.py b/screen_locker/tests/test_walk_tcx.py new file mode 100644 index 0000000..6f7a285 --- /dev/null +++ b/screen_locker/tests/test_walk_tcx.py @@ -0,0 +1,134 @@ +"""Tests for ``_walk_tcx``: which exports are walks, and their moving time.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import logging +from typing import TYPE_CHECKING, Any +import xml.etree.ElementTree as ET + +import pytest + +from screen_locker._walk_tcx import ( + is_walking_export, + track_summary, + walk_day, + walk_shortfall, +) + +if TYPE_CHECKING: + from collections.abc import Sequence + +_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" +_START = datetime(2026, 10, 10, 8, 0, tzinfo=UTC) + + +def activity(points: Sequence[tuple[float, str | None, str | None]]) -> ET.Element: + """An Activity of ``(seconds after start, metres, raw time override)``. + + A raw time of ``"-"`` leaves the point without a time. + """ + root = ET.Element(f"{{{_NS}}}Activity", Sport="Other") + ET.SubElement(root, f"{{{_NS}}}Id").text = "A1" + track = ET.SubElement(ET.SubElement(root, f"{{{_NS}}}Lap"), f"{{{_NS}}}Track") + for sec, metres, raw in points: + point = ET.SubElement(track, f"{{{_NS}}}Trackpoint") + if raw != "-": + stamp = raw or (_START + timedelta(seconds=sec)).isoformat() + ET.SubElement(point, f"{{{_NS}}}Time").text = stamp + if metres: + ET.SubElement(point, f"{{{_NS}}}DistanceMeters").text = metres + return root + + +def walk(seconds: int, speed: float = 1.3) -> ET.Element: + """A steady walk sampled every 5 s.""" + return activity([(s, f"{s * speed:.1f}", None) for s in range(0, seconds + 1, 5)]) + + +class TestIsWalkingExport: + def test_the_file_name_names_the_sport(self) -> None: + """RunnerUp's default export writes Sport="Other"; the name says Walking.""" + assert is_walking_export("RunnerUp_2026-10-10-08-00-00_Walking.tcx", "Other") + + def test_the_strava_variant_tags_the_activity(self) -> None: + assert is_walking_export("RunnerUp_2026-10-10-08-00-00.tcx", "Walking") + + def test_other_alone_is_not_a_walk(self) -> None: + assert not is_walking_export("RunnerUp_2026-10-10_Running.tcx", "Other") + + +class TestTrackSummary: + def test_walking_speed_is_moving(self) -> None: + summary = track_summary(walk(600)) + assert summary["moving_seconds"] == pytest.approx(600) + assert summary["activity_id"] == "A1" + assert summary["ended_at"] == (_START + timedelta(seconds=600)).timestamp() + + def test_standing_and_vehicles_are_not_moving(self) -> None: + assert track_summary(walk(600, speed=0.1))["moving_seconds"] == 0 + assert track_summary(walk(600, speed=8.0))["moving_seconds"] == 0 + + def test_a_gps_dropout_earns_nothing(self) -> None: + """Two points 100 s apart at walking pace: motion unseen, not counted.""" + summary = track_summary(activity([(0, "0.1", None), (100, "130", None)])) + assert summary["moving_seconds"] == 0 + + def test_unusable_points_are_skipped( + self, caplog: pytest.LogCaptureFixture + ) -> None: + points = [ + (0, "0.1", None), + (5, None, None), # no distance + (6, "8", "-"), # no time + (7, "9", "not-a-time"), + (10, "13", None), + ] + with caplog.at_level(logging.WARNING): + summary = track_summary(activity(points)) + assert summary["moving_seconds"] == pytest.approx(10) + assert "unparsable time" in caplog.text + + def test_no_points_has_no_end(self) -> None: + summary = track_summary(activity([])) + assert summary == {"activity_id": "A1", "moving_seconds": 0.0, "ended_at": None} + + +def leg( + ident: str, minutes: float, ended: float, metres: float = 2000 +) -> dict[str, Any]: + return { + "activity_id": ident, + "moving_seconds": minutes * 60, + "ended_at": ended, + "distance_m": metres, + } + + +class TestWalkDay: + def test_legs_sum_and_complete_when_the_bar_is_crossed(self) -> None: + day = walk_day([leg("b", 20, 200.0), leg("a", 25, 100.0), leg("c", 5, 300.0)]) + assert day.legs == 3 + assert day.moving_minutes == pytest.approx(50) + assert day.distance_km == pytest.approx(6) + assert day.completed_at == 200.0 # a (25) + b (20) crosses 40 at b's end + assert day.qualifies + assert day.message() == "Walking: 3 walk(s), 6.0 km, 50 min moving" + + def test_the_same_walk_from_two_sources_counts_once(self) -> None: + copy = leg("same", 30, 100.0) + day = walk_day([copy, dict(copy)]) + assert day.legs == 1 + assert not day.qualifies + assert day.completed_at is None + + def test_a_walk_without_id_or_end_still_counts(self) -> None: + anonymous = {"moving_seconds": 45 * 60} + day = walk_day([anonymous]) + assert day.legs == 1 + assert day.qualifies + assert day.completed_at is None + + def test_shortfall_says_what_is_missing(self) -> None: + text = walk_shortfall(walk_day([leg("a", 30, 1.0)])) + assert text.endswith("need 40+ min moving") diff --git a/screen_locker/tests/test_workout_ledger.py b/screen_locker/tests/test_workout_ledger.py new file mode 100644 index 0000000..0932089 --- /dev/null +++ b/screen_locker/tests/test_workout_ledger.py @@ -0,0 +1,197 @@ +"""Tests for ``_workout_ledger`` and its backfill CLI: one signed row per credit.""" + +from __future__ import annotations + +from datetime import date, datetime, time +import json +import logging +from typing import TYPE_CHECKING, Any + +import earned_time +import pytest + +from screen_locker import _workout_ledger +from screen_locker._log_mixin import write_signed_entry +from screen_locker._workout_ledger import ( + append_rows, + credit_row, + entry_done_at, + existing_rows, + ledger_path, + new_rows, + rest_row, +) +from screen_locker.tests._earned_fixtures import KEY, signing_key + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_DAY = date(2026, 10, 4) +_ISO = _DAY.isoformat() + + +def local(hhmm: str, day: date = _DAY) -> float: + hours, minutes = map(int, hhmm.split(":")) + return datetime.combine(day, time(hours, minutes)).astimezone().timestamp() + + +@pytest.fixture +def key(tmp_path: Path) -> Iterator[Path]: + with signing_key(tmp_path) as path: + yield path + + +def entry( + data: dict[str, Any], stamp: str = "2026-10-04T20:00:00+00:00", **extra: Any +) -> dict[str, Any]: + return {"timestamp": stamp, "workout_data": data, **extra} + + +def ledger() -> list[dict[str, Any]]: + return json.loads(ledger_path().read_text())["entries"] + + +class TestEntryDoneAt: + def test_the_tcx_end_wins(self) -> None: + done = entry({"type": "runnerup_verified", "completed_at": 123.5}) + assert entry_done_at(done, _DAY) == 123.5 + + def test_a_numeric_string_and_iso_are_read(self) -> None: + assert entry_done_at(entry({"completed_at": "77.5"}), _DAY) == 77.5 + iso = entry({"completed_at": "2026-10-04T10:00:00+00:00"}) + assert ( + entry_done_at(iso, _DAY) + == datetime.fromisoformat("2026-10-04T10:00:00+00:00").timestamp() + ) + + def test_a_manual_end_time_is_on_its_day(self) -> None: + manual = entry({"type": "manual_workout", "end_time": "18:40"}) + assert entry_done_at(manual, _DAY) == local("18:40") + + def test_a_pc_session_is_its_start_plus_duration(self) -> None: + session = entry( + { + "type": "pc_workout_verified", + "sync_record_id": "2026-10-04T15:42:00", + "duration_minutes": "90.0", + } + ) + assert entry_done_at(session, _DAY) == local("17:12") + + def test_a_session_without_a_duration_falls_back_to_logging( + self, caplog: pytest.LogCaptureFixture + ) -> None: + session = entry( + {"type": "pc_workout_verified", "sync_record_id": "2026-10-04T15:42:00"} + ) + with caplog.at_level(logging.WARNING): + done = entry_done_at(session, _DAY) + assert done == datetime.fromisoformat("2026-10-04T20:00:00+00:00").timestamp() + assert "no usable duration" in caplog.text + + def test_nothing_usable_is_none(self, caplog: pytest.LogCaptureFixture) -> None: + bare = {"workout_data": "?", "timestamp": "never"} + with caplog.at_level(logging.WARNING): + assert entry_done_at(bare, _DAY) is None + assert "Unusable workout time" in caplog.text + + +class TestCreditRow: + def test_each_source_is_named(self) -> None: + # No workout_id on the entry: the credit slot names the row. + cases = { + "runnerup_verified": ("runnerup_tcx", f"runnerup_verified:{_ISO}#0"), + "phone_verified": ("stronglifts", f"stronglifts:{_ISO}"), + "manual_workout": ("manual", f"manual_workout:{_ISO}#0"), + } + for wtype, (source, entry_id) in cases.items(): + row = credit_row(_ISO, [entry({"type": wtype})], 0) + assert row is not None + assert row["detail"]["source"] == source + assert row["entry_id"] == entry_id + + def test_a_non_credit_has_no_row(self) -> None: + assert credit_row(_ISO, [entry({"type": "relaxed_day_skip"})], 0) is None + + def test_both_stronglifts_paths_share_one_row_and_the_real_end(self) -> None: + phone = entry({"type": "phone_verified"}, workout_id="phone_verified:x") + pc = entry( + { + "type": "pc_workout_verified", + "sync_record_id": "2026-10-04T15:42:00", + "duration_minutes": "90", + } + ) + first = credit_row(_ISO, [phone, pc], 0) + second = credit_row(_ISO, [phone, pc], 1) + assert first == second + assert first is not None + assert first["entry_id"] == "phone_verified:x" + assert float(first["detail"]["completed_at"]) == local("17:12") + + def test_a_credit_with_no_time_writes_no_row( + self, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.WARNING): + row = credit_row(_ISO, [{"workout_data": {"type": "phone_verified"}}], 0) + assert row is None + assert "no usable time" in caplog.text + + def test_a_rest_row_is_stamped_at_its_midnight(self) -> None: + row = rest_row(_DAY, 5.0) + assert row["detail"] == { + "completed_at": str(local("00:00")), + "declared_at": "5.0", + "source": "rest_day", + } + + +class TestAppend: + def test_rows_are_signed_once(self, key: Path) -> None: + del key + row = credit_row(_ISO, [entry({"type": "phone_verified"})], 0) + assert row is not None + assert append_rows([row, dict(row)]) == 1 + assert append_rows([dict(row)]) == 0 + (stored,) = ledger() + assert earned_time.verified(stored, KEY) + + def test_no_key_writes_nothing(self, caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING): + assert append_rows([rest_row(_DAY, 1.0)]) == 0 + assert "NOT updated" in caplog.text + assert not ledger_path().exists() + + @pytest.mark.usefixtures("key") + def test_a_malformed_ledger_is_left_alone( + self, caplog: pytest.LogCaptureFixture + ) -> None: + ledger_path().parent.mkdir(parents=True) + ledger_path().write_text("[]") + with caplog.at_level(logging.WARNING): + assert append_rows([rest_row(_DAY, 1.0)]) == 0 + assert existing_rows(ledger_path()) is None + assert ledger_path().read_text() == "[]" + + def test_new_rows_dedups_within_and_against_the_ledger(self) -> None: + a, b = rest_row(_DAY, 1.0), rest_row(date(2026, 10, 5), 1.0) + assert new_rows([a, b, dict(b)], [a]) == [b] + + @pytest.mark.usefixtures("key") + def test_every_log_credit_reaches_the_ledger(self, tmp_path: Path) -> None: + """The log write chokepoint is the ledger's only writer for credits.""" + log = tmp_path / "log.json" + write_signed_entry( + log, + _ISO, + {"type": "manual_workout", "start_time": "17:00", "end_time": "18:00"}, + ) + write_signed_entry(log, _ISO, {"type": "relaxed_day_skip"}) + (row,) = ledger() + assert row["detail"]["source"] == "manual" + assert float(row["detail"]["completed_at"]) == local("18:00") + + +def test_the_ledger_lives_under_the_redirected_home() -> None: + assert ledger_path().is_relative_to(_workout_ledger._earned.LEDGER_HOME) diff --git a/screen_locker/tests/test_workout_ledger_cli.py b/screen_locker/tests/test_workout_ledger_cli.py new file mode 100644 index 0000000..03d6ba8 --- /dev/null +++ b/screen_locker/tests/test_workout_ledger_cli.py @@ -0,0 +1,95 @@ +"""Tests for ``--backfill-workout-ledger``: idempotent, per source, range-aware.""" + +from __future__ import annotations + +from datetime import date, datetime +import json +from typing import TYPE_CHECKING, Any +from unittest.mock import patch + +import pytest + +from screen_locker._rest_day import declare +from screen_locker._workout_ledger import ledger_path +from screen_locker._workout_ledger_cli import backfill_rows, run_backfill +from screen_locker.tests._earned_fixtures import signing_key + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_DAY = date(2026, 10, 4) +_ISO = _DAY.isoformat() + + +@pytest.fixture +def key(tmp_path: Path) -> Iterator[Path]: + with signing_key(tmp_path) as path: + yield path + + +def entry(data: dict[str, Any]) -> dict[str, Any]: + return {"timestamp": "2026-10-04T20:00:00+00:00", "workout_data": data} + + +def ledger() -> list[dict[str, Any]]: + return json.loads(ledger_path().read_text())["entries"] + + +@pytest.mark.usefixtures("key") +class TestBackfill: + def _log(self, tmp_path: Path) -> Path: + log = tmp_path / "log.json" + log.write_text( + json.dumps( + { + _ISO: [ + entry({"type": "phone_verified"}), + entry({"type": "relaxed_day_skip"}), + ], + "2026-10-05": [ + entry({"type": "runnerup_verified", "completed_at": 9.0}) + ], + } + ) + ) + return log + + def test_rows_for_a_range_and_its_rest_days(self, tmp_path: Path) -> None: + with patch("screen_locker._rest_day.datetime") as clock: + clock.now.return_value = datetime(2026, 10, 1).astimezone() + assert declare(date(2026, 10, 6)).ok + ledger_path().unlink() # the declaration mirrored it; backfill must too + rows = backfill_rows(self._log(tmp_path), _DAY, date(2026, 10, 6)) + assert [r["detail"]["source"] for r in rows] == [ + "stronglifts", + "runnerup_tcx", + "rest_day", + ] + + def test_dry_run_prints_and_writes_nothing( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + argv = ["--dry-run", "--from", _ISO, "--to", "2026-10-05"] + assert run_backfill(self._log(tmp_path), argv) == 0 + out = capsys.readouterr().out + assert "(missing)" in out + assert "per source: {'runnerup_tcx': 1, 'stronglifts': 1}" in out + assert not ledger_path().exists() + + def test_a_real_run_is_idempotent( + self, tmp_path: Path, capsys: pytest.CaptureFixture[str] + ) -> None: + log = self._log(tmp_path) + assert run_backfill(log, ["--from", _ISO, "--to", "2026-10-05"]) == 0 + assert run_backfill(log, ["--from", _ISO, "--to", "2026-10-05"]) == 0 + out = capsys.readouterr().out.splitlines() + assert out[0].startswith("2 new row(s)") + assert out[1].startswith("0 new row(s)") + assert run_backfill(log, ["--dry-run"]) == 0 + assert "(present)" in capsys.readouterr().out + + def test_default_is_today(self, tmp_path: Path) -> None: + with patch("screen_locker._workout_ledger_cli.today_str", return_value=_ISO): + assert run_backfill(self._log(tmp_path), []) == 0 + assert len(ledger()) == 1 diff --git a/scripts/_preview_sim.py b/scripts/_preview_sim.py new file mode 100644 index 0000000..be5674f --- /dev/null +++ b/scripts/_preview_sim.py @@ -0,0 +1,105 @@ +"""Simulation doubles for ``preview_shutdown.py``: real code, temporary state. + +Split out to keep the script under the 250-line cap. Every class here runs +the production methods it inherits; only the I/O edges (the drop directory, +adb, the sudo-written config, the sick history) point at a temporary place. +""" + +from __future__ import annotations + +from datetime import date, datetime, time +import json +from typing import TYPE_CHECKING + +import earned_time + +from screen_locker import _grace_floor, _sick_tracker +from screen_locker._earned import first_minutes +from screen_locker._runnerup_verification import RunnerUpVerificationMixin +from screen_locker._shutdown import ShutdownMixin + +if TYPE_CHECKING: + from pathlib import Path + +_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" +HELPER_CEILING = 23 * 60 # adjust_shutdown_schedule.sh's RESTORE_CEILING + + +def stamp(day: date, hhmm_text: str) -> float: + """Unix time of ``HH:MM`` local on ``day``.""" + hours, _, minutes = hhmm_text.partition(":") + moment = datetime.combine(day, time(int(hours), int(minutes))).astimezone() + return moment.timestamp() + + +def _iso(stamp: float) -> str: + return datetime.fromtimestamp(stamp).astimezone().isoformat() + + +def write_walk(folder: Path, day: date, start: str, minutes: int) -> None: + """A RunnerUp-style walk export: Sport="Other", the sport in the name.""" + begin = stamp(day, start) + points = "".join( + f"" + f"{s * 1.3:.1f}" + for s in range(0, minutes * 60 + 1, 5) + ) + stem = f"RunnerUp_Pixel_6a_{day}-{start.replace(':', '-')}-00_Walking.tcx" + (folder / stem).write_text( + f'' + f"{day}T{start}:00Z{minutes * 60}" + f"{minutes * 78}" + f"{points}", + encoding="utf-8", + ) + + +class SimRunnerUp(RunnerUpVerificationMixin): + """The real backfill, reading only a temporary drop directory.""" + + def __init__(self, folder: Path) -> None: + self.folder = folder + + def _has_adb_device(self) -> bool: + return False + + def fill(self, day: str, log: Path) -> bool: + """The real RunnerUp backfill for ``day`` into ``log``.""" + return self._try_fill_runnerup_for_date(day, log) + + def _find_runnerup_exports_for_date(self, date_str: str) -> list[str]: + return sorted(str(p) for p in self.folder.glob(f"*{date_str}*.tcx")) + + +class SimShutdown(ShutdownMixin): + """The real add/absorb paths against an in-memory config.""" + + def __init__(self, minutes: int) -> None: + self.config = (minutes, minutes, 5 * 60) + + def _read_shutdown_config(self) -> tuple[int, int, int] | None: + return self.config + + def _write_shutdown_config( + self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = False + ) -> bool: + del restore + clamp = min(HELPER_CEILING, mon_wed), min(HELPER_CEILING, thu_sun) + self.config = (*clamp, morning_end) + return True + + def credit(self, name: str, day: date, first_done: float) -> None: + """One live pass: the earner's add (lift-absorbing), then the floor.""" + if name == earned_time.WORKOUT.name: + self._adjust_shutdown_time_later() + else: + self._adjust_shutdown_time_by(first_minutes(earned_time.earner(name), day)) + self._apply_grace_floor(None, first_done=first_done) + + +def sick_history(day: date, tmp: Path, *, sick_day: bool) -> bool: + """A temp sick history naming ``day``, read back through the real check.""" + history = tmp / "sick_history.json" + history.write_text(json.dumps({"sick_days": [day.isoformat()] if sick_day else []})) + vars(_sick_tracker)["SICK_HISTORY_FILE"] = history + return _grace_floor.sick_on(day) diff --git a/scripts/preview_shutdown.py b/scripts/preview_shutdown.py new file mode 100755 index 0000000..2f509a3 --- /dev/null +++ b/scripts/preview_shutdown.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Print the shutdown the locker would derive -- today, or a simulated day. + +READ-ONLY on the real system: it never runs the sudo helper, never writes +``/etc``, ``shutdown_base.json``, ``grace_floor.json``, ``rest_days.json``, +``log.json`` or the real workout ledger. Simulations run the real code paths +(TCX parsing + walk summing via the RunnerUp backfill, rest-day signing and +verification, ``_shutdown_target.derive``, the grace floor and its lift +bookkeeping) against a temporary directory, with an in-memory shutdown config. + + python -m scripts.preview_shutdown # today, real ledgers + python -m scripts.preview_shutdown --day D --walk 25,20 + python -m scripts.preview_shutdown --day D --rest-day + python -m scripts.preview_shutdown --day D --first-done 18:55 --done anki + python -m scripts.preview_shutdown --day D --first-done 18:55 --sick-day + python -m scripts.preview_shutdown --day D --sequence anki@21:00,reading@21:20 + +Run it with ``PYTHONPATH=~/src/utils/earned_time`` to preview the ladder +(earned_time working tree) instead of the installed version. +""" + +from __future__ import annotations + +import argparse +from datetime import date, datetime, time, timedelta +from pathlib import Path +import sys +import tempfile + +import earned_time + +from screen_locker import _earned, _grace_floor, _shutdown +from screen_locker._constants import EXTRA_BENEFITS_FILE, SHUTDOWN_CONFIG_FILE +from screen_locker._day import today_str +from screen_locker._earned import hhmm +from screen_locker._extra_benefits import weekly_shutdown_bonus_hours +from screen_locker._rest_day import declare, is_rest_day +from screen_locker._shutdown import read_shutdown_config +from screen_locker._shutdown_target import ( + DayInputs, + Target, + day_credit_count, + derive, + gather, +) +from scripts._preview_sim import ( + HELPER_CEILING, + SimRunnerUp, + SimShutdown, + sick_history, + stamp, + write_walk, +) + +_LOG = Path(__file__).resolve().parents[1] / "screen_locker" / "log.json" +_SICK_STEP, _SICK_FLOOR = 60, 18 * 60 # _shutdown's sick-day step and floor + + +def _say(text: str) -> None: + sys.stdout.write(f"{text}\n") + + +def _report(label: str, target: Target) -> None: + res = target.resolution + terms = ", ".join( + f"{t.earner.name}={t.answer}:+{t.shutdown_minutes}" for t in res.terms + ) + first = ( + datetime.fromtimestamp(target.first_done_at).astimezone().strftime("%H:%M") + if target.first_done_at + else "-" + ) + grace = hhmm(target.grace) if target.grace is not None else "-" + _say(f"[{label}] {res.day} base {hhmm(res.base.shutdown_minutes)} | {terms}") + _say( + f" rest_day={target.rest_day} earned={hhmm(target.earned)} " + f"first_done={first} grace={grace} -> SHUTDOWN {hhmm(target.minutes)}" + ) + + +def _simulate(args: argparse.Namespace, tmp: Path) -> None: + day = date.fromisoformat(args.day) + # Every ledger (the workout one included) resolves under the temp dir. + vars(_earned)["LEDGER_HOME"] = tmp + log = tmp / "log.json" + if args.walk: + drop = tmp / "drop" + drop.mkdir() + for index, minutes in enumerate(int(m) for m in args.walk.split(",")): + write_walk(drop, day, f"{8 + index * 9:02d}:15", minutes) + SimRunnerUp(drop).fill(day.isoformat(), log) + rest_file = tmp / "rest_days.json" + if args.rest_day: + eve = datetime.combine(day - timedelta(days=1), time(12)).astimezone() + _say(f" declare: {declare(day, now=eve, path=rest_file).reason}") + rest = is_rest_day(day, rest_file) + done = {n for n in (args.done or "").split(",") if n} + flat = {e.name: e.name in done for e in _earned.flat_earners()} + first = stamp(day, args.first_done) if args.first_done else None + logged = _grace_floor.workout_done_at(log, day) + times = [t for t in (first, logged) if t is not None] + target = derive( + DayInputs( + day, + flat, + day_credit_count(log, day), + rest_day=rest, + first_done=min(times, default=None), + sick_day=sick_history(day, tmp, sick_day=args.sick_day), + ) + ) + _report("simulated", target) + if args.sick_day: + # What the lock screen's sick-day choice then does to that (the + # existing _apply_earlier_shutdown rule); no grace floor lifts it back. + after = max(_SICK_FLOOR, target.minutes - _SICK_STEP) + _say(f" sick-day step at the lock screen -> SHUTDOWN {hhmm(after)}") + + +def _sequence(args: argparse.Namespace, tmp: Path) -> None: + day = date.fromisoformat(args.day) + for module in (_grace_floor, _shutdown): + vars(module)["today_str"] = lambda: args.day + vars(_grace_floor)["GRACE_STATE_FILE"] = tmp / "grace_floor.json" + done = {e.name: False for e in _earned.flat_earners()} + start = derive(DayInputs(day, done)) + sim = SimShutdown(start.minutes) + _say(f"[sequence] {day} start {hhmm(start.minutes)} (nothing done)") + workouts, first = 0, None + for step in args.sequence.split(","): + name, _, at = step.partition("@") + first = min(first or stamp(day, at), stamp(day, at)) + if name == earned_time.WORKOUT.name: + workouts += 1 + else: + done[name] = True + sim.credit(name, day, first) + want = derive(DayInputs(day, done, workouts, first_done=first)) + verdict = "" if sim.config[0] == want.minutes else " MISMATCH" + _say( + f" {name}@{at}: config {hhmm(sim.config[0])} " + f"(derived {hhmm(want.minutes)}{verdict})" + ) + + +def main() -> int: + """Print the requested preview. Returns 0.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--day", default=None, help="simulate this day (YYYY-MM-DD)") + parser.add_argument("--walk", help="moving minutes per walk leg, e.g. 25,20") + parser.add_argument("--rest-day", action="store_true") + parser.add_argument("--sick-day", action="store_true") + parser.add_argument("--first-done", help="HH:MM a ledger earner was first done") + parser.add_argument("--done", help="flat earners done, e.g. anki,leetcode") + parser.add_argument("--sequence", help="name@HH:MM,... live passes in order") + args = parser.parse_args() + ladder = getattr(earned_time, "on_ladder", None) + _say(f"earned_time from {Path(earned_time.__file__).parent}") + if args.day is None: + today = date.fromisoformat(today_str()) + on = ladder(today) if ladder else False + _say(f" ladder in force today: {on}") + live = read_shutdown_config(SHUTDOWN_CONFIG_FILE) + _say( + f" live config: {tuple(hhmm(m) for m in live) if live else 'unreadable'}" + ) + target = gather(today, _LOG) + _report("today, real ledgers", target) + weekly = weekly_shutdown_bonus_hours(EXTRA_BENEFITS_FILE) * 60 + _say( + f" + weekly streak bonus {weekly} min (layered after the reset, " + f"absorbs the grace lift) -> expected config " + f"{hhmm(min(HELPER_CEILING, max(target.earned + weekly, target.minutes)))}" + ) + return 0 + day = date.fromisoformat(args.day) + _say(f" ladder in force on {day}: {ladder(day) if ladder else False}") + with tempfile.TemporaryDirectory(prefix="preview_shutdown_") as raw: + if args.sequence: + _sequence(args, Path(raw)) + else: + _simulate(args, Path(raw)) + return 0 + + +if __name__ == "__main__": + sys.exit(main())