#!/usr/bin/env python3 """Canonical Apple Health Analytics v2 for JARVIS reports. The raw ``apple_health_records`` table remains an immutable audit trail. This module provides the only supported aggregation layer for dashboards/reports: - exact observations duplicated across overlapping exports are deduplicated; - fine-grained exports supersede coarse weekly/monthly fallback rows; - one deterministic source class is selected per metric/day to avoid summing overlapping Watch/phone/app observations; - aware timestamps are converted to Europe/Zurich and naive timestamps are explicitly treated as local Europe/Zurich time; - metric-specific sum/average/last semantics and unit normalization apply; - current-day values are provisional and excluded by default; - coverage summaries expose counts/quality only, never raw values. """ from __future__ import annotations import re import sqlite3 from math import ceil from collections import Counter, defaultdict from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from pathlib import Path from statistics import mean from typing import Any, Iterable from zoneinfo import ZoneInfo BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" DB = BASE / "health_data.db" LOCAL_TZ = ZoneInfo("Europe/Zurich") ANALYTICS_VERSION = 2 SUM_METRICS = { "step_count", "walking_running_distance", "active_energy", "basal_energy_burned", "apple_exercise_time", "apple_stand_time", "flights_climbed", "sleep_analysis", "dietary_energy", "dietary_water", "carbohydrates", "protein", "total_fat", "dietary_sugar", "fiber", "saturated_fat", "monounsaturated_fat", "polyunsaturated_fat", } LAST_METRICS = { "weight_body_mass", "body_mass_index", "blood_pressure", } ENERGY_METRICS = {"active_energy", "basal_energy_burned", "dietary_energy"} CANONICAL_UNITS = { **{metric: "kcal" for metric in ENERGY_METRICS}, "walking_running_distance": "km", "weight_body_mass": "kg", "dietary_water": "mL", "sleep_analysis": "hr", "apple_exercise_time": "min", "apple_stand_time": "min", "resting_heart_rate": "bpm", "heart_rate_variability": "ms", "step_count": "count", "blood_oxygen_saturation": "%", "respiratory_rate": "breaths/min", "body_mass_index": "kg/m²", } @dataclass(frozen=True) class MetricSpec: aggregation: str canonical_unit: str | None = None expected_cadence_days: int = 1 @dataclass(frozen=True) class DailyPoint: value: float unit: str quality: str source_class: str timezone_assumption: str selected_records: int deduplicated_records: int discarded_overlap_records: int discarded_source_records: int discarded_unit_records: int @dataclass(frozen=True) class _Observation: row: sqlite3.Row timestamp: datetime local_day: str value: float unit: str source_class: str source_rank: int coarse_factor: int duplicate_count: int timezone_assumption: str def conn() -> sqlite3.Connection: connection = sqlite3.connect(DB) connection.row_factory = sqlite3.Row return connection def fnum(value: Any) -> float | None: try: if value is None: return None return float(str(value).replace(",", ".").strip()) except (TypeError, ValueError): return None def metric_spec(metric: str, mode: str | None = None) -> MetricSpec: if mode is not None: aggregation = mode elif metric in SUM_METRICS: aggregation = "sum" elif metric in LAST_METRICS: aggregation = "last" else: aggregation = "avg" if aggregation not in {"sum", "avg", "last"}: raise ValueError(f"Unsupported aggregation mode: {aggregation}") canonical_unit = CANONICAL_UNITS.get(metric) expected_cadence_days = 7 if metric in LAST_METRICS else 1 return MetricSpec( aggregation=aggregation, canonical_unit=canonical_unit, expected_cadence_days=expected_cadence_days, ) def _parse_timestamp(value: Any) -> tuple[datetime | None, str]: text = str(value or "").strip() if not text: return None, "missing" try: parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) except ValueError: return None, "unparseable" if parsed.tzinfo is None: return parsed.replace(tzinfo=LOCAL_TZ), "naive_assumed_europe_zurich" return parsed.astimezone(LOCAL_TZ), "timezone_aware_converted_europe_zurich" def _source_info(source: Any) -> tuple[str, str, int]: raw = str(source or "").replace("\u00a0", " ").strip() normalized = re.sub(r"\s+", " ", raw).casefold() if "|" in normalized: return normalized, "composite", 0 if "watch" in normalized: return normalized, "watch", 1 if "iphone" in normalized or "phone" in normalized: return normalized, "phone", 2 if any(token in normalized for token in ("health", "santé", "gesundheit")): return normalized, "health_app", 3 if normalized: return normalized, "app", 4 return "", "unknown", 5 def _normalize_value(metric: str, value: float, unit: Any) -> tuple[float, str]: raw_unit = str(unit or "").strip() normalized_unit = raw_unit.casefold().replace(" ", "") if metric in ENERGY_METRICS: if normalized_unit == "kj": return value / 4.184, "kcal" if normalized_unit == "kcal": return value, "kcal" if metric == "walking_running_distance": if normalized_unit in {"m", "meter", "metre"}: return value / 1000, "km" if normalized_unit in {"mi", "mile", "miles"}: return value * 1.609344, "km" if normalized_unit in {"km", "kilometer", "kilometre"}: return value, "km" if metric == "weight_body_mass": if normalized_unit in {"g", "gram", "grams"}: return value / 1000, "kg" if normalized_unit in {"lb", "lbs", "pound", "pounds"}: return value * 0.45359237, "kg" if normalized_unit in {"kg", "kilogram", "kilograms"}: return value, "kg" if metric == "dietary_water": if normalized_unit in {"l", "liter", "litre"}: return value * 1000, "mL" if normalized_unit in {"ml", "milliliter", "millilitre"}: return value, "mL" if metric == "sleep_analysis": if normalized_unit in {"s", "sec", "second", "seconds"}: return value / 3600, "hr" if normalized_unit in {"min", "minute", "minutes"}: return value / 60, "hr" if normalized_unit in {"h", "hr", "hour", "hours"}: return value, "hr" if metric in {"apple_exercise_time", "apple_stand_time"}: if normalized_unit in {"s", "sec", "second", "seconds"}: return value / 60, "min" if normalized_unit in {"h", "hr", "hour", "hours"}: return value * 60, "min" if normalized_unit in {"min", "minute", "minutes"}: return value, "min" if metric == "resting_heart_rate" and normalized_unit in { "count/min", "counts/min", "bpm", }: return value, "bpm" if metric == "heart_rate_variability" and normalized_unit in {"ms", "millisecond", "milliseconds"}: return value, "ms" if metric == "step_count" and normalized_unit in {"count", "counts"}: return value, "count" if metric == "blood_oxygen_saturation" and normalized_unit in {"%", "percent"}: return value, "%" if metric == "respiratory_rate" and normalized_unit in { "count/min", "counts/min", "breaths/min", "breath/min", }: return value, "breaths/min" if metric == "physical_effort" and normalized_unit in { "kcal/hr·kg", "kcal/h/kg", "kcal/hr/kg", }: return value, "kcal/h/kg" if metric == "body_mass_index" and normalized_unit in {"count", "kg/m²", "kg/m2"}: return value, "kg/m²" return value, raw_unit def normalize_value(metric: str, value: float, unit: Any) -> tuple[float, str]: """Return the canonical value/unit pair used by Analytics v2 and API contracts.""" return _normalize_value(metric, value, unit) def _row_key(row: sqlite3.Row) -> tuple[Any, ...]: start, _ = _parse_timestamp(row["start_date"]) end, _ = _parse_timestamp(row["end_date"]) raw_value = fnum(row["value"]) _, canonical_unit = _normalize_value(str(row["metric"]), raw_value or 0.0, row["unit"]) _, source_class, _ = _source_info(row["source_name"]) return ( row["metric"], source_class, start.astimezone(timezone.utc).isoformat() if start else str(row["start_date"] or ""), end.astimezone(timezone.utc).isoformat() if end else str(row["end_date"] or ""), canonical_unit.casefold(), ) def _canonical_value_key(row: sqlite3.Row) -> float: raw_value = fnum(row["value"]) canonical_value, _ = _normalize_value(str(row["metric"]), raw_value or 0.0, row["unit"]) return round(canonical_value, 12) def _load_rows( metric: str | None = None, start_date: str | None = None, end_date: str | None = None, ) -> list[sqlite3.Row]: connection = conn() try: sql = """SELECT id,metric,value,unit,start_date,end_date,source_name,file_name,file_hash FROM apple_health_records WHERE value IS NOT NULL AND start_date IS NOT NULL""" params: list[Any] = [] if metric: sql += " AND metric=?" params.append(metric) if start_date: sql += " AND substr(start_date,1,10)>=?" params.append(start_date) if end_date: sql += " AND substr(start_date,1,10)<=?" params.append(end_date) sql += " ORDER BY metric, start_date, end_date, source_name, id" return list(connection.execute(sql, params)) finally: connection.close() def canonical_records(metric: str | None = None) -> list[sqlite3.Row]: """Return one canonical row per exact observation key. For sum metrics, direct-resolution rows win over coarse grouped-export rows even when the coarse export was imported later. """ if metric is not None: rows = _load_rows(metric) factors = _coarse_factors(rows, metric) if metric in SUM_METRICS else {} return [row for row, _ in _deduplicated_rows(metric, rows, factors)] rows_by_metric: dict[str, list[sqlite3.Row]] = defaultdict(list) for row in _load_rows(): rows_by_metric[str(row["metric"])].append(row) canonical: list[sqlite3.Row] = [] for metric_name, rows in rows_by_metric.items(): factors = _coarse_factors(rows, metric_name) if metric_name in SUM_METRICS else {} canonical.extend(row for row, _ in _deduplicated_rows(metric_name, rows, factors)) return canonical def _deduplicated_rows( metric: str, rows: list[sqlite3.Row], coarse_factors: dict[tuple[str, str], int], ) -> list[tuple[sqlite3.Row, int]]: groups: dict[tuple[Any, ...], list[sqlite3.Row]] = defaultdict(list) for row in rows: groups[_row_key(row)].append(row) result = [] for group_rows in groups.values(): # Repeated exports from the exact same source are revisions: retain the # best-resolution newest row even when the value changed. by_exact_source: dict[str, list[sqlite3.Row]] = defaultdict(list) for row in group_rows: source_key, _, _ = _source_info(row["source_name"]) by_exact_source[source_key].append(row) source_selected: list[tuple[sqlite3.Row, int]] = [] for source_rows in by_exact_source.values(): selected = min( source_rows, key=lambda row: ( coarse_factors.get((str(row["file_hash"] or row["file_name"] or ""), metric), 1), -int(row["id"]), ), ) source_selected.append((selected, len(source_rows) - 1)) # Across aliases of the same source class, equivalent normalized values # are duplicates; different values remain distinct observations. by_value: dict[float, list[tuple[sqlite3.Row, int]]] = defaultdict(list) for selected, duplicate_count in source_selected: by_value[_canonical_value_key(selected)].append((selected, duplicate_count)) for value_rows in by_value.values(): selected, _ = min( value_rows, key=lambda item: ( coarse_factors.get( (str(item[0]["file_hash"] or item[0]["file_name"] or ""), metric), 1, ), -int(item[0]["id"]), ), ) duplicate_count = sum(count for _, count in value_rows) + len(value_rows) - 1 result.append((selected, duplicate_count)) return result def _coarse_factors(rows: Iterable[sqlite3.Row], metric: str) -> dict[tuple[str, str], int]: """Return deterministic factors for explicitly grouped export files. Historical ``jahr-/year-YYYY`` exports in this project contain weekly aggregates. Explicit weekly/monthly names remain classifiable even when an interrupted export contains only one or two rows. Dated daily export names such as ``jahr-YYYY-MM-DD`` intentionally do not match. """ factors: dict[tuple[str, str], int] = {} for row in rows: file_key = str(row["file_hash"] or row["file_name"] or "") key = (file_key, metric) file_name = str(row["file_name"] or "").casefold() if any(token in file_name for token in ("monthly", "monat")): factors[key] = 30 elif ( any(token in file_name for token in ("weekly", "woche", "annual")) or re.search(r"(?:jahr|year)[-_ ]?\d{4}(?:\.|$)", file_name) ): factors[key] = 7 return factors def _observations( metric: str, start_date: str | None = None, end_date: str | None = None, rows: Iterable[sqlite3.Row] | None = None, ) -> list[_Observation]: raw_rows = list(rows) if rows is not None else _load_rows(metric, start_date=start_date, end_date=end_date) coarse_factors = _coarse_factors(raw_rows, metric) if metric in SUM_METRICS else {} deduplicated = _deduplicated_rows(metric, raw_rows, coarse_factors) observations: list[_Observation] = [] for row, duplicate_count in deduplicated: timestamp, timezone_assumption = _parse_timestamp(row["start_date"]) raw_value = fnum(row["value"]) if timestamp is None or raw_value is None: continue value, unit = _normalize_value(metric, raw_value, row["unit"]) _, source_class, source_rank = _source_info(row["source_name"]) file_key = str(row["file_hash"] or row["file_name"] or "") observations.append( _Observation( row=row, timestamp=timestamp, local_day=timestamp.date().isoformat(), value=value, unit=unit, source_class=source_class, source_rank=source_rank, coarse_factor=coarse_factors.get((file_key, metric), 1), duplicate_count=duplicate_count, timezone_assumption=timezone_assumption, ) ) return observations def _select_source(observations: list[_Observation]) -> tuple[list[_Observation], int]: by_source: dict[str, list[_Observation]] = defaultdict(list) for observation in observations: by_source[observation.source_class].append(observation) selected_key = min( by_source, key=lambda key: ( min(item.source_rank for item in by_source[key]), -len(by_source[key]), key, ), ) selected = by_source[selected_key] return selected, len(observations) - len(selected) def _select_unit(observations: list[_Observation]) -> tuple[list[_Observation], int]: counts = Counter(item.unit for item in observations) selected_unit = min(counts, key=lambda unit: (-counts[unit], unit)) selected = [item for item in observations if item.unit == selected_unit] return selected, len(observations) - len(selected) def daily_points( metric: str, mode: str | None = None, stable_only: bool = True, now: datetime | None = None, start_date: str | None = None, end_date: str | None = None, _rows: Iterable[sqlite3.Row] | None = None, ) -> dict[str, DailyPoint]: spec = metric_spec(metric, mode) local_now = now or datetime.now(LOCAL_TZ) if local_now.tzinfo is None: local_now = local_now.replace(tzinfo=LOCAL_TZ) else: local_now = local_now.astimezone(LOCAL_TZ) current_day = local_now.date().isoformat() by_day: dict[str, list[_Observation]] = defaultdict(list) for observation in _observations(metric, start_date=start_date, end_date=end_date, rows=_rows): if stable_only and observation.local_day >= current_day: continue if start_date and observation.local_day < start_date: continue if end_date and observation.local_day > end_date: continue by_day[observation.local_day].append(observation) result: dict[str, DailyPoint] = {} for day, all_observations in sorted(by_day.items()): direct = [item for item in all_observations if item.coarse_factor == 1] if direct: resolution_selected = direct discarded_overlap = len(all_observations) - len(direct) quality = "direct" else: resolution_selected = all_observations discarded_overlap = 0 quality = "coarse_fallback" source_selected, discarded_source = _select_source(resolution_selected) unit_selected, discarded_unit = _select_unit(source_selected) adjusted = [item.value / item.coarse_factor for item in unit_selected] if not adjusted: continue if spec.aggregation == "sum": value = sum(adjusted) chosen = unit_selected elif spec.aggregation == "last": latest = max(unit_selected, key=lambda item: item.timestamp) value = latest.value / latest.coarse_factor chosen = [latest] else: value = mean(adjusted) chosen = unit_selected timezone_assumption = ( "naive_assumed_europe_zurich" if any(item.timezone_assumption == "naive_assumed_europe_zurich" for item in chosen) else "timezone_aware_converted_europe_zurich" ) result[day] = DailyPoint( value=round(value, 2), unit=chosen[0].unit, quality=quality, source_class=chosen[0].source_class, timezone_assumption=timezone_assumption, selected_records=len(chosen), deduplicated_records=sum(item.duplicate_count for item in all_observations), discarded_overlap_records=discarded_overlap, discarded_source_records=discarded_source, discarded_unit_records=discarded_unit, ) return result def daily_series( metric: str, mode: str | None = None, stable_only: bool = True, now: datetime | None = None, _rows: Iterable[sqlite3.Row] | None = None, ) -> dict[str, float]: """Backward-compatible day -> value view over Analytics v2 points.""" return { day: point.value for day, point in daily_points( metric, mode=mode, stable_only=stable_only, now=now, _rows=_rows, ).items() } def coverage_summary( metric: str, *, start_date: str | None = None, end_date: str | None = None, mode: str | None = None, stable_only: bool = True, now: datetime | None = None, _rows: Iterable[sqlite3.Row] | None = None, ) -> dict[str, Any]: query_start = (date.fromisoformat(start_date) - timedelta(days=1)).isoformat() if start_date else None query_end = (date.fromisoformat(end_date) + timedelta(days=1)).isoformat() if end_date else None points = daily_points( metric, mode=mode, stable_only=stable_only, now=now, start_date=query_start, end_date=query_end, _rows=_rows, ) dated_points = {date.fromisoformat(day): point for day, point in points.items()} observed_dates = sorted(dated_points) if start_date: start = date.fromisoformat(start_date) elif observed_dates: start = observed_dates[0] else: start = None if end_date: end = date.fromisoformat(end_date) elif observed_dates: end = observed_dates[-1] else: end = None if start is None or end is None or end < start: expected_days = 0 window_points: dict[date, DailyPoint] = {} else: expected_days = (end - start).days + 1 window_points = {day: point for day, point in dated_points.items() if start <= day <= end} observed_days = len(window_points) missing_days = max(expected_days - observed_days, 0) spec = metric_spec(metric, mode) expected_observations = ceil(expected_days / spec.expected_cadence_days) if expected_days else 0 missing_observations = max(expected_observations - observed_days, 0) return { "analytics_version": ANALYTICS_VERSION, "metric": metric, "aggregation": spec.aggregation, "expected_cadence_days": spec.expected_cadence_days, "expected_days": expected_days, "expected_observations": expected_observations, "observed_days": observed_days, "missing_days": missing_days, "missing_observations": missing_observations, "coverage_ratio": round(min(observed_days / expected_observations, 1.0), 3) if expected_observations else 0.0, "direct_days": sum(point.quality == "direct" for point in window_points.values()), "coarse_fallback_days": sum(point.quality == "coarse_fallback" for point in window_points.values()), "days_with_source_conflict": sum(point.discarded_source_records > 0 for point in window_points.values()), "days_with_unit_conflict": sum(point.discarded_unit_records > 0 for point in window_points.values()), "deduplicated_records": sum(point.deduplicated_records for point in window_points.values()), "discarded_overlap_records": sum(point.discarded_overlap_records for point in window_points.values()), } def unit_for(metric: str, _rows: Iterable[sqlite3.Row] | None = None) -> str: spec = metric_spec(metric) if spec.canonical_unit: unit = spec.canonical_unit elif _rows is not None: unit_counts = Counter( str(row["unit"] or "") for row in _rows if str(row["metric"]) == metric ) unit = sorted(unit_counts, key=lambda value: (-unit_counts[value], value))[0] if unit_counts else "" else: connection = conn() try: row = connection.execute( "SELECT unit, count(*) n FROM apple_health_records WHERE metric=? GROUP BY unit ORDER BY n DESC LIMIT 1", (metric,), ).fetchone() finally: connection.close() unit = str(row["unit"] or "") if row else "" if spec.aggregation == "sum": return (unit + " / Tag*").strip() return unit def summary() -> dict[str, Any]: all_rows = _load_rows() rows_by_metric: dict[str, list[sqlite3.Row]] = defaultdict(list) for row in all_rows: rows_by_metric[str(row["metric"])].append(row) metrics = sorted(rows_by_metric) canonical_count = 0 for metric, rows in rows_by_metric.items(): factors = _coarse_factors(rows, metric) if metric in SUM_METRICS else {} canonical_count += len(_deduplicated_rows(metric, rows, factors)) return { "analytics_version": ANALYTICS_VERSION, "raw_records": len(all_rows), "canonical_records": canonical_count, "metric_count": len(metrics), "quality": {metric: coverage_summary(metric, _rows=rows_by_metric[metric]) for metric in metrics}, } if __name__ == "__main__": import json print(json.dumps(summary(), ensure_ascii=False, indent=2))