diff --git a/earned_time/README.md b/earned_time/README.md index b81873e..4df2669 100644 --- a/earned_time/README.md +++ b/earned_time/README.md @@ -8,8 +8,76 @@ gaming time and shutdown time, read by both consumers. | steam-backlog-enforcer | `Resolution.gaming_minutes` | `_budget_resolve.resolve_budget` | | screen-locker | `Resolution.shutdown_minutes` | `_shutdown_base.reset_to_base_if_new_day` + live pass | -A gate (leetcode-guard, book-guard, ...) never touches either. It publishes a -fact — HMAC-signed `credit` rows in its ledger — and is registered here once. +A gate (leetcode-guard, book-guard, anki-guard, screen-locker's workout +credit, ...) never touches either. It publishes a fact — HMAC-signed `credit` +rows in its ledger — and is registered here once. + +## The model + +**Gaming** (unchanged): a base of 5 h, minus every earner's `gaming_minutes` +from its `penalty_from` date; each earner done adds its `gaming_minutes` back. +Capped at 8 h. + +**Shutdown** has two regimes, switched by date (`LADDER_FROM`, 2026-10-10 — +never a same-day cut): + +- **Before `LADDER_FROM`:** base 20:00 minus the penalties in force, plus + each earner's `shutdown_minutes` (a second workout +60), capped at 23:00. +- **From `LADDER_FROM` — the sleep ladder.** The ceiling is eight hours of + sleep before the alarm: `WAKE_MINUTES` (07:00) − 8 h = 23:00. The floor is + that ceiling minus the sum of every registered earner's first-unit rung + (`LADDER`, a `Rung` per earner name), so doing everything lands exactly on + the ceiling: + + | done | shutdown | + |---|---| + | nothing | 19:00 | + | workout (110) | 20:50 | + | + LeetCode (50) | 21:40 | + | + reading (30) | 22:10 | + | + Anki (25) | 22:35 | + | + Automation (25) | 23:00 | + + A second workout earns nothing more (`Rung.extra` defaults to 0). + Move the alarm (`WAKE_MINUTES = 6 * 60`) and the whole ladder shifts an hour + earlier on its own. Days before `LADDER_FROM` keep their frozen 23:00 + ceiling, so the change never rewrites history. + +The two floors are derived separately (`_resolve._shutdown_floor`), so the +ladder cannot move a gaming minute. + +## API + +- `resolve(answers, day=None, earners=EARNERS) -> Resolution` — the one sum. +- `base_for(day=None, earners=EARNERS) -> Base` — the day's floors. +- `done_today(earner, ledger, key_file, *, now=None) -> bool | None` +- `first_credit_at(earner, ledger, key_file, day) -> float | None` — unix time + of the earliest verified credit that counts for `day`. The time is the + stamp the earner's `match` decides on (`_match.CREDIT_STAMPS`: LeetCode + `submitted_at`, reading `ended_at`, workout `completed_at`; Anki/Automation + fall back to `created_at`). It can lie outside `day` when a gate's day is + not the calendar day (Anki's rollover, a rest day declared the evening + before). +- `credit_units(earner, ledger, key_file, day) -> int | None` — how many + verified rows count for `day`: a counted earner's units. +- `shutdown_minutes_for(earner, day) -> int`, + `extra_shutdown_minutes_for(earner, day) -> int`, + `shutdown_ceiling_for(day) -> int`, `on_ladder(day) -> bool` — what a status + view needs to list what is still left to earn. `Earner.shutdown_for(units, + day=None)` is the day-aware sum (default today). Never read + `earner.shutdown_minutes` directly: it is the pre-ladder value. +- `day_window(day)`, `today_window(now=None)`, `entry_signature`, `verified`. + +## The workout ledger + +`WORKOUT` is `kind="counted"` with a ledger at +`~/.local/share/workout_locker/ledger.json`, written by screen-locker: one +signed credit row per credited unit, +`{"kind": "credit", "entry_id", "day", "detail": {"completed_at", "source"}}`. +`workout_match` counts a `runnerup_tcx` row when `completed_at` falls in the +window, and a `rest_day` row on the `day` it names — only if +`detail.declared_at` is before that day's local midnight. A rest day declared +on the day itself is logged and does not count. ## Adding a gate @@ -17,45 +85,50 @@ fact — HMAC-signed `credit` rows in its ledger — and is registered here once `kind: "credit"`, a `detail` dict, and an `hmac` signed with `/etc/workout-locker/hmac.key` (`earned_time.entry_signature`, the same as gatelock's `log_integrity`). -2. Add an `Earner` to `earned_time/_policy.py` and to `EARNERS`: +2. Add a matcher to `earned_time/_match.py` and an `Earner` to + `earned_time/_policy.py` and `EARNERS`: ```python - ANKI = Earner( - name="anki", # also screen-locker's "anki_bonus_date" stamp - label="Anki", + PIANO = Earner( + name="piano", # also screen-locker's "piano_bonus_date" stamp + label="piano", gaming_minutes=30, - shutdown_minutes=30, - penalty_from=date(2026, 10, 6), # base drops by the same 30/30 from then on - ledger=".local/share/anki_guard/ledger.json", - match=_anki_match, # which verified credit rows count today + shutdown_minutes=30, # pre-ladder value + penalty_from=date(2026, 11, 1), # gaming base drops by 30 from then on + ledger=".local/share/piano_guard/ledger.json", + match=piano_match, # which verified credit rows count today missing_ledger_is_no=True, ) ``` + Then give it a rung in `LADDER` (`PIANO.name: Rung(20)`), and, if its + matcher decides on an event stamp, an entry in `_match.CREDIT_STAMPS`. + + On the ladder a new earner's rung comes out of the floor the day its pin + lands (the ceiling is fixed): re-split the rungs and ship it the day the + gate can actually be satisfied. 3. Bump the version, tag `earned-time-vX.Y.Z`, and bump the pin in both consumers. Neither needs code for a ledger-backed flat earner. -`penalty_from` is the gate's start date, so the base is never cut before the -reward that pays it back exists. The best case stays the same, because the -base drops by exactly what the new earner adds. - ## Rules the consumers rely on - **Minutes everywhere.** Shutdown is minutes after local midnight. - screen-locker still stores whole hours and refuses (raises) a remainder - rather than flooring it. -- **Paths are passed in.** `done_today(earner, ledger, key_file)` never - resolves a real path itself; consumers' test suites redirect their own - constants. +- **Paths are passed in.** No reader resolves a real path itself; consumers' + test suites redirect their own constants. - **`None` is "could not check", never "no".** It earns nothing (fail closed), but consumers log it. A missing answer in `resolve()` is logged as well; an unknown earner name raises. +- **The HMAC key is root-owned and 0644 on purpose.** The gates run as the + user and must sign their rows, so the user can read the key; root owns it + so a user process cannot replace or rotate it. The signature therefore + stops casual edits and corruption, not a determined forger with a shell -- + the accepted trade-off of user-run gates. - **Stdlib only.** The gaming daemon imports this as root, headless. ## Install ```console -pip install --user "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time" +pip install --user --break-system-packages --no-deps "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.4.0#subdirectory=earned_time" ``` The root `steam-backlog-enforcer.service` sets `HOME=/home/kuhy`, so it imports diff --git a/earned_time/earned_time/__init__.py b/earned_time/earned_time/__init__.py index e9db012..f529ae7 100644 --- a/earned_time/earned_time/__init__.py +++ b/earned_time/earned_time/__init__.py @@ -28,6 +28,22 @@ Adding a gate is one :class:`Earner` in :mod:`earned_time._policy`; with from __future__ import annotations +from earned_time._credits import ( + credit_units, + day_window, + first_credit_at, +) +from earned_time._ladder import ( + LADDER, + LADDER_FROM, + SHUTDOWN_CEILING_MINUTES, + WAKE_MINUTES, + Rung, + extra_shutdown_minutes_for, + on_ladder, + shutdown_ceiling_for, + shutdown_minutes_for, +) from earned_time._ledger import done_today, entry_signature, today_window, verified from earned_time._policy import ( ANKI, @@ -38,7 +54,6 @@ from earned_time._policy import ( LEETCODE, READING, SHUTDOWN_BASE_MINUTES, - SHUTDOWN_CEILING_MINUTES, WORKOUT, Earner, earner, @@ -51,20 +66,31 @@ __all__ = [ "EARNERS", "GAMING_BASE_MINUTES", "GAMING_CEILING_MINUTES", + "LADDER", + "LADDER_FROM", "LEETCODE", "READING", "SHUTDOWN_BASE_MINUTES", "SHUTDOWN_CEILING_MINUTES", + "WAKE_MINUTES", "WORKOUT", "Base", "Earner", "Resolution", + "Rung", "Term", "base_for", + "credit_units", + "day_window", "done_today", "earner", "entry_signature", + "extra_shutdown_minutes_for", + "first_credit_at", + "on_ladder", "resolve", + "shutdown_ceiling_for", + "shutdown_minutes_for", "today_window", "verified", ] diff --git a/earned_time/earned_time/_credits.py b/earned_time/earned_time/_credits.py new file mode 100644 index 0000000..e9977eb --- /dev/null +++ b/earned_time/earned_time/_credits.py @@ -0,0 +1,107 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Day-scoped questions about a gate's signed ledger: when, and how many. + +:func:`~earned_time._ledger.done_today` asks "is there a credit yet today?". +These ask about a whole local day: when the first counting credit happened +(:func:`first_credit_at`) and how many count (:func:`credit_units`, a counted +earner's units). The same rules hold: paths are passed in, and ``None`` is +"could not check", never "no". +""" + +from __future__ import annotations + +from datetime import date, datetime, time +import logging +from typing import TYPE_CHECKING, Final + +from earned_time._ledger import counting_rows +from earned_time._match import CREDIT_STAMPS + +if TYPE_CHECKING: + from pathlib import Path + + from earned_time._policy import Earner, Window + +_logger: Final = logging.getLogger(__name__) + + +def day_window(day: date) -> Window: + """``day``'s local midnight to its last microsecond, as unix seconds. + + Each end is localised on its own, so a DST day gets its real length. + """ + start = datetime.combine(day, time.min).astimezone() + end = datetime.combine(day, time.max).astimezone() + return start.timestamp(), end.timestamp() + + +def credit_time(earner: Earner, row: dict[str, object]) -> float | None: + """When a credit row's work happened, as unix seconds; ``None`` if unusable. + + The detail field its matcher decides on (``CREDIT_STAMPS``) (LeetCode's + ``submitted_at``, reading's ``ended_at`` -- the stamp ``match`` decides + on), else the row's ``created_at``. + """ + detail = row.get("detail") + stamp = CREDIT_STAMPS.get(earner.match) if earner.match is not None else None + if stamp is not None and isinstance(detail, dict): + try: + return float(str(detail.get(stamp))) + except ValueError: + pass + raw = row.get("created_at") + try: + return datetime.fromisoformat(str(raw)).timestamp() + except ValueError: + _logger.warning( + "%s credit %r has no usable time", earner.label, row.get("entry_id") + ) + return None + + +def first_credit_at( + earner: Earner, ledger: Path, key_file: Path, day: date +) -> float | None: + """When ``earner``'s first verified credit that counts for ``day`` happened. + + Args: + earner: A ledger-backed earner (it must have ``match``). + ledger: The gate's ledger file, resolved by the caller. + key_file: The shared HMAC key, resolved by the caller. + day: The local day the credit must count for (by ``match``). + + Returns: + The earliest :func:`credit_time` among the counting rows, as unix + seconds. ``None`` when no row counts, or when the ledger or key could + not be read -- never a time that was not checked. + + Raises: + ValueError: ``earner`` has no ``match``; its answer is the consumer's. + """ + rows = counting_rows(earner, ledger, key_file, day_window(day)) + if rows is None: + return None + times = [t for t in (credit_time(earner, row) for row in rows) if t is not None] + return min(times, default=None) + + +def credit_units(earner: Earner, ledger: Path, key_file: Path, day: date) -> int | None: + """How many verified credit rows count for ``day`` -- a counted earner's units. + + Args: + earner: A ledger-backed earner (it must have ``match``). + ledger: The gate's ledger file, resolved by the caller. + key_file: The shared HMAC key, resolved by the caller. + day: The local day the rows must count for (by ``match``). + + Returns: + The count (``0`` included), or ``None`` when the ledger or key could + not be read -- never "no units". + + Raises: + ValueError: ``earner`` has no ``match``; its answer is the consumer's. + """ + rows = counting_rows(earner, ledger, key_file, day_window(day)) + if rows is None: + return None + return sum(1 for _ in rows) diff --git a/earned_time/earned_time/_ladder.py b/earned_time/earned_time/_ladder.py new file mode 100644 index 0000000..9961ac8 --- /dev/null +++ b/earned_time/earned_time/_ladder.py @@ -0,0 +1,94 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""The shutdown ladder: from ``LADDER_FROM``, shutdown is read down from sleep. + +The ceiling is ``WAKE_MINUTES`` minus eight hours (23:00 for a 07:00 wake); +the floor is that ceiling minus every earner's first-unit :class:`Rung`, so +doing everything lands exactly on the ceiling. Move the alarm and the whole +ladder moves with it. Days before ``LADDER_FROM`` keep their own values and a +frozen ceiling, so the switch never rewrites how an old day resolved. Gaming +is not on the ladder. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date +from types import MappingProxyType +from typing import TYPE_CHECKING, Final + +if TYPE_CHECKING: + from collections.abc import Mapping + + from earned_time._policy import Earner + +# The shutdown ceiling of every day before ``LADDER_FROM``. Frozen: a later +# ``WAKE_MINUTES`` change must not rewrite how an old day resolved. +SHUTDOWN_CEILING_MINUTES: Final = 23 * 60 + +# From this day shutdown follows the sleep ladder. A date, never a same-day +# cut: the day it was approved (2026-10-09) still resolves the old way. +LADDER_FROM: Final = date(2026, 10, 10) +# Alarm time, minutes after midnight. The ladder's ceiling is eight hours of +# sleep before it, so moving the alarm moves the whole ladder with it. +WAKE_MINUTES: Final = 7 * 60 +_SLEEP_MINUTES: Final = 8 * 60 +_DAY_MINUTES: Final = 24 * 60 + + +@dataclass(frozen=True) +class Rung: + """One earner's step on the shutdown ladder. + + Attributes: + first: Shutdown minutes the first unit earns. + extra: Each further unit (counted earners); the ladder pays none. + """ + + first: int + extra: int = 0 + + +# The rungs, keyed by earner name: they sum to the gap between the floor and +# the ceiling, so doing everything lands exactly on 23:00. A second workout +# earns nothing more. +LADDER: Final[Mapping[str, Rung]] = MappingProxyType( + { + "workout": Rung(110), + "leetcode": Rung(50), + "reading": Rung(30), + "anki": Rung(25), + "automation": Rung(25), + } +) + + +def on_ladder(day: date) -> bool: + """Whether ``day`` resolves shutdown on the sleep ladder.""" + return day >= LADDER_FROM + + +def _rung(item: Earner, day: date) -> Rung | None: + return LADDER.get(item.name) if on_ladder(day) else None + + +def shutdown_minutes_for(item: Earner, day: date) -> int: + """What ``item``'s first unit pushes shutdown later by on ``day``. + + Its ladder rung from ``LADDER_FROM`` on; before that, or for an earner + without a rung, its own ``shutdown_minutes``. + """ + rung = _rung(item, day) + return item.shutdown_minutes if rung is None else rung.first + + +def extra_shutdown_minutes_for(item: Earner, day: date) -> int: + """What each of ``item``'s further units earns on ``day`` (counted only).""" + rung = _rung(item, day) + return item.extra_shutdown_minutes if rung is None else rung.extra + + +def shutdown_ceiling_for(day: date) -> int: + """The latest shutdown ``day`` can earn, minutes after its midnight.""" + if on_ladder(day): + return _DAY_MINUTES + WAKE_MINUTES - _SLEEP_MINUTES + return SHUTDOWN_CEILING_MINUTES diff --git a/earned_time/earned_time/_ledger.py b/earned_time/earned_time/_ledger.py index 1dae65f..a6220f1 100644 --- a/earned_time/earned_time/_ledger.py +++ b/earned_time/earned_time/_ledger.py @@ -30,9 +30,10 @@ import logging from typing import TYPE_CHECKING, Final if TYPE_CHECKING: + from collections.abc import Iterator from pathlib import Path - from earned_time._policy import Earner + from earned_time._policy import Earner, Window _logger: Final = logging.getLogger(__name__) @@ -115,6 +116,36 @@ def read_rows(ledger: Path, earner: Earner) -> list[object] | None: return rows +def counting_rows( + earner: Earner, ledger: Path, key_file: Path, window: Window +) -> Iterator[dict[str, object]] | None: + """The verified ``credit`` rows that ``earner.match`` counts in ``window``. + + ``None`` when the key or the ledger cannot be read. + + Raises: + ValueError: ``earner`` has no ``match``; its answer is the consumer's. + """ + if earner.match is None: + msg = f"earner {earner.name!r} has no shared reader; supply its answer" + raise ValueError(msg) + key = read_key(key_file) + if key is None: + return None + rows = read_rows(ledger, earner) + if rows is None: + return None + match = earner.match + return ( + row + for row in rows + if isinstance(row, dict) + and row.get("kind") == _CREDIT + and verified(row, key) + and match(row, window) + ) + + def done_today( earner: Earner, ledger: Path, @@ -137,21 +168,7 @@ def done_today( Raises: ValueError: ``earner`` has no ``match``; its answer is the consumer's. """ - if earner.match is None: - msg = f"earner {earner.name!r} has no shared reader; supply its answer" - raise ValueError(msg) - key = read_key(key_file) - if key is None: - return None - rows = read_rows(ledger, earner) + rows = counting_rows(earner, ledger, key_file, today_window(now)) if rows is None: return None - window = today_window(now) - match = earner.match - return any( - isinstance(row, dict) - and row.get("kind") == _CREDIT - and verified(row, key) - and match(row, window) - for row in rows - ) + return any(True for _ in rows) diff --git a/earned_time/earned_time/_match.py b/earned_time/earned_time/_match.py new file mode 100644 index 0000000..b225377 --- /dev/null +++ b/earned_time/earned_time/_match.py @@ -0,0 +1,126 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Which HMAC-verified ``credit`` rows count for a window, per gate. + +Each ledger-backed :class:`~earned_time._policy.Earner` names one of these as +its ``match``. They see only rows that already passed the signature check. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +import logging +from types import MappingProxyType +from typing import TYPE_CHECKING, Final + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + Window = tuple[float, float] + CreditMatch = Callable[[dict[str, object], Window], bool] + +_logger: Final = logging.getLogger(__name__) + + +def _within(raw: object, window: Window) -> bool | None: + """Whether a unix-seconds stamp falls in ``window``; ``None`` if unusable.""" + try: + stamp = float(str(raw)) + except ValueError: + return None + start, end = window + return start <= stamp <= end + + +def leetcode_match(row: dict[str, object], window: Window) -> bool: + """A solve counts on the day LeetCode accepted it (``submitted_at``). + + ``day`` is the *harvesting* run's date and runs late, so it is only the + fallback for a row without a usable stamp: it can miss a late solve but + never invent one. + """ + detail = row.get("detail") + raw = detail.get("submitted_at") if isinstance(detail, dict) else None + if raw is not None: + landed = _within(raw, window) + if landed is not None: + return landed + _logger.warning( + "LeetCode credit %r has an unparsable submitted_at (%r); " + "falling back to its day key", + row.get("entry_id"), + raw, + ) + local = datetime.fromtimestamp(window[0], tz=UTC).astimezone() + return row.get("day") == local.date().isoformat() + + +def reading_match(row: dict[str, object], window: Window) -> bool: + """A reading credit counts if it earned the bonus and ended today.""" + detail = row.get("detail") + if not isinstance(detail, dict) or detail.get("bonus") != "1": + return False + landed = _within(detail.get("ended_at"), window) + if landed is None: + _logger.warning("reading credit %r has no usable ended_at", row.get("entry_id")) + return False + return landed + + +def anki_match(row: dict[str, object], window: Window) -> bool: + """An Anki credit counts on its Anki day (``detail.anki_day``). + + anki-guard writes one row per Anki day, keyed on the collection's own day + boundary, so the day it names is the day it pays -- no stamp to parse. + """ + detail = row.get("detail") + local = datetime.fromtimestamp(window[0], tz=UTC).astimezone().date() + return isinstance(detail, dict) and detail.get("anki_day") == local.isoformat() + + +def workout_match(row: dict[str, object], window: Window) -> bool: + """A workout credit counts when it was completed in the window. + + screen-locker writes one row per credited unit. A ``rest_day`` row is a + credit for the day it names (``day``), not for when it was written, and + only if it was declared (``declared_at``) before that day began: a rest + day declared on the day itself is a skipped workout, not a rest day. + """ + detail = row.get("detail") + if not isinstance(detail, dict): + return False + if detail.get("source") == "rest_day": + local = datetime.fromtimestamp(window[0], tz=UTC).astimezone().date() + if row.get("day") != local.isoformat(): + return False + try: + declared = float(str(detail.get("declared_at"))) + except ValueError: + declared = None + if declared is None or declared >= window[0]: + _logger.warning( + "rest-day credit %r was not declared before its day (declared_at " + "%r); it does not count", + row.get("entry_id"), + detail.get("declared_at"), + ) + return False + return True + landed = _within(detail.get("completed_at"), window) + if landed is None: + _logger.warning( + "workout credit %r has no usable completed_at", row.get("entry_id") + ) + return False + return landed + + +# The ``detail`` field each matcher decides on: the unix second the work +# happened, which :func:`earned_time._credits.credit_time` reports. A matcher +# without one (Anki's day key) falls back to the row's ``created_at``. +CREDIT_STAMPS: Final[Mapping[CreditMatch, str]] = MappingProxyType( + { + leetcode_match: "submitted_at", + reading_match: "ended_at", + workout_match: "completed_at", + } +) diff --git a/earned_time/earned_time/_policy.py b/earned_time/earned_time/_policy.py index 40028da..ae719ac 100644 --- a/earned_time/earned_time/_policy.py +++ b/earned_time/earned_time/_policy.py @@ -13,15 +13,25 @@ reader could not yet earn back was a same-day loss on 2026-09-26. **Ceilings stay put.** Adding a penalised earner leaves the best case unchanged: the base drops by what the new earner adds. + +From ``LADDER_FROM`` shutdown follows the sleep ladder instead +(:mod:`earned_time._ladder`); gaming keeps the penalty derivation unchanged. """ from __future__ import annotations from dataclasses import dataclass from datetime import UTC, date, datetime -import logging from typing import TYPE_CHECKING, Final, Literal +from earned_time._ladder import extra_shutdown_minutes_for, shutdown_minutes_for +from earned_time._match import ( + anki_match, + leetcode_match, + reading_match, + workout_match, +) + if TYPE_CHECKING: from collections.abc import Callable @@ -29,8 +39,6 @@ if TYPE_CHECKING: # Whether one HMAC-verified ``credit`` row counts for the window. CreditMatch = Callable[[dict[str, object], Window], bool] -_logger: Final = logging.getLogger(__name__) - # The day before any earner's penalty: 5h of gaming, shutdown at 20:00. # Minutes, both of them -- shutdown is minutes after local midnight. GAMING_BASE_MINUTES: Final = 5 * 60 @@ -38,7 +46,6 @@ SHUTDOWN_BASE_MINUTES: Final = 20 * 60 # Hard ceilings, whatever the terms add up to. GAMING_CEILING_MINUTES: Final = 8 * 60 -SHUTDOWN_CEILING_MINUTES: Final = 23 * 60 @dataclass(frozen=True) @@ -63,6 +70,10 @@ class Earner: match: Which verified ``credit`` rows count for today. missing_ledger_is_no: A ledger that does not exist yet is an honest "no" (the gate never ran), not a fault. + + ``shutdown_minutes`` and ``extra_shutdown_minutes`` are the pre-ladder + values; from ``LADDER_FROM`` the earner's :class:`Rung` in ``LADDER`` + applies (:func:`shutdown_minutes_for`). """ name: str @@ -84,67 +95,14 @@ class Earner: """Gaming minutes ``units`` earn: the first unit only, never more.""" return self.gaming_minutes if units > 0 else 0 - def shutdown_for(self, units: int) -> int: - """Shutdown minutes ``units`` earn: first unit, then each extra one.""" + def shutdown_for(self, units: int, day: date | None = None) -> int: + """Shutdown minutes ``units`` earn on ``day`` (default today).""" if units <= 0: return 0 - return self.shutdown_minutes + (units - 1) * self.extra_shutdown_minutes - - -def _within(raw: object, window: Window) -> bool | None: - """Whether a unix-seconds stamp falls in ``window``; ``None`` if unusable.""" - try: - stamp = float(str(raw)) - except ValueError: - return None - start, end = window - return start <= stamp <= end - - -def _leetcode_match(row: dict[str, object], window: Window) -> bool: - """A solve counts on the day LeetCode accepted it (``submitted_at``). - - ``day`` is the *harvesting* run's date and runs late, so it is only the - fallback for a row without a usable stamp: it can miss a late solve but - never invent one. - """ - detail = row.get("detail") - raw = detail.get("submitted_at") if isinstance(detail, dict) else None - if raw is not None: - landed = _within(raw, window) - if landed is not None: - return landed - _logger.warning( - "LeetCode credit %r has an unparsable submitted_at (%r); " - "falling back to its day key", - row.get("entry_id"), - raw, + target = day or datetime.now(tz=UTC).astimezone().date() + return shutdown_minutes_for(self, target) + (units - 1) * ( + extra_shutdown_minutes_for(self, target) ) - local = datetime.fromtimestamp(window[0], tz=UTC).astimezone() - return row.get("day") == local.date().isoformat() - - -def _reading_match(row: dict[str, object], window: Window) -> bool: - """A reading credit counts if it earned the bonus and ended today.""" - detail = row.get("detail") - if not isinstance(detail, dict) or detail.get("bonus") != "1": - return False - landed = _within(detail.get("ended_at"), window) - if landed is None: - _logger.warning("reading credit %r has no usable ended_at", row.get("entry_id")) - return False - return landed - - -def _anki_match(row: dict[str, object], window: Window) -> bool: - """An Anki credit counts on its Anki day (``detail.anki_day``). - - anki-guard writes one row per Anki day, keyed on the collection's own day - boundary, so the day it names is the day it pays -- no stamp to parse. - """ - detail = row.get("detail") - local = datetime.fromtimestamp(window[0], tz=UTC).astimezone().date() - return isinstance(detail, dict) and detail.get("anki_day") == local.isoformat() WORKOUT: Final = Earner( @@ -154,6 +112,11 @@ WORKOUT: Final = Earner( shutdown_minutes=2 * 60, kind="counted", extra_shutdown_minutes=60, + # screen-locker writes one signed row per credited unit (a verified + # RunnerUp TCX, or a rest day); ``credit_units`` counts them. + ledger=".local/share/workout_locker/ledger.json", + match=workout_match, + missing_ledger_is_no=True, ) LEETCODE: Final = Earner( name="leetcode", @@ -161,7 +124,7 @@ LEETCODE: Final = Earner( gaming_minutes=60, shutdown_minutes=60, ledger=".local/share/leetcode_guard/ledger.json", - match=_leetcode_match, + match=leetcode_match, ) READING: Final = Earner( name="reading", @@ -170,7 +133,7 @@ READING: Final = Earner( shutdown_minutes=60, penalty_from=date(2026, 10, 1), ledger=".local/share/book_guard/ledger.json", - match=_reading_match, + match=reading_match, missing_ledger_is_no=True, ) ANKI: Final = Earner( @@ -180,7 +143,7 @@ ANKI: Final = Earner( shutdown_minutes=30, penalty_from=date(2026, 10, 6), ledger=".local/share/anki_guard/ledger.json", - match=_anki_match, + match=anki_match, missing_ledger_is_no=True, ) # Same gate, same row shape: anki-guard's ``automation`` quota (the Automation @@ -192,7 +155,7 @@ AUTOMATION: Final = Earner( shutdown_minutes=30, penalty_from=date(2026, 10, 6), ledger=".local/share/anki_guard/automation_ledger.json", - match=_anki_match, + match=anki_match, missing_ledger_is_no=True, ) diff --git a/earned_time/earned_time/_resolve.py b/earned_time/earned_time/_resolve.py index 83f6405..4934768 100644 --- a/earned_time/earned_time/_resolve.py +++ b/earned_time/earned_time/_resolve.py @@ -19,12 +19,12 @@ from datetime import UTC, date, datetime import logging from typing import TYPE_CHECKING, Final +from earned_time._ladder import on_ladder, shutdown_ceiling_for, shutdown_minutes_for from earned_time._policy import ( EARNERS, GAMING_BASE_MINUTES, GAMING_CEILING_MINUTES, SHUTDOWN_BASE_MINUTES, - SHUTDOWN_CEILING_MINUTES, Earner, ) @@ -101,14 +101,33 @@ def _today() -> date: return datetime.now(tz=UTC).astimezone().date() +def _shutdown_floor(day: date, earners: tuple[Earner, ...]) -> int: + """Shutdown of a day that earned nothing. + + On the ladder: the ceiling minus every registered earner's first unit, + so doing everything lands exactly on the ceiling. Before it: the old base + minus the penalties in force. + """ + if on_ladder(day): + return shutdown_ceiling_for(day) - sum( + shutdown_minutes_for(e, day) for e in earners + ) + penalised = [e for e in earners if e.penalised_on(day)] + return SHUTDOWN_BASE_MINUTES - sum(e.shutdown_minutes for e in penalised) + + def base_for(day: date | None = None, earners: tuple[Earner, ...] = EARNERS) -> Base: - """The floor for ``day`` (default today): every penalty in force, taken off.""" + """The floor for ``day`` (default today). + + Gaming: every penalty in force, taken off the gaming base. Shutdown: see + :func:`_shutdown_floor` -- the two are derived separately so the ladder + can never move a gaming minute. + """ target = day or _today() penalised = [e for e in earners if e.penalised_on(target)] return Base( gaming_minutes=GAMING_BASE_MINUTES - sum(e.gaming_minutes for e in penalised), - shutdown_minutes=SHUTDOWN_BASE_MINUTES - - sum(e.shutdown_minutes for e in penalised), + shutdown_minutes=_shutdown_floor(target, earners), ) @@ -129,7 +148,8 @@ def resolve( answers: Per earner name, a unit count or ``None`` ("could not check"). A registered earner missing from it earns nothing and is logged: the consumer forgot to ask, which must not pass as a "no". - day: The day to resolve (default today); only the base depends on it. + day: The day to resolve (default today): it picks the base, the + ceiling, and (on the ladder) each earner's shutdown minutes. earners: The registry; tests pass their own. Returns: @@ -159,7 +179,7 @@ def resolve( earner=item, answer=units, gaming_minutes=item.gaming_for(count), - shutdown_minutes=item.shutdown_for(count), + shutdown_minutes=item.shutdown_for(count, target), ) ) gaming = base.gaming_minutes + sum(t.gaming_minutes for t in terms) @@ -169,5 +189,5 @@ def resolve( base=base, terms=tuple(terms), gaming_minutes=min(GAMING_CEILING_MINUTES, gaming), - shutdown_minutes=min(SHUTDOWN_CEILING_MINUTES, shutdown), + shutdown_minutes=min(shutdown_ceiling_for(target), shutdown), ) diff --git a/earned_time/earned_time/tests/test_credits.py b/earned_time/earned_time/tests/test_credits.py new file mode 100644 index 0000000..5371c38 --- /dev/null +++ b/earned_time/earned_time/tests/test_credits.py @@ -0,0 +1,200 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Day-scoped ledger reads (first credit, units) and the workout's credit rule.""" + +from __future__ import annotations + +from datetime import date, datetime, time, timedelta +import json +import logging +from typing import TYPE_CHECKING + +import pytest + +from earned_time import ( + ANKI, + LEETCODE, + WORKOUT, + Earner, + credit_units, + day_window, + entry_signature, + first_credit_at, +) +from earned_time._credits import credit_time + +if TYPE_CHECKING: + from pathlib import Path + +KEY = b"test-key" +DAY = date(2026, 10, 10) +START, END = day_window(DAY) + + +def _at(hour: int, minute: int = 0, day: date = DAY) -> float: + return datetime.combine(day, time(hour, minute)).astimezone().timestamp() + + +def _signed(row: dict[str, object]) -> dict[str, object]: + return {**row, "hmac": entry_signature(row, KEY)} + + +def _workout(entry: str, done: float) -> dict[str, object]: + detail = {"completed_at": str(done), "source": "runnerup_tcx"} + row = {"kind": "credit", "entry_id": entry, "day": "", "detail": detail} + return _signed(row) + + +def _rest(entry: str, day: date, declared: object) -> dict[str, object]: + detail: dict[str, object] = { + "completed_at": str(_at(0, 0, day)), + "source": "rest_day", + } + if declared is not None: + detail["declared_at"] = declared + row = { + "kind": "credit", + "entry_id": entry, + "day": day.isoformat(), + "detail": detail, + } + return _signed(row) + + +@pytest.fixture +def key_file(tmp_path: Path) -> Path: + path = tmp_path / "hmac.key" + path.write_bytes(KEY) + return path + + +def _ledger(tmp_path: Path, rows: list[object]) -> Path: + path = tmp_path / "ledger.json" + path.write_text(json.dumps({"entries": rows}), encoding="utf-8") + return path + + +def test_day_window_spans_the_local_day() -> None: + assert _at(0) == START + next_midnight = datetime.combine(DAY + timedelta(days=1), time.min).astimezone() + assert next_midnight.timestamp() - 1 < END < next_midnight.timestamp() + + +def test_units_and_first_credit(tmp_path: Path, key_file: Path) -> None: + rows: list[object] = [ + _workout("evening", _at(18)), + _workout("morning", _at(8, 15)), + _workout("yesterday", _at(23, 0, DAY - timedelta(days=1))), + {**_workout("forged", _at(6)), "hmac": "forged"}, + "junk", + ] + path = _ledger(tmp_path, rows) + assert credit_units(WORKOUT, path, key_file, DAY) == 2 + assert first_credit_at(WORKOUT, path, key_file, DAY) == _at(8, 15) + later = DAY + timedelta(days=1) + assert credit_units(WORKOUT, path, key_file, later) == 0 + assert first_credit_at(WORKOUT, path, key_file, later) is None + + +def test_missing_workout_ledger_is_zero_units(tmp_path: Path, key_file: Path) -> None: + assert credit_units(WORKOUT, tmp_path / "none.json", key_file, DAY) == 0 + + +def test_cannot_check_is_none(tmp_path: Path, key_file: Path) -> None: + path = _ledger(tmp_path, [_workout("a", _at(9))]) + assert credit_units(WORKOUT, path, tmp_path / "nokey", DAY) is None + assert first_credit_at(WORKOUT, path, tmp_path / "nokey", DAY) is None + assert first_credit_at(LEETCODE, tmp_path / "none.json", key_file, DAY) is None + + +def test_earner_without_reader_is_refused(tmp_path: Path, key_file: Path) -> None: + bare = Earner(name="bare", label="bare", gaming_minutes=0, shutdown_minutes=0) + with pytest.raises(ValueError, match="no shared reader"): + credit_units(bare, tmp_path / "x.json", key_file, DAY) + with pytest.raises(ValueError, match="no shared reader"): + first_credit_at(bare, tmp_path / "x.json", key_file, DAY) + + +def test_credit_time_prefers_the_earners_stamp() -> None: + row: dict[str, object] = { + "created_at": "2026-10-10T20:27:00+02:00", + "detail": {"submitted_at": "1791484015"}, + } + assert credit_time(LEETCODE, row) == 1791484015.0 + created = datetime.fromisoformat("2026-10-10T20:27:00+02:00").timestamp() + assert credit_time(ANKI, row) == created + assert credit_time(LEETCODE, {**row, "detail": {"submitted_at": "?"}}) == created + assert credit_time(LEETCODE, {**row, "detail": "nope"}) == created + + +def test_credit_time_without_a_matcher_uses_created_at() -> None: + bare = Earner(name="bare", label="bare", gaming_minutes=0, shutdown_minutes=0) + row: dict[str, object] = {"created_at": "2026-10-10T08:00:00+00:00"} + assert ( + credit_time(bare, row) + == datetime.fromisoformat("2026-10-10T08:00:00+00:00").timestamp() + ) + + +def test_credit_time_unusable_is_logged(caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING): + assert credit_time(ANKI, {"entry_id": "x", "created_at": "never"}) is None + assert "no usable time" in caplog.text + + +def test_first_credit_skips_rows_without_a_time(tmp_path: Path, key_file: Path) -> None: + row = _signed({"kind": "credit", "detail": {"anki_day": DAY.isoformat()}}) + assert first_credit_at(ANKI, _ledger(tmp_path, [row]), key_file, DAY) is None + + +def _match(row: dict[str, object]) -> bool: + assert WORKOUT.match is not None + return WORKOUT.match(row, (START, END)) + + +@pytest.mark.parametrize( + ("detail", "expected"), + [ + ({"completed_at": str(_at(12))}, True), + ({"completed_at": str(START - 1)}, False), + ("not a dict", False), + ], +) +def test_workout_match(detail: object, *, expected: bool) -> None: + assert _match({"detail": detail}) is expected + + +def test_workout_without_completion_is_logged(caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING): + assert not _match({"entry_id": "w", "detail": {"source": "runnerup_tcx"}}) + assert "no usable completed_at" in caplog.text + + +def test_rest_day_declared_the_evening_before_counts() -> None: + assert _match(_rest("r", DAY, str(START - 3600))) + + +def test_rest_day_for_another_day_does_not_count( + caplog: pytest.LogCaptureFixture, +) -> None: + with caplog.at_level(logging.WARNING): + assert not _match(_rest("r", DAY + timedelta(days=1), str(START - 3600))) + assert not caplog.text + + +@pytest.mark.parametrize("declared", [str(_at(9)), str(START), "soon", None]) +def test_rest_day_not_declared_before_its_day_does_not_count( + caplog: pytest.LogCaptureFixture, declared: object +) -> None: + with caplog.at_level(logging.WARNING): + assert not _match(_rest("r", DAY, declared)) + assert "not declared before its day" in caplog.text + + +def test_rest_day_unit_via_the_ledger(tmp_path: Path, key_file: Path) -> None: + rows: list[object] = [ + _rest("ok", DAY, str(START - 60)), + _rest("same-day", DAY + timedelta(days=1), str(_at(9, 0, DAY + timedelta(1)))), + ] + path = _ledger(tmp_path, rows) + assert credit_units(WORKOUT, path, key_file, DAY) == 1 + assert credit_units(WORKOUT, path, key_file, DAY + timedelta(days=1)) == 0 diff --git a/earned_time/earned_time/tests/test_ladder.py b/earned_time/earned_time/tests/test_ladder.py new file mode 100644 index 0000000..73faf07 --- /dev/null +++ b/earned_time/earned_time/tests/test_ladder.py @@ -0,0 +1,139 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""The shutdown ladder: dated switch, wake anchor, and gaming left untouched.""" + +from __future__ import annotations + +from datetime import date + +import pytest + +from earned_time import ( + ANKI, + AUTOMATION, + EARNERS, + LADDER, + LADDER_FROM, + LEETCODE, + READING, + SHUTDOWN_CEILING_MINUTES, + WAKE_MINUTES, + WORKOUT, + Earner, + Rung, + base_for, + extra_shutdown_minutes_for, + on_ladder, + resolve, + shutdown_ceiling_for, + shutdown_minutes_for, +) +import earned_time._ladder as ladder + +EVE = date(2026, 10, 9) +FIRST = date(2026, 10, 10) +NAMES = [e.name for e in EARNERS] + + +def _shutdown(day: date, **done: int) -> str: + answers = {name: done.get(name, 0) for name in NAMES} + minutes = resolve(answers, day).shutdown_minutes + return f"{minutes // 60:02d}:{minutes % 60:02d}" + + +def _gaming(day: date, **done: int) -> int: + return resolve({name: done.get(name, 0) for name in NAMES}, day).gaming_minutes + + +def test_switch_date_and_wake_anchor() -> None: + assert LADDER_FROM == FIRST + assert WAKE_MINUTES == 7 * 60 + assert not on_ladder(EVE) + assert on_ladder(FIRST) + + +def test_ceiling_is_eight_hours_before_wake() -> None: + assert shutdown_ceiling_for(FIRST) == 23 * 60 + assert shutdown_ceiling_for(EVE) == SHUTDOWN_CEILING_MINUTES + + +def test_ceiling_follows_the_alarm_but_old_days_stay( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(ladder, "WAKE_MINUTES", 6 * 60) + assert shutdown_ceiling_for(FIRST) == 22 * 60 + assert base_for(FIRST).shutdown_minutes == 18 * 60 + assert shutdown_ceiling_for(EVE) == 23 * 60 + assert base_for(EVE).shutdown_minutes == 18 * 60 + + +@pytest.mark.parametrize( + ("item", "eve", "first"), + [ + (WORKOUT, 120, 110), + (LEETCODE, 60, 50), + (READING, 60, 30), + (ANKI, 30, 25), + (AUTOMATION, 30, 25), + ], +) +def test_shutdown_minutes_for(item: Earner, eve: int, first: int) -> None: + assert shutdown_minutes_for(item, EVE) == eve + assert shutdown_minutes_for(item, FIRST) == first + + +def test_every_earner_has_a_rung() -> None: + assert set(LADDER) == {e.name for e in EARNERS} + assert LADDER["workout"] == Rung(110, 0) + + +def test_first_units_fill_the_gap_exactly() -> None: + total = sum(shutdown_minutes_for(e, FIRST) for e in EARNERS) + assert total == 240 + assert base_for(FIRST).shutdown_minutes == shutdown_ceiling_for(FIRST) - total + + +def test_extra_workouts_earn_nothing_on_the_ladder() -> None: + assert extra_shutdown_minutes_for(WORKOUT, EVE) == 60 + assert extra_shutdown_minutes_for(WORKOUT, FIRST) == 0 + assert extra_shutdown_minutes_for(LEETCODE, FIRST) == 0 + assert [WORKOUT.shutdown_for(n, FIRST) for n in range(4)] == [0, 110, 110, 110] + + +def test_earner_without_ladder_values_keeps_its_own() -> None: + plain = Earner( + name="plain", + label="plain", + gaming_minutes=0, + shutdown_minutes=15, + kind="counted", + extra_shutdown_minutes=5, + ) + assert shutdown_minutes_for(plain, FIRST) == 15 + assert extra_shutdown_minutes_for(plain, FIRST) == 5 + + +def test_shutdown_for_defaults_to_today(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(ladder, "LADDER_FROM", date(2000, 1, 1)) + assert WORKOUT.shutdown_for(2) == 110 + + +def test_the_ladder() -> None: + assert _shutdown(FIRST) == "19:00" + steps = ["workout", "leetcode", "reading", "anki", "automation"] + expected = ["20:50", "21:40", "22:10", "22:35", "23:00"] + for count, want in enumerate(expected, start=1): + assert _shutdown(FIRST, **dict.fromkeys(steps[:count], 1)) == want + assert _shutdown(FIRST, workout=2) == "20:50" + assert _shutdown(FIRST, **dict.fromkeys(steps, 1) | {"workout": 3}) == "23:00" + + +def test_the_eve_resolves_the_old_way() -> None: + assert _shutdown(EVE) == "18:00" + assert _shutdown(EVE, workout=2) == "21:00" + assert _shutdown(EVE, workout=1, leetcode=1, reading=1, anki=1) == "22:30" + + +@pytest.mark.parametrize("done", [{}, {"workout": 2}, dict.fromkeys(NAMES, 1)]) +def test_gaming_is_the_same_on_both_sides_of_the_switch(done: dict[str, int]) -> None: + assert _gaming(EVE, **done) == _gaming(FIRST, **done) + assert base_for(EVE).gaming_minutes == base_for(FIRST).gaming_minutes == 180 diff --git a/earned_time/earned_time/tests/test_ledger.py b/earned_time/earned_time/tests/test_ledger.py index 1a8eba4..1a0c0c9 100644 --- a/earned_time/earned_time/tests/test_ledger.py +++ b/earned_time/earned_time/tests/test_ledger.py @@ -10,7 +10,7 @@ from typing import TYPE_CHECKING import pytest -from earned_time import LEETCODE, READING, WORKOUT, done_today, entry_signature +from earned_time import LEETCODE, READING, Earner, done_today, entry_signature from earned_time._ledger import read_key, read_rows, today_window, verified if TYPE_CHECKING: @@ -114,5 +114,6 @@ def test_cannot_check(tmp_path: Path, key_file: Path) -> None: def test_earner_without_reader_is_refused(tmp_path: Path, key_file: Path) -> None: + bare = Earner(name="bare", label="bare", gaming_minutes=0, shutdown_minutes=0) with pytest.raises(ValueError, match="no shared reader"): - done_today(WORKOUT, tmp_path / "x.json", key_file) + done_today(bare, tmp_path / "x.json", key_file) diff --git a/earned_time/earned_time/tests/test_policy.py b/earned_time/earned_time/tests/test_policy.py index 33f6800..801ab41 100644 --- a/earned_time/earned_time/tests/test_policy.py +++ b/earned_time/earned_time/tests/test_policy.py @@ -57,13 +57,17 @@ def test_penalty_starts_on_its_day() -> None: assert not LEETCODE.penalised_on(date(2030, 1, 1)) +PRE_LADDER = date(2026, 10, 9) + + def test_flat_earner_pays_once() -> None: assert [LEETCODE.gaming_for(n) for n in range(3)] == [0, 60, 60] - assert [LEETCODE.shutdown_for(n) for n in range(3)] == [0, 60, 60] + assert [LEETCODE.shutdown_for(n, PRE_LADDER) for n in range(3)] == [0, 60, 60] def test_counted_earner_pays_each_extra_unit_in_shutdown_only() -> None: - assert [WORKOUT.shutdown_for(n) for n in range(4)] == [0, 120, 180, 240] + units = [WORKOUT.shutdown_for(n, PRE_LADDER) for n in range(4)] + assert units == [0, 120, 180, 240] assert [WORKOUT.gaming_for(n) for n in range(3)] == [0, 120, 120] diff --git a/earned_time/earned_time/tests/test_resolve.py b/earned_time/earned_time/tests/test_resolve.py index 7f50a73..23d9f4a 100644 --- a/earned_time/earned_time/tests/test_resolve.py +++ b/earned_time/earned_time/tests/test_resolve.py @@ -3,7 +3,7 @@ from __future__ import annotations -from datetime import date +from datetime import UTC, date, datetime import logging import pytest @@ -22,7 +22,8 @@ def test_base_before_and_after_the_reading_cut() -> None: def test_base_defaults_to_today() -> None: - assert base_for().gaming_minutes in {240, 300} + today = datetime.now(tz=UTC).astimezone().date() + assert base_for() == base_for(today) def test_nothing_earned_is_the_base() -> None: @@ -74,19 +75,19 @@ def test_term_lookup_unknown_raises() -> None: def test_new_penalised_earner_keeps_the_best_case() -> None: # A hypothetical next earner on top of the real registry (reading and - # anki and automation already penalised by 2026-10-11): - # 5h - 1h - 30m - 30m - 30m. + # anki and automation already penalised by 2026-10-09, the last day + # before the shutdown ladder): 5h - 1h - 30m - 30m - 30m. piano = Earner( name="piano", label="piano", gaming_minutes=30, shutdown_minutes=30, - penalty_from=date(2026, 10, 10), + penalty_from=date(2026, 10, 8), ) from earned_time import EARNERS earners = (*EARNERS, piano) - later = date(2026, 10, 11) + later = date(2026, 10, 9) assert base_for(later, earners).gaming_minutes == 150 assert base_for(later, earners).shutdown_minutes == 17 * 60 + 30 others = {e.name: 0 for e in EARNERS} diff --git a/earned_time/pyproject.toml b/earned_time/pyproject.toml index fbcd29d..e34190e 100644 --- a/earned_time/pyproject.toml +++ b/earned_time/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "earned-time" -version = "0.3.0" +version = "0.4.0" description = "Penalty-then-reward registry: what every gate app earns in gaming time and shutdown time" requires-python = ">=3.14" dependencies = []