Deploy ecg-only 332caae0: ECG-derived HR, source labels

.pytest_cache/v/cache/nodeids

@@ -1,6 +1,12 @@
 [
   "tests/test_database.py::test_connect_db_does_not_raise_when_cluster_dns_missing",
   "tests/test_database.py::test_health_reports_db_disconnected_when_cluster_dns_missing",
+  "tests/test_ecg_heart_rate.py::test_ignores_missing_or_implausible_values",
+  "tests/test_ecg_heart_rate.py::test_uses_ecg_measured_heart_rate_from_model_features",
+  "tests/test_firmware_ecg.py::test_constant_input_settles_to_the_2048_centre",
+  "tests/test_firmware_ecg.py::test_matches_values_in_device_garbage_fixture_range",
+  "tests/test_firmware_ecg.py::test_scaling_keeps_clipped_device_peaks_off_the_rails",
+  "tests/test_firmware_ecg.py::test_splits_into_full_ten_second_windows",
   "tests/test_signal_quality.py::test_atrial_fibrillation_is_not_rejected_for_irregularity[100]",
   "tests/test_signal_quality.py::test_atrial_fibrillation_is_not_rejected_for_irregularity[250]",
   "tests/test_signal_quality.py::test_heart_rate_agrees_between_sample_rates",
@@ -46,5 +52,8 @@
   "tests/test_signal_quality.py::test_too_short_window_is_poor",
   "tests/test_signal_quality.py::test_upload_marks_lead_off",
   "tests/test_signal_quality.py::test_upload_runs_prediction_for_good_signal",
-  "tests/test_signal_quality.py::test_upload_skips_prediction_for_poor_signal"
+  "tests/test_signal_quality.py::test_upload_skips_prediction_for_poor_signal",
+  "tests/test_vitals_source.py::test_response_includes_source_label",
+  "tests/test_vitals_source.py::test_response_source_is_none_for_live_device_docs",
+  "tests/test_vitals_source.py::test_upload_accepts_optional_source_label"
 ]

app/models/vitals.py

@@ -13,6 +13,7 @@ class VitalsCreate(BaseModel):
     ecg_lead_off: bool = Field(default=False)
     ecg_samples: List[int] = Field(..., min_length=100, max_length=6000)
     beat_timestamps_ms: List[int] = Field(default_factory=list)
+    source: Optional[str] = Field(None, max_length=120, description="Origin label for playback data")
 
 
 class VitalsResponse(BaseModel):
@@ -20,6 +21,7 @@ class VitalsResponse(BaseModel):
     device_id: str
     timestamp: datetime
     heart_rate_bpm: float
+    heart_rate_source: Optional[str] = None  # "sensor" (MAX30100) or "ecg" (derived from R-peaks)
     spo2_percent: int
     ecg_lead_off: bool
     sample_count: int
@@ -27,6 +29,7 @@ class VitalsResponse(BaseModel):
     sample_rate_hz: Optional[int] = None
     prediction: Optional[dict] = None
     signal_quality: Optional[str] = None
+    source: Optional[str] = None
     created_at: datetime
 
 

app/routes/vitals.py

@@ -11,6 +11,12 @@ from app.services.signal_quality import assess_ecg_quality
 router = APIRouter()
 
 
+def _ecg_heart_rate(features: dict):
+    """Heart rate the feature extractor measured from the ECG R-peaks, if plausible."""
+    hr = (features or {}).get("mean_hr_ecg")
+    return round(float(hr), 1) if hr and 30 <= hr <= 220 else None
+
+
 def _vitals_doc_to_response(doc: dict, prediction: dict = None) -> VitalsResponse:
     ecg = doc.get("ecg_samples")
     return VitalsResponse(
@@ -18,6 +24,7 @@ def _vitals_doc_to_response(doc: dict, prediction: dict = None) -> VitalsRespons
         device_id=doc["device_id"],
         timestamp=doc["timestamp"],
         heart_rate_bpm=doc["heart_rate_bpm"],
+        heart_rate_source=doc.get("heart_rate_source"),
         spo2_percent=doc["spo2_percent"],
         ecg_lead_off=doc["ecg_lead_off"],
         sample_count=len(ecg) if ecg else 0,
@@ -25,6 +32,7 @@ def _vitals_doc_to_response(doc: dict, prediction: dict = None) -> VitalsRespons
         sample_rate_hz=doc.get("sample_rate_hz"),
         prediction=prediction,
         signal_quality=doc.get("signal_quality"),
+        source=doc.get("source"),
         created_at=doc["created_at"],
     )
 
@@ -65,12 +73,14 @@ async def upload_vitals(data: VitalsCreate, _=Depends(verify_api_key)):
         "window_ms": data.window_ms,
         "sample_rate_hz": data.sample_rate_hz,
         "heart_rate_bpm": data.heart_rate_bpm,
