diff --git a/README.md b/README.md index cb49da2c..2faee18a 100644 --- a/README.md +++ b/README.md @@ -157,6 +157,7 @@ innovation-lab-examples/ | Example | Description | Tech Stack | Difficulty | |---------|-------------|------------|------------| | [contributors/community_agent](contributors/community_agent/) | AI community growth agent for events and hackathons | Python, uAgents, ASI:One, Tavily | 🟡 Intermediate | +| [contributors/cardiopulse-agent](contributors/cardiopulse-agent/) | Live cardiovascular fitness test from a Garmin BLE heart-rate stream | Python, uAgents, Bleak, ASI:One | 🟡 Intermediate | | [contributors/news-summarizer-agent](contributors/news-summarizer-agent/) | Fetches top headlines for a topic via NewsAPI and summarizes them with ASI:One, via Chat Protocol | Python, uAgents, NewsAPI, ASI:One | 🟡 Intermediate | ### 🌐 Web3 & Blockchain diff --git a/contributors/CHANGELOG.md b/contributors/CHANGELOG.md index e3f5b8ab..0b915bea 100644 --- a/contributors/CHANGELOG.md +++ b/contributors/CHANGELOG.md @@ -11,6 +11,8 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `gemini-research-agent/`: Added Gemini-powered research assistant demonstrating the standard Agent Chat Protocol (@Kavurubuvanesh) - `contributors/` folder and contribution guide for community agent examples - `contributors/community_agent/` — moved from repository root; AI community growth agent for events and hackathons +- `contributors/cardiopulse-agent/` — live cardiovascular fitness-test agent that streams a Garmin watch's heart rate over BLE and runs a three-phase test through ASI:One +- `contributors/community_agent/` — moved from repository root; AI community growth agent for events and hackathons ### Fixed - Fixed sandbox validation in `scan_directory` to properly reject paths outside the demo sandbox using `Path.relative_to()` (#159) - `contributors/news-summarizer-agent/` — beginner-friendly agent that fetches top headlines via NewsAPI and summarizes them with ASI:One; now a uAgent with Chat Protocol support diff --git a/contributors/cardiopulse-agent/.env.example b/contributors/cardiopulse-agent/.env.example new file mode 100644 index 00000000..a855b321 --- /dev/null +++ b/contributors/cardiopulse-agent/.env.example @@ -0,0 +1,40 @@ +# ============================================================================ +# CardioPulse — environment configuration +# Copy this file to `.env` and fill in the values. Never commit your `.env`. +# ============================================================================ + +# ---- Main CardioPulse agent ------------------------------------------------- + +# Seed phrase that deterministically generates the agent's address. +# Use any unique random string. Change this before deploying. +AGENT_SEED=change-me-to-a-unique-random-string + +# ASI:One API key — powers the natural-language coaching summary after a test. +# Optional: without it the agent falls back to a built-in static summary and +# still works end to end. Get a key at https://asi1.ai +ASI1_API_KEY=your_asi1_api_key_here +ASI1_MODEL=asi1 + +# ---- Bridge agent (runs locally on the machine with the watch) -------------- + +# Seed phrase for the bridge agent. Use a different unique random string. +BRIDGE_SEED=change-me-to-a-different-unique-random-string + +# Substring the bridge matches when scanning for your watch over BLE. +# A Garmin Forerunner advertises as e.g. "Forerunner 165 12345". +GARMIN_NAME=Forerunner + +# ---- Optional --------------------------------------------------------------- + +# Where the bridge POSTs heart-rate readings. Defaults to the main agent's +# local REST endpoint, which is correct when both run on the same machine. +# Only change this if you move the agent to a different host/port. +# AGENT_URL=http://127.0.0.1:8001/bpm + +# Reserved for a future cloud-hosted setup where the bridge would message the +# agent by address instead of POSTing to localhost. Not used in local mode. +# CARDIOPULSE_ADDRESS=agent1q... + +# Chart delivery works out of the box via catbox.moe (no key required). +# Optionally set an Imgur Client-ID to use Imgur as a fallback image host. +# IMGUR_CLIENT_ID=your_imgur_client_id_here diff --git a/contributors/cardiopulse-agent/.gitignore b/contributors/cardiopulse-agent/.gitignore new file mode 100644 index 00000000..24cd0b95 --- /dev/null +++ b/contributors/cardiopulse-agent/.gitignore @@ -0,0 +1,12 @@ +venv/ +__pycache__/ +*.pyc +.env +.DS_Store +*.log + +# Per-user test history — personal data, never commit +data/ + +# Regenerated chart artifact (charts are produced in-memory at runtime) +assets/last_test.png diff --git a/contributors/cardiopulse-agent/README.md b/contributors/cardiopulse-agent/README.md new file mode 100644 index 00000000..ce09ce26 --- /dev/null +++ b/contributors/cardiopulse-agent/README.md @@ -0,0 +1,171 @@ +# CardioPulse + +A live cardiovascular fitness-test agent for Agentverse and ASI:One. + +From a chat message, the agent reads your live heart rate from a Garmin watch +over Bluetooth, walks you through a short three-phase test, and returns an +estimated **Cardio Fitness Age** with reference ranges, an inline chart, and a +plain-language summary. Repeated runs build a per-user trend. + +> **Scope / disclaimer:** This is a proof of concept for a *live, stream-driven +> agent* — one that ingests a real-time sensor feed, reasons on it, and acts, +> all inside a chat. It is **not a medical device and not health advice.** The +> readings are illustrative estimates (roughly ±5 years); the trend across +> repeated tests matters far more than any single number. + +## Overview + +- **Category:** Live data, BLE, agent-to-agent +- **Tech stack:** Python, uAgents, Bleak (BLE), ASI:One Agent Chat Protocol, ASI-1, matplotlib +- **Status:** demo + +## Features + +- Ingests a live 1 Hz heart-rate stream from a Garmin watch over standard BLE. +- Runs a timed three-phase autonomic protocol (resting baseline → orthostatic → + paced breathing) entirely through a chat conversation. +- Returns a structured result with reference ranges, an inline HR-timeline chart, + and an ASI-1 natural-language coaching summary. +- Persists results per user and renders a trend chart across repeated tests. +- Two-agent architecture: a local BLE bridge agent + a mailbox-registered main + agent reachable through ASI:One. + +## Prerequisites + +- **Python 3.12** (uAgents is incompatible with Python 3.14's asyncio changes) +- A **Garmin watch** with "Broadcast Heart Rate" (Forerunner, Fenix, Venu, …), + or any standard BLE heart-rate strap (e.g. Polar H10) +- An [Agentverse](https://agentverse.ai) account +- *(Optional)* an [ASI:One](https://asi1.ai) API key for the AI coaching summary + +## Installation + +```bash +git clone +cd contributors/cardiopulse-agent +python3.12 -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +``` + +## Environment Variables + +```bash +cp .env.example .env +``` + +| Variable | Required | Purpose | +|---|---|---| +| `AGENT_SEED` | yes | Seed that generates the main agent's address | +| `BRIDGE_SEED` | yes | Seed for the bridge agent (use a different value) | +| `GARMIN_NAME` | yes | Substring matched when scanning for the watch over BLE | +| `ASI1_API_KEY` | optional | Enables the AI coaching summary (static fallback otherwise) | +| `AGENT_URL` | optional | Where the bridge POSTs readings (default `http://127.0.0.1:8001/bpm`) | +| `IMGUR_CLIENT_ID` | optional | Fallback image host; charts use catbox.moe by default | + +## Run the Agent + +Start the main agent first so its REST endpoint is up before the bridge POSTs to +it. Both run on the same machine. + +```bash +# 1. Main agent (chat protocol + local /bpm endpoint) +python agent.py + +# 2. On the watch: enable Broadcast Heart Rate and keep that screen active. + +# 3. Bridge agent (BLE -> POST to the main agent), in a second terminal +python bridge_agent.py +``` + +Then chat with the agent via ASI:One (or the Agentverse Manual Test panel): + +``` +age 25, start test +``` + +> For a clean, deterministic session, use the **Manual Test** panel on the +> agent's Agentverse profile rather than the main ASI:One chat. + +## Expected Output + +``` +User: age 27, start test +Agent: Got it — age 27. Starting the test now. +Agent: Connected. Latest BPM: 64. + Phase 1 — Resting baseline (2 minutes) ... + (3 minutes later) + Cardio Fitness Test — Results + Cardio Fitness Age: 24 (you are 27) + Resting HR: 58 bpm — a well-rested value for your age is typically under 65 bpm. + ... + [inline HR-timeline chart] + Coach's read: ... + [trend chart, from the second test onward] +``` + +Commands the agent understands: `help`, `age N`, `start test` +(`age N, start test` does both), `status`. + +## Demo + +![CardioPulse demo](./assets/demo.png) + +## Agent Profile + +Deployed on Agentverse via mailbox (the address is derived from your +`AGENT_SEED`). See [agentverse.ai](https://agentverse.ai). + +## Architecture + +``` +[ Garmin Forerunner ] + │ BLE — standard Heart Rate Service + ▼ +[ bridge_agent.py ] ── HTTP POST (localhost:8001/bpm) ──► [ agent.py — CardioPulse ] + local uAgent uAgent: REST + mailbox + │ + Agent Chat Protocol + (Agentverse mailbox) + ▼ + [ ASI:One chat ] +``` + +- **Bridge agent** (`bridge_agent.py`) subscribes to the watch's BLE Heart Rate + Measurement characteristic via Bleak and POSTs each reading to the main + agent's local REST endpoint, reconnecting on its own if the watch drops. +- **Main agent** (`agent.py`) receives the stream on a local `/bpm` endpoint and + is registered on Agentverse via mailbox for the ASI:One chat. Both agents run + on the same machine, so the localhost POST keeps up with the 1 Hz stream — + agent-to-agent mailbox routing is too slow for that rate, which is why HR data + uses REST while the conversation uses the mailbox. +- Readings accumulate in a thread-safe in-memory ring buffer (`session_state`). + The chat handler runs the test as a sequential coroutine and samples HR + windows from that buffer at each phase boundary — the telemetry stream + populates state but does not drive control flow. + +## Verifying scoring without a watch + +```bash +python test_scoring.py +``` + +Generates synthetic HR data for several hypothetical users and runs the scoring +engine end to end — useful for sanity-checking formula changes. Use +`python diagnose_ble.py` to confirm the watch is visible over BLE. + +## Troubleshooting + +- **"I'm not receiving any heart rate data"** — the watch isn't broadcasting. + Enable Broadcast Heart Rate and keep that screen awake (tap occasionally). +- **Bridge can't find the watch** — make sure `GARMIN_NAME` matches what the + watch advertises, and grant the terminal/Python Bluetooth permission + (macOS: System Settings → Privacy & Security → Bluetooth). +- **`RuntimeError: no running event loop` on startup** — you're on Python 3.14; + recreate the venv with Python 3.12. +- **Agent not reachable on ASI:One** — confirm the agent process is running and + its mailbox is connected; the mailbox only exists while the agent is up. + +## License + +Apache-2.0 (repository default). diff --git a/contributors/cardiopulse-agent/agent.py b/contributors/cardiopulse-agent/agent.py new file mode 100644 index 00000000..12a933d5 --- /dev/null +++ b/contributors/cardiopulse-agent/agent.py @@ -0,0 +1,69 @@ +""" +CardioAgent: a live cardiovascular fitness test agent on Agentverse + ASI:One. + +The agent guides users through a 3-minute test using live heart rate readings +that a companion `bridge_agent.py` POSTs to a local REST endpoint. (Agentverse +mailbox routing is too slow for 1Hz streaming, so the same-machine bridge uses +a direct localhost POST; the mailbox is used only for the ASI:One chat.) + +Run: + python agent.py + +Then, on the same machine, run the bridge agent (which talks to the Garmin +watch over BLE and POSTs readings to this agent): + + python bridge_agent.py +""" + +from __future__ import annotations + +import os + +from dotenv import load_dotenv + +# Load env vars BEFORE importing modules that read them at import time. +load_dotenv() + +from uagents import Agent, Context # noqa: E402 + +from chat_handler import chat_proto # noqa: E402 +from models import HRAck, HRReading # noqa: E402 +import session_state # noqa: E402 + + +AGENT_SEED = os.environ.get("AGENT_SEED", "cardio-agent-default-seed-change-me") + + +agent = Agent( + name="cardio-test-agent", + seed=AGENT_SEED, + port=8001, + mailbox=True, +) + +agent.include(chat_proto, publish_manifest=True) + + +# Receive HR readings via a fast local REST endpoint instead of agent-to-agent +# messaging. Agent messaging routes via Agentverse mailbox which is too slow +# for 1Hz streaming data; the bridge runs on the same machine, so direct +# localhost POST is both faster and simpler. +@agent.on_rest_post("/bpm", HRReading, HRAck) +async def receive_bpm(ctx: Context, req: HRReading) -> HRAck: + """REST endpoint that the local bridge POSTs heart rate readings to.""" + session_state.receive_bpm(req.bpm, req.rr_intervals, req.ts) + return HRAck(ok=True) + + +@agent.on_event("startup") +async def on_startup(ctx: Context) -> None: + ctx.logger.info("CardioAgent online") + ctx.logger.info(f"Agent address: {agent.address}") + ctx.logger.info( + "Copy this address into your bridge_agent .env as CARDIOPULSE_ADDRESS, " + "then run `python bridge_agent.py` on the machine with the watch." + ) + + +if __name__ == "__main__": + agent.run() diff --git a/contributors/cardiopulse-agent/assets/demo.png b/contributors/cardiopulse-agent/assets/demo.png new file mode 100644 index 00000000..ead38e7a Binary files /dev/null and b/contributors/cardiopulse-agent/assets/demo.png differ diff --git a/contributors/cardiopulse-agent/bridge_agent.py b/contributors/cardiopulse-agent/bridge_agent.py new file mode 100644 index 00000000..6844a22f --- /dev/null +++ b/contributors/cardiopulse-agent/bridge_agent.py @@ -0,0 +1,213 @@ +""" +Bridge agent: reads heart rate over BLE from a Garmin watch and POSTs each +reading to the CardioPulse agent's local REST endpoint (default +http://127.0.0.1:8001/bpm). + +This runs on the user's local machine (BLE is local hardware). The bridge and +the main agent run on the same machine, so a direct localhost POST keeps up +with the 1Hz heart-rate stream — agent-to-agent mailbox messaging is too slow +for that rate. Override AGENT_URL in .env only if you move the agent elsewhere. + + python bridge_agent.py +""" + +from __future__ import annotations + +import asyncio +import os +import time + +import httpx +from bleak import BleakClient, BleakScanner +from dotenv import load_dotenv +from uagents import Agent, Context + +print("Bridge agent starting up...", flush=True) + +load_dotenv() + + +# ---------- Configuration ---------------------------------------------------- + +# Local URL the bridge POSTs HR readings to. Both agents run on the same +# machine, so localhost is fast and direct (1Hz streaming). +AGENT_URL = os.environ.get("AGENT_URL", "http://127.0.0.1:8001/bpm") +BRIDGE_SEED = os.environ.get("BRIDGE_SEED", "cardio-bridge-default-seed-change-me") +DEVICE_HINT = os.environ.get("GARMIN_NAME", "Forerunner") + +# Kept for reference / future use when CardioPulse moves to a cloud host. +CARDIOPULSE_ADDRESS = os.environ.get("CARDIOPULSE_ADDRESS", "").strip() + +# Standard BLE Heart Rate Service / Measurement characteristic UUIDs. +HEART_RATE_SERVICE = "0000180d-0000-1000-8000-00805f9b34fb" +HEART_RATE_MEASUREMENT = "00002a37-0000-1000-8000-00805f9b34fb" + + +# ---------- BLE parsing ------------------------------------------------------ + + +def parse_hr_measurement(data: bytearray) -> dict: + """Decode the standard BLE Heart Rate Measurement payload.""" + flags = data[0] + hr_16bit = flags & 0x01 + energy_present = (flags >> 3) & 0x01 + rr_present = (flags >> 4) & 0x01 + + offset = 1 + if hr_16bit: + bpm = int.from_bytes(data[offset : offset + 2], byteorder="little") + offset += 2 + else: + bpm = data[offset] + offset += 1 + if energy_present: + offset += 2 # skip 2 bytes of energy + + rr_intervals: list[float] = [] + if rr_present: + while offset + 1 < len(data): + rr_raw = int.from_bytes(data[offset : offset + 2], byteorder="little") + rr_intervals.append(round(rr_raw * 1000 / 1024, 2)) + offset += 2 + + return {"bpm": bpm, "rr_intervals": rr_intervals} + + +async def find_garmin(): + """Scan once for a Garmin watch advertising the standard HR service.""" + devices = await BleakScanner.discover(timeout=15.0) + for d in devices: + name = d.name or "" + if DEVICE_HINT.lower() in name.lower() or "garmin" in name.lower(): + return d + return None + + +async def wait_for_garmin(ctx: Context): + """Keep scanning until the watch shows up. Don't exit if it's offline — + that would crash-loop under KeepAlive. Just wait patiently.""" + attempt = 0 + while True: + attempt += 1 + ctx.logger.info( + f"Scanning for BLE devices matching '{DEVICE_HINT}' " + f"(attempt {attempt}, 15s)..." + ) + device = await find_garmin() + if device: + ctx.logger.info(f"Found: {device.name} ({device.address})") + return device + ctx.logger.info( + "No matching device. Sleeping 30s before next scan. " + "Tip: enable Broadcast Heart Rate on your watch." + ) + await asyncio.sleep(30) + + +# ---------- Agent runtime ---------------------------------------------------- + +bridge = Agent( + name="cardio-bridge", + seed=BRIDGE_SEED, + port=8003, + mailbox=False, # Bridge talks to a known address; no mailbox needed. +) + +# Shared queue from BLE callback -> agent send loop. Using a queue lets the +# BLE callback stay fast (non-blocking) while the agent does the network I/O. +_send_queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=200) + +# Stats for the periodic stream status line. Counters and the last-error +# string are kept in separate containers so each value has a single type. +_stats: dict[str, int] = {"sent": 0, "ack": 0, "fail": 0} +_last_err: dict[str, str] = {"msg": ""} + + +@bridge.on_event("startup") +async def on_startup(ctx: Context) -> None: + ctx.logger.info(f"Bridge online (POSTing HR readings to {AGENT_URL})") + + # Spin up the BLE scanner + sender loop as background tasks. + asyncio.create_task(_ble_loop(ctx)) + asyncio.create_task(_sender_loop(ctx)) + asyncio.create_task(_status_loop(ctx)) + + +async def _ble_loop(ctx: Context) -> None: + """Outer loop: find the watch, hold the connection, reconnect if dropped. + + This loop runs forever. If the watch goes offline (broadcast disabled, + out of range, dead battery), the BLE client will raise and we just go + back to scanning instead of exiting. That way KeepAlive never has to + restart us. + """ + while True: + device = await wait_for_garmin(ctx) + try: + async with BleakClient(device) as ble: + ctx.logger.info(f"Connected to {device.name}. Streaming HR...") + + def on_hr(_handle, data: bytearray) -> None: + parsed = parse_hr_measurement(data) + parsed["ts"] = time.time() + try: + _send_queue.put_nowait(parsed) + except asyncio.QueueFull: + # Drop old readings rather than block the BLE callback. + pass + + await ble.start_notify(HEART_RATE_MEASUREMENT, on_hr) + # Stay connected as long as the BLE link is alive. + while ble.is_connected: + await asyncio.sleep(5) + ctx.logger.warning("BLE connection dropped. Will rescan and reconnect.") + except Exception as e: + ctx.logger.warning( + f"BLE loop error ({type(e).__name__}): {e}. Retrying in 10s." + ) + await asyncio.sleep(10) + + +async def _sender_loop(ctx: Context) -> None: + """Pull readings off the queue and POST them to the local agent endpoint. + + Uses a single httpx.AsyncClient with keep-alive so each POST reuses the + same TCP connection — fast enough to keep up with 1Hz BLE notifications. + """ + async with httpx.AsyncClient(timeout=2.0) as http: + while True: + reading = await _send_queue.get() + try: + resp = await http.post(AGENT_URL, json=reading) + if resp.status_code < 300: + _stats["sent"] += 1 + _stats["ack"] += 1 + else: + _stats["fail"] += 1 + _last_err["msg"] = f"HTTP {resp.status_code}: {resp.text[:120]}" + except Exception as e: + _stats["fail"] += 1 + _last_err["msg"] = f"{type(e).__name__}: {e}" + + +async def _status_loop(ctx: Context) -> None: + """Print a heartbeat status line every 10 seconds so the user can see flow.""" + while True: + await asyncio.sleep(10) + sent = _stats["sent"] + ack = _stats["ack"] + fail = _stats["fail"] + if fail > 0: + ctx.logger.info( + "stream: %d sent / %d ack / %d failed -> last error: %s", + sent, + ack, + fail, + _last_err["msg"], + ) + else: + ctx.logger.info("stream: %d sent / %d ack / 0 failed", sent, ack) + + +if __name__ == "__main__": + bridge.run() diff --git a/contributors/cardiopulse-agent/chart.py b/contributors/cardiopulse-agent/chart.py new file mode 100644 index 00000000..16f45252 --- /dev/null +++ b/contributors/cardiopulse-agent/chart.py @@ -0,0 +1,533 @@ +""" +HR-timeline chart generation. + +Builds a single matplotlib figure showing BPM over the duration of a test, +with each phase (baseline / orthostatic / breathing) shaded distinctly so the +user can visually see their cardiac response. Saves the image to disk and +returns the path. + +When a `TestResult` is supplied to `build()`, the chart is annotated with: + - A horizontal dashed line at the computed Resting HR + - A vertical bracket at the orthostatic peak showing the delta from RHR + - A stats panel in the upper-right corner with the headline numbers + grades +""" + +from __future__ import annotations + +import base64 +import io +from pathlib import Path +from typing import Sequence, TYPE_CHECKING + +import matplotlib + +matplotlib.use("Agg") # Headless backend — no display required. +import matplotlib.pyplot as plt + +if TYPE_CHECKING: + from history import TestRecord + from scoring import TestResult + +ASSETS_DIR = Path(__file__).parent / "assets" + + +def build( + baseline_samples: Sequence[tuple[float, int]], + orthostatic_samples: Sequence[tuple[float, int]], + breathing_samples: Sequence[tuple[float, int]], + output_path: Path | None = None, + result: "TestResult | None" = None, +) -> Path: + """ + Render an HR timeline with phase shading. + + Each `*_samples` argument is a sequence of (timestamp_seconds, bpm) tuples. + The chart shifts the first timestamp to 0 so the x-axis reads as + elapsed seconds. + + Returns the absolute path to the saved PNG. + """ + ASSETS_DIR.mkdir(exist_ok=True) + if output_path is None: + output_path = ASSETS_DIR / "last_test.png" + + all_samples = ( + list(baseline_samples) + list(orthostatic_samples) + list(breathing_samples) + ) + if not all_samples: + # Empty test — generate a placeholder so the agent still has something to send. + fig, ax = plt.subplots(figsize=(8, 4), dpi=110) + ax.text( + 0.5, + 0.5, + "No HR data collected", + ha="center", + va="center", + transform=ax.transAxes, + fontsize=14, + color="#888", + ) + ax.set_axis_off() + fig.savefig(output_path, bbox_inches="tight") + plt.close(fig) + return output_path + + t0 = all_samples[0][0] + + def normalise(samples): + return [(t - t0, bpm) for t, bpm in samples] + + baseline_norm = normalise(baseline_samples) + ortho_norm = normalise(orthostatic_samples) + breath_norm = normalise(breathing_samples) + + fig, ax = plt.subplots(figsize=(9, 4.5), dpi=110) + + # Phase boundaries + if baseline_norm: + b_end = baseline_norm[-1][0] + else: + b_end = 0 + if ortho_norm: + o_end = ortho_norm[-1][0] + else: + o_end = b_end + if breath_norm: + br_end = breath_norm[-1][0] + else: + br_end = o_end + + # Background shading per phase + ax.axvspan(0, b_end, color="#E3F2FD", alpha=0.6, label="_nolegend_") + ax.axvspan(b_end, o_end, color="#FFF3E0", alpha=0.6, label="_nolegend_") + ax.axvspan(o_end, br_end, color="#E8F5E9", alpha=0.6, label="_nolegend_") + + # Plot HR lines per phase with distinct colours + if baseline_norm: + xs, ys = zip(*baseline_norm) + ax.plot(xs, ys, color="#1976D2", linewidth=1.8, label="Resting baseline") + if ortho_norm: + xs, ys = zip(*ortho_norm) + ax.plot(xs, ys, color="#F57C00", linewidth=1.8, label="Standing") + if breath_norm: + xs, ys = zip(*breath_norm) + ax.plot(xs, ys, color="#388E3C", linewidth=1.8, label="Paced breathing") + + # Phase labels at the top + if b_end > 0: + ax.text( + b_end / 2, + ax.get_ylim()[1] if False else 1.02, + "Baseline", + transform=ax.get_xaxis_transform(), + ha="center", + fontsize=10, + color="#1976D2", + fontweight="bold", + ) + if o_end > b_end: + ax.text( + (b_end + o_end) / 2, + 1.02, + "Stand", + transform=ax.get_xaxis_transform(), + ha="center", + fontsize=10, + color="#F57C00", + fontweight="bold", + ) + if br_end > o_end: + ax.text( + (o_end + br_end) / 2, + 1.02, + "Breathe", + transform=ax.get_xaxis_transform(), + ha="center", + fontsize=10, + color="#388E3C", + fontweight="bold", + ) + + # Mark orthostatic peak if present + peak_t, peak_bpm = None, None + if ortho_norm: + peak_t, peak_bpm = max(ortho_norm, key=lambda p: p[1]) + ax.annotate( + f"peak {peak_bpm} bpm", + xy=(peak_t, peak_bpm), + xytext=(peak_t + 5, peak_bpm + 6), + fontsize=9, + color="#333", + arrowprops=dict(arrowstyle="->", color="#666", lw=0.8), + ) + + # --- Annotations driven by the scored result ----------------------------- + if result is not None: + rhr = result.resting_hr + + # 1. Horizontal dashed line at the computed Resting HR. + ax.axhline( + y=rhr, + color="#1976D2", + linestyle="--", + linewidth=1.1, + alpha=0.65, + zorder=2, + ) + # Right-edge label so the line means something visually. + x_right = ax.get_xlim()[1] + ax.text( + x_right * 0.985, + rhr - 1.2, + f"Resting HR {rhr} bpm", + color="#1976D2", + fontsize=9, + fontweight="semibold", + ha="right", + va="top", + bbox=dict( + facecolor="white", + edgecolor="#1976D2", + boxstyle="round,pad=0.25", + alpha=0.85, + ), + ) + + # 2. Vertical bracket at the orthostatic peak showing the delta. + if peak_t is not None and peak_bpm is not None and result.orthostatic_delta > 0: + bracket_x = ( + peak_t - 4 + ) # offset a bit left of the peak so it doesn't overlap + # Stem of the bracket + ax.annotate( + "", + xy=(bracket_x, peak_bpm), + xytext=(bracket_x, rhr), + arrowprops=dict(arrowstyle="-", color="#F57C00", linewidth=1.4), + ) + # Tick marks at top and bottom + tick_w = 2 + ax.plot( + [bracket_x - tick_w, bracket_x + tick_w], + [peak_bpm, peak_bpm], + color="#F57C00", + linewidth=1.4, + zorder=3, + ) + ax.plot( + [bracket_x - tick_w, bracket_x + tick_w], + [rhr, rhr], + color="#F57C00", + linewidth=1.4, + zorder=3, + ) + # Delta label + mid_y = (rhr + peak_bpm) / 2 + ax.text( + bracket_x - 5, + mid_y, + f"+{result.orthostatic_delta}\nbpm", + color="#F57C00", + fontsize=9, + fontweight="semibold", + ha="right", + va="center", + ) + + # 3. Stats panel — small overlay in the upper-left. + verdict_short = ( + "younger" + if result.cardio_fitness_age < result.age + else "aligned" + if result.cardio_fitness_age == result.age + else "older" + ) + delta_years = abs(result.cardio_fitness_age - result.age) + if verdict_short == "aligned": + cardio_summary = f"= age {result.age}" + elif verdict_short == "younger": + cardio_summary = f"{delta_years}y younger" + else: + cardio_summary = f"{delta_years}y older" + + stats_text = ( + f"Cardio Age: {result.cardio_fitness_age} ({cardio_summary})\n" + f"Resting HR: {result.resting_hr} bpm ({result.rhr_grade})\n" + f"Orthostatic: +{result.orthostatic_delta} bpm\n" + f"Breathing: {result.breathing_variance} bpm RSA" + ) + ax.text( + 0.018, + 0.97, + stats_text, + transform=ax.transAxes, + fontsize=9, + family="monospace", + verticalalignment="top", + horizontalalignment="left", + bbox=dict( + facecolor="white", + edgecolor="#999", + boxstyle="round,pad=0.45", + alpha=0.92, + ), + ) + + ax.set_xlabel("Elapsed time (seconds)", fontsize=10) + ax.set_ylabel("Heart rate (bpm)", fontsize=10) + ax.set_title("Your HR across the cardio fitness test", fontsize=13, pad=18) + ax.legend(loc="lower right", fontsize=9, framealpha=0.9) + ax.grid(True, alpha=0.25, linestyle="--") + + fig.tight_layout() + fig.savefig(output_path, bbox_inches="tight", facecolor="white") + plt.close(fig) + return output_path + + +def build_png_bytes( + baseline_samples: Sequence[tuple[float, int]], + orthostatic_samples: Sequence[tuple[float, int]], + breathing_samples: Sequence[tuple[float, int]], + result: "TestResult | None" = None, +) -> bytes: + """Render the chart and return raw PNG bytes — no disk writes. + + Use this when you want to upload the chart somewhere (Agentverse storage, + imgur, S3) without saving locally first. + """ + return _render_png_buf( + baseline_samples, orthostatic_samples, breathing_samples, result + ).getvalue() + + +def build_trend_png_bytes(records: "list[TestRecord]") -> bytes: + """Render a trend chart showing Cardio Fitness Age + Resting HR over the + last N tests. Returns PNG bytes. Returns empty bytes if fewer than 2 + records (nothing to trend yet). + """ + if not records or len(records) < 2: + return b"" + + # Index x-axis as 1..N (test number) rather than dates — clearer in chat. + xs = list(range(1, len(records) + 1)) + cardio_ages = [r.cardio_fitness_age for r in records] + rhrs = [r.resting_hr for r in records] + chronological_age = records[-1].age # most recent test's chrono age + + fig, ax_left = plt.subplots(figsize=(7, 3.5), dpi=80) + + # Left axis: Cardio Fitness Age (blue line + dots) + color_left = "#1976D2" + ax_left.plot( + xs, + cardio_ages, + color=color_left, + marker="o", + linewidth=2, + label="Cardio Fitness Age", + ) + ax_left.set_xlabel("Test number", fontsize=10) + ax_left.set_ylabel("Cardio Fitness Age (years)", color=color_left, fontsize=10) + ax_left.tick_params(axis="y", labelcolor=color_left) + + # Reference line: user's chronological age + ax_left.axhline( + y=chronological_age, + color=color_left, + linestyle=":", + alpha=0.45, + linewidth=1.0, + ) + ax_left.text( + xs[-1], + chronological_age + 0.2, + f"chronological age ({chronological_age})", + color=color_left, + fontsize=8, + ha="right", + va="bottom", + alpha=0.75, + ) + + # Right axis: Resting HR (orange line + dots) + color_right = "#F57C00" + ax_right = ax_left.twinx() + ax_right.plot( + xs, rhrs, color=color_right, marker="s", linewidth=2, label="Resting HR" + ) + ax_right.set_ylabel("Resting HR (bpm)", color=color_right, fontsize=10) + ax_right.tick_params(axis="y", labelcolor=color_right) + + # Annotate the latest point with both values + latest_x = xs[-1] + latest_ca = cardio_ages[-1] + latest_rhr = rhrs[-1] + ax_left.annotate( + f"{latest_ca}", + xy=(latest_x, latest_ca), + xytext=(6, 6), + textcoords="offset points", + color=color_left, + fontsize=10, + fontweight="bold", + ) + ax_right.annotate( + f"{latest_rhr}", + xy=(latest_x, latest_rhr), + xytext=(6, -14), + textcoords="offset points", + color=color_right, + fontsize=10, + fontweight="bold", + ) + + # Trend direction text in upper-left + first_ca, last_ca = cardio_ages[0], cardio_ages[-1] + first_rhr, last_rhr = rhrs[0], rhrs[-1] + ca_delta = last_ca - first_ca + rhr_delta = last_rhr - first_rhr + + if ca_delta < 0: + ca_trend = f"Cardio Age: {abs(ca_delta)} years younger" + elif ca_delta > 0: + ca_trend = f"Cardio Age: {ca_delta} years older" + else: + ca_trend = "Cardio Age: stable" + + rhr_dir = "down" if rhr_delta < 0 else "up" if rhr_delta > 0 else "flat" + rhr_trend = f"Resting HR: {rhr_dir} {abs(rhr_delta)} bpm" + + summary = f"Across {len(records)} tests\n{ca_trend}\n{rhr_trend}" + ax_left.text( + 0.02, + 0.97, + summary, + transform=ax_left.transAxes, + fontsize=8, + family="monospace", + verticalalignment="top", + bbox=dict( + facecolor="white", edgecolor="#999", boxstyle="round,pad=0.35", alpha=0.9 + ), + ) + + ax_left.set_title("Your trend across recent tests", fontsize=12, pad=10) + ax_left.grid(True, alpha=0.25, linestyle="--") + ax_left.set_xticks(xs) + + fig.tight_layout() + buf = io.BytesIO() + fig.savefig(buf, format="png", bbox_inches="tight", facecolor="white") + plt.close(fig) + return buf.getvalue() + + +def build_data_uri( + baseline_samples: Sequence[tuple[float, int]], + orthostatic_samples: Sequence[tuple[float, int]], + breathing_samples: Sequence[tuple[float, int]], + result: "TestResult | None" = None, +) -> str: + """Render the same chart and return it as a base64 data URI. + + Suitable for embedding directly in a markdown image tag inside a chat + message: `![chart](data:image/png;base64,...)`. No upload needed. + + We re-render at slightly smaller size + lower DPI than build() to keep the + encoded payload modest (~40-60 KB) so chat protocols don't choke. + """ + buf = _render_png_buf( + baseline_samples, orthostatic_samples, breathing_samples, result + ) + if buf.getbuffer().nbytes == 0: + return "" + b64 = base64.b64encode(buf.getvalue()).decode() + return f"data:image/png;base64,{b64}" + + +def _render_png_buf( + baseline_samples: Sequence[tuple[float, int]], + orthostatic_samples: Sequence[tuple[float, int]], + breathing_samples: Sequence[tuple[float, int]], + result: "TestResult | None" = None, +) -> "io.BytesIO": + """Internal: render the chart into an in-memory BytesIO buffer. + + Returns an empty buffer if there's no sample data to plot. + """ + all_samples = ( + list(baseline_samples) + list(orthostatic_samples) + list(breathing_samples) + ) + if not all_samples: + return io.BytesIO() + + t0 = all_samples[0][0] + + def normalise(samples): + return [(t - t0, bpm) for t, bpm in samples] + + baseline_norm = normalise(baseline_samples) + ortho_norm = normalise(orthostatic_samples) + breath_norm = normalise(breathing_samples) + + fig, ax = plt.subplots(figsize=(7, 3.3), dpi=80) + + b_end = baseline_norm[-1][0] if baseline_norm else 0 + o_end = ortho_norm[-1][0] if ortho_norm else b_end + br_end = breath_norm[-1][0] if breath_norm else o_end + + ax.axvspan(0, b_end, color="#E3F2FD", alpha=0.6) + ax.axvspan(b_end, o_end, color="#FFF3E0", alpha=0.6) + ax.axvspan(o_end, br_end, color="#E8F5E9", alpha=0.6) + + if baseline_norm: + xs, ys = zip(*baseline_norm) + ax.plot(xs, ys, color="#1976D2", linewidth=1.6, label="Resting baseline") + if ortho_norm: + xs, ys = zip(*ortho_norm) + ax.plot(xs, ys, color="#F57C00", linewidth=1.6, label="Standing") + if breath_norm: + xs, ys = zip(*breath_norm) + ax.plot(xs, ys, color="#388E3C", linewidth=1.6, label="Paced breathing") + + if result is not None: + ax.axhline( + y=result.resting_hr, + color="#1976D2", + linestyle="--", + linewidth=1.0, + alpha=0.6, + ) + stats = ( + f"Cardio Age: {result.cardio_fitness_age} (age {result.age})\n" + f"RHR: {result.resting_hr} ({result.rhr_grade})\n" + f"Ortho: +{result.orthostatic_delta} bpm\n" + f"RSA: {result.breathing_variance} bpm" + ) + ax.text( + 0.018, + 0.97, + stats, + transform=ax.transAxes, + fontsize=8, + family="monospace", + verticalalignment="top", + bbox=dict( + facecolor="white", + edgecolor="#999", + boxstyle="round,pad=0.3", + alpha=0.92, + ), + ) + + ax.set_xlabel("Elapsed time (s)", fontsize=9) + ax.set_ylabel("Heart rate (bpm)", fontsize=9) + ax.set_title("Your HR across the cardio fitness test", fontsize=11) + ax.legend(loc="lower right", fontsize=8, framealpha=0.9) + ax.grid(True, alpha=0.25, linestyle="--") + + buf = io.BytesIO() + fig.savefig(buf, format="png", bbox_inches="tight", facecolor="white") + plt.close(fig) + return buf diff --git a/contributors/cardiopulse-agent/chat_handler.py b/contributors/cardiopulse-agent/chat_handler.py new file mode 100644 index 00000000..1177a19d --- /dev/null +++ b/contributors/cardiopulse-agent/chat_handler.py @@ -0,0 +1,374 @@ +""" +Chat protocol handler. + +Drives the 3-phase cardio fitness test through ASI:One chat: + 1. Resting baseline (2 minutes) + 2. Orthostatic challenge (30 seconds) + 3. Paced breathing (30 seconds) +""" + +from __future__ import annotations + +import asyncio +import re +import time +from datetime import datetime, timezone +from uuid import uuid4 + +from uagents import Context, Protocol +from uagents_core.contrib.protocols.chat import ( + ChatAcknowledgement, + ChatMessage, + EndSessionContent, + StartSessionContent, + TextContent, + chat_protocol_spec, +) + +import chart as chart_module +import coach as coach_module +import history +import image_host +import scoring +import session_state + +chat_proto = Protocol(spec=chat_protocol_spec) + + +WELCOME = ( + "Hi! I'm a Cardio Fitness Test agent. " + "In about 3 minutes I can estimate your cardiovascular age from your " + "heart rate using your Garmin watch.\n\n" + "**Before we start:**\n" + "1. Wear your Garmin watch.\n" + "2. Enable Broadcast Heart Rate on the watch " + "(Settings → Health and Wellness → Wrist Heart Rate → Broadcast Heart Rate → Start).\n" + "3. Sit somewhere quiet.\n\n" + "**Tell me your age** (e.g. `age 25`), then say `start test`.\n\n" + "_You can also just say `age 25, start test` to do both at once._" +) + +# Per-session state. Each ASI:One conversation has its own session id. +# Keys: session id (str). Values: dict with the user's age (or None until set) +# and a flag for whether a test is currently running. +SESSIONS: dict[str, dict] = {} + +BASELINE_SEC = 120 +ORTHOSTATIC_SEC = 30 +BREATHING_SEC = 30 + + +def _new_session() -> dict: + return {"age": None, "running": False} + + +def _text(text: str, end_session: bool = False) -> ChatMessage: + content: list = [TextContent(type="text", text=text)] + if end_session: + content.append(EndSessionContent(type="end-session")) + return ChatMessage( + timestamp=datetime.now(timezone.utc), + msg_id=uuid4(), + content=content, + ) + + +def _parse_age(text: str) -> int | None: + """Pull a plausible age out of a free-text message.""" + match = re.search(r"\b(\d{2})\b", text) + if not match: + return None + age = int(match.group(1)) + if 10 <= age <= 100: + return age + return None + + +def _strip_mention(text: str) -> str: + """Remove a leading @ mention so keyword matching still works. + + ASI:One main chat requires explicit @
mentions to force-route a + message to a specific agent. The mention shouldn't change how we interpret + the rest of the text. + """ + return re.sub(r"^@\S+\s*", "", text).strip() + + +async def _run_test(ctx: Context, sender: str, session_id: str) -> None: + """Run the 3-phase test and report results.""" + + sess = SESSIONS.setdefault(session_id, _new_session()) + + # Prevent overlapping tests within the same session. + if sess["running"]: + await ctx.send( + sender, + _text("A test is already in progress in this session. Hold tight."), + ) + return + + # Age is required for scoring — refuse to guess. + if sess["age"] is None: + await ctx.send( + sender, + _text( + "I need your age before I can score the test. " + "Reply with something like `age 27`, then say `start test`." + ), + ) + return + + if not session_state.is_streaming(): + await ctx.send( + sender, + _text( + "I'm not receiving any heart rate data yet. " + "Make sure your Garmin watch is on and Broadcast Heart Rate is " + "active (Settings → Health and Wellness → Wrist Heart Rate → " + "Broadcast Heart Rate → Start). Keep that broadcast screen open, " + "then say `start test` again." + ), + ) + return + + sess["running"] = True + try: + age = sess["age"] + ctx.logger.info(f"Starting test for session {session_id}, age={age}") + + await ctx.send( + sender, + _text( + f"Connected. Latest BPM: {session_state.latest_bpm()}.\n\n" + "**Phase 1 — Resting baseline (2 minutes)**\n" + "Sit calmly, breathe normally, keep your wrist still." + ), + ) + + baseline_start = time.time() + await asyncio.sleep(BASELINE_SEC / 2) + await ctx.send( + sender, + _text( + "Halfway through baseline. Keep sitting calmly. " + f"Current BPM: {session_state.latest_bpm()}." + ), + ) + await asyncio.sleep(BASELINE_SEC / 2) + baseline_end = time.time() + + await ctx.send( + sender, + _text( + "**Phase 2 — Stand up (30 seconds)**\n" + "Stand up now and stay still. Don't move your arms." + ), + ) + ortho_start = time.time() + await asyncio.sleep(ORTHOSTATIC_SEC) + ortho_end = time.time() + + await ctx.send( + sender, + _text( + "**Phase 3 — Paced breathing (30 seconds)**\n" + "Breathe in for 5 seconds, out for 5 seconds. " + "Three full breath cycles." + ), + ) + breath_start = time.time() + await asyncio.sleep(BREATHING_SEC) + breath_end = time.time() + + baseline = session_state.bpm_in_window(baseline_start, baseline_end) + ortho = session_state.bpm_in_window(ortho_start, ortho_end) + breath = session_state.bpm_in_window(breath_start, breath_end) + + # Pull timestamped series for the chart (separate from the BPM-only + # arrays the scorer wants). + baseline_series = session_state.bpm_series_in_window( + baseline_start, baseline_end + ) + ortho_series = session_state.bpm_series_in_window(ortho_start, ortho_end) + breath_series = session_state.bpm_series_in_window(breath_start, breath_end) + + ctx.logger.info( + f"Samples collected: baseline={len(baseline)}, " + f"ortho={len(ortho)}, breath={len(breath)}" + ) + + try: + result = scoring.compute( + age=age, + baseline_bpm=baseline, + orthostatic_bpm=ortho, + breathing_bpm=breath, + ) + except ValueError as e: + await ctx.send( + sender, + _text( + f"Couldn't score the test: {e}\n\n" + "This usually means the watch broadcast stopped mid-test. " + "Re-enable HR broadcast on the watch and try again." + ), + ) + return + + # Build EVERYTHING first, then deliver ONE consolidated message. + # ASI:One's main chat merges and re-renders multiple agent messages + # into a jumbled blob; a single message stays intact. Charts go to + # Imgur and come back as public URLs that render as markdown images. + + # Chart 1: this test's HR timeline. + chart_md = "" + try: + png_bytes = chart_module.build_png_bytes( + baseline_samples=baseline_series, + orthostatic_samples=ortho_series, + breathing_samples=breath_series, + result=result, + ) + if png_bytes: + url = image_host.upload_image(png_bytes) + if url: + chart_md = ( + f"\n\n![Your HR timeline]({url})\n" + "_Your heart rate across the three phases: resting " + "baseline, standing, paced breathing._" + ) + ctx.logger.info(f"Test chart posted: {url}") + else: + ctx.logger.warning("Chart skipped — image upload failed.") + except Exception as e: + ctx.logger.warning(f"Chart generation failed: {e}") + + # Coaching paragraph, with trend context from THIS USER's history only. + coach_text = "" + try: + prev = history.previous(sender) + coach_text = coach_module.coach(result, previous=prev) + except Exception as e: + ctx.logger.warning(f"Coach paragraph failed: {e}") + + # Persist this test under the sender's key so future trend + # comparisons never mix different users' tests. + try: + history.append(result, sender) + except Exception as e: + ctx.logger.warning(f"History append failed: {e}") + + # Chart 2: trend across this user's tests (needs 2+ on file). + trend_md = "" + try: + records = history.recent(limit=10, sender=sender) + if len(records) >= 2: + trend_bytes = chart_module.build_trend_png_bytes(records) + if trend_bytes: + trend_url = image_host.upload_image(trend_bytes) + if trend_url: + trend_md = ( + f"\n\n![Your trend]({trend_url})\n" + f"_Cardio Fitness Age and Resting HR across your " + f"last {len(records)} tests._" + ) + ctx.logger.info(f"Trend chart posted: {trend_url}") + except Exception as e: + ctx.logger.warning(f"Trend chart failed: {e}") + + # Assemble and send as ONE message. + final = scoring.format_result(result) + final += chart_md + if coach_text: + final += f"\n\n**Coach's read**\n{coach_text}" + final += trend_md + + await ctx.send(sender, _text(final)) + ctx.logger.info(f"History updated. {history.count()} test(s) on file.") + finally: + sess["running"] = False + + +async def _handle_text(ctx: Context, sender: str, session_id: str, text: str) -> None: + sess = SESSIONS.setdefault(session_id, _new_session()) + + # Strip any @ prefix so keyword matching works whether the + # user is in Manual Test (no mention) or ASI:One main chat (mention required). + cleaned = _strip_mention(text) + lower = cleaned.lower().strip() + + has_age_keyword = "age" in lower + has_start = "start" in lower and "test" in lower + + if has_age_keyword: + age = _parse_age(cleaned) + if age is not None: + sess["age"] = age + if has_start: + # Combined "age N, start test" — set age and immediately kick off. + await ctx.send( + sender, + _text(f"Got it — age {age}. Starting the test now."), + ) + await _run_test(ctx, sender, session_id) + return + await ctx.send( + sender, + _text(f"Got it — age {age}. When you're ready, say `start test`."), + ) + return + if not has_start: + await ctx.send(sender, _text("Couldn't parse that age. Try `age 25`.")) + return + + if has_start: + await _run_test(ctx, sender, session_id) + return + + if lower in {"hi", "hello", "hey", "?", "help", ""}: + await ctx.send(sender, _text(WELCOME)) + return + + if lower in {"status", "bpm", "ping"}: + if session_state.is_streaming(): + await ctx.send( + sender, + _text(f"Streaming. Latest BPM: {session_state.latest_bpm()}."), + ) + else: + await ctx.send( + sender, + _text( + "Not receiving HR data. Make sure your Garmin watch is on " + "and Broadcast Heart Rate is active." + ), + ) + return + + # Default fallback: re-show the welcome message. + await ctx.send(sender, _text(WELCOME)) + + +@chat_proto.on_message(ChatMessage) +async def on_chat(ctx: Context, sender: str, msg: ChatMessage) -> None: + await ctx.send( + sender, + ChatAcknowledgement( + timestamp=datetime.now(timezone.utc), + acknowledged_msg_id=msg.msg_id, + ), + ) + + session_id = str(ctx.session) + for item in msg.content: + if isinstance(item, StartSessionContent): + await ctx.send(sender, _text(WELCOME)) + elif isinstance(item, TextContent): + await _handle_text(ctx, sender, session_id, item.text) + elif isinstance(item, EndSessionContent): + SESSIONS.pop(session_id, None) + + +@chat_proto.on_message(ChatAcknowledgement) +async def on_ack(ctx: Context, sender: str, msg: ChatAcknowledgement) -> None: + ctx.logger.debug(f"ACK from {sender} for {msg.acknowledged_msg_id}") diff --git a/contributors/cardiopulse-agent/coach.py b/contributors/cardiopulse-agent/coach.py new file mode 100644 index 00000000..5d1ac7a5 --- /dev/null +++ b/contributors/cardiopulse-agent/coach.py @@ -0,0 +1,163 @@ +""" +Personalized coaching layer. + +Takes the deterministic TestResult plus the user's age and turns it into a +short, conversational coach paragraph using ASI:One. The structured numbers +are still shown first — the coach paragraph is added below for color and +context. + +Falls back to a static encouraging message if ASI:One is unreachable. +""" + +from __future__ import annotations + +import os + +from openai import OpenAI + +from history import TestRecord, time_since +from scoring import TestResult + +ASI1_BASE_URL = "https://api.asi1.ai/v1" + + +SYSTEM_PROMPT = """You are a careful, honest cardiovascular health coach. + +You will be given: +- A user's age +- Their measured Cardio Fitness Age estimate from today's test +- Resting HR, orthostatic delta (HR jump on standing), and breathing-driven + HR variance from today's 3-minute test +- The user-facing grade for resting HR +- Optionally, the same metrics from their previous test plus when it was taken + +Your job: write a 3-5 sentence coach paragraph. + +Rules: +- If a DATA QUALITY WARNING is present, lead with THAT: tell the user plainly + that this reading is low-confidence and why, and do not draw strong + conclusions or alarming comparisons from it. Suggest how to get a clean + reading instead. +- Otherwise, if a previous test is provided, lead with the trend: did the + score improve, stay flat, or worsen, and by how much. Make the trend the + most prominent part of the message — it's what makes the coaching feel + personalised. If the trend and an individual metric disagree, say which + one the user should trust and why, in one sentence — never present a + contradiction without resolving it. +- If no previous test is provided, lead with today's result, acknowledged + honestly. If the score is alarming, name the most likely benign cause + (caffeine, stress, recent activity, NOT a true rested state) before + suggesting anything. +- Reference at least one specific number so the user knows you read the data. +- Suggest ONE concrete action they can take in the next 24 hours (not a list). +- Be encouraging but not patronising. No emojis. No motivational fluff. +- Never claim clinical diagnosis. Use "may", "could suggest", "likely". +- Total length: 3-5 sentences. No headings, no bullet lists. +""" + + +def _build_prompt(result: TestResult, previous: TestRecord | None = None) -> str: + lines = [ + "TODAY'S TEST:", + f" Chronological age: {result.age}", + f" Cardio Fitness Age: {result.cardio_fitness_age}", + f" Verdict: {result.verdict}", + f" Resting HR: {result.resting_hr} bpm (grade: {result.rhr_grade})", + f" Orthostatic HR delta: +{result.orthostatic_delta} bpm", + f" Breathing-driven HR variance: {result.breathing_variance} bpm", + ] + + if result.quality_note: + lines.append(f" DATA QUALITY WARNING: {result.quality_note}") + + if previous is not None: + cardio_delta = result.cardio_fitness_age - previous.cardio_fitness_age + rhr_delta = result.resting_hr - previous.resting_hr + when = time_since(previous) + + if cardio_delta < 0: + direction = f"improved by {abs(cardio_delta)} year(s)" + elif cardio_delta > 0: + direction = f"worsened by {cardio_delta} year(s)" + else: + direction = "stayed the same" + + lines.extend( + [ + "", + f"PREVIOUS TEST ({when}):", + f" Cardio Fitness Age: {previous.cardio_fitness_age}", + f" Resting HR: {previous.resting_hr} bpm", + f" Orthostatic HR delta: +{previous.orthostatic_delta} bpm", + "", + "TREND:", + f" Cardio Fitness Age has {direction} since the previous test.", + f" Resting HR is {'down' if rhr_delta < 0 else 'up' if rhr_delta > 0 else 'flat'} " + f"{abs(rhr_delta)} bpm.", + ] + ) + + return "\n".join(lines) + + +def _fallback(result: TestResult, previous: TestRecord | None = None) -> str: + """Static encouraging message when ASI:One is unavailable.""" + if previous is not None: + cardio_delta = result.cardio_fitness_age - previous.cardio_fitness_age + when = time_since(previous) + if cardio_delta < 0: + return ( + f"Your Cardio Fitness Age improved by {abs(cardio_delta)} year(s) " + f"since your previous test ({when}). Keep doing whatever you've been " + "doing — the trend is what matters." + ) + elif cardio_delta > 0: + return ( + f"Your Cardio Fitness Age is {cardio_delta} year(s) higher than your " + f"previous test ({when}). Could be a less-rested baseline today — " + "try re-testing in the same conditions to confirm the trend." + ) + else: + return ( + f"Your Cardio Fitness Age is the same as your previous test ({when}). " + "Steady is a good sign. The needle moves slowly — keep at it." + ) + + if result.cardio_fitness_age <= result.age: + return ( + "Your cardiovascular function is reading at or below your age. " + "Keep doing what you're doing. The number that matters is the trend over time, " + "so re-test in a few weeks to see how interventions move it." + ) + delta = result.cardio_fitness_age - result.age + return ( + f"Your reading came back {delta} years above your age. Most often this means the " + "baseline wasn't truly rested (caffeine, stress, sitting at a desk). " + "Try re-testing first thing tomorrow morning before getting out of bed. " + "What matters is the trend over multiple readings, not the single number." + ) + + +def coach(result: TestResult, previous: TestRecord | None = None) -> str: + """Generate a personalised coach paragraph. Returns fallback on any error. + + If `previous` is provided, the coaching includes a trend comparison. + """ + api_key = os.environ.get("ASI1_API_KEY") + if not api_key: + return _fallback(result, previous) + + try: + client = OpenAI(base_url=ASI1_BASE_URL, api_key=api_key) + resp = client.chat.completions.create( + model=os.environ.get("ASI1_MODEL", "asi1"), + messages=[ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": _build_prompt(result, previous)}, + ], + temperature=0.4, + ) + text = (resp.choices[0].message.content or "").strip() + return text or _fallback(result, previous) + except Exception: + return _fallback(result, previous) diff --git a/contributors/cardiopulse-agent/diagnose_ble.py b/contributors/cardiopulse-agent/diagnose_ble.py new file mode 100644 index 00000000..291b8e71 --- /dev/null +++ b/contributors/cardiopulse-agent/diagnose_ble.py @@ -0,0 +1,92 @@ +""" +BLE diagnostic. + +Scans for ALL Bluetooth Low Energy devices for 20 seconds and prints whatever +it finds. Use this to confirm: + 1. macOS / your OS will let Python use Bluetooth + 2. Your Garmin watch is actually broadcasting + 3. What name the watch is advertising (so bridge_agent.py can match it) + + python diagnose_ble.py +""" + +from __future__ import annotations + +import asyncio + +from bleak import BleakScanner + +HEART_RATE_SERVICE = "0000180d-0000-1000-8000-00805f9b34fb" + + +async def main() -> None: + print("Scanning for BLE devices for 20 seconds...") + print("(Make sure your watch is broadcasting HR or in an indoor activity.)\n") + + devices = await BleakScanner.discover(timeout=20.0, return_adv=True) + + if not devices: + print("No devices found at all.") + print() + print("If this surprises you, the most likely cause is macOS not having") + print("granted Python permission to use Bluetooth.") + print("Fix: System Settings -> Privacy & Security -> Bluetooth") + print("Add Terminal (or whatever shell you're using) to the list.") + return + + print(f"Found {len(devices)} device(s):\n") + + has_garmin = False + has_hr = False + + for addr, (device, adv) in devices.items(): + name = device.name or adv.local_name or "(no name)" + services = list(adv.service_uuids or []) + rssi = adv.rssi + + is_garmin = "garmin" in name.lower() or "forerunner" in name.lower() + advertises_hr = HEART_RATE_SERVICE in services + + if is_garmin: + has_garmin = True + if advertises_hr: + has_hr = True + + marker = "" + if is_garmin and advertises_hr: + marker = " <-- THIS IS YOUR WATCH BROADCASTING HR" + elif is_garmin: + marker = " <-- Garmin device but NOT broadcasting HR right now" + elif advertises_hr: + marker = " <-- BLE device advertising HR (possible match)" + + print(f" {name}") + print(f" address: {addr}") + print(f" RSSI: {rssi} dBm") + if services: + print(f" services: {services}") + if marker: + print(f" {marker}") + print() + + print("---") + if has_garmin and has_hr: + print("Watch detected and broadcasting HR. bridge_agent.py should work.") + elif has_garmin and not has_hr: + print("Found your Garmin watch but it isn't broadcasting HR.") + print("Fix: on the watch, enable Broadcast HR or start an indoor activity.") + elif has_hr and not has_garmin: + print("Found an HR-broadcasting device but its name doesn't contain") + print("'Garmin' or 'Forerunner'. Note the device name above and update") + print("the GARMIN_NAME variable in your .env file to match.") + else: + print("No Garmin watch and no HR broadcaster found.") + print("Most likely: HR broadcast isn't actually on. Try:") + print(" 1. On the watch, hold the upper-left button") + print(" 2. Settings -> Sensors & Accessories -> Wrist Heart Rate") + print(" 3. Select 'Broadcast Heart Rate' -> Broadcast") + print(" 4. Keep that screen open while you rerun this script") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/contributors/cardiopulse-agent/history.py b/contributors/cardiopulse-agent/history.py new file mode 100644 index 00000000..ab13901f --- /dev/null +++ b/contributors/cardiopulse-agent/history.py @@ -0,0 +1,143 @@ +""" +Persistent test history. + +Saves every completed cardio fitness test to a JSON file so subsequent runs +can compare against prior results — "your Cardio Age dropped from 30 to 27 in +the last two weeks" beats a single isolated number every time. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, asdict +from datetime import datetime, timezone +from pathlib import Path +from threading import Lock + +from scoring import TestResult + +_DATA_DIR = Path(__file__).parent / "data" +_HISTORY_FILE = _DATA_DIR / "test_history.json" +_LOCK = Lock() + + +@dataclass +class TestRecord: + """One stored test result, with a timestamp. + + `sender` is the chat counterpart's agent address — it scopes history to + one user so trend comparisons never mix different people's tests. + Records written before this field existed have sender == "". + """ + + timestamp: str # ISO 8601 UTC + age: int + cardio_fitness_age: int + resting_hr: int + orthostatic_delta: int + breathing_variance: float + rhr_grade: str + sender: str = "" + + @classmethod + def from_result(cls, result: TestResult, sender: str = "") -> "TestRecord": + return cls( + timestamp=datetime.now(timezone.utc).isoformat(), + age=result.age, + cardio_fitness_age=result.cardio_fitness_age, + resting_hr=result.resting_hr, + orthostatic_delta=result.orthostatic_delta, + breathing_variance=result.breathing_variance, + rhr_grade=result.rhr_grade, + sender=sender, + ) + + +def _load() -> list[dict]: + if not _HISTORY_FILE.exists(): + return [] + try: + return json.loads(_HISTORY_FILE.read_text()) + except Exception: + return [] + + +def _save(records: list[dict]) -> None: + _DATA_DIR.mkdir(exist_ok=True) + _HISTORY_FILE.write_text(json.dumps(records, indent=2)) + + +def append(result: TestResult, sender: str = "") -> None: + """Persist a new test result to history, scoped to `sender`.""" + with _LOCK: + records = _load() + records.append(asdict(TestRecord.from_result(result, sender))) + _save(records) + + +def recent(limit: int = 5, sender: str | None = None) -> list[TestRecord]: + """Return the most recent `limit` records, oldest first. + + If `sender` is given, only that user's records are returned. + """ + with _LOCK: + raw = _load() + if sender is not None: + raw = [e for e in raw if e.get("sender", "") == sender] + out: list[TestRecord] = [] + for entry in raw[-limit:]: + try: + out.append(TestRecord(**entry)) + except Exception: + continue + return out + + +def previous(sender: str | None = None) -> TestRecord | None: + """Return the most recent prior test for this sender, or None. + + Call this BEFORE history.append(current_result) so it returns the actual + previous test rather than the one just recorded. + """ + with _LOCK: + raw = _load() + if sender is not None: + raw = [e for e in raw if e.get("sender", "") == sender] + if not raw: + return None + try: + return TestRecord(**raw[-1]) + except Exception: + return None + + +def count() -> int: + with _LOCK: + return len(_load()) + + +def humanize_delta(days: float) -> str: + """Friendly time-since label (e.g. '3 days ago', 'just now').""" + if days < 0.04: # ~1 hour + return "less than an hour ago" + if days < 1: + hours = round(days * 24) + return f"{hours} hour{'s' if hours != 1 else ''} ago" + if days < 7: + d = round(days) + return f"{d} day{'s' if d != 1 else ''} ago" + if days < 30: + w = round(days / 7) + return f"{w} week{'s' if w != 1 else ''} ago" + m = round(days / 30) + return f"{m} month{'s' if m != 1 else ''} ago" + + +def time_since(record: TestRecord) -> str: + """Human-friendly 'X ago' string for a record.""" + try: + ts = datetime.fromisoformat(record.timestamp) + except Exception: + return "an unknown time ago" + delta_seconds = (datetime.now(timezone.utc) - ts).total_seconds() + return humanize_delta(delta_seconds / 86400) diff --git a/contributors/cardiopulse-agent/image_host.py b/contributors/cardiopulse-agent/image_host.py new file mode 100644 index 00000000..2d99ea6d --- /dev/null +++ b/contributors/cardiopulse-agent/image_host.py @@ -0,0 +1,93 @@ +""" +Public image hosting for inline chart delivery. + +ASI:One renders markdown images (`![alt](https://...)`) from public URLs +reliably, but it does NOT render Agentverse `ResourceContent` inline. So we +upload the generated chart PNG to a public host and embed the returned URL. + +Primary host: catbox.moe — anonymous, no API key, permanent hosting, widely +used by bots. Just POST the file, get a URL back as plain text. + +Optional fallback: Imgur, only if IMGUR_CLIENT_ID happens to be set. + +`upload_image(png_bytes)` returns a public URL string, or None on failure. +""" + +from __future__ import annotations + +import base64 +import logging +import os + +import httpx + +logger = logging.getLogger(__name__) + +CATBOX_API = "https://catbox.moe/user/api.php" +IMGUR_API = "https://api.imgur.com/3/image" + + +def _upload_catbox(png_bytes: bytes) -> str | None: + """Upload to catbox.moe anonymously. No key required.""" + try: + files = {"fileToUpload": ("chart.png", png_bytes, "image/png")} + data = {"reqtype": "fileupload"} + with httpx.Client(timeout=30.0) as client: + resp = client.post(CATBOX_API, data=data, files=files) + text = (resp.text or "").strip() + if resp.status_code == 200 and text.startswith("https://"): + return text + logger.warning( + "catbox upload failed: status=%d body=%s", + resp.status_code, + text[:200], + ) + return None + except Exception as e: + logger.warning("catbox upload exception: %s: %s", type(e).__name__, e) + return None + + +def _upload_imgur(png_bytes: bytes, title: str) -> str | None: + """Fallback: upload to Imgur if IMGUR_CLIENT_ID is set.""" + client_id = os.environ.get("IMGUR_CLIENT_ID") + if not client_id: + return None + try: + headers = {"Authorization": f"Client-ID {client_id}"} + payload = { + "image": base64.b64encode(png_bytes).decode(), + "type": "base64", + "title": title[:128], + } + with httpx.Client(timeout=15.0) as client: + resp = client.post(IMGUR_API, headers=headers, data=payload) + if resp.status_code == 200: + url = (resp.json() or {}).get("data", {}).get("link") + if url: + return url + logger.warning("imgur upload failed: %d %s", resp.status_code, resp.text[:200]) + return None + except Exception as e: + logger.warning("imgur upload exception: %s: %s", type(e).__name__, e) + return None + + +def upload_image(png_bytes: bytes, title: str = "CardioPulse chart") -> str | None: + """Upload PNG bytes to a public host. Returns a public URL or None. + + Tries catbox.moe first (no key needed). Falls back to Imgur only if a + Client-ID is configured. + """ + url = _upload_catbox(png_bytes) + if url: + logger.info("Chart hosted at %s", url) + return url + + url = _upload_imgur(png_bytes, title) + if url: + logger.info("Chart hosted at %s (imgur fallback)", url) + return url + + logger.warning("All image hosts failed — chart will be skipped.") + return None diff --git a/contributors/cardiopulse-agent/models.py b/contributors/cardiopulse-agent/models.py new file mode 100644 index 00000000..5d507927 --- /dev/null +++ b/contributors/cardiopulse-agent/models.py @@ -0,0 +1,29 @@ +""" +Shared message models for CardioPulse + Bridge agent communication. + +These types define the contract between the bridge (which talks to the watch +over BLE on the user's local machine) and the main CardioPulse agent (which +runs on Agentverse and handles chat, scoring, and coaching). +""" + +from __future__ import annotations + +from uagents import Model + + +class HRReading(Model): + """One heart rate reading from the watch. + + Sent from bridge_agent -> cardiopulse_agent, typically once per second + while the watch is broadcasting. + """ + + bpm: int + rr_intervals: list[float] = [] + ts: float # unix timestamp from the bridge's clock + + +class HRAck(Model): + """Trivial acknowledgement so the bridge knows the message landed.""" + + ok: bool diff --git a/contributors/cardiopulse-agent/requirements.txt b/contributors/cardiopulse-agent/requirements.txt new file mode 100644 index 00000000..22c53f2c --- /dev/null +++ b/contributors/cardiopulse-agent/requirements.txt @@ -0,0 +1,8 @@ +uagents>=0.22.0 +uagents-core>=0.3.0 +bleak>=0.22.0 +httpx>=0.27.0 +python-dotenv>=1.0.0 +pydantic>=2.0.0 +openai>=1.40.0 +matplotlib>=3.8.0 diff --git a/contributors/cardiopulse-agent/scoring.py b/contributors/cardiopulse-agent/scoring.py new file mode 100644 index 00000000..6347159f --- /dev/null +++ b/contributors/cardiopulse-agent/scoring.py @@ -0,0 +1,198 @@ +""" +Cardio fitness scoring. + +Given the BPM samples from each phase of the test, compute: +- Resting HR (median of last 60s of baseline) +- Orthostatic delta (peak BPM in stand-up phase minus resting HR) +- Breathing-driven HR variance (stdev of BPM during paced breathing) +- A composite "Cardio Fitness Age" estimate adjusted from chronological age + +The age estimate is rough (±5 years). It's meant for trend tracking — the +absolute number shouldn't be taken as a clinical reading. +""" + +from __future__ import annotations + +import statistics +from dataclasses import dataclass + +# Age-norm bands for resting HR. Numbers drawn from ACSM / AHA general +# guidelines. These are coarse buckets, not clinical thresholds. +RHR_NORMS: dict[str, tuple[int, int, int, int]] = { + # bucket -> (excellent_at_or_below, good, average, anything-above = below avg) + "20-29": (60, 65, 70, 75), + "30-39": (62, 67, 72, 77), + "40-49": (64, 69, 74, 79), + "50-59": (66, 71, 76, 81), + "60+": (68, 73, 78, 83), +} + + +@dataclass +class TestResult: + age: int + resting_hr: int + orthostatic_delta: int + breathing_variance: float + cardio_fitness_age: int + rhr_grade: str + verdict: str + quality_note: str | None = None + + +def _bucket(age: int) -> str: + if age < 30: + return "20-29" + if age < 40: + return "30-39" + if age < 50: + return "40-49" + if age < 60: + return "50-59" + return "60+" + + +def _rhr_grade(rhr: int, age: int) -> str: + excellent, good, average, _ = RHR_NORMS[_bucket(age)] + if rhr <= excellent: + return "excellent" + if rhr <= good: + return "good" + if rhr <= average: + return "typical" + # "elevated", not "below average" — the old label read as if the NUMBER + # was low, when a high resting HR is the unfavourable direction. + return "elevated" + + +def compute( + age: int, + baseline_bpm: list[int], + orthostatic_bpm: list[int], + breathing_bpm: list[int], +) -> TestResult: + if len(baseline_bpm) < 5: + raise ValueError( + "Not enough baseline data. Make sure the BLE bridge is streaming." + ) + + # Resting HR: median of the last ~60s of baseline (steadier than mean). + tail = baseline_bpm[-60:] if len(baseline_bpm) >= 60 else baseline_bpm + rhr = int(statistics.median(tail)) + + # Orthostatic delta: peak BPM after standing minus resting HR. + if orthostatic_bpm: + ortho_peak = max(orthostatic_bpm) + ortho_delta = max(0, ortho_peak - rhr) + else: + ortho_delta = 0 + + # Breathing variance: stdev of BPM during paced breathing — proxy for + # respiratory sinus arrhythmia (RSA). + breath_var = statistics.stdev(breathing_bpm) if len(breathing_bpm) > 1 else 0.0 + + # Composite "fitness age" adjustment relative to chronological age. + # Each adjustment is small — the metric tracks trends more than absolutes. + adjust = 0 + + if rhr < 55: + adjust -= 5 + elif rhr < 60: + adjust -= 3 + elif rhr < 65: + adjust -= 1 + elif rhr > 80: + adjust += 5 + elif rhr > 75: + adjust += 3 + + if ortho_delta >= 25: + adjust -= 2 + elif ortho_delta < 8: + adjust += 3 + + if breath_var > 5: + adjust -= 2 + elif breath_var < 1.5: + adjust += 2 + + cardio_age = max(18, age + adjust) + + rhr_grade = _rhr_grade(rhr, age) + + if cardio_age < age - 2: + verdict = ( + f"Your cardiovascular function is reading " + f"{age - cardio_age} years younger than your chronological age." + ) + elif cardio_age > age + 2: + verdict = ( + f"Your cardiovascular function is reading " + f"{cardio_age - age} years older than your chronological age. " + "Room to improve." + ) + else: + verdict = ( + "Your cardiovascular function is roughly aligned with your " + "chronological age." + ) + + # Data-quality detection: a visibly bad baseline should LOWER our + # confidence rather than silently produce a scary score. + notes: list[str] = [] + if rhr >= 90: + notes.append( + "Your baseline HR was unusually high for seated rest — stress, " + "caffeine, talking, or moving around just before the test commonly " + "inflate it. Treat this score as a low-confidence reading and " + "re-test when you're calm and have been seated for 10+ minutes." + ) + if len(baseline_bpm) < 90: + notes.append( + "Fewer heart-rate samples arrived than expected during the " + "baseline — the watch broadcast may have dropped mid-test." + ) + quality_note = " ".join(notes) if notes else None + + return TestResult( + age=age, + resting_hr=rhr, + orthostatic_delta=ortho_delta, + breathing_variance=round(breath_var, 1), + cardio_fitness_age=cardio_age, + rhr_grade=rhr_grade, + verdict=verdict, + quality_note=quality_note, + ) + + +def format_result(r: TestResult) -> str: + """User-facing result block. Every number ships with its typical range so + the reader can judge it without prior knowledge.""" + _, _, average, _ = RHR_NORMS[_bucket(r.age)] + + lines = [ + "**Cardio Fitness Test — Results**", + "", + f"**Cardio Fitness Age: {r.cardio_fitness_age}** (you are {r.age})", + "", + r.verdict, + "", + "**Key readings**", + f"- Resting HR: **{r.resting_hr} bpm** — {r.rhr_grade}. " + f"A well-rested value for your age is typically under {average} bpm.", + f"- Standing response: **+{r.orthostatic_delta} bpm** — " + "a typical jump on standing is +10 to +30 bpm.", + f"- Breathing-driven HR variation: **{r.breathing_variance} bpm** — " + "during slow paced breathing, roughly 3-8 bpm is common.", + ] + + if r.quality_note: + lines += ["", f"⚠️ **Data quality:** {r.quality_note}"] + + lines += [ + "", + "_Estimates carry roughly ±5 years of uncertainty. The trend across " + "repeated tests matters far more than any single number._", + ] + return "\n".join(lines) diff --git a/contributors/cardiopulse-agent/session_state.py b/contributors/cardiopulse-agent/session_state.py new file mode 100644 index 00000000..aa47c77c --- /dev/null +++ b/contributors/cardiopulse-agent/session_state.py @@ -0,0 +1,61 @@ +""" +Shared in-memory state for incoming BPM readings. + +Both the REST endpoint (which receives readings from bridge_agent.py) and the +chat handler (which reads them during a test) live in the same Python process, +so a simple module-level buffer with a lock is enough. +""" + +from __future__ import annotations + +import time +from collections import deque +from threading import Lock + +# Last ~15 minutes of readings — plenty for a 3-minute test plus margin. +_BPM_BUFFER: deque[tuple[float, int]] = deque(maxlen=900) +_RR_BUFFER: deque[tuple[float, float]] = deque(maxlen=3000) +_LAST_TS: float = 0.0 +_LOCK = Lock() + + +def receive_bpm(bpm: int, rr_intervals: list[float], ts: float) -> None: + """Record one reading from the BLE bridge.""" + global _LAST_TS + with _LOCK: + _BPM_BUFFER.append((ts, bpm)) + for rr in rr_intervals: + _RR_BUFFER.append((ts, rr)) + _LAST_TS = ts + + +def bpm_in_window(start_ts: float, end_ts: float) -> list[int]: + """Return just the BPM values recorded between two timestamps.""" + with _LOCK: + return [bpm for ts, bpm in _BPM_BUFFER if start_ts <= ts <= end_ts] + + +def bpm_series_in_window(start_ts: float, end_ts: float) -> list[tuple[float, int]]: + """Return (timestamp, BPM) tuples between two timestamps — for charting.""" + with _LOCK: + return [(ts, bpm) for ts, bpm in _BPM_BUFFER if start_ts <= ts <= end_ts] + + +def rr_in_window(start_ts: float, end_ts: float) -> list[float]: + """Return RR intervals recorded between two timestamps.""" + with _LOCK: + return [rr for ts, rr in _RR_BUFFER if start_ts <= ts <= end_ts] + + +def is_streaming(max_age_sec: float = 5.0) -> bool: + """True if a reading arrived in the last `max_age_sec` seconds.""" + with _LOCK: + return (time.time() - _LAST_TS) < max_age_sec + + +def latest_bpm() -> int | None: + """Most recent BPM reading, or None if none have arrived yet.""" + with _LOCK: + if not _BPM_BUFFER: + return None + return _BPM_BUFFER[-1][1] diff --git a/contributors/cardiopulse-agent/test_scoring.py b/contributors/cardiopulse-agent/test_scoring.py new file mode 100644 index 00000000..f2a59977 --- /dev/null +++ b/contributors/cardiopulse-agent/test_scoring.py @@ -0,0 +1,69 @@ +""" +Offline smoke test for scoring.py. + +Generates synthetic BPM data for each phase, runs the scoring engine, and +prints the formatted result. Useful for sanity-checking changes to the +scoring formulas without needing a real watch. + + python test_scoring.py +""" + +from __future__ import annotations + +import random + +import scoring + +random.seed(42) + + +def synth_baseline(rhr: int, n: int = 120, jitter: int = 2) -> list[int]: + """Resting BPM hovering around `rhr` with small jitter.""" + return [rhr + random.randint(-jitter, jitter) for _ in range(n)] + + +def synth_orthostatic(rhr: int, peak_delta: int, n: int = 30) -> list[int]: + """BPM rises sharply on standing, then partially recovers.""" + out = [] + for i in range(n): + # Rise over first 5 samples, plateau, then slight recovery. + if i < 5: + out.append(rhr + (peak_delta * i // 5)) + elif i < 20: + out.append(rhr + peak_delta + random.randint(-2, 2)) + else: + out.append(rhr + peak_delta - (i - 20)) + return out + + +def synth_breathing(rhr: int, swing: int, n: int = 30) -> list[int]: + """BPM oscillates with the breath cycle (proxy for RSA).""" + import math + + return [int(rhr + swing * math.sin(2 * math.pi * i / 10)) for i in range(n)] + + +def run_case(label: str, age: int, rhr: int, ortho_peak: int, breath_swing: int): + print(f"\n=== {label} (age={age}, rhr={rhr}) ===") + baseline = synth_baseline(rhr) + ortho = synth_orthostatic(rhr, ortho_peak) + breath = synth_breathing(rhr, breath_swing) + + result = scoring.compute( + age=age, + baseline_bpm=baseline, + orthostatic_bpm=ortho, + breathing_bpm=breath, + ) + print(scoring.format_result(result)) + + +def main() -> None: + run_case("Fit 30-year-old", age=30, rhr=52, ortho_peak=22, breath_swing=6) + run_case("Average 30-year-old", age=30, rhr=70, ortho_peak=15, breath_swing=3) + run_case("Deconditioned 30-year-old", age=30, rhr=82, ortho_peak=6, breath_swing=1) + run_case("Fit 50-year-old", age=50, rhr=58, ortho_peak=20, breath_swing=5) + + +if __name__ == "__main__": + main()