"""Read-only, non-causal Explorer comparison contract for Dashboard V5.""" from __future__ import annotations import hashlib import json import math import sqlite3 from datetime import date, datetime, timedelta from typing import Any, Callable from zoneinfo import ZoneInfo CONTRACT_VERSION = "health.explorer_comparison.v1" LOCAL_TZ = ZoneInfo("Europe/Zurich") MAX_DAYS = 366 MAX_HEALTH_SERIES = 2 MAX_INFLUENCES = 2 MAX_EVENTS = 800 HEALTH_METRICS = ( ("apple.sleep", "Gesamtschlaf"), ("apple.resting_heart_rate", "Ruhepuls"), ("apple.hrv", "HRV"), ("symptom.total", "Dokumentierte Beschwerden"), ("symptom.aphthae", "Dokumentierte Aphten/Aphthen"), ("apple.respiratory_rate", "Atemfrequenz"), ("apple.active_energy", "Aktive Energie"), ("apple.exercise_time", "Trainingsminuten"), ("apple.blood_pressure.systolic", "Blutdruck systolisch"), ("apple.blood_pressure.diastolic", "Blutdruck diastolisch"), ) INFLUENCES = ( ("nutrition.profile", "Ernährungstage", "nutrition"), ( "nutrition.histamine", "Dokumentierte histaminbezogene Ernährungseinträge", "nutrition_exposure", ), ("event.medication", "Medikamente", "event"), ("event.training", "Training", "event"), ("event.sauna", "Sauna/Erholung", "event"), ("event.symptom", "Beschwerden", "event"), ("event.other", "Sonstige Ereignisse", "event"), ) PRESETS = ( { "id": "sleep_recovery", "label": "Schlaf und Erholung", "metrics": ["apple.sleep", "apple.resting_heart_rate"], "influences": ["event.training", "event.sauna"], }, { "id": "nutrition_symptoms", "label": "Ernährung und Beschwerden", "metrics": ["symptom.total"], "influences": ["nutrition.profile", "event.medication"], }, { "id": "activity_recovery", "label": "Aktivität und Erholung", "metrics": ["apple.exercise_time", "apple.sleep"], "influences": ["event.training", "event.sauna"], }, ) NUTRITION_COMPONENTS = ( ("nutrition.energy", "Dokumentierte Energie", "kcal", "kcal"), ("nutrition.protein", "Protein", "g", "protein_g"), ("nutrition.carbohydrate", "Kohlenhydrate", "g", "carb_g"), ("nutrition.fat", "Fett", "g", "fat_g"), ("nutrition.meal_items", "Dokumentierte Einträge", "Einträge", "item_count"), ) AFFIRMATIVE_MEDICATION_EVENTS = { "administered", "verabreicht", "planmaessige injektion", "planmässige injektion", "planmäßige injektion", "erste injektion / therapiebeginn", } def _table_exists(connection: sqlite3.Connection, name: str) -> bool: return connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,) ).fetchone() is not None def _table_columns(connection: sqlite3.Connection, name: str) -> set[str]: return {str(row[1]) for row in connection.execute(f"PRAGMA table_info({name})")} def _number(value: Any) -> float | int | None: try: result = float(value) except (TypeError, ValueError): return None if not math.isfinite(result): return None return int(result) if result.is_integer() else round(result, 3) def _text(value: Any, maximum: int = 120) -> str: text = " ".join(str(value or "").split()) if any(marker in text.casefold() for marker in ("/home/", "/tmp/", "traceback", ".db")): return "" return text[:maximum] def _local_timestamp(value: Any, fallback_day: Any = None) -> tuple[str, str | None]: raw = str(value or "").strip() if not raw and fallback_day: raw = str(fallback_day) if not raw: raise ValueError("invalid_event_time") if len(raw) == 10: observed = date.fromisoformat(raw) return observed.isoformat(), None parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=LOCAL_TZ) local = parsed.astimezone(LOCAL_TZ) return local.date().isoformat(), local.strftime("%H:%M") def _opaque_event_id(*parts: Any) -> str: payload = "\x1f".join(str(part or "") for part in parts).encode("utf-8") return "cmp_" + hashlib.sha256(b"health-comparison-event-v1\0" + payload).hexdigest()[:24] def _detail(label: str, value: Any, unit: str = "") -> dict[str, str] | None: number = _number(value) if number is not None: return {"label": label, "value": f"{number:g}{(' ' + unit) if unit else ''}"} text = _text(value) return {"label": label, "value": text} if text else None def catalog(availability: dict[str, int] | None = None) -> dict[str, Any]: availability = availability or {} return { "contract_version": CONTRACT_VERSION, "timezone": "Europe/Zurich", "limits": { "health_series": MAX_HEALTH_SERIES, "influences": MAX_INFLUENCES, "days": MAX_DAYS, }, "health_metrics": [ {"id": metric_id, "label": label, "available": availability.get(metric_id, 0) > 0} for metric_id, label in HEALTH_METRICS ], "influences": [ {"id": influence_id, "label": label, "kind": kind} for influence_id, label, kind in INFLUENCES ], "nutrition_components": [ { "id": component_id, "label": label, "unit": unit, "source": "YAZIO-Import", "aggregation": "daily_sum" if field != "item_count" else "daily_count", } for component_id, label, unit, field in NUTRITION_COMPONENTS ], "presets": list(PRESETS), "statement": ( "Explorative zeitliche Gegenüberstellung. Zeitliche Nähe ist kein Nachweis " "von Ursache, Wirkung oder persönlicher Verträglichkeit." ), "stress_statement": ( "Diese Werte können zeitliche Muster zeigen, messen aber nicht automatisch " "psychischen Stress." ), } def _parse_selection(raw: str, allowed: set[str], maximum: int, required: bool = False) -> list[str]: values = [item for item in str(raw or "").split(",") if item] if len(values) != len(set(values)) or len(values) > maximum or (required and not values): raise ValueError("comparison_selection_invalid") if any(value not in allowed for value in values): raise ValueError("comparison_selection_invalid") return values def _nutrition_rows(connection: sqlite3.Connection, start: date, end: date) -> list[sqlite3.Row]: if not _table_exists(connection, "nutrition_daily_summary_v2"): return [] return list( connection.execute( """SELECT datum,kcal,protein_g,carb_g,fat_g,item_count, histamine_max,histamine_unknown_count FROM nutrition_daily_summary_v2 WHERE datum>=? AND datum<=? AND item_count>0 ORDER BY datum LIMIT ?""", (start.isoformat(), end.isoformat(), MAX_EVENTS + 1), ) ) def _nutrition_profile_events(connection: sqlite3.Connection, start: date, end: date) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: events: list[dict[str, Any]] = [] components = [ {"id": component_id, "label": label, "unit": unit, "source": "YAZIO-Import", "aggregation": "daily_sum" if field != "item_count" else "daily_count", "points": []} for component_id, label, unit, field in NUTRITION_COMPONENTS ] for row in _nutrition_rows(connection, start, end): day = date.fromisoformat(str(row["datum"])).isoformat() details = [] for component, (_, label, unit, field) in zip(components, NUTRITION_COMPONENTS, strict=True): value = _number(row[field]) if value is not None: component["points"].append({"date": day, "value": value}) item = _detail(label, value, unit) if item: details.append(item) item_count = int(row["item_count"] or 0) unknown = max(0, int(row["histamine_unknown_count"] or 0)) mapped = max(0, item_count - unknown) coverage = round(mapped / item_count, 4) if item_count else None details.extend( [ {"label": "Mapping-Abdeckung", "value": f"{round(coverage * 100)} %"} if coverage is not None else {"label": "Mapping-Abdeckung", "value": "Unbekannt"}, {"label": "Noch ungeprüfte Einträge", "value": str(unknown)}, ] ) events.append( { "id": _opaque_event_id("nutrition.profile", day), "influence_id": "nutrition.profile", "date": day, "time": None, "category": "Ernährung", "title": "Ernährung dokumentiert", "details": details, "source": "YAZIO-Import", "data_status": "partially_documented", "review_status": "complete_mapping" if unknown == 0 else "review_open", "mapping_coverage": coverage, "mapped_entries": mapped, "documented_entries": item_count, } ) return events, components def _histamine_events(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]: events = [] for row in _nutrition_rows(connection, start, end): item_count = int(row["item_count"] or 0) unknown = max(0, int(row["histamine_unknown_count"] or 0)) mapped = max(0, item_count - unknown) day = date.fromisoformat(str(row["datum"])).isoformat() maximum = _number(row["histamine_max"]) coverage = round(mapped / item_count, 4) if item_count else None details = [ {"label": "Bestätigt zugeordnete Einträge", "value": str(mapped)}, {"label": "Noch ungeprüfte Einträge", "value": str(unknown)}, {"label": "Mapping-Abdeckung", "value": f"{round(coverage * 100)} %" if coverage is not None else "Unbekannt"}, ] if maximum is not None: details.append({"label": "Höchste vorhandene Zuordnungskategorie", "value": f"Kategorie {maximum:g}"}) events.append( { "id": _opaque_event_id("nutrition.histamine", day), "influence_id": "nutrition.histamine", "date": day, "time": None, "category": ( "Histaminbezogene Ernährungseinträge" if mapped > 0 else "Offene Histaminzuordnung" ), "title": ( "Dokumentierte histaminbezogene Ernährungseinträge" if mapped > 0 else "Ernährung dokumentiert – Zuordnung offen" ), "details": details, "source": "Lokale, bestehende Lebensmittelzuordnungen", "data_status": "documented_mapping" if mapped > 0 else "unmapped", "review_status": "complete_mapping" if unknown == 0 else "review_open", "mapping_coverage": coverage, "mapped_entries": mapped, "documented_entries": item_count, } ) return events def _medication_events(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]: if not _table_exists(connection, "medication_administrations"): return [] events = [] columns = _table_columns(connection, "medication_administrations") expressions = [name if name in columns else f"NULL AS {name}" for name in ("datum", "occurred_at", "medication_name", "dose", "route", "event_type", "source", "corrected_target_status")] latest_clause = " AND NOT EXISTS (SELECT 1 FROM medication_administrations correction WHERE correction.corrects_event_id=medication_administrations.id)" if "corrects_event_id" in columns else "" rows = connection.execute( f"""SELECT {','.join(expressions)} FROM medication_administrations WHERE datum>=? AND datum<=?{latest_clause} ORDER BY datum,id LIMIT ?""", (start.isoformat(), end.isoformat(), MAX_EVENTS + 1), ) for row in rows: event_type = _text(row["event_type"], 80).casefold() if event_type in {"corrected", "korrigiert", "correction"}: event_type = _text(row["corrected_target_status"], 80).casefold() if event_type not in AFFIRMATIVE_MEDICATION_EVENTS: continue day, time = _local_timestamp(row["occurred_at"], row["datum"]) if day < start.isoformat() or day > end.isoformat(): continue details = [item for item in ( _detail("Dokumentierte Dosis", row["dose"]), _detail("Applikationsweg", row["route"]), ) if item] title = _text(row["medication_name"], 80) or "Medikament dokumentiert" events.append( { "id": _opaque_event_id("medication", day, time, title), "influence_id": "event.medication", "date": day, "time": time, "category": "Medikament", "title": title, "details": details, "source": "Dokumentierte Medikamentengabe", "data_status": "documented", "review_status": "none", "mapping_coverage": None, } ) return events def _capture_events(connection: sqlite3.Connection, kind: str, start: date, end: date) -> list[dict[str, Any]]: if not _table_exists(connection, "capture_entries"): return [] events = [] for row in connection.execute( """SELECT occurred_at,payload_json,source,created_at FROM capture_entries WHERE capture_type='event' AND status='active' AND substr(occurred_at,1,10)>=? AND substr(occurred_at,1,10)<=? ORDER BY occurred_at,created_at LIMIT ?""", ((start - timedelta(days=1)).isoformat(), (end + timedelta(days=1)).isoformat(), MAX_EVENTS + 1), ): try: payload = json.loads(str(row["payload_json"] or "{}")) except json.JSONDecodeError: continue if payload.get("event_kind") != kind: continue day, time = _local_timestamp(row["occurred_at"]) if day < start.isoformat() or day > end.isoformat(): continue details = [] fields = ( (("Dauer", "duration_minutes", "min"), ("Saunagänge", "rounds", ""), ("Temperatur", "temperature_c", "°C"), ("Abkühlung/Kaltphase", "cooling", ""), ("Flüssigkeitszufuhr", "hydration_ml", "ml")) if kind == "sauna" else (("Aktivitätsart", "activity_type", ""), ("Dauer", "duration_minutes", "min"), ("Aktive Energie", "active_kcal", "kcal"), ("Distanz", "distance_km", "km")) ) for label, key, unit in fields: item = _detail(label, payload.get(key), unit) if item: details.append(item) note = _detail("Notiz", payload.get("note")) if note: details.append(note) title = _text(payload.get("title"), 80) or ("Sauna" if kind == "sauna" else "Training") events.append( { "id": _opaque_event_id(kind, day, time, title), "influence_id": f"event.{kind}", "date": day, "time": time, "category": "Sauna/Erholung" if kind == "sauna" else "Training", "title": title, "details": details, "source": "Manuell in Dashboard V5 erfasst", "data_status": "documented", "review_status": "none", "mapping_coverage": None, } ) return events def _health_event_rows(connection: sqlite3.Connection, start: date, end: date) -> list[sqlite3.Row]: if not _table_exists(connection, "health_events"): return [] columns = _table_columns(connection, "health_events") expressions = [ name if name in columns else f"NULL AS {name}" for name in ("date", "category", "parameter", "value", "unit", "source", "notes", "occurred_at") ] return list( connection.execute( f"""SELECT {','.join(expressions)} FROM health_events WHERE date>=? AND date<=? ORDER BY date,id LIMIT ?""", (start.isoformat(), end.isoformat(), MAX_EVENTS + 1), ) ) def _legacy_health_events(connection: sqlite3.Connection, kind: str, start: date, end: date) -> list[dict[str, Any]]: events = [] training = {"sport", "training", "physical_load", "besondere belastung", "manual_training"} sauna = {"sauna", "heat", "hitze", "sauna_recovery"} other = {"ereignis", "lifestyle", "stress", "sleep_disruption", "infection", "travel"} allowed = training if kind == "training" else sauna if kind == "sauna" else other for row in _health_event_rows(connection, start, end): if str(row["source"] or "") == "dashboard_v5_manual_capture": continue values = {str(row["category"] or "").strip().casefold(), str(row["parameter"] or "").strip().casefold()} if not values & allowed: continue day, time = _local_timestamp(row["occurred_at"], row["date"]) title = _text(row["parameter"], 80) or ("Training" if kind == "training" else "Sauna" if kind == "sauna" else "Ereignis") details = [item for item in ( _detail("Dokumentierter Wert", row["value"], _text(row["unit"], 20)), _detail("Notiz", row["notes"]), ) if item] events.append( { "id": _opaque_event_id(kind, day, time, title), "influence_id": f"event.{kind}" if kind in {"training", "sauna"} else "event.other", "date": day, "time": time, "category": "Training" if kind == "training" else "Sauna/Erholung" if kind == "sauna" else "Sonstiges Ereignis", "title": title, "details": details, "source": "Dokumentiertes Gesundheitsereignis", "data_status": "documented", "review_status": "none", "mapping_coverage": None, } ) return events def _symptom_events(connection: sqlite3.Connection, start: date, end: date) -> list[dict[str, Any]]: if not _table_exists(connection, "symptom_log"): return [] events = [] columns = _table_columns(connection, "symptom_log") expressions = [name if name in columns else f"NULL AS {name}" for name in ("datum", "occurred_at", "symptom", "schwergrad", "duration_minutes")] for row in connection.execute( f"""SELECT {','.join(expressions)} FROM symptom_log WHERE datum>=? AND datum<=? ORDER BY datum,id LIMIT ?""", (start.isoformat(), end.isoformat(), MAX_EVENTS + 1), ): severity = _number(row["schwergrad"]) if severity is None or severity <= 0: continue day, time = _local_timestamp(row["occurred_at"], row["datum"]) if day < start.isoformat() or day > end.isoformat(): continue details = [item for item in ( _detail("Dokumentierte Stärke", severity), _detail("Dauer", row["duration_minutes"], "min"), ) if item] title = _text(row["symptom"], 80) or "Beschwerde dokumentiert" events.append( { "id": _opaque_event_id("symptom", day, time, title), "influence_id": "event.symptom", "date": day, "time": time, "category": "Beschwerde", "title": title, "details": details, "source": "Symptomtagebuch", "data_status": "documented", "review_status": "none", "mapping_coverage": None, } ) return events def _influence(connection: sqlite3.Connection, influence_id: str, start: date, end: date) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: if influence_id == "nutrition.profile": return _nutrition_profile_events(connection, start, end) if influence_id == "nutrition.histamine": return _histamine_events(connection, start, end), [] if influence_id == "event.medication": return _medication_events(connection, start, end), [] if influence_id == "event.training": return _capture_events(connection, "training", start, end) + _legacy_health_events(connection, "training", start, end), [] if influence_id == "event.sauna": return _capture_events(connection, "sauna", start, end) + _legacy_health_events(connection, "sauna", start, end), [] if influence_id == "event.symptom": return _symptom_events(connection, start, end), [] if influence_id == "event.other": return _legacy_health_events(connection, "other", start, end), [] raise ValueError("comparison_selection_invalid") def _source_label(payload: dict[str, Any]) -> str: source = str(payload.get("source", {}).get("type") or "") return { "apple_health": "Apple Health", "symptom_log": "Symptomtagebuch", }.get(source, "Dokumentierte Quelle") def build( connection: sqlite3.Connection, params: dict[str, str], load_series: Callable[[str, date, date], dict[str, Any]], ) -> dict[str, Any]: start = date.fromisoformat(params["from"]) end = date.fromisoformat(params["to"]) if end < start or (end - start).days + 1 > MAX_DAYS: raise ValueError("comparison_range_invalid") metric_ids = _parse_selection(params.get("metrics", ""), {item[0] for item in HEALTH_METRICS}, MAX_HEALTH_SERIES) influence_ids = _parse_selection(params.get("influences", ""), {item[0] for item in INFLUENCES}, MAX_INFLUENCES) if not metric_ids and not influence_ids: raise ValueError("comparison_selection_invalid") labels = dict(HEALTH_METRICS) series, errors = [], [] for metric_id in metric_ids: try: payload = load_series(metric_id, start, end) series.append( { "id": metric_id, "label": payload.get("label") or labels[metric_id], "unit": payload.get("unit") or "", "source": _source_label(payload), "aggregation": payload.get("aggregation_rule", {}).get("id", "daily_documented"), "coverage": payload.get("coverage") or {}, "missingness": "observed_only_no_interpolation", "data_status": "no_data" if not payload.get("points") else "documented", "points": payload.get("points") or [], } ) except Exception: errors.append({"kind": "health_series", "id": metric_id, "code": "series_load_failed"}) events, nutrition_components = [], [] for influence_id in influence_ids: try: loaded, components = _influence(connection, influence_id, start, end) events.extend(loaded) nutrition_components.extend(components) except Exception: errors.append({"kind": "influence", "id": influence_id, "code": "influence_load_failed"}) if len(events) > MAX_EVENTS: raise ValueError("comparison_response_too_large") events.sort(key=lambda item: (item["date"], item.get("time") or "", item["id"])) point_maps = {item["id"]: {point["date"]: point for point in item["points"]} for item in series} for event in events: if event["influence_id"] not in {"event.training", "event.sauna"}: event["follow_up_values"] = [] continue following = (date.fromisoformat(event["date"]) + timedelta(days=1)).isoformat() event["follow_up_values"] = [ { "date": following, "label": item["label"], "value": point_maps[item["id"]][following]["value"], "unit": item["unit"], "source": item["source"], } for item in series if following in point_maps[item["id"]] and point_maps[item["id"]][following].get("value") is not None ] events_by_day: dict[str, list[dict[str, Any]]] = {} for event in events: events_by_day.setdefault(event["date"], []).append(event) days = [] cursor = start while cursor <= end: day = cursor.isoformat() values = [] for item in series: point = point_maps[item["id"]].get(day) values.append( { "id": item["id"], "label": item["label"], "unit": item["unit"], "source": item["source"], "value": point.get("value") if point else None, "data_status": "documented" if point and point.get("value") is not None else "missing", "quality": point.get("quality") if point else None, } ) days.append({"date": day, "values": values, "events": events_by_day.get(day, [])}) cursor += timedelta(days=1) return { "contract_version": CONTRACT_VERSION, "timezone": "Europe/Zurich", "period": {"from": start.isoformat(), "to": end.isoformat()}, "selected": {"health_metrics": metric_ids, "influences": influence_ids}, "health_series": series, "nutrition_components": nutrition_components, "events": events, "days": days, "component_errors": errors, "partial": bool(errors), "statement": "Explorative zeitliche Gegenüberstellung; keine Kausalitätsaussage.", "missingness": "Fehlende Tage bleiben unbekannt und werden nicht als Null ergänzt.", "review_statement": "Ungeprüfte Ernährungseinträge werden nicht automatisch histaminbezogen klassifiziert.", }