+        "heart_rate_source": "sensor" if data.heart_rate_bpm > 0 else None,
         "spo2_percent": data.spo2_percent,
         "ecg_lead_off": data.ecg_lead_off,
         "ecg_samples": data.ecg_samples,
         "beat_timestamps_ms": data.beat_timestamps_ms,
         "signal_quality": signal_quality,
         "signal_quality_reason": signal_quality_reason,
+        "source": data.source,
         "created_at": datetime.utcnow(),
     }
     result = await db.vitals.insert_one(vitals_doc)
@@ -153,6 +163,16 @@ async def upload_vitals(data: VitalsCreate, _=Depends(verify_api_key)):
                 history_features=history_features,
             )
 
+            # No pulse sensor reading (e.g. MAX30100 disabled): use the HR measured from the ECG
+            if data.heart_rate_bpm <= 0:
+                ecg_hr = _ecg_heart_rate(ml_result.get("features"))
+                if ecg_hr:
+                    vitals_doc.update(heart_rate_bpm=ecg_hr, heart_rate_source="ecg")
+                    await db.vitals.update_one(
+                        {"_id": result.inserted_id},
+                        {"$set": {"heart_rate_bpm": ecg_hr, "heart_rate_source": "ecg"}},
+                    )
+
             if ml_result["risk_label"] != "unknown":
                 pred_doc = {
                     "vitals_id": str(result.inserted_id),

scripts/firmware_ecg.py

@@ -0,0 +1,62 @@
+"""Python port of the ESP32 ECG processing (firmware/src/ecg_filter.h + sensor_manager.cpp).
+
+Turns raw 250 Hz AD8232 ADC samples into the exact 10 s windows the firmware uploads:
+50 Hz notch -> 40 Hz low-pass -> DC removal, scaled by ECG_OUTPUT_SCALE, re-centred
+at 2048 and clamped to 12 bits, with filter state reset at every window boundary.
+Keep in sync with firmware/include/config.h.
+"""
+
+SAMPLE_RATE_HZ = 250
+WINDOW_SAMPLES = 2500
+ECG_OUTPUT_SCALE = 0.5
+
+
+class _Notch50:
+    b0, b1, b2, a1, a2 = 0.981334, -0.606498, 0.981334, -0.606498, 0.962668
+
+    def __init__(self):
+        self.x1 = self.x2 = self.y1 = self.y2 = 0.0
+
+    def step(self, x):
+        y = self.b0 * x + self.b1 * self.x1 + self.b2 * self.x2 - self.a1 * self.y1 - self.a2 * self.y2
+        self.x2, self.x1, self.y2, self.y1 = self.x1, x, self.y1, y
+        return y
+
+
+class _LowPass40:
+    b0, b1, b2, a1, a2 = 0.145310, 0.290620, 0.145310, -0.670919, 0.252160
+
+    def __init__(self):
+        self.z1 = self.z2 = 0.0
+
+    def step(self, x):
+        y = self.b0 * x + self.z1
+        self.z1 = self.b1 * x - self.a1 * y + self.z2
+        self.z2 = self.b2 * x - self.a2 * y
+        return y
+
+
+class _DcRemover:
+    def __init__(self, alpha=0.9875):
+        self.alpha, self.w = alpha, 0.0
+
+    def step(self, x):
+        old = self.w
+        self.w = x + self.alpha * self.w
+        return self.w - old
+
+
+def firmware_windows(raw, scale=ECG_OUTPUT_SCALE, reset_each_window=True):
+    """Split raw ADC samples into firmware-processed 10 s windows (lists of ints)."""
+    chain = [_Notch50(), _LowPass40(), _DcRemover()]
+    windows = []
+    for start in range(0, len(raw) - WINDOW_SAMPLES + 1, WINDOW_SAMPLES):
+        if reset_each_window:
+            chain = [_Notch50(), _LowPass40(), _DcRemover()]
+        window = []
+        for v in raw[start:start + WINDOW_SAMPLES]:
+            for f in chain:
+                v = f.step(v)
+            window.append(int(min(4095, max(0, round(v * scale + 2048)))))
+        windows.append(window)
+    return windows

scripts/playback_ptbxl.py

@@ -36,6 +36,7 @@ from scipy.signal import resample_poly
 DEVICE_ID = "PLAYBACK_PTBXL"
 DEVICE_NAME = "PTB-XL playback (recorded clinical ECGs)"
 MI_CODES = {"IMI", "AMI", "ASMI", "ALMI", "ILMI", "LMI", "PMI", "IPLMI"}
+LABEL_TEXT = {"NORM": "normal ECG", "MI": "myocardial infarction"}
 COUNTS_PER_MV = 400.0
 CENTER = 2048
 SAMPLE_RATE_HZ = 250
@@ -131,6 +132,7 @@ def main():
                 "spo2_percent": 0,
                 "ecg_lead_off": False,
                 "ecg_samples": samples,
+                "source": f"PTB-XL #{row.ecg_id} · label: {LABEL_TEXT.get(row.label, row.label)}",
             })
             r.raise_for_status()
             body = r.json()

