|
#!/usr/bin/env python3 |
|
import sys |
|
import argparse |
|
import os |
|
import sqlite3 |
|
try: |
|
from lxml import etree as ET |
|
except ImportError: |
|
import xml.etree.ElementTree as ET |
|
from collections import defaultdict |
|
from datetime import datetime, timedelta |
|
|
|
SLEEP_TYPE = "HKCategoryTypeIdentifierSleepAnalysis" |
|
HRV_TYPE = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN" |
|
HR_TYPE = "HKQuantityTypeIdentifierHeartRate" |
|
RESP_TYPE = "HKQuantityTypeIdentifierRespiratoryRate" |
|
VO2_TYPE = "HKQuantityTypeIdentifierVO2Max" |
|
STEPS_TYPE = "HKQuantityTypeIdentifierStepCount" |
|
ENERGY_TYPE = "HKQuantityTypeIdentifierActiveEnergyBurned" |
|
EXERCISE_TYPE = "HKQuantityTypeIdentifierAppleExerciseTime" |
|
STAND_TYPE = "HKCategoryTypeIdentifierAppleStandHour" |
|
QUANTITY_TABLES = { |
|
HRV_TYPE: "hrv", HR_TYPE: "hr", |
|
RESP_TYPE: "resp", VO2_TYPE: "vo2", |
|
STEPS_TYPE: "steps", ENERGY_TYPE: "energy", EXERCISE_TYPE: "exercise", |
|
} |
|
MIN_REM_CYCLE_MIN = 3 # minimum total REM to count as a cycle boundary |
|
MAX_AWAKE_MIN = 60 # awake stretches longer than this mark a session break |
|
|
|
|
|
def main(): |
|
parser = argparse.ArgumentParser() |
|
parser.add_argument("export", metavar="export.xml") |
|
parser.add_argument("--from", dest="date_from", metavar="DATE") |
|
parser.add_argument("--to", dest="date_to", metavar="DATE") |
|
parser.add_argument("--source", help="Filter by source name (e.g. \"Vlad's Apple Watch\")") |
|
parser.add_argument("--extended", action="store_true", help="Show per-reading detail for series metrics") |
|
args = parser.parse_args() |
|
|
|
records_by_date, metrics = load_all_sleep_records(args.export, args.source) |
|
|
|
if args.date_from: |
|
date_to = args.date_to or args.date_from |
|
dates = [d for d in sorted(records_by_date) if args.date_from <= d <= date_to] |
|
else: |
|
dates = sorted(records_by_date) |
|
|
|
all_cycle_durations = [] |
|
for date in dates: |
|
records = records_by_date.get(date) |
|
if not records: |
|
if args.date_from: |
|
print(f"No sleep data for {date}") |
|
continue |
|
cycles = print_cycles(date, records, metrics, args.extended) |
|
all_cycle_durations.extend(dur for _, _, dur, _ in cycles) |
|
|
|
if all_cycle_durations: |
|
all_cycle_durations.sort() |
|
n = len(all_cycle_durations) |
|
p90 = all_cycle_durations[min(int(n * 0.9), n - 1)] |
|
median = all_cycle_durations[n // 2] |
|
def fmt(m): h, m = divmod(m, 60); return f"{h}h {m:02d}m" |
|
print(f"Cycle duration ({n} cycles): median {fmt(median)}, P90 {fmt(p90)}") |
|
|
|
|
|
def db_path(xml_path): |
|
return os.path.splitext(xml_path)[0] + ".db" |
|
|
|
|
|
def build_db(xml_path, path): |
|
print(f"Building cache {path} ...", file=sys.stderr) |
|
root = ET.parse(xml_path).getroot() |
|
con = sqlite3.connect(path) |
|
con.execute(""" |
|
CREATE TABLE sleep ( |
|
startDate TEXT, |
|
endDate TEXT, |
|
stage TEXT, |
|
source TEXT |
|
) |
|
""") |
|
for tname in QUANTITY_TABLES.values(): |
|
con.execute(f"CREATE TABLE {tname} (startDate TEXT, value REAL)") |
|
con.execute("CREATE TABLE stand (startDate TEXT, value TEXT)") |
|
|
|
sleep_rows = [] |
|
quantity_rows = {t: [] for t in QUANTITY_TABLES.values()} |
|
stand_rows = [] |
|
for r in root.findall("Record"): |
|
rtype = r.get("type") |
|
if rtype == SLEEP_TYPE: |
|
if "InBed" in r.get("value", ""): |
|
continue |
|
stage = ( |
|
r.get("value", "") |
|
.replace("HKCategoryValueSleepAnalysisAsleep", "") |
|
.replace("HKCategoryValueSleepAnalysis", "") |
|
) |
|
sleep_rows.append((r.get("startDate"), r.get("endDate"), stage, r.get("sourceName", ""))) |
|
elif rtype in QUANTITY_TABLES: |
|
tname = QUANTITY_TABLES[rtype] |
|
src = r.get("sourceName", "") |
|
if tname in ("steps", "energy", "exercise") and "Watch" not in src: |
|
continue |
|
quantity_rows[tname].append( |
|
(r.get("startDate"), float(r.get("value", 0))) |
|
) |
|
elif rtype == STAND_TYPE: |
|
stand_rows.append((r.get("startDate"), r.get("value", ""))) |
|
|
|
con.executemany("INSERT INTO sleep VALUES (?,?,?,?)", sleep_rows) |
|
con.execute("CREATE INDEX idx_end ON sleep (endDate)") |
|
for tname, rows in quantity_rows.items(): |
|
con.executemany(f"INSERT INTO {tname} VALUES (?,?)", rows) |
|
con.execute(f"CREATE INDEX idx_{tname}_start ON {tname} (startDate)") |
|
con.executemany("INSERT INTO stand VALUES (?,?)", stand_rows) |
|
con.execute("CREATE INDEX idx_stand_start ON stand (startDate)") |
|
con.commit() |
|
con.close() |
|
|
|
|
|
def load_all_sleep_records(xml_path, source=None): |
|
path = db_path(xml_path) |
|
|
|
needs_rebuild = ( |
|
not os.path.exists(path) |
|
or os.path.getmtime(path) < os.path.getmtime(xml_path) |
|
) |
|
if not needs_rebuild: |
|
con = sqlite3.connect(path) |
|
tables = {r[0] for r in con.execute( |
|
"SELECT name FROM sqlite_master WHERE type='table'" |
|
)} |
|
con.close() |
|
if not (set(QUANTITY_TABLES.values()) | {"stand"}).issubset(tables): |
|
needs_rebuild = True |
|
if needs_rebuild: |
|
if os.path.exists(path): |
|
os.remove(path) |
|
build_db(xml_path, path) |
|
|
|
con = sqlite3.connect(path) |
|
query = "SELECT startDate, endDate, stage FROM sleep" |
|
params = [] |
|
if source: |
|
query += " WHERE source = ?" |
|
params.append(source) |
|
rows = con.execute(query, params).fetchall() |
|
hrv_rows = con.execute("SELECT startDate, value FROM hrv").fetchall() |
|
hr_rows = con.execute("SELECT startDate, value FROM hr").fetchall() |
|
resp_rows = con.execute("SELECT startDate, value FROM resp").fetchall() |
|
vo2_rows = con.execute("SELECT startDate, value FROM vo2").fetchall() |
|
steps_rows = con.execute("SELECT startDate, value FROM steps").fetchall() |
|
energy_rows = con.execute("SELECT startDate, value FROM energy").fetchall() |
|
exercise_rows = con.execute("SELECT startDate, value FROM exercise").fetchall() |
|
stand_rows = con.execute("SELECT startDate, value FROM stand").fetchall() |
|
con.close() |
|
|
|
by_date = defaultdict(list) |
|
for start_str, end_str, stage in rows: |
|
s = parse_dt(start_str) |
|
e = parse_dt(end_str) |
|
# Records ending before noon belong to that morning's night; |
|
# records ending at noon or later are evening/onset records for the next night. |
|
if e.hour >= 12: |
|
night = (e + timedelta(days=1)).strftime("%Y-%m-%d") |
|
else: |
|
night = e.strftime("%Y-%m-%d") |
|
by_date[night].append((s, e, stage)) |
|
for records in by_date.values(): |
|
records.sort() |
|
|
|
sleep_windows = {night: sleep_window(records) for night, records in by_date.items()} |
|
|
|
def overnight_key(dt): |
|
if dt.hour >= 12: |
|
return (dt + timedelta(days=1)).strftime("%Y-%m-%d") |
|
return dt.strftime("%Y-%m-%d") |
|
|
|
hrv_by_date = defaultdict(list) |
|
hrv_series_by_date = defaultdict(list) |
|
for start_str, value in hrv_rows: |
|
dt = parse_dt(start_str) |
|
key = overnight_key(dt) |
|
window = sleep_windows.get(key) |
|
if window and window[0] <= dt <= window[1]: |
|
hrv_by_date[key].append(value) |
|
hrv_series_by_date[key].append((dt, value)) |
|
|
|
resp_by_date = defaultdict(list) |
|
for start_str, value in resp_rows: |
|
dt = parse_dt(start_str) |
|
resp_by_date[overnight_key(dt)].append((dt, value)) |
|
|
|
hr_series_by_date = defaultdict(list) |
|
for start_str, value in hr_rows: |
|
dt = parse_dt(start_str) |
|
hr_series_by_date[overnight_key(dt)].append((dt, value)) |
|
|
|
vo2_series = sorted( |
|
(parse_dt(s).strftime("%Y-%m-%d"), v) for s, v in vo2_rows |
|
) |
|
|
|
steps_by_date = defaultdict(float) |
|
for start_str, value in steps_rows: |
|
steps_by_date[parse_dt(start_str).strftime("%Y-%m-%d")] += value |
|
|
|
energy_by_date = defaultdict(float) |
|
for start_str, value in energy_rows: |
|
energy_by_date[parse_dt(start_str).strftime("%Y-%m-%d")] += value |
|
|
|
exercise_by_date = defaultdict(float) |
|
for start_str, value in exercise_rows: |
|
exercise_by_date[parse_dt(start_str).strftime("%Y-%m-%d")] += value |
|
|
|
stand_by_date = defaultdict(int) |
|
for start_str, value in stand_rows: |
|
if value == "HKCategoryValueAppleStandHourStood": |
|
stand_by_date[parse_dt(start_str).strftime("%Y-%m-%d")] += 1 |
|
|
|
metrics = { |
|
"hrv": hrv_by_date, |
|
"hrv_series": hrv_series_by_date, |
|
"sleep_window": sleep_windows, |
|
"resp": resp_by_date, |
|
"hr": hr_series_by_date, |
|
"vo2": vo2_series, |
|
"steps": steps_by_date, |
|
"energy": energy_by_date, |
|
"exercise": exercise_by_date, |
|
"stand": stand_by_date, |
|
} |
|
return by_date, metrics |
|
|
|
|
|
def split_sessions(records): |
|
"""Split records into sessions at long Awake gaps; drop the Awake record itself.""" |
|
sessions, current = [], [] |
|
for s, e, stage in records: |
|
if stage == "Awake" and (e - s).total_seconds() / 60 > MAX_AWAKE_MIN: |
|
if current: |
|
sessions.append(current) |
|
current = [] |
|
else: |
|
current.append((s, e, stage)) |
|
if current: |
|
sessions.append(current) |
|
return sessions |
|
|
|
|
|
def sleep_window(records): |
|
"""Return (start, end) of the main sleep session for a night's records.""" |
|
main_records = split_sessions(records)[-1] |
|
return main_records[0][0], main_records[-1][1] |
|
|
|
|
|
def print_cycles(date, records, metrics, extended=False): |
|
sessions = split_sessions(records) |
|
naps, main_records = sessions[:-1], sessions[-1] |
|
|
|
print(f"Sleep cycles for {date}") |
|
print() |
|
|
|
cycles = find_cycles(main_records) |
|
labels = [f"Cycle {i+1}" for i in range(len(cycles))] |
|
|
|
print(f"{'':8} {'Start':>5} {'End':>5} {'Duration':>10} Composition") |
|
print("-" * 70) |
|
for nap in naps: |
|
nap_start, nap_end = nap[0][0], nap[-1][1] |
|
dur = int((nap_end - nap_start).total_seconds() / 60) |
|
h, m = divmod(dur, 60) |
|
dur_str = f"{h}h {m:02d}m" if h else f"{m}m" |
|
print(f"{'Nap':8} {nap_start.strftime('%H:%M'):>5} {nap_end.strftime('%H:%M'):>5} {dur_str:>10} {summarize_stages(nap)}") |
|
for label, (start, end, dur, recs) in zip(labels, cycles): |
|
composition = summarize_stages(recs) |
|
h, m = divmod(dur, 60) |
|
dur_str = f"{h}h {m:02d}m" if h else f"{m}m" |
|
print(f"{label:8} {start.strftime('%H:%M'):>5} {end.strftime('%H:%M'):>5} {dur_str:>10} {composition}") |
|
|
|
total = sum(dur for _, _, dur, _ in cycles) |
|
h, m = divmod(total, 60) |
|
print(f"\n{'Total sleep':>32} {h}h {m:02d}m") |
|
sleep_start = main_records[0][0] |
|
sleep_end = main_records[-1][1] |
|
print_trend_metric("HRV", date, metrics["hrv"], "ms", decimals=0) |
|
if extended: |
|
print_series_pairs(date, metrics["hrv_series"], sleep_start, sleep_end, decimals=1, cycles=cycles, records=main_records, per_line=4) |
|
print_overnight_hr_metric(date, metrics["hr"], metrics["sleep_window"]) |
|
if extended: |
|
print_hr_per_cycle(date, metrics["hr"], cycles) |
|
print_resp_series(date, metrics["resp"], sleep_start, sleep_end, extended) |
|
print_vo2(date, metrics["vo2"]) |
|
print_activity(date, metrics) |
|
print() |
|
return cycles |
|
|
|
|
|
def print_trend_metric(label, date, by_date, unit, decimals): |
|
values = by_date.get(date) |
|
if not values: |
|
print(f"{label:>32} (no data)") |
|
return |
|
values = sorted(values) |
|
current = values[len(values) // 2] |
|
prior_dates = sorted(d for d in by_date if d < date)[-5:] |
|
prior_vals = sorted(v for d in prior_dates for v in by_date[d]) |
|
trend_str = "" |
|
if prior_vals: |
|
prior_median = prior_vals[len(prior_vals) // 2] |
|
if decimals == 0: |
|
delta = int(current) - int(prior_median) |
|
else: |
|
delta = round(current - prior_median, decimals) |
|
arrow = "↑" if delta > 0 else "↓" if delta < 0 else "→" |
|
trend_str = f" {arrow}{abs(delta):.{decimals}f} vs 5-night median ({prior_median:.{decimals}f})" |
|
print(f"{label:>32} {current:.{decimals}f} {unit}{trend_str}") |
|
|
|
|
|
def print_vo2(date, vo2_series): |
|
latest = None |
|
for d, v in vo2_series: |
|
if d <= date: |
|
latest = (d, v) |
|
else: |
|
break |
|
if not latest: |
|
return |
|
d, v = latest |
|
print(f"{'VO2 max':>32} {v:.1f} mL/(kg·min) as of {d}") |
|
|
|
|
|
def print_overnight_hr_metric(date, hr_series_by_date, sleep_windows): |
|
series = hr_series_by_date.get(date) |
|
window = sleep_windows.get(date) |
|
if not series or not window: |
|
print(f"{'Overnight HR':>32} (no data)") |
|
return |
|
sleep_start, sleep_end = window |
|
vals = sorted(v for t, v in series if sleep_start <= t <= sleep_end) |
|
if not vals: |
|
print(f"{'Overnight HR':>32} (no data)") |
|
return |
|
current = int(vals[len(vals) // 2]) |
|
prior_dates = sorted(d for d in hr_series_by_date if d < date and d in sleep_windows)[-5:] |
|
prior_vals = [] |
|
for d in prior_dates: |
|
p_start, p_end = sleep_windows[d] |
|
prior_vals.extend(v for t, v in hr_series_by_date[d] if p_start <= t <= p_end) |
|
prior_vals.sort() |
|
trend_str = "" |
|
if prior_vals: |
|
prior_median = int(prior_vals[len(prior_vals) // 2]) |
|
delta = current - prior_median |
|
arrow = "↑" if delta > 0 else "↓" if delta < 0 else "→" |
|
trend_str = f" {arrow}{abs(delta)} vs 5-night median ({prior_median})" |
|
print(f"{'Overnight HR':>32} {current} bpm{trend_str}") |
|
|
|
|
|
def print_hr_per_cycle(date, hr_by_date, cycles): |
|
entries = hr_by_date.get(date) |
|
if not entries: |
|
return |
|
entries.sort() |
|
stats = [] |
|
global_min = (float("inf"), None, -1) |
|
for idx, (cstart, cend, _dur, _recs) in enumerate(cycles): |
|
in_cycle = [(t, v) for t, v in entries if cstart <= t <= cend] |
|
if not in_cycle: |
|
stats.append(None) |
|
continue |
|
vals = sorted(v for _, v in in_cycle) |
|
cmin, cmax = vals[0], vals[-1] |
|
cmed = vals[len(vals) // 2] |
|
min_time = next(t for t, v in in_cycle if v == cmin) |
|
stats.append((cstart, cend, cmin, cmax, cmed, min_time)) |
|
if cmin < global_min[0]: |
|
global_min = (cmin, min_time, idx) |
|
for idx, stat in enumerate(stats): |
|
if stat is None: |
|
continue |
|
cstart, cend, cmin, cmax, cmed, min_time = stat |
|
annot = "" |
|
if idx == global_min[2]: |
|
annot = f" ← min {int(cmin)} at {min_time.strftime('%H:%M')}" |
|
print(f" Cycle {idx+1} ({cstart.strftime('%H:%M')}–{cend.strftime('%H:%M')}): min {int(cmin)}, max {int(cmax)}, med {int(cmed)} bpm{annot}") |
|
|
|
|
|
def print_series_pairs(date, series_by_date, sleep_start, sleep_end, decimals=1, cycles=None, records=None, per_line=7): |
|
entries = series_by_date.get(date) |
|
if not entries: |
|
return |
|
entries = sorted(e for e in entries if sleep_start <= e[0] <= sleep_end) |
|
if not entries: |
|
return |
|
pairs = [] |
|
for t, v in entries: |
|
s = f"{t.strftime('%H:%M')} {v:>4.{decimals}f}" |
|
if cycles is not None and records is not None: |
|
s += f" ({cycle_stage_at(t, cycles, records)})" |
|
pairs.append(s) |
|
for i in range(0, len(pairs), per_line): |
|
print(" " + " ".join(pairs[i:i + per_line])) |
|
|
|
|
|
def cycle_stage_at(t, cycles, records): |
|
cycle = next((f"C{i+1}" for i, (cs, ce, _, _) in enumerate(cycles) if cs <= t <= ce), "?") |
|
stage = next((st for s, e, st in records if s <= t <= e), "?") |
|
return f"{cycle}-{stage}" |
|
|
|
|
|
def print_resp_series(date, resp_by_date, sleep_start, sleep_end, extended=False): |
|
entries = resp_by_date.get(date) |
|
if entries: |
|
entries = sorted(e for e in entries if sleep_start <= e[0] <= sleep_end) |
|
if not entries: |
|
print(f"{'Resp rate':>32} (no data)") |
|
return |
|
values = [v for _, v in entries] |
|
vmin, vmax = min(values), max(values) |
|
print(f"{'Resp rate':>32} {vmin:.1f}–{vmax:.1f} br/min, {len(entries)} readings") |
|
if extended: |
|
print_series_pairs(date, resp_by_date, sleep_start, sleep_end, decimals=1) |
|
|
|
|
|
def print_activity(date, metrics): |
|
prev_day = (datetime.strptime(date, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d") |
|
energy = metrics["energy"].get(prev_day) |
|
exercise = metrics["exercise"].get(prev_day) |
|
stand = metrics["stand"].get(prev_day) |
|
steps = metrics["steps"].get(prev_day) |
|
if energy is None and exercise is None and stand is None and steps is None: |
|
print(f"{'Activity':>32} (no data)") |
|
return |
|
parts = [] |
|
if energy is not None: parts.append(f"Move {int(energy)} kcal") |
|
if exercise is not None: parts.append(f"Exercise {int(exercise)} min") |
|
if stand is not None: parts.append(f"Stand {stand} h") |
|
if steps is not None: parts.append(f"Steps {int(steps):,}") |
|
print(f"{'Activity':>32} {' · '.join(parts)}") |
|
|
|
|
|
def find_cycles(records): |
|
"""Split records into cycles using REM blocks as cycle boundaries.""" |
|
cycles = [] |
|
current_start = records[0][0] if records else None |
|
|
|
i = 0 |
|
while i < len(records): |
|
s, e, stage = records[i] |
|
if stage == "REM": |
|
# Merge consecutive REM blocks (possibly separated by tiny Core/Awake) |
|
rem_end = e |
|
rem_total = (e - s).total_seconds() / 60 |
|
j = i + 1 |
|
while j < len(records): |
|
ns, ne, nstage = records[j] |
|
gap = (ns - rem_end).total_seconds() / 60 |
|
if gap <= 5 and nstage in ("REM", "Core", "Awake"): |
|
if nstage == "REM": |
|
rem_end = ne |
|
rem_total += (ne - ns).total_seconds() / 60 |
|
j += 1 |
|
else: |
|
break |
|
|
|
if rem_total < MIN_REM_CYCLE_MIN: |
|
i = j |
|
continue |
|
|
|
# Collect all records up to and including this REM block |
|
cycle_records = [r for r in records if current_start <= r[0] < rem_end] |
|
dur = int((rem_end - current_start).total_seconds() / 60) |
|
cycles.append((current_start, rem_end, dur, cycle_records)) |
|
# Find the next record starting at or after rem_end (don't skip |
|
# records that the merge loop may have advanced j past). |
|
j = next((k for k in range(i + 1, len(records)) if records[k][0] >= rem_end), len(records)) |
|
current_start = records[j][0] if j < len(records) else None |
|
i = j |
|
else: |
|
i += 1 |
|
|
|
# Tail: remaining records after last REM |
|
if current_start is not None: |
|
tail = [r for r in records if r[0] >= current_start] |
|
if tail: |
|
tail_end = tail[-1][1] |
|
dur = int((tail_end - current_start).total_seconds() / 60) |
|
cycles.append((current_start, tail_end, dur, tail)) |
|
|
|
return cycles |
|
|
|
|
|
def summarize_stages(cycle_records): |
|
# Resolve overlaps: when records overlap, the later-starting one takes precedence. |
|
# Build a list of non-overlapping segments by clipping earlier records. |
|
segments = [] |
|
for s, e, stage in sorted(cycle_records): |
|
if segments and s < segments[-1][1]: |
|
ps, _, pstage = segments[-1] |
|
segments[-1] = (ps, s, pstage) |
|
segments.append((s, e, stage)) |
|
|
|
totals = {} |
|
for s, e, stage in segments: |
|
dur = (e - s).total_seconds() / 60 |
|
if dur >= 1: |
|
totals[stage] = totals.get(stage, 0) + dur |
|
|
|
parts = [] |
|
for stage in ("Deep", "Core", "REM", "Awake"): |
|
if stage in totals: |
|
parts.append(f"{stage} {int(totals[stage])}m") |
|
return " → ".join(parts) |
|
|
|
|
|
def parse_dt(s): |
|
return datetime.strptime(s, "%Y-%m-%d %H:%M:%S %z") |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |
