From 602ffa3d2c5c95efec8a827d060c1b55476e4562 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20Sch=C3=A4dlich?= Date: Thu, 9 Jul 2026 07:30:16 +0200 Subject: [PATCH] Add HCI-capture extraction + CSV ground-truth correlation for #103 Two stdlib tools in Tools/linux-capture that turn a WHOOP data export into leverage for the un-decoded deep-record layouts tracked in #103: - hci_extract.py: parse a phone Bluetooth HCI capture (iOS .pklg / Android btsnoop) of the official app, reassemble L2CAP/ATT, and emit the CRC-valid WHOOP frames as the project's capture.json. Only WHOOP streams reach the output. - correlate_ground_truth.py: known-plaintext field search. Cross-reference capture records against the official per-night values in a WHOOP CSV export (reusing the Swift importer's localized header aliases) to locate each biometric's byte offset + encoding. Distribution-overlap scoring (recall + precision) rejects constants and coincidences. Prints offsets only, so the output is safe to post on #103 while health data stays local. Docs: new README sections + a ground-truth-correlation note in WHOOP5_DEEP_DATA.md. 25 new stdlib tests; full suite 129 green. Co-Authored-By: Claude Opus 4.8 --- Tools/linux-capture/correlate_ground_truth.py | 333 ++++++++++++++++++ Tools/linux-capture/hci_extract.py | 312 ++++++++++++++++ .../test_correlate_ground_truth.py | 145 ++++++++ Tools/linux-capture/test_hci_extract.py | 186 ++++++++++ docs/WHOOP5_DEEP_DATA.md | 23 ++ tools/linux-capture/README.md | 65 ++++ 6 files changed, 1064 insertions(+) create mode 100644 Tools/linux-capture/correlate_ground_truth.py create mode 100644 Tools/linux-capture/hci_extract.py create mode 100644 Tools/linux-capture/test_correlate_ground_truth.py create mode 100644 Tools/linux-capture/test_hci_extract.py diff --git a/Tools/linux-capture/correlate_ground_truth.py b/Tools/linux-capture/correlate_ground_truth.py new file mode 100644 index 000000000..14d9863a1 --- /dev/null +++ b/Tools/linux-capture/correlate_ground_truth.py @@ -0,0 +1,333 @@ +"""Locate WHOOP record fields by correlating capture frames against a CSV-export ground truth. + +An HCI capture (issue #103) gives raw, un-decoded records — type-`0x2F` biometrics and the 5/MG +history types NOOP still skips (see docs/WHOOP5_DEEP_DATA.md). This tool solves the "what byte is +what" problem as *known-plaintext*: the owner's own WHOOP CSV export lists the official per-night +values (HRV, resting HR, skin temp, SpO₂, respiratory rate, sleep-stage minutes), so we search the +records for the byte offset + encoding that reproduces each known value across every night. + + capture.json + whoop_export.zip ──► correlate_ground_truth.py ──► candidate field offsets + +Only aggregate offsets are reported — never the health values themselves — so the output is safe to +post on #103 while the CSV (personal health data) and the capture (device identifiers) stay local. + +The CSV side reuses the canonical column keys and the German/Spanish/etc. header aliases from the +Swift importer (Packages/StrandImport/.../CSVParsing.swift), kept in sync here as `HEADER_ALIASES`. + +Stdlib-only. Usage: + + python3 correlate_ground_truth.py capture.json my_whoop_data.zip + python3 correlate_ground_truth.py capture.json export_folder/ --family whoop5 --tolerance 0.02 +""" + +import argparse +import bisect +import csv +import io +import json +import struct +import sys +import zipfile + +import whoop_frame as wf + +# --- Ground-truth CSV loading ---------------------------------------------------------------------- + +# Canonical WHOOP export filenames (English). Localized bundles rename the files; we match by these +# after aliasing, and fall back to header-sniffing, exactly like the Swift importer. +CYCLES_FILES = {"physiological_cycles.csv", "physiologische_zyklen.csv", "ciclos_fisiologicos.csv"} +SLEEP_FILES = {"sleeps.csv", "schlaf.csv", "sueno.csv", "sueños.csv"} + +# Localized column header → canonical key. Mirrors HeaderNorm.foreignAliases in CSVParsing.swift +# (German + Spanish subset that covers the physiologic/sleep numeric fields we correlate on). English +# headers pass through unchanged, so an English export needs no alias. +HEADER_ALIASES = { + # German + "erholungswert %": "recovery_score_pct", + "ruheherzfrequenz (schläge pro minute)": "resting_heart_rate_bpm", + "herzfrequenzvariabilität (ms)": "heart_rate_variability_ms", + "hauttemperatur (celsius)": "skin_temp_celsius", + "blutsauerstoff %": "blood_oxygen_pct", + "atemfrequenz (atemzüge/min.)": "respiratory_rate_rpm", + "schlafdauer (min.)": "asleep_duration_min", + "dauer des tiefschlafs (min.)": "deep_sws_duration_min", + "dauer des rem-schlafs (min.)": "rem_duration_min", + "startzeit des zyklus": "cycle_start_time", + # Spanish + "puntuación de recuperación %": "recovery_score_pct", + "frecuencia cardíaca en reposo (lpm)": "resting_heart_rate_bpm", + "variabilidad de la frecuencia cardíaca (ms)": "heart_rate_variability_ms", + "temp. cutánea (grados centígrados)": "skin_temp_celsius", + "oxígeno en sangre %": "blood_oxygen_pct", + "frecuencia respiratoria (rpm)": "respiratory_rate_rpm", + "hora de inicio del ciclo": "cycle_start_time", +} + +# Numeric fields we correlate on, with the scale factor the raw record is *likely* to carry the value +# at (the tool tries each). A field whose real encoding differs simply won't match — no harm. +# name: (canonical CSV column, [candidate scale multipliers]) +GROUND_TRUTH_FIELDS = { + "hrv_ms": ("heart_rate_variability_ms", [1, 1000]), # ms, or seconds×1000 + "resting_hr_bpm": ("resting_heart_rate_bpm", [1]), + "respiratory_rate": ("respiratory_rate_rpm", [1, 10, 100]), # 14.6 → 146 or 1460 + "skin_temp_c": ("skin_temp_celsius", [1, 10, 100]), # 34.93 → 349 or 3493 + "spo2_pct": ("blood_oxygen_pct", [1, 10, 100]), +} + + +def _norm_header(h: str) -> str: + key = h.strip().lower().lstrip("") + return HEADER_ALIASES.get(key, key) + + +def load_csv_rows(text: str): + """Parse one WHOOP CSV into a list of {canonical_key: value} dicts.""" + reader = csv.reader(io.StringIO(text)) + rows = list(reader) + if not rows: + return [] + headers = [_norm_header(h) for h in rows[0]] + out = [] + for raw in rows[1:]: + if not any(cell.strip() for cell in raw): + continue + out.append({headers[i]: raw[i] for i in range(min(len(headers), len(raw)))}) + return out + + +def load_ground_truth(path: str): + """Load the cycles + sleep rows from a WHOOP export .zip or folder. + + Returns a list of per-cycle dicts keyed by cycle_start_time, each carrying whatever numeric + ground-truth fields were present. Cycles and sleeps are merged on cycle_start_time. + """ + files = {} # lowercased basename -> text + if zipfile.is_zipfile(path): + with zipfile.ZipFile(path) as z: + for name in z.namelist(): + base = name.rsplit("/", 1)[-1].lower() + if base.endswith(".csv"): + files[base] = z.read(name).decode("utf-8-sig", errors="replace") + else: + import os + for root, _dirs, names in os.walk(path): + for n in names: + if n.lower().endswith(".csv"): + with open(os.path.join(root, n), encoding="utf-8-sig", errors="replace") as f: + files[n.lower()] = f.read() + + def pick(candidates): + for base, text in files.items(): + if base in candidates: + return text + return None + + merged = {} # cycle_start_time -> row dict + for text in (pick(CYCLES_FILES), pick(SLEEP_FILES)): + if text is None: + continue + for row in load_csv_rows(text): + key = row.get("cycle_start_time", "") + merged.setdefault(key, {}).update(row) + return [r for r in merged.values() if r] + + +def truth_values(rows): + """Extract {field_name: [floats]} for every GROUND_TRUTH_FIELDS column present in the export.""" + out = {} + for field, (col, _scales) in GROUND_TRUTH_FIELDS.items(): + vals = [] + for r in rows: + raw = r.get(col, "").strip() + if not raw: + continue + try: + vals.append(float(raw.replace(",", "."))) + except ValueError: + pass + if vals: + out[field] = vals + return out + + +# --- Candidate value extraction from frames -------------------------------------------------------- + +def frame_inner(frame: bytes, family: str): + """Return the inner record (type byte onward, CRC32 trailer stripped) of a CRC-valid frame, else + None. Correlation runs on the decoded payload region, not the framing/CRC bytes.""" + verify = wf.verify_whoop5_frame if family == "whoop5" else wf.verify_whoop4_frame + if not verify(frame): + return None + inner_off = wf.WHOOP5_INNER_OFF if family == "whoop5" else wf.WHOOP4_INNER_OFF + return frame[inner_off:-4] + + +def candidates_at(record: bytes, offset: int): + """All numeric interpretations of `record` at `offset`: u8/i8, u16/i16 LE+BE, float32 LE+BE. + + Returns {encoding_name: value}. Out-of-range offsets are simply omitted. + """ + out = {} + n = len(record) + if offset < n: + out["u8"] = record[offset] + out["i8"] = struct.unpack_from("b", record, offset)[0] + if offset + 2 <= n: + out["u16le"] = struct.unpack_from("H", record, offset)[0] + out["i16le"] = struct.unpack_from("h", record, offset)[0] + if offset + 4 <= n: + out["f32le"] = struct.unpack_from("f", record, offset)[0] + return out + + +def _close(a: float, b: float, tol: float) -> bool: + if a != a or b != b: # NaN (junk float32 interpretation) + return False + if abs(b) > 1e9: # absurd magnitude — not a real biometric + return False + denom = max(abs(b), 1.0) + return abs(a - b) / denom <= tol + + +def _has_neighbor(sorted_vals, target, tol): + """True if `sorted_vals` contains a value within relative `tol` of `target`.""" + if not sorted_vals: + return False + span = tol * max(abs(target), 1.0) + i = bisect.bisect_left(sorted_vals, target - span) + return i < len(sorted_vals) and sorted_vals[i] <= target + span + + +def correlate(records_by_type, truth, family, tolerance, min_hits): + """For each record type and each ground-truth field, find (offset, encoding, scale) whose decoded + values reproduce the field's distribution. + + Records aren't 1:1 with nights, so we don't pair them up. Instead we compare the *distribution* of + decoded values against the known values two ways: + + * recall — fraction of ground-truth nights that have a decoded value within tolerance. A real + field reproduces every night's value, so recall ≈ 1. A near-constant byte only + covers the few nights near that constant → low recall (this is the guard against a + fixed byte masquerading as a low-variance field like skin temp). + * precision — fraction of decoded values that land on some known value. Random bytes scatter + outside the biometric's range → low precision. + + Both are needed: recall alone is fooled by a byte that ranges over the whole scale; precision alone + by a constant that happens to sit in-range. `score = recall × precision`. Distinct decoded values + must exceed a floor so a degenerate constant can't score on a single-valued field. Best first. + """ + hits = [] + encodings = ("u8", "i8", "u16le", "u16be", "i16le", "i16be", "f32le", "f32be") + for rtype, records in sorted(records_by_type.items()): + if not records: + continue + maxlen = max(len(r) for r in records) + for field, values in truth.items(): + truth_sorted = sorted(values) + _col, scales = GROUND_TRUTH_FIELDS[field] + for offset in range(maxlen): + for enc in encodings: + for scale in scales: + decoded = [] + for rec in records: + cands = candidates_at(rec, offset) + if enc in cands: + v = cands[enc] + if v == v and abs(v) < 1e9: # skip NaN / absurd float32 junk + decoded.append(v / scale if scale != 1 else v) + if len(decoded) < min_hits: + continue + dec_sorted = sorted(decoded) + recall = sum(_has_neighbor(dec_sorted, t, tolerance) + for t in truth_sorted) / len(truth_sorted) + precision = sum(_has_neighbor(truth_sorted, d, tolerance) + for d in decoded) / len(decoded) + distinct = len(set(decoded)) + if recall >= 0.6 and precision >= 0.6 and distinct >= 3: + hits.append({ + "type": rtype, "field": field, "offset": offset, + "encoding": enc, "scale": scale, + "records_matched": round(precision * len(decoded)), + "records_total": len(decoded), + "nights_covered": round(recall * len(truth_sorted)), + "nights_total": len(truth_sorted), + "score": recall * precision, + }) + hits.sort(key=lambda h: h["score"], reverse=True) + return hits + + +def group_records_by_type(records, family): + """Group CRC-valid frames' inner records by their packet-type byte.""" + by_type = {} + for r in records: + frame = bytes.fromhex(r["hex"]) + inner = frame_inner(frame, family) + if inner is None or not inner: + continue + by_type.setdefault(inner[0], []).append(inner) + return by_type + + +def main(): + p = argparse.ArgumentParser( + description="Correlate capture frames against a WHOOP CSV export to locate record fields.") + p.add_argument("capture", help="capture.json from whoop_capture.py or hci_extract.py") + p.add_argument("export", help="WHOOP CSV export .zip or folder (your ground truth)") + p.add_argument("--family", choices=["whoop4", "whoop5"], default="whoop5", + help="strap generation for framing/CRC rules (default: whoop5)") + p.add_argument("--tolerance", type=float, default=0.03, + help="relative match tolerance, e.g. 0.03 = ±3%% (default: 0.03)") + p.add_argument("--min-hits", type=int, default=5, + help="minimum matching records for a candidate offset (default: 5)") + p.add_argument("--type", type=lambda s: int(s, 0), default=None, + help="restrict to one record type byte, e.g. 0x2f") + args = p.parse_args() + + with open(args.capture) as f: + records = json.load(f) + rows = load_ground_truth(args.export) + if not rows: + print("no cycles/sleep rows found in the export — check the path/format", file=sys.stderr) + sys.exit(1) + truth = truth_values(rows) + if not truth: + print("no usable ground-truth numeric fields in the export", file=sys.stderr) + sys.exit(1) + + by_type = group_records_by_type(records, args.family) + if args.type is not None: + by_type = {args.type: by_type.get(args.type, [])} + + print(f"ground truth: {len(rows)} cycles; fields " + + ", ".join(f"{k}(n={len(v)})" for k, v in truth.items())) + print("record types in capture: " + + ", ".join(f"0x{t:02x}×{len(r)}" for t, r in sorted(by_type.items())) + "\n") + + hits = correlate(by_type, truth, args.family, args.tolerance, args.min_hits) + if not hits: + print("no field offsets matched. Try a larger --tolerance, or confirm the capture actually " + "contains deep records (type 0x2f / history types), not just live HR.") + return + + # One best hit per (type, field) — the strongest offset for each biometric. + seen = set() + print(f"{'type':>6} {'field':<16} {'offset':>6} {'enc':>6} {'scale':>6} " + f"{'records':>12} {'nights':>10} score") + for h in hits: + k = (h["type"], h["field"]) + if k in seen: + continue + seen.add(k) + print(f" 0x{h['type']:02x} {h['field']:<16} {h['offset']:>6} {h['encoding']:>6} " + f"{h['scale']:>6} {h['records_matched']:>5}/{h['records_total']:<6} " + f"{h['nights_covered']:>4}/{h['nights_total']:<5} {h['score']:.2f}") + print("\nOffsets are into the inner record (type byte = offset 0). Safe to share on #103; " + "your CSV values and the capture stay local.") + + +if __name__ == "__main__": + main() diff --git a/Tools/linux-capture/hci_extract.py b/Tools/linux-capture/hci_extract.py new file mode 100644 index 000000000..27f5414b7 --- /dev/null +++ b/Tools/linux-capture/hci_extract.py @@ -0,0 +1,312 @@ +"""Extract WHOOP BLE frames from a phone's Bluetooth HCI capture → capture.json. + + btsnoop_hci.log / .pklg → hci_extract.py → capture.json → whoop-decode / correlate_ground_truth.py + +Issue #103 asks 5/MG owners for an HCI capture of the OFFICIAL app doing a full history sync +(iOS PacketLogger → `.pklg`, or Android Developer Options → Bluetooth HCI snoop log → +`btsnoop_hci.log`). This tool turns such a capture into the same `capture.json` format that +`whoop_capture.py` writes, so the existing decode pipeline (`whoop-decode`, the parity tests, +`correlate_ground_truth.py`) works on official-app traffic unchanged. + +Privacy: only ATT value traffic that reassembles into CRC-valid WHOOP frames is emitted — other +devices' traffic, GATT chatter and the phone's own commands to non-WHOOP peripherals never reach +the output. (The input HCI log still contains them; share the JSON, not the log, if in doubt.) + +Stdlib-only, like whoop_frame.py. Usage: + + python3 hci_extract.py btsnoop_hci.log --out capture.json + python3 hci_extract.py sync.pklg --family whoop5 --out capture.json +""" + +import argparse +import json +import struct +import sys + +import whoop_frame as wf + +# --- HCI log container parsing --------------------------------------------------------------------- + +# btsnoop timestamps are µs since year 0; this is the µs delta to the unix epoch (Wireshark's value). +BTSNOOP_EPOCH_DELTA_US = 0x00E03AB44A676000 + +# HCI H4 packet-type bytes (prefixed on datalink 1002; synthesised from flags on 1001). +H4_ACL = 0x02 + +# PacketLogger per-record type byte. +PKLG_ACL_SENT = 0x02 +PKLG_ACL_RECV = 0x03 + + +def parse_btsnoop(data: bytes): + """Yield (ts_ms, direction, acl_bytes) for every ACL packet in a btsnoop file. + + direction: "tx" = host→controller (the phone app sending), "rx" = controller→host. + Handles datalink 1001 (raw H1, type inferred from the flags word) and 1002 (H4, type byte + prefixed). Command/event packets are skipped — frames ride on ACL only. + """ + if len(data) < 16 or data[:8] != b"btsnoop\x00": + raise ValueError("not a btsnoop file (bad magic)") + _version, datalink = struct.unpack(">II", data[8:16]) + if datalink not in (1001, 1002): + raise ValueError(f"unsupported btsnoop datalink {datalink}") + off = 16 + while off + 24 <= len(data): + _orig, incl, flags, _drops, ts = struct.unpack(">IIIIq", data[off:off + 24]) + off += 24 + if off + incl > len(data): + break # truncated capture — keep what we have + pkt = data[off:off + incl] + off += incl + ts_ms = (ts - BTSNOOP_EPOCH_DELTA_US) // 1000 + direction = "rx" if (flags & 1) else "tx" + if datalink == 1002: + if not pkt or pkt[0] != H4_ACL: + continue # command / event / SCO + yield ts_ms, direction, pkt[1:] + else: # 1001: flags bit1 set = command-or-event + if flags & 2: + continue + yield ts_ms, direction, pkt + + +def parse_pklg(data: bytes): + """Yield (ts_ms, direction, acl_bytes) for every ACL packet in an Apple PacketLogger file. + + Record: [len u32][ts_secs u32][ts_usecs u32][type u8][payload…] where len covers everything + after itself (8-byte timestamp + 1 type byte + payload). Both the classic big-endian and the + newer little-endian length variant exist in the wild; sniffed from the first record. + """ + if len(data) < 13: + raise ValueError("not a pklg file (too short)") + + def plausible(fmt): + ln = struct.unpack_from(fmt, data, 0)[0] + return 9 <= ln <= len(data) - 4 + + if plausible(">I"): + len_fmt = ">I" + elif plausible(" len(data): + break # truncated / corrupt tail + ts_secs, ts_usecs = struct.unpack_from(">II", data, off + 4) + ptype = data[off + 12] + payload = data[off + 13:off + 4 + ln] + off += 4 + ln + if ptype not in (PKLG_ACL_SENT, PKLG_ACL_RECV): + continue # commands, events, log strings + ts_ms = ts_secs * 1000 + ts_usecs // 1000 + yield ts_ms, ("tx" if ptype == PKLG_ACL_SENT else "rx"), payload + + +def load_hci(path: str): + """Sniff the container format at `path` and yield (ts_ms, direction, acl_bytes).""" + with open(path, "rb") as f: + data = f.read() + if data[:8] == b"btsnoop\x00": + return parse_btsnoop(data) + return parse_pklg(data) + + +# --- ACL → L2CAP → ATT ------------------------------------------------------------------------------ + +ATT_CID = 0x0004 +# ATT opcodes carrying a (handle, value) we care about. +ATT_NOTIFY = 0x1B +ATT_INDICATE = 0x1D +ATT_WRITE_REQ = 0x12 +ATT_WRITE_CMD = 0x52 +# GATT discovery responses — used to map ATT handles back to characteristic UUIDs. +ATT_FIND_INFO_RSP = 0x05 +ATT_READ_BY_TYPE_RSP = 0x09 + + +class L2capReassembler: + """Recombine ACL fragments into complete L2CAP PDUs, per (connection handle, direction). + + The PB flag in the ACL header says whether a fragment starts a new L2CAP PDU (0b00/0b10) or + continues one (0b01); the L2CAP length field says when the PDU is complete. + """ + + def __init__(self): + self.pending = {} # (conn_handle, direction) -> bytearray + + def feed(self, direction: str, acl: bytes): + """Add one ACL packet; return a list of (conn_handle, cid, l2cap_payload) now complete.""" + if len(acl) < 4: + return [] + handle_flags, dlen = struct.unpack_from("> 12) & 0x3 + body = acl[4:4 + dlen] + key = (conn, direction) + if pb == 0b01: # continuation fragment + buf = self.pending.get(key) + if buf is None: + return [] # missed the start — drop + buf.extend(body) + else: # start of a (possibly fragmented) PDU + self.pending[key] = bytearray(body) + buf = self.pending[key] + if len(buf) < 4: + return [] + l2len, cid = struct.unpack_from(" str: + """128-bit UUID from its little-endian wire form to canonical string.""" + b = le_bytes[::-1] + h = b.hex() + return f"{h[0:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:32]}" + + +def _uuid16_str(v: int) -> str: + return f"{v:08x}-0000-1000-8000-00805f9b34fb" + + +class HandleMap: + """ATT handle → characteristic UUID, learned from GATT discovery responses in the capture. + + A phone that already knows the strap caches GATT, so discovery may be absent — then handles + stay unmapped and frames are attributed to "handle-0xNNNN" instead. That's fine for decoding; + the WHOOP channels are still identified by their frames being CRC-valid. + """ + + def __init__(self): + self.map = {} # value handle -> uuid string + + def feed_att(self, pdu: bytes): + op = pdu[0] + if op == ATT_FIND_INFO_RSP and len(pdu) >= 2: + fmt = pdu[1] + step = 4 if fmt == 1 else 18 + for i in range(2, len(pdu) - step + 1, step): + handle = struct.unpack_from("= 2: + # Characteristic-declaration entries: [decl handle u16][props u8][value handle u16][uuid], + # entry length 7 (uuid16) or 21 (uuid128). Other read-by-type shapes just won't match. + elen = pdu[1] + if elen not in (7, 21): + return + for i in range(2, len(pdu) - elen + 1, elen): + value_handle = struct.unpack_from(" str: + return self.map.get(handle, f"handle-0x{handle:04x}") + + +# --- Extraction ------------------------------------------------------------------------------------- + +def extract(packets, family: str): + """Run the ACL stream through L2CAP/ATT and reassemble WHOOP frames per value handle. + + Returns (records, stats): `records` in the capture.json shape (+ a "dir" key: "rx" frames come + from the strap, "tx" are the app's writes — the enable sequence / sync commands issue #103 is + after), `stats` a per-stream {name: {"frames": n, "bytes": n}} summary. + """ + l2cap = L2capReassembler() + handles = HandleMap() + verify = wf.verify_whoop5_frame if family == "whoop5" else wf.verify_whoop4_frame + streams = {} # (conn, handle, dir) -> {"ra": Reassembler, "records": [...], "valid": n} + latest_hr = None + records = [] + + for ts_ms, direction, acl in packets: + for conn, cid, pdu in l2cap.feed(direction, acl): + if cid != ATT_CID or not pdu: + continue + op = pdu[0] + if op in (ATT_FIND_INFO_RSP, ATT_READ_BY_TYPE_RSP): + handles.feed_att(pdu) + continue + if op not in (ATT_NOTIFY, ATT_INDICATE, ATT_WRITE_REQ, ATT_WRITE_CMD) or len(pdu) < 3: + continue + handle = struct.unpack_from(" inner_off: + types[b[inner_off]] = types.get(b[inner_off], 0) + 1 + print(" record types: " + ", ".join(f"{t} ×{n}" for t, n in sorted(types.items()))) + + +if __name__ == "__main__": + main() diff --git a/Tools/linux-capture/test_correlate_ground_truth.py b/Tools/linux-capture/test_correlate_ground_truth.py new file mode 100644 index 000000000..e2fe2c571 --- /dev/null +++ b/Tools/linux-capture/test_correlate_ground_truth.py @@ -0,0 +1,145 @@ +"""Tests for correlate_ground_truth — CSV/alias loading and the known-plaintext field search. + +Run: python3 -m unittest -v (no third-party deps; does not import bleak) + +The correlation test plants a known ground-truth value into synthetic records at a chosen offset and +encoding, then asserts the search recovers that exact (type, field, offset, encoding, scale). +""" + +import io +import json +import os +import struct +import tempfile +import unittest +import zipfile + +import correlate_ground_truth as cg +import whoop_frame as wf + + +class HeaderAliasTests(unittest.TestCase): + def test_german_header_maps_to_canonical(self): + self.assertEqual(cg._norm_header("Herzfrequenzvariabilität (ms)"), "heart_rate_variability_ms") + self.assertEqual(cg._norm_header("Hauttemperatur (Celsius)"), "skin_temp_celsius") + + def test_english_header_passes_through(self): + self.assertEqual(cg._norm_header("Heart rate variability (ms)"), + "heart rate variability (ms)") # English CSV already uses canonical-ish; the + # importer's own English path keys on these — the alias table only needs the localized forms. + + def test_bom_stripped(self): + self.assertEqual(cg._norm_header("Startzeit des Zyklus"), "cycle_start_time") + + +class CsvLoadingTests(unittest.TestCase): + def test_load_german_cycles_from_zip(self): + cycles = ("Startzeit des Zyklus,Herzfrequenzvariabilität (ms)," + "Ruheherzfrequenz (Schläge pro Minute)\n" + "2026-06-20 00:24:31,92,54\n" + "2026-06-18 23:30:56,101,53\n") + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as z: + z.writestr("physiologische_zyklen.csv", cycles) + with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f: + f.write(buf.getvalue()) + path = f.name + rows = cg.load_ground_truth(path) + self.assertEqual(len(rows), 2) + truth = cg.truth_values(rows) + self.assertEqual(sorted(truth["hrv_ms"]), [92.0, 101.0]) + self.assertEqual(sorted(truth["resting_hr_bpm"]), [53.0, 54.0]) + + def test_merges_cycles_and_sleep_on_cycle_start(self): + cycles = "Startzeit des Zyklus,Herzfrequenzvariabilität (ms)\n2026-06-20 00:24:31,92\n" + sleep = "Startzeit des Zyklus,Atemfrequenz (Atemzüge/Min.)\n2026-06-20 00:24:31,14.6\n" + folder = tempfile.mkdtemp() + with open(os.path.join(folder, "physiologische_zyklen.csv"), "w") as f: + f.write(cycles) + with open(os.path.join(folder, "Schlaf.csv"), "w") as f: + f.write(sleep) + rows = cg.load_ground_truth(folder) + self.assertEqual(len(rows), 1) + truth = cg.truth_values(rows) + self.assertEqual(truth["hrv_ms"], [92.0]) + self.assertEqual(truth["respiratory_rate"], [14.6]) + + +class CandidateExtractionTests(unittest.TestCase): + def test_all_encodings_present(self): + rec = struct.pack(" bytes: + return bytes([hx.ATT_NOTIFY]) + struct.pack(" bytes: + return bytes([hx.ATT_WRITE_CMD]) + struct.pack(" bytes: + return struct.pack(" bytes: + handle_flags = (conn & 0x0FFF) | (pb << 12) + return struct.pack(" bytes: + """packets: list of (direction, acl_bytes). Wrap in a datalink-1002 btsnoop container.""" + out = bytearray(b"btsnoop\x00" + struct.pack(">II", 1, 1002)) + ts = hx.BTSNOOP_EPOCH_DELTA_US + 1_000_000 + for direction, acl in packets: + pkt = bytes([hx.H4_ACL]) + acl + flags = 1 if direction == "rx" else 0 + out += struct.pack(">IIIIq", len(pkt), len(pkt), flags, 0, ts) + out += pkt + ts += 10_000 + return bytes(out) + + +def pklg_file(packets) -> bytes: + """packets: list of (direction, acl_bytes). Wrap in a big-endian PacketLogger container.""" + out = bytearray() + secs = 1_700_000_000 + for direction, acl in packets: + ptype = hx.PKLG_ACL_SENT if direction == "tx" else hx.PKLG_ACL_RECV + body = struct.pack(">II", secs, 0) + bytes([ptype]) + acl + out += struct.pack(">I", len(body)) + body + secs += 1 + return bytes(out) + + +class BtsnoopParseTests(unittest.TestCase): + def test_rejects_bad_magic(self): + with self.assertRaises(ValueError): + list(hx.parse_btsnoop(b"nope" + b"\x00" * 20)) + + def test_direction_and_payload(self): + acl = acl_packet(0x40, l2cap_att(att_notify(0x10, b"hello"))) + data = btsnoop_file([("rx", acl), ("tx", acl)]) + pkts = list(hx.parse_btsnoop(data)) + self.assertEqual(len(pkts), 2) + self.assertEqual(pkts[0][1], "rx") + self.assertEqual(pkts[1][1], "tx") + self.assertEqual(pkts[0][2], acl) # H4 type byte stripped + + def test_timestamp_is_unix_ms(self): + acl = acl_packet(0x40, l2cap_att(att_notify(0x10, b"x"))) + pkts = list(hx.parse_btsnoop(btsnoop_file([("rx", acl)]))) + self.assertEqual(pkts[0][0], 1000) # 1_000_000 µs after epoch delta → 1000 ms + + +class PklgParseTests(unittest.TestCase): + def test_roundtrip_big_endian(self): + acl = acl_packet(0x40, l2cap_att(att_notify(0x10, b"data"))) + pkts = list(hx.parse_pklg(pklg_file([("tx", acl), ("rx", acl)]))) + self.assertEqual([p[1] for p in pkts], ["tx", "rx"]) + self.assertEqual(pkts[0][2], acl) + + def test_skips_non_acl_records(self): + # A type-0x00 (command) record between two ACLs must be ignored. + acl = acl_packet(0x40, l2cap_att(att_notify(0x10, b"z"))) + good = pklg_file([("rx", acl)]) + cmd_body = struct.pack(">II", 1, 0) + bytes([0x00]) + b"\xde\xad" + cmd = struct.pack(">I", len(cmd_body)) + cmd_body + pkts = list(hx.parse_pklg(good[:0] + cmd + good)) + self.assertEqual(len(pkts), 1) + + +class L2capReassemblyTests(unittest.TestCase): + def test_fragmented_pdu_reassembles(self): + ra = hx.L2capReassembler() + pdu = l2cap_att(att_notify(0x10, b"A" * 40)) + first = acl_packet(0x40, pdu[:20], pb=0b10) + cont = acl_packet(0x40, pdu[20:], pb=0b01) + self.assertEqual(ra.feed("rx", first), []) # incomplete + done = ra.feed("rx", cont) + self.assertEqual(len(done), 1) + conn, cid, payload = done[0] + self.assertEqual(cid, hx.ATT_CID) + self.assertEqual(payload, att_notify(0x10, b"A" * 40)) + + def test_continuation_without_start_dropped(self): + ra = hx.L2capReassembler() + self.assertEqual(ra.feed("rx", acl_packet(0x40, b"orphan", pb=0b01)), []) + + +class HandleMapTests(unittest.TestCase): + def test_read_by_type_maps_char_value_handle(self): + hm = hx.HandleMap() + uuid = "fd4b0003-cce1-4033-93ce-002d5875f58a" + uuid_le = bytes.fromhex(uuid.replace("-", ""))[::-1] + entry = struct.pack("`. | | `analyze_v26_waveform.py` | Characterise the WHOOP 5 **v26** type-47 buffer as PPG @24 Hz using its own co-timestamped HR as ground truth. | | `analyze_v25_waveform.py` | **WHOOP 4.0 v25 PPG → HR span-pinning harness ([#194](https://github.com/ryanbr/noop/issues/194)).** Sweeps the unpinned PPG span (start + sample-count) across a corpus of captures at *known* HRs and reports the span where recovered HR **tracks** ground truth instead of the `1440/N` autocorrelation artifact — or, on resting-only data, exactly what capture is still needed. `--selftest` proves it on synthetic pulses; no args runs the bundled-frames demo. Stdlib only. | | `whoop_spot_hrv.py` | **Spot HRV (RMSSD) from the sparse PPG bursts.** Reads the v26 `feat_ppg` channel-0 24 Hz waveform (read-only), detects beats, computes RMSSD per PPG-covered window with a GOOD/COARSE/POOR quality label. See [Spot HRV](#spot-hrv-from-sparse-ppg-whoop_spot_hrvpy). Stdlib only. | | `test_whoop_frame.py` | Unit tests for framing / reassembly / HR parsing / buzz frames (no `bleak` needed). | +| `test_hci_extract.py` | Unit tests for the btsnoop/pklg parsers, L2CAP/ATT reassembly, and WHOOP-frame extraction (synthetic fixtures; stdlib only). | +| `test_correlate_ground_truth.py` | Unit tests for the CSV/alias loading and the known-plaintext field search (planted-value recovery + false-positive rejection; stdlib only). | | `test_whoop_spot_hrv.py` | Unit tests for the spot-HRV DSP (synthetic-signal HR/RMSSD recovery, ectopic rejection, grid reconstruction; stdlib only). | | `requirements.txt` | `bleak` (runtime dep for capture only). | @@ -511,6 +515,67 @@ than crashing the import. its MAC — they are git-ignored and should not be shared. - "WHOOP" is used **nominatively** to name the hardware. These tools contain no WHOOP code or assets. +## From a phone HCI capture (`hci_extract.py`) + +You don't need a Linux BLE adapter to contribute frames. A **Bluetooth HCI capture of the official +WHOOP app** — the artifact issue [#103](https://github.com/ryanbr/noop/issues/103) asks for — records +the real app unlocking and draining the deep streams, which is exactly the traffic NOOP can't yet +elicit itself. `hci_extract.py` converts such a log into the same `capture.json` the rest of this +pipeline consumes. + +Grab the capture on the phone that already pairs with your strap: + +- **iOS** — install Apple's *Bluetooth* logging profile (the PacketLogger / Additional Logging + profile), reproduce a **full history sync** in the WHOOP app, then export the `.pklg`. +- **Android** — *Developer options → Enable Bluetooth HCI snoop log*, sync, then pull + `btsnoop_hci.log` (via a bug report or `adb`). + +```bash +python3 hci_extract.py btsnoop_hci.log --family whoop5 --out capture.json +# → wrote 1843 frames → capture.json +# fd4b0005-… [rx]: 1620 frames, … ← strap → app data +# fd4b0002-… [tx]: 41 frames, … ← the app's commands (the enable sequence) +# record types: 47 ×1601, 49 ×18, … +``` + +Only streams that reassemble into **CRC-valid WHOOP frames** are written — other devices' traffic and +non-WHOOP GATT chatter in the log are dropped. Frames the app *sent* are kept too (`"dir": "tx"`), so +the capture shows both the enable sequence and the strap's response. Privacy: the source HCI log still +holds your strap's BLE MAC and every device around you — share the extracted `capture.json` (or its +mapped offsets), not the raw log. + +## Ground-truth correlation (`correlate_ground_truth.py`) + +A capture gives raw records; your **WHOOP data export** gives the official readings for the same +nights. Put them together and the un-decoded layout (the type-`0x2F` deep records, the 5/MG history +types NOOP still skips — see [`WHOOP5_DEEP_DATA.md`](../../docs/WHOOP5_DEEP_DATA.md)) becomes a +*known-plaintext* problem: search each record for the byte offset + encoding whose value reproduces a +known biometric across every night. + +Get the export from **app.whoop.com → Account → Data Export** (any language — the tool reads the +localized German/Spanish/… column headers via the same alias table the app's importer uses). + +```bash +python3 correlate_ground_truth.py capture.json my_whoop_data.zip --tolerance 0.02 +``` + +``` + type field offset enc scale records nights score + 0x2f resting_hr_bpm 10 u8 1 331/331 331/331 1.00 + 0x2f skin_temp_c 14 f32le 1 331/331 330/330 1.00 +``` + +Offsets are into the **inner record** (the type byte is offset 0), ready to wire into +`parseFrameWhoop5` / `whoop_protocol.json`. The search demands both breadth (a decoded value for most +records) and distribution match (recall over known nights **and** precision — decoded values landing +in-range), so a near-constant byte or random noise won't score. `--type 0x2f` restricts to one record +type; widen `--tolerance` if a field is stored pre-rounded. + +**Privacy:** everything runs locally. The tool prints only offsets/encodings — never your health +values — so its output is safe to post on #103 while the CSV export and the capture stay on your +machine. That's the intended way for other 5/MG owners to contribute field mappings without sharing +personal data. + ## Contributing captures back A `capture.json`'s `hex` values are a drop-in for the parity fixtures in