scripts/replay_recording.py

@@ -0,0 +1,104 @@
+"""Replay a raw AD8232 recording from the ESP32 through the vitals API, window by window.
+
+Each 10 s window is processed exactly as the firmware does (firmware_ecg.py) and sent
+from a playback device (default PLAYBACK_DEVICE_REC) with a source label, so stored
+data is never mistaken for a live session. HR/SpO2 are sent as 0 (no pulse sensor);
+the server derives HR from the ECG.
+
+Input: a serial capture with "D:v,v,..." lines (ECG diagnostic sketch, 250 Hz raw) or
+a JSON file {"sample_rate_hz": 250, "samples": [...]}.
+
+Usage (from backend/):
+  API_KEY=... uv run --quiet --no-project --python 3.11 --with httpx \
+      python scripts/replay_recording.py recording.raw --api http://localhost:7860 \
+      --label "Recorded on this device · 2026-10-11 · healthy adult at rest" \
+      [--email demo@example.com --password ...] [--interval 10]
+"""
+
+import argparse
+import json
+import os
+import time
+from collections import Counter
+
+import httpx
+
+from firmware_ecg import SAMPLE_RATE_HZ, firmware_windows
+
+DEVICE_ID = "PLAYBACK_DEVICE_REC"
+DEVICE_NAME = "Recorded session from the ESP32 (replayed)"
+
+
+def load_raw(path: str) -> list:
+    text = open(path, "rb").read().decode("ascii", "ignore")
+    if text.lstrip().startswith("{"):
+        data = json.loads(text)
+        assert data.get("sample_rate_hz", SAMPLE_RATE_HZ) == SAMPLE_RATE_HZ
+        return data["samples"]
+    return [int(v) for line in text.replace("\r", "").splitlines() if line.startswith("D:")
+            for v in line[2:].split(",") if v.isdigit()]
+
+
+def register_device(client: httpx.Client, device_id: str, email: str, password: str):
+    r = client.post("/api/v1/auth/login", json={"email": email, "password": password})
+    r.raise_for_status()
+    headers = {"Authorization": f"Bearer {r.json()['access_token']}"}
+    r = client.post("/api/v1/devices/register", headers=headers,
+                    json={"device_id": device_id, "name": DEVICE_NAME})
+    r.raise_for_status()
+    print(f"Registered {device_id} to {email}")
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    p.add_argument("recording", help="serial capture (.raw) or JSON recording")
+    p.add_argument("--api", required=True, help="API base URL")
+    p.add_argument("--api-key", default=os.environ.get("API_KEY"), help="device API key (default: $API_KEY)")
+    p.add_argument("--label", required=True, help="source label shown with every window")
+    p.add_argument("--device-id", default=DEVICE_ID)
+    p.add_argument("--email", help="account to register the playback device to")
+    p.add_argument("--password")
+    p.add_argument("--interval", type=float, default=0, help="seconds between windows")
+    args = p.parse_args()
+    if not args.api_key:
+        p.error("--api-key or $API_KEY is required")
+    if len(args.label) > 120:
+        p.error("--label must be at most 120 characters")
+
+    windows = firmware_windows(load_raw(args.recording))
+    print(f"{len(windows)} windows of 10 s from {args.recording}")
+
+    with httpx.Client(base_url=args.api, timeout=120) as client:
+        if args.email:
+            register_device(client, args.device_id, args.email, args.password)
+
+        labels = Counter()
+        print(f"{'t':>5}  {'quality':8}  {'HR':>5}  {'risk':>6}  label")
+        for n, window in enumerate(windows):
+            if n and args.interval:
+                time.sleep(args.interval)
+            r = client.post("/api/v1/vitals", headers={"X-API-Key": args.api_key}, json={
+                "device_id": args.device_id,
+                "timestamp": int(time.time()),
+                "window_ms": 10000,
+                "sample_rate_hz": SAMPLE_RATE_HZ,
+                "heart_rate_bpm": 0,
+                "spo2_percent": 0,
+                "ecg_lead_off": False,
+                "ecg_samples": window,
+                "source": args.label,
+            })
+            r.raise_for_status()
+            body = r.json()
+            pred = body.get("prediction") or {}
+            label = pred.get("risk_label", "-")
+            labels[label] += 1
+            score = f"{pred['risk_score']:.3f}" if pred else "-"
+            quality = body.get("signal_quality") or "-"
+            print(f"{n * 10:>4}s  {quality:8}  {body['heart_rate_bpm']:>5}  {score:>6}  {label}")
+
+    print(f"\nSummary: {dict(labels)}")
+
+
+if __name__ == "__main__":
+    main()

