diff --git a/rivaflow/rivaflow/api/routes/whoop.py b/rivaflow/rivaflow/api/routes/whoop.py index e910a49..1940529 100644 --- a/rivaflow/rivaflow/api/routes/whoop.py +++ b/rivaflow/rivaflow/api/routes/whoop.py @@ -276,6 +276,21 @@ def list_tags_endpoint( return WhoopRepository.list_tags(current_user["id"], start or None, end or None) +@router.get("/tags/vocabulary") +@route_error_handler("whoop_tag_vocabulary", detail="Failed to load tag vocabulary") +def tag_vocabulary_endpoint(current_user: dict = Depends(get_current_user)) -> dict: + """The tag VOCABULARY for the "Tag Today" picker (B4/Wave 2.4) — built-in + suggestions (see db.repositories.whoop_repo.BUILTIN_TAGS) plus any tag this + user has actually applied before, so a custom tag is offered again next time. + + Deliberately NOT `/tags` — that path already serves the per-day tag HISTORY + (`list_tags_endpoint` above); this is the picker's OPTIONS list, a distinct + concept that would collide with (and shadow) the existing route at the same + path if it reused it. + """ + return {"tags": WhoopRepository.tag_vocabulary(current_user["id"])} + + @router.delete("/tag") @route_error_handler("whoop_remove_tag", detail="Failed to remove tag") def remove_tag_endpoint( diff --git a/rivaflow/rivaflow/core/scheduler.py b/rivaflow/rivaflow/core/scheduler.py index 27d6943..8911146 100644 --- a/rivaflow/rivaflow/core/scheduler.py +++ b/rivaflow/rivaflow/core/scheduler.py @@ -457,6 +457,10 @@ def start_scheduler() -> None: # 06/09/18/21 Melbourne-local (wake / post-morning-training / evening / post-evening-training) — # pre-compute the WHOOP cockpit snapshot. Explicit tz so these are Ruby's local hours, not UTC. + # misfire_grace_time: the build takes 1-2min and a deploy can restart the app around a tick — + # without a generous grace, apscheduler logs "missed" and SKIPS the run (observed 2026-07-10: + # double-restart deploy left the snapshot 8h stale until the next tick, so /summary fell back + # to the ~45s live path for every phone refresh). _scheduler.add_job( _cockpit_snapshot_job, "cron", @@ -465,6 +469,8 @@ def start_scheduler() -> None: timezone=_MELBOURNE_TZ, id="cockpit_snapshot", replace_existing=True, + misfire_grace_time=600, + coalesce=True, ) # Warm the snapshot shortly after startup so a fresh deploy serves instantly without waiting for a tick. @@ -474,6 +480,7 @@ def start_scheduler() -> None: run_date=utcnow() + timedelta(seconds=30), id="cockpit_snapshot_warmup", replace_existing=True, + misfire_grace_time=600, ) _scheduler.start() diff --git a/rivaflow/rivaflow/core/whoop_analytics.py b/rivaflow/rivaflow/core/whoop_analytics.py index 1fbd958..6ec9dcc 100644 --- a/rivaflow/rivaflow/core/whoop_analytics.py +++ b/rivaflow/rivaflow/core/whoop_analytics.py @@ -72,6 +72,7 @@ ) from rivaflow.core.whoop_digest import compile_digest from rivaflow.core.whoop_profile import get_whoop_profile, is_rest_day +from rivaflow.core.whoop_tiles import _compose_hero, _compose_tiles from rivaflow.db.repositories.whoop_repo import WhoopRepository MAX_HR = 190 # FALLBACK ONLY — real ceiling comes from user_max_hr() (B1). Kept so @@ -388,7 +389,7 @@ def whoop_summary(user_id: int, today_is_sabbath: bool = False) -> dict: load_series = [ c["cardio_load"] for c in daily_cardio_load(user_id, days=28, max_hr=mx) ] - return { + result = { "readiness": readiness, "strain_target": strain, "acwr": acwr(load_series), @@ -407,6 +408,12 @@ def whoop_summary(user_id: int, today_is_sabbath: bool = False) -> dict: user_id ), # P4a — powers the slim-app safety banner (amber/red) } + # Wave 2.4 (B4/B3) — server-composed tiles + hero, additive: a new field above reaches + # the phone as a tile/hero input with zero app changes, since these are derived FROM + # the dict just built (no new DB reads). See core/whoop_tiles.py. + result["tiles"] = _compose_tiles(result) + result["hero"] = _compose_hero(result) + return result def _chronic_load(cardio: list[dict]) -> float | None: @@ -1072,7 +1079,8 @@ def daily_narrative( # Version of the structured metrics contract stored beside the cockpit HTML. # Bump when whoop_summary's shape changes so a stored snapshot is attributable. -COCKPIT_DERIVER_VERSION = "whoop-summary-v1" +# v2 (Wave 2.4/B4/B3) — added the additive `tiles` + `hero` keys. +COCKPIT_DERIVER_VERSION = "whoop-summary-v2" def build_cockpit_snapshot(user_id: int) -> tuple[str, str, str]: diff --git a/rivaflow/rivaflow/core/whoop_tiles.py b/rivaflow/rivaflow/core/whoop_tiles.py new file mode 100644 index 0000000..0c7679d --- /dev/null +++ b/rivaflow/rivaflow/core/whoop_tiles.py @@ -0,0 +1,265 @@ +"""B4/B3 — server-composed cockpit tiles + hero block (Wave 2.4). + +The phone currently hardcodes which whoop_summary fields become tiles, so a new VPS +metric needs an app release before it's visible. This module composes a `tiles` list +and a `hero` block FROM an already-computed whoop_summary() dict (no new DB reads, +no side effects) so a thin display client can render generically: iterate `tiles` in +`order`, colour each by `accent`, and headline the screen from `hero`. + +Two ranked scales back the accent/severity logic, both "worse wins": + - COLOR_RANK: neutral/green (0) < amber (1) < red (2) + - SEVERITY_RANK: info (0) < caution (1) < alert (2) + +`prevention_watch` is the SAFETY channel (fires every day, incl. the Sabbath, and +outranks readiness for urgency — see whoop_analytics.prevention_watch's docstring). +Both the readiness tile's accent and the hero block apply the same override: an +amber/red prevention tier can only escalate colour/severity, never downgrade what +readiness itself already earned (e.g. a Rundown/red day stays red under an amber +prevention tier). The hero's headline is replaced by prevention's own headline +whenever prevention fires amber/red — the safety message must reach the user even +when the performance readiness verdict alone wouldn't have said anything alarming. +""" + +from __future__ import annotations + +COLOR_RANK: dict[str, int] = {"neutral": 0, "green": 0, "blue": 0, "amber": 1, "red": 2} +SEVERITY_RANK: dict[str, int] = {"info": 0, "caution": 1, "alert": 2} + +# Stable tile ids in their fixed display order — readiness always leads. +TILE_ORDER: dict[str, int] = { + "readiness": 0, + "hrv": 1, + "resting_hr": 2, + "sleep": 3, + "strain": 4, + "respiratory": 5, + "stress": 6, +} + +_READINESS_COLOR = { + "Prime": "green", + "Balanced": "green", + "Strained": "amber", + "Rundown": "red", + "Rest": "neutral", + "Building": "neutral", +} +_READINESS_SEVERITY = { + "Prime": "info", + "Balanced": "info", + "Strained": "caution", + "Rundown": "alert", + "Rest": "info", + "Building": "info", +} + +# readiness.contributors[*]['signal'] name for each tile that rides the same +# deviation-from-baseline blend (see readiness.py) — reused so a tile's colour +# agrees with whether that signal is currently helping or hurting recovery. +_CONTRIBUTOR_SIGNAL = { + "hrv": "hrv", + "resting_hr": "rhr", + "sleep": "sleep", + "respiratory": "resp", +} + + +def _prevention_forced_color(tier: str | None) -> str | None: + return {"amber": "amber", "red": "red"}.get(tier or "") + + +def _prevention_forced_severity(tier: str | None) -> str | None: + return {"amber": "caution", "red": "alert"}.get(tier or "") + + +def _apply_prevention_color(base: str, tier: str | None) -> str: + """Prevention can only push a colour UP the COLOR_RANK scale, never down.""" + forced = _prevention_forced_color(tier) + if forced is None: + return base + return forced if COLOR_RANK[forced] > COLOR_RANK.get(base, 0) else base + + +def _apply_prevention_severity(base: str, tier: str | None) -> str: + """Prevention can only push severity UP the SEVERITY_RANK scale, never down.""" + forced = _prevention_forced_severity(tier) + if forced is None: + return base + return forced if SEVERITY_RANK[forced] > SEVERITY_RANK.get(base, 0) else base + + +def _contributor_accent(readiness: dict, signal: str) -> str: + """green if `signal` currently supports recovery, amber if it drags it, else + neutral (no baseline yet — Building/Rest/thin data).""" + for c in readiness.get("contributors") or []: + if c.get("signal") == signal: + return "green" if c.get("effect", 0) >= 0 else "amber" + return "neutral" + + +def _stress_accent(value: float) -> str: + if value <= 33: + return "green" + if value <= 66: + return "amber" + return "red" + + +def _readiness_tile(summary: dict) -> dict: + readiness = summary.get("readiness") or {} + prevention = summary.get("prevention") or {} + state = readiness.get("state") + accent = _READINESS_COLOR.get(state or "", "neutral") + if prevention.get("available"): + accent = _apply_prevention_color(accent, prevention.get("tier")) + return { + "id": "readiness", + "label": "Readiness", + "value": state, + "unit": None, + "accent": accent, + "order": TILE_ORDER["readiness"], + } + + +def _hrv_tile(summary: dict) -> dict | None: + hrv = summary.get("hrv_today") + if not hrv or hrv.get("rmssd") is None: + return None + readiness = summary.get("readiness") or {} + return { + "id": "hrv", + "label": "HRV", + "value": hrv["rmssd"], + "unit": "ms", + "accent": _contributor_accent(readiness, _CONTRIBUTOR_SIGNAL["hrv"]), + "order": TILE_ORDER["hrv"], + } + + +def _resting_hr_tile(summary: dict) -> dict | None: + rhr = summary.get("resting_hr_today") + if not rhr or rhr.get("resting_hr") is None: + return None + readiness = summary.get("readiness") or {} + return { + "id": "resting_hr", + "label": "Resting HR", + "value": rhr["resting_hr"], + "unit": "bpm", + "accent": _contributor_accent(readiness, _CONTRIBUTOR_SIGNAL["resting_hr"]), + "order": TILE_ORDER["resting_hr"], + } + + +def _sleep_tile(summary: dict) -> dict | None: + sleep = summary.get("sleep") + if not sleep or not sleep.get("available") or sleep.get("quality_score") is None: + return None + readiness = summary.get("readiness") or {} + return { + "id": "sleep", + "label": "Sleep", + "value": sleep["quality_score"], + "unit": "pts", + "accent": _contributor_accent(readiness, _CONTRIBUTOR_SIGNAL["sleep"]), + "order": TILE_ORDER["sleep"], + } + + +def _strain_tile(summary: dict) -> dict | None: + """No tile on the Sabbath (or Building) — prescribe_strain already returns + available=False for those states ("rest is prescribed"), so gating on + `available` alone mirrors the phone's current hiding without a separate + sabbath check here.""" + strain = summary.get("strain_target") + if not strain or not strain.get("available"): + return None + accent = ( + "amber" + if strain.get("capped") + else ("green" if strain.get("state") == "Prime" else "neutral") + ) + return { + "id": "strain", + "label": "Strain Target", + "value": strain["target_load"], + "unit": "load", + "accent": accent, + "order": TILE_ORDER["strain"], + } + + +def _respiratory_tile(summary: dict) -> dict | None: + resp = summary.get("respiratory_rate") + if not resp or not resp.get("available") or resp.get("respiratory_rate") is None: + return None + readiness = summary.get("readiness") or {} + return { + "id": "respiratory", + "label": "Respiratory Rate", + "value": resp["respiratory_rate"], + "unit": "rpm", + "accent": _contributor_accent(readiness, _CONTRIBUTOR_SIGNAL["respiratory"]), + "order": TILE_ORDER["respiratory"], + } + + +def _stress_tile(summary: dict) -> dict | None: + stress = summary.get("stress") + if not stress or not stress.get("available") or stress.get("stress") is None: + return None + value = stress["stress"] + return { + "id": "stress", + "label": "Stress", + "value": value, + "unit": "pts", + "accent": _stress_accent(float(value)), + "order": TILE_ORDER["stress"], + } + + +def _compose_tiles(summary: dict) -> list[dict]: + """Build the ordered tile list from an already-computed whoop_summary() dict. + Readiness always contributes a tile (its state is never None); every other + tile is omitted outright when its underlying metric is unavailable/None — + never emitted with a blank value.""" + builders = ( + _readiness_tile, + _hrv_tile, + _resting_hr_tile, + _sleep_tile, + _strain_tile, + _respiratory_tile, + _stress_tile, + ) + tiles = [t for t in (b(summary) for b in builders) if t is not None] + tiles.sort(key=lambda t: t["order"]) + return tiles + + +def _compose_hero(summary: dict) -> dict: + """The single headline block a thin client renders at the top of the screen: + readiness state/headline, prevention-overridden colour + severity (safety + outranks performance — see module docstring), in + {state, state_color, headline, severity}.""" + readiness = summary.get("readiness") or {} + prevention = summary.get("prevention") or {} + state = readiness.get("state") + color = _READINESS_COLOR.get(state or "", "neutral") + severity = _READINESS_SEVERITY.get(state or "", "info") + headline = readiness.get("headline", "") + + tier = prevention.get("tier") if prevention.get("available") else None + if tier in ("amber", "red"): + color = _apply_prevention_color(color, tier) + severity = _apply_prevention_severity(severity, tier) + headline = prevention.get("headline", headline) + + return { + "state": state, + "state_color": color, + "headline": headline, + "severity": severity, + } diff --git a/rivaflow/rivaflow/db/repositories/whoop_repo.py b/rivaflow/rivaflow/db/repositories/whoop_repo.py index 7732a32..5cbcbee 100644 --- a/rivaflow/rivaflow/db/repositories/whoop_repo.py +++ b/rivaflow/rivaflow/db/repositories/whoop_repo.py @@ -29,6 +29,19 @@ def _is_hex(s: str) -> bool: return False +# Suggested first-class tags surfaced in the tag-vocabulary endpoint — mirrors the +# examples add_tag_endpoint's docstring lists (routes/whoop.py) so the "Tag Today" +# picker always offers them even before the user has ever applied one. +BUILTIN_TAGS: tuple[str, ...] = ( + "alcohol", + "late-training", + "ill", + "poor-sleep", + "travel", + "sabbath-rest", +) + + class WhoopRepository: """Idempotent, user-scoped ingest for the whoop_* stream tables.""" @@ -345,6 +358,27 @@ def tagged_days(user_id: int, tag: str) -> set[str]: ) return {str(r["day"])[:10] for r in rows} + @staticmethod + def distinct_tags(user_id: int) -> list[str]: + """The sorted, deduped set of tag strings this user has ever applied — feeds the + tag-vocabulary endpoint (B4) alongside the built-in suggestions.""" + rows = BaseRepository._fetchall( + convert_query( + "SELECT DISTINCT tag FROM whoop_tags WHERE user_id = ? ORDER BY tag" + ), + (user_id,), + ) + return [str(r["tag"]) for r in rows] + + @staticmethod + def tag_vocabulary(user_id: int) -> list[str]: + """BUILTIN_TAGS (in their canonical order) + any distinct tag this user has + actually applied that isn't already a built-in — so a one-off custom tag is + offered again next time, and the picker never starts empty.""" + used = WhoopRepository.distinct_tags(user_id) + extras = [t for t in used if t not in BUILTIN_TAGS] + return list(BUILTIN_TAGS) + extras + # ── Timestamped workout sessions (real start/end window so HR can attach) ── @staticmethod diff --git a/rivaflow/tests/unit/test_whoop_tiles.py b/rivaflow/tests/unit/test_whoop_tiles.py new file mode 100644 index 0000000..6410833 --- /dev/null +++ b/rivaflow/tests/unit/test_whoop_tiles.py @@ -0,0 +1,415 @@ +"""Wave 2.4 (B4/B3) — server-composed cockpit tiles + hero, and the tag vocabulary +endpoint. Pure unit tests: `_compose_tiles`/`_compose_hero` take an already-built +whoop_summary()-shaped dict (no DB), and the route/repo layers are exercised with +mocks, mirroring tests/unit/test_summary_snapshot.py's asyncio.run(route(...)) +pattern for the `@route_error_handler`-wrapped endpoint. +""" + +from __future__ import annotations + +import asyncio + +from rivaflow.api.routes import whoop as whoop_route +from rivaflow.core import whoop_analytics +from rivaflow.core.whoop_tiles import _compose_hero, _compose_tiles +from rivaflow.db.repositories.whoop_repo import BUILTIN_TAGS, WhoopRepository + + +def _contributor(signal: str, effect: float) -> dict: + return { + "signal": signal, + "label": signal, + "z": effect, + "weight": 0.25, + "effect": effect, + "direction": "supports recovery" if effect >= 0 else "drags recovery", + } + + +def _full_summary(*, state: str = "Prime", prevention_tier: str = "green") -> dict: + return { + "readiness": { + "state": state, + "headline": "Above your baseline — green light to train hard.", + "score": 87, + "contributors": [ + _contributor("hrv", 0.6), + _contributor("rhr", 0.2), + _contributor("sleep", -0.045), + _contributor("resp", 0.01), + ], + }, + "prevention": { + "available": True, + "tier": prevention_tier, + "headline": "SAFETY: sustained deviation across two signal families.", + }, + "hrv_today": {"day": "2026-07-10", "rmssd": 55.2, "ln_rmssd": 4.0}, + "resting_hr_today": {"day": "2026-07-10", "resting_hr": 48}, + "sleep": { + "available": True, + "quality_score": 88, + "quality_version": "sleep-quality-v1", + "duration_hours": 7.8, + }, + "strain_target": { + "available": True, + "state": state, + "target_load": 14.5, + "capped": False, + "headline": "Recovered — you can push. Target strain 14.5.", + }, + "respiratory_rate": {"available": True, "respiratory_rate": 14.2}, + "stress": {"available": True, "stress": 20}, + } + + +# ── _compose_tiles — full fixture ─────────────────────────────────────────── + + +def test_full_fixture_tile_ids_order_and_accents(): + tiles = _compose_tiles(_full_summary()) + + assert [t["id"] for t in tiles] == [ + "readiness", + "hrv", + "resting_hr", + "sleep", + "strain", + "respiratory", + "stress", + ] + assert [t["order"] for t in tiles] == [0, 1, 2, 3, 4, 5, 6] + + by_id = {t["id"]: t for t in tiles} + assert by_id["readiness"]["value"] == "Prime" + assert by_id["readiness"]["accent"] == "green" + assert by_id["hrv"] == { + "id": "hrv", + "label": "HRV", + "value": 55.2, + "unit": "ms", + "accent": "green", # hrv contributor effect 0.6 >= 0 + "order": 1, + } + assert by_id["resting_hr"]["accent"] == "green" # rhr contributor effect 0.2 >= 0 + assert by_id["sleep"]["accent"] == "amber" # sleep contributor effect -0.045 < 0 + assert by_id["sleep"]["value"] == 88 + assert by_id["strain"]["accent"] == "green" # Prime, not capped + assert by_id["strain"]["value"] == 14.5 + assert ( + by_id["respiratory"]["accent"] == "green" + ) # resp contributor effect 0.01 >= 0 + assert by_id["stress"]["accent"] == "green" # 20 <= 33 + + +# ── _compose_tiles — sparse fixture (missing metrics contribute no tile) ─── + + +def test_sparse_fixture_omits_unavailable_tiles_without_blanks(): + summary = { + "readiness": {"state": "Building", "headline": "Building…", "contributors": []}, + "prevention": {"available": False}, + "hrv_today": None, + "resting_hr_today": None, + "sleep": {"available": False, "reason": "no overnight data"}, + "strain_target": {"available": False, "state": "Building", "reason": "…"}, + "respiratory_rate": {"available": False}, + "stress": {"available": False}, + } + + tiles = _compose_tiles(summary) + + assert tiles == [ + { + "id": "readiness", + "label": "Readiness", + "value": "Building", + "unit": None, + "accent": "neutral", + "order": 0, + } + ] + + +# ── _compose_tiles — sabbath fixture ──────────────────────────────────────── + + +def test_sabbath_fixture_has_no_strain_tile_and_readiness_reads_rest(): + summary = { + "readiness": { + "state": "Rest", + "headline": "Sabbath — rest is prescribed. No score today.", + "driver": "sabbath", + "score": None, + "caveat": None, + }, + "prevention": {"available": False}, + "hrv_today": {"day": "2026-07-13", "rmssd": 50.0}, + "resting_hr_today": {"day": "2026-07-13", "resting_hr": 49}, + "sleep": {"available": True, "quality_score": 80}, + # prescribe_strain returns exactly this shape for state == "Rest". + "strain_target": { + "available": False, + "state": "Rest", + "reason": "Sabbath — rest is prescribed.", + }, + "respiratory_rate": {"available": False}, + "stress": {"available": False}, + } + + tiles = _compose_tiles(summary) + ids = [t["id"] for t in tiles] + + assert "strain" not in ids + by_id = {t["id"]: t for t in tiles} + assert by_id["readiness"]["value"] == "Rest" + assert ( + by_id["readiness"]["accent"] == "neutral" + ) # no baseline z, no prevention firing + + +def test_sabbath_prevention_still_recolors_readiness_tile_safety_fires_on_sabbath(): + summary = { + "readiness": { + "state": "Rest", + "headline": "Sabbath — rest is prescribed. No score today.", + "driver": "sabbath", + "score": None, + "caveat": None, + }, + "prevention": {"available": True, "tier": "amber", "headline": "SAFETY"}, + "hrv_today": None, + "resting_hr_today": None, + "sleep": {"available": False}, + "strain_target": {"available": False, "state": "Rest", "reason": "…"}, + "respiratory_rate": {"available": False}, + "stress": {"available": False}, + } + + tiles = _compose_tiles(summary) + + assert tiles[0]["id"] == "readiness" + assert tiles[0]["value"] == "Rest" + assert ( + tiles[0]["accent"] == "amber" + ) # prevention fires (safety) even on the Sabbath + + +# ── prevention override — readiness tile accent ───────────────────────────── + + +def test_prevention_red_overrides_green_readiness_tile_accent(): + tiles = _compose_tiles(_full_summary(state="Balanced", prevention_tier="red")) + readiness_tile = next(t for t in tiles if t["id"] == "readiness") + assert readiness_tile["accent"] == "red" + + +def test_prevention_amber_does_not_downgrade_already_red_readiness_tile(): + tiles = _compose_tiles(_full_summary(state="Rundown", prevention_tier="amber")) + readiness_tile = next(t for t in tiles if t["id"] == "readiness") + assert ( + readiness_tile["accent"] == "red" + ) # Rundown is already red; amber can't downgrade it + + +def test_prevention_green_tier_does_not_touch_readiness_accent(): + tiles = _compose_tiles(_full_summary(state="Balanced", prevention_tier="green")) + readiness_tile = next(t for t in tiles if t["id"] == "readiness") + assert readiness_tile["accent"] == "green" + + +# ── _compose_hero ──────────────────────────────────────────────────────────── + + +def test_hero_severity_mapping_without_prevention_firing(): + expected = { + "Prime": ("green", "info"), + "Balanced": ("green", "info"), + "Strained": ("amber", "caution"), + "Rundown": ("red", "alert"), + "Rest": ("neutral", "info"), + "Building": ("neutral", "info"), + } + for state, (color, severity) in expected.items(): + summary = _full_summary(state=state, prevention_tier="green") + hero = _compose_hero(summary) + assert hero["state"] == state + assert hero["state_color"] == color + assert hero["severity"] == severity + + +def test_hero_prevention_red_overrides_green_readiness(): + summary = _full_summary(state="Balanced", prevention_tier="red") + hero = _compose_hero(summary) + assert hero["state_color"] == "red" + assert hero["severity"] == "alert" + assert hero["headline"] == summary["prevention"]["headline"] + + +def test_hero_prevention_amber_lifts_severity_and_recolors(): + summary = _full_summary(state="Balanced", prevention_tier="amber") + hero = _compose_hero(summary) + assert hero["state_color"] == "amber" + assert hero["severity"] == "caution" + assert hero["headline"] == summary["prevention"]["headline"] + + +def test_hero_prevention_amber_does_not_downgrade_rundown_but_swaps_headline(): + summary = _full_summary(state="Rundown", prevention_tier="amber") + hero = _compose_hero(summary) + assert hero["state_color"] == "red" # unchanged — amber can't downgrade red + assert hero["severity"] == "alert" # unchanged — amber can't downgrade alert + # Safety headline still wins even when it doesn't need to escalate colour/severity. + assert hero["headline"] == summary["prevention"]["headline"] + + +def test_hero_prevention_unavailable_or_green_leaves_readiness_headline(): + summary = _full_summary(state="Balanced", prevention_tier="green") + hero = _compose_hero(summary) + assert hero["headline"] == summary["readiness"]["headline"] + + summary["prevention"] = {"available": False} + hero2 = _compose_hero(summary) + assert hero2["headline"] == summary["readiness"]["headline"] + + +# ── whoop_summary wiring — tiles/hero ride along additively ──────────────── + + +def test_whoop_summary_output_carries_tiles_and_hero(monkeypatch): + class _FakeProfile: + tz = whoop_analytics.LOCAL_TZ + sleep_need_h = 8.0 + age = 40 + rest_weekday = 6 + max_hr_override = None + + monkeypatch.setattr( + whoop_analytics, "get_whoop_profile", lambda uid: _FakeProfile() + ) + monkeypatch.setattr(whoop_analytics, "user_max_hr", lambda uid: {"max_hr": 180.0}) + monkeypatch.setattr( + whoop_analytics, + "compute_readiness", + lambda uid, today_is_sabbath=False: { + "state": "Prime", + "headline": "green light to push", + "score": 87, + "contributors": [_contributor("hrv", 0.6)], + }, + ) + monkeypatch.setattr( + whoop_analytics, + "daily_resting_rmssd", + lambda uid, days=14, max_hr=None: [{"day": "2026-07-10", "rmssd": 55.2}], + ) + monkeypatch.setattr( + whoop_analytics, + "daily_resting_hr", + lambda uid, days=14, max_hr=None: [{"day": "2026-07-10", "resting_hr": 48}], + ) + monkeypatch.setattr( + whoop_analytics, + "daily_cardio_load", + lambda uid, days=14, max_hr=None: [{"day": "2026-07-10", "cardio_load": 12.0}], + ) + monkeypatch.setattr( + whoop_analytics, + "nightly_sleep", + lambda uid: { + "available": True, + "duration_hours": 7.8, + "coverage_pct": 95, + "fragmented": False, + "sleep_start": "2026-07-09T22:00:00+10:00", + "sleep_end": "2026-07-10T06:00:00+10:00", + }, + ) + monkeypatch.setattr( + whoop_analytics, "nightly_sleep_history", lambda uid, nights=7, max_hr=None: [] + ) + monkeypatch.setattr( + whoop_analytics, + "respiratory_rate", + lambda uid: {"available": True, "respiratory_rate": 14.2}, + ) + monkeypatch.setattr( + whoop_analytics, + "today_stress", + lambda uid, max_hr=None: {"available": True, "stress": 20}, + ) + monkeypatch.setattr(whoop_analytics, "capture_coverage", lambda uid, days=21: {}) + monkeypatch.setattr( + whoop_analytics, + "prevention_watch", + lambda uid: {"available": True, "tier": "green", "headline": "In range."}, + ) + monkeypatch.setattr(whoop_analytics, "_chronic_load", lambda cardio: 12.0) + monkeypatch.setattr( + whoop_analytics, + "prescribe_strain", + lambda state, chronic, acute: { + "available": True, + "state": state, + "target_load": 14.5, + "capped": False, + "headline": "Recovered — you can push. Target strain 14.5.", + }, + ) + monkeypatch.setattr(whoop_analytics, "acwr", lambda series: {"available": False}) + + result = whoop_analytics.whoop_summary(1) + + assert "tiles" in result and "hero" in result + assert [t["id"] for t in result["tiles"]] == [ + "readiness", + "hrv", + "resting_hr", + "sleep", + "strain", + "respiratory", + "stress", + ] + assert result["hero"]["state"] == "Prime" + assert result["hero"]["headline"] == "green light to push" + # Additive — every pre-existing key untouched. + assert result["readiness"]["state"] == "Prime" + assert result["strain_target"]["target_load"] == 14.5 + + +def test_deriver_version_bumped_to_v2(): + assert whoop_analytics.COCKPIT_DERIVER_VERSION == "whoop-summary-v2" + + +# ── tag vocabulary — repo layer ───────────────────────────────────────────── + + +def test_tag_vocabulary_returns_builtins_when_nothing_used(monkeypatch): + monkeypatch.setattr(WhoopRepository, "distinct_tags", staticmethod(lambda uid: [])) + assert WhoopRepository.tag_vocabulary(1) == list(BUILTIN_TAGS) + + +def test_tag_vocabulary_appends_custom_tags_without_duplicating_builtins(monkeypatch): + monkeypatch.setattr( + WhoopRepository, + "distinct_tags", + staticmethod(lambda uid: ["ill", "big-comp", "alcohol"]), + ) + result = WhoopRepository.tag_vocabulary(1) + assert result == [*BUILTIN_TAGS, "big-comp"] + + +# ── tag vocabulary — route layer ──────────────────────────────────────────── + + +def test_tag_vocabulary_endpoint_returns_tags_dict(monkeypatch): + monkeypatch.setattr( + WhoopRepository, + "tag_vocabulary", + staticmethod(lambda uid: ["alcohol", "ill", "big-comp"]), + ) + + result = asyncio.run(whoop_route.tag_vocabulary_endpoint(current_user={"id": 42})) + + assert result == {"tags": ["alcohol", "ill", "big-comp"]} diff --git a/tests/test_whoop_cockpit_metrics.py b/tests/test_whoop_cockpit_metrics.py index ee1a518..5408eea 100644 --- a/tests/test_whoop_cockpit_metrics.py +++ b/tests/test_whoop_cockpit_metrics.py @@ -25,7 +25,7 @@ def test_build_cockpit_snapshot_returns_html_and_metrics_json(monkeypatch): html, metrics_json, version = whoop_analytics.build_cockpit_snapshot(1) assert html == "ok" - assert version == whoop_analytics.COCKPIT_DERIVER_VERSION == "whoop-summary-v1" + assert version == whoop_analytics.COCKPIT_DERIVER_VERSION == "whoop-summary-v2" # metrics_json is a valid JSON string carrying the structured summary contract parsed = json.loads(metrics_json) assert parsed["readiness"]["score"] == 87