""" Whoop access library for Cameron's fleet of agents. Other agents (e.g. glasses) can import this directly. Handles token refresh, caching, and v2 API access. Same-user (cameronsmith) only — relies on chmod 600 on creds/tokens. USAGE (from another agent on the same box): import sys sys.path.insert(0, "/data/cameron/life/scripts") from whoop_lib import ( get_recovery, get_sleep, get_cycle_strain, get_workouts, get_latest_summary, refresh_if_stale, ) summary = get_latest_summary() # cached <60s, instant after first call print(summary["recovery_score"]) # 77.0 print(summary["hours_asleep"]) # 8.23 print(summary["strain_today"]) # 4.78 Functions auto-refresh the access token via the stored refresh token when needed. Cached calls hit a JSON snapshot on disk and avoid spamming Whoop's API. DATA AVAILABILITY: - Recovery score, HRV, RHR, SpO2, skin temp: YES - Sleep hours, REM, deep, performance, efficiency, disturbances: YES - Current cycle strain (kJ + avg/max HR): YES - Workouts list / current workout: REQUIRES re-OAuth with `read:workout` scope (see EXTEND_WORKOUT_SCOPE below) - Step count: NOT exposed by Whoop API (they measure strain in kJ, not steps). Closest proxy: kilojoule from cycle.score. - Start/stop workout: NOT supported by Whoop API. User starts/stops on device/app; you can read it after the fact. RATE LIMITS: Whoop rate limits per app, ~100 req/min. The cached helper functions are safe for tight polling (e.g. 1 Hz read of cached file). Hard-refresh functions hit the API — keep those polls >30s apart. """ import json, os, time from datetime import datetime, timezone from pathlib import Path from typing import Any import requests # ---- Paths ---- LIFE = Path("/data/cameron/life") CREDS = LIFE / ".whoop_creds.json" TOKENS = LIFE / ".whoop_tokens.json" READINESS = LIFE / ".whoop_readiness.json" LOG = LIFE / ".whoop_pull.log" # ---- API ---- TOKEN_URL = "https://api.prod.whoop.com/oauth/oauth2/token" API_BASE = "https://api.prod.whoop.com/developer/v2" # Cached-summary staleness threshold (seconds). Above this, re-pull. DEFAULT_STALE_S = 60 def _log(msg: str) -> None: ts = datetime.now(timezone.utc).isoformat() with LOG.open("a") as f: f.write(f"{ts} [lib] {msg}\n") def _refresh_access_token() -> dict: creds = json.loads(CREDS.read_text()) tokens = json.loads(TOKENS.read_text()) r = requests.post( TOKEN_URL, data={ "grant_type": "refresh_token", "refresh_token": tokens["refresh_token"], "client_id": creds["client_id"], "client_secret": creds["client_secret"], "scope": "offline", }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30, ) r.raise_for_status() new_tokens = r.json() TOKENS.write_text(json.dumps(new_tokens, indent=2)) os.chmod(TOKENS, 0o600) return new_tokens def _get_with_retry(path: str, params: dict | None = None) -> dict: """GET with one auto-refresh on 401.""" tokens = json.loads(TOKENS.read_text()) for attempt in range(2): r = requests.get( f"{API_BASE}{path}", headers={"Authorization": f"Bearer {tokens['access_token']}"}, params=params or {}, timeout=30, ) if r.status_code == 401 and attempt == 0: tokens = _refresh_access_token() continue r.raise_for_status() return r.json() raise RuntimeError("unreachable") # ---- High-level helpers ---- def get_recovery(force_refresh: bool = False) -> dict: """Latest recovery score record. Returns dict with score_state and score subdict.""" data = _get_with_retry("/recovery", {"limit": 1}) return (data.get("records") or [{}])[0] def get_cycle(force_refresh: bool = False) -> dict: """Latest cycle (current day). Includes strain, kilojoule, avg/max HR.""" data = _get_with_retry("/cycle", {"limit": 1}) return (data.get("records") or [{}])[0] def get_sleep(force_refresh: bool = False) -> dict: """Latest sleep activity. Includes stage_summary + sleep_performance %.""" data = _get_with_retry("/activity/sleep", {"limit": 1}) return (data.get("records") or [{}])[0] def get_cycle_strain() -> float | None: """Today's strain so far (0-21 scale).""" cycle = get_cycle() return (cycle.get("score") or {}).get("strain") def get_kilojoules_today() -> float | None: """Total kJ burned today (closest Whoop has to 'calories' — divide by ~4.184 for kcal).""" cycle = get_cycle() return (cycle.get("score") or {}).get("kilojoule") def get_workouts(limit: int = 5) -> list[dict]: """List recent workouts. NOTE: requires `read:workout` scope. Currently the OAuth handshake doesn't include this — call will 401 until Cameron re-auths. See EXTEND_WORKOUT_SCOPE below. """ try: data = _get_with_retry("/activity/workout", {"limit": limit}) return data.get("records") or [] except requests.HTTPError as e: if e.response is not None and e.response.status_code == 401: raise PermissionError( "Whoop OAuth doesn't include read:workout scope. " "Cameron needs to re-run whoop_oauth.py with the scope added. " "See EXTEND_WORKOUT_SCOPE in whoop_lib.py for details." ) from e raise def get_current_workout() -> dict | None: """Workout in progress (end == None), if any. Same scope caveat as get_workouts.""" for w in get_workouts(limit=3): if not w.get("end"): return w return None def get_latest_summary(stale_s: int = DEFAULT_STALE_S) -> dict: """ Fast path: returns the parsed summary from .whoop_readiness.json. Refreshes via full pull if file is older than `stale_s`. Schema (every key may be None if Whoop hasn't scored yet): recovery_score, hrv_rmssd_milli, resting_heart_rate, spo2_percentage, skin_temp_celsius, sleep_performance_percentage, sleep_efficiency_percentage, sleep_consistency_percentage, respiratory_rate, hours_asleep, disturbance_count, rem_minutes, deep_minutes, strain_today """ if READINESS.exists(): age = time.time() - READINESS.stat().st_mtime if age < stale_s: return json.loads(READINESS.read_text())["summary"] refresh_if_stale(0) # force refresh return json.loads(READINESS.read_text())["summary"] def refresh_if_stale(stale_s: int = DEFAULT_STALE_S) -> bool: """Re-run the full pull if the readiness file is older than `stale_s`. Returns True if a fresh pull happened.""" if READINESS.exists(): age = time.time() - READINESS.stat().st_mtime if age < stale_s: return False # Re-implements whoop_pull.py inline to avoid subprocess cycle = get_cycle() recovery = get_recovery() sleep = get_sleep() rec_score = recovery.get("score") or {} sleep_score = sleep.get("score") or {} sleep_stages = sleep_score.get("stage_summary") or {} in_bed_ms = sleep_stages.get("total_in_bed_time_milli", 0) awake_ms = sleep_stages.get("total_awake_time_milli", 0) asleep_hrs = (in_bed_ms - awake_ms) / 3600000 if in_bed_ms else None summary = { "recovery_score": rec_score.get("recovery_score"), "hrv_rmssd_milli": rec_score.get("hrv_rmssd_milli"), "resting_heart_rate": rec_score.get("resting_heart_rate"), "spo2_percentage": rec_score.get("spo2_percentage"), "skin_temp_celsius": rec_score.get("skin_temp_celsius"), "sleep_performance_percentage": sleep_score.get("sleep_performance_percentage"), "sleep_efficiency_percentage": sleep_score.get("sleep_efficiency_percentage"), "sleep_consistency_percentage": sleep_score.get("sleep_consistency_percentage"), "respiratory_rate": sleep_score.get("respiratory_rate"), "hours_asleep": round(asleep_hrs, 2) if asleep_hrs is not None else None, "disturbance_count": sleep_stages.get("disturbance_count"), "rem_minutes": round(sleep_stages.get("total_rem_sleep_time_milli", 0) / 60000) or None, "deep_minutes": round(sleep_stages.get("total_slow_wave_sleep_time_milli", 0) / 60000) or None, "strain_today": (cycle.get("score") or {}).get("strain"), "kilojoules_today": (cycle.get("score") or {}).get("kilojoule"), } result = { "pulled_at": datetime.now(timezone.utc).isoformat(), "summary": summary, "cycle": cycle, "recovery": recovery, "sleep": sleep, } READINESS.write_text(json.dumps(result, indent=2)) os.chmod(READINESS, 0o600) _log(f"refresh_if_stale -> pulled (recovery={summary['recovery_score']})") return True # ---- Notes for extending ---- EXTEND_WORKOUT_SCOPE = """ To enable workout reads: 1. Cameron updates Whoop app at https://developer-dashboard.whoop.com to include `read:workout` scope. 2. Edit /data/cameron/life/scripts/whoop_oauth.py SCOPES string to add `read:workout`. 3. Re-run `python3 /data/cameron/life/scripts/whoop_oauth.py` (one-time browser auth). 4. After that, get_workouts() and get_current_workout() will work. WHOOP DOES NOT EXPOSE: step count, ability to start/stop workouts via API. Workouts must be triggered on the device/app; the API can only read them after. """ if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "--summary": print(json.dumps(get_latest_summary(), indent=2)) elif len(sys.argv) > 1 and sys.argv[1] == "--fresh": refresh_if_stale(0) print(json.dumps(get_latest_summary(0), indent=2)) else: print(__doc__)