tests/test_ecg_heart_rate.py

@@ -0,0 +1,13 @@
+from app.routes.vitals import _ecg_heart_rate
+
+
+def test_uses_ecg_measured_heart_rate_from_model_features():
+    assert _ecg_heart_rate({"mean_hr_ecg": 71.3456, "rmssd": 30.1}) == 71.3
+
+
+def test_ignores_missing_or_implausible_values():
+    assert _ecg_heart_rate({}) is None
+    assert _ecg_heart_rate(None) is None
+    assert _ecg_heart_rate({"mean_hr_ecg": 0}) is None
+    assert _ecg_heart_rate({"mean_hr_ecg": 12.0}) is None
+    assert _ecg_heart_rate({"mean_hr_ecg": 260.0}) is None

tests/test_firmware_ecg.py

@@ -0,0 +1,40 @@
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
+
+from firmware_ecg import WINDOW_SAMPLES, firmware_windows  # noqa: E402
+
+
+def test_splits_into_full_ten_second_windows():
+    windows = firmware_windows([1800] * (WINDOW_SAMPLES * 3 + 100))
+    assert len(windows) == 3
+    assert all(len(w) == WINDOW_SAMPLES for w in windows)
+
+
+def test_constant_input_settles_to_the_2048_centre():
+    window = firmware_windows([1800] * WINDOW_SAMPLES)[0]
+    assert abs(window[-1] - 2048) <= 1
+    assert all(0 <= v <= 4095 for v in window)
+
+
+def test_scaling_keeps_clipped_device_peaks_off_the_rails():
+    # Raw device-like window: baseline 1774 with R-peaks clipped at the ADC limit
+    raw = [1774] * WINDOW_SAMPLES
+    for beat in range(100, WINDOW_SAMPLES, 210):
+        raw[beat - 2:beat + 3] = [2600, 3600, 4095, 3600, 2600]
+    scaled = firmware_windows(raw)[0]
+    unscaled = firmware_windows(raw, scale=1.0)[0]
+    assert max(scaled) < 4093
+    assert max(unscaled) >= 4093
+
+
+def test_matches_values_in_device_garbage_fixture_range():
+    # The firmware output is 12-bit, so it must be a valid VitalsCreate payload
+    path = os.path.join(os.path.dirname(__file__), "fixtures", "device_garbage.json")
+    raw = json.load(open(path))
+    windows = raw["windows"] if isinstance(raw, dict) else raw
+    samples = [v for w in windows for v in w["ecg_samples"]]
+    out = firmware_windows(samples)
+    assert out and all(0 <= v <= 4095 for w in out for v in w)

tests/test_vitals_source.py

@@ -0,0 +1,48 @@
+from datetime import datetime
+
+from bson import ObjectId
+
+from app.models.vitals import VitalsCreate
+from app.routes.vitals import _vitals_doc_to_response
+
+PLAYBACK_SOURCE = "PTB-XL #8473 · cardiologist label: MI"
+
+
+def _payload(**overrides):
+    payload = {
+        "device_id": "PLAYBACK_PTBXL",
+        "timestamp": 1760000000,
+        "heart_rate_bpm": 71.3,
+        "spo2_percent": 0,
+        "ecg_samples": [2048] * 2500,
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _doc(**overrides):
+    doc = {
+        "_id": ObjectId(),
+        "device_id": "PLAYBACK_PTBXL",
+        "timestamp": datetime(2026, 10, 9),
+        "heart_rate_bpm": 71.3,
+        "spo2_percent": 0,
+        "ecg_lead_off": False,
+        "ecg_samples": [2048] * 2500,
+        "created_at": datetime(2026, 10, 9),
+    }
+    doc.update(overrides)
+    return doc
+
+
+def test_upload_accepts_optional_source_label():
+    assert VitalsCreate(**_payload(source=PLAYBACK_SOURCE)).source == PLAYBACK_SOURCE
+    assert VitalsCreate(**_payload()).source is None
+
+
+def test_response_includes_source_label():
+    assert _vitals_doc_to_response(_doc(source=PLAYBACK_SOURCE)).source == PLAYBACK_SOURCE
+
+
+def test_response_source_is_none_for_live_device_docs():
+    assert _vitals_doc_to_response(_doc(device_id="ESP32_21E5F0")).source is None