make_sample_data.py

#!/usr/bin/env python3
"""Generate a realistic synthetic overnight O2Ring CSV.

This lets anyone try o2ring-analyzer immediately without uploading their own
health data. The output mimics an O2 Insight Pro export: a header row followed by
per-few-seconds rows of timestamp, SpO2, pulse rate and motion, including a few
plausible desaturation dips, a short sensor dropout, and motion bursts.

Usage:
    python make_sample_data.py --output sample_night.csv
    python make_sample_data.py --hours 8 --interval 4 --seed 42 --locale eu
"""

from __future__ import annotations

import argparse
import csv
from datetime import datetime, timedelta

import numpy as np


def generate_session(
    hours: float = 8.0,
    interval_s: float = 4.0,
    seed: int = 42,
    start: datetime | None = None,
) -> list[dict]:
    """Return a list of per-sample dict rows for a synthetic overnight session."""
    rng = np.random.default_rng(seed)
    start = start or datetime(2024, 1, 15, 23, 15, 0)
    n = int(hours * 3600 / interval_s)

    # Baseline SpO2 wanders slowly around ~96.5 with small second-to-second noise.
    slow = 96.5 + 0.8 * np.sin(np.linspace(0, 6 * np.pi, n))
    noise = rng.normal(0, 0.4, n)
    spo2 = slow + noise

    # Inject a handful of desaturation events at random-ish positions.
    n_events = max(3, int(hours * 2))
    for _ in range(n_events):
        center = rng.integers(int(0.05 * n), int(0.95 * n))
        depth = rng.uniform(4, 9)
        width = int(rng.uniform(15, 40) / interval_s)  # samples
        idx = np.arange(max(0, center - width), min(n, center + width))
        dip = depth * np.exp(-((idx - center) ** 2) / (2 * (width / 2.5) ** 2))
        spo2[idx] -= dip

    spo2 = np.clip(np.round(spo2), 80, 100)

    # Pulse: ~58 bpm resting with slow drift, respiratory-ish variability, and a
    # small rise accompanying each desaturation recovery.
    pulse = 58 + 4 * np.sin(np.linspace(0, 8 * np.pi, n)) + rng.normal(0, 2.5, n)
    pulse = np.clip(np.round(pulse), 40, 110)

    # Motion: mostly still with occasional bursts.
    motion = np.zeros(n)
    for _ in range(int(hours * 5)):
        c = rng.integers(0, n)
        w = int(rng.uniform(2, 8))
        motion[c : c + w] = rng.integers(1, 5)

    rows = []
    for i in range(n):
        ts = start + timedelta(seconds=i * interval_s)
        # A short sensor dropout around the 40% mark -> zeros (treated as missing).
        dropout = int(0.40 * n) <= i < int(0.40 * n) + int(30 / interval_s)
        rows.append(
            {
                "Time": ts.strftime("%Y-%m-%d %H:%M:%S"),
                "SpO2": 0 if dropout else int(spo2[i]),
                "Pulse Rate": 0 if dropout else int(pulse[i]),
                "Motion": int(motion[i]),
            }
        )
    return rows


def write_csv(rows: list[dict], path: str, locale: str = "us") -> None:
    """Write rows to a CSV in either US (comma) or EU (semicolon) style."""
    delimiter = ";" if locale == "eu" else ","
    with open(path, "w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(
            fh, fieldnames=["Time", "SpO2", "Pulse Rate", "Motion"], delimiter=delimiter
        )
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--output", default="sample_night.csv", help="Output CSV path.")
    p.add_argument("--hours", type=float, default=8.0)
    p.add_argument("--interval", type=float, default=4.0, help="Seconds between samples.")
    p.add_argument("--seed", type=int, default=42)
    p.add_argument("--locale", choices=["us", "eu"], default="us")
    args = p.parse_args()
    rows = generate_session(hours=args.hours, interval_s=args.interval, seed=args.seed)
    write_csv(rows, args.output, locale=args.locale)
    print(f"Wrote {len(rows)} rows to {args.output} ({args.locale} locale).")


if __name__ == "__main__":
    main()