"""Metric Catalog v2: explicit source evidence, search aliases, and inventories.""" from __future__ import annotations import json import math import re import sqlite3 import unicodedata from dataclasses import asdict, dataclass from datetime import date, datetime from typing import Any from zoneinfo import ZoneInfo from apple_health_analytics import normalize_value from dashboard_v5.data_provider import SYMPTOM_LABELS, parse_score from dashboard_v5.lab_registry import EXACT_LAB_NUMBER, LAB_ALLOWLIST from dashboard_v5.nutrition_contract import NUTRIENT_CONTRACTS, normalize_nutrient @dataclass(frozen=True) class MetricV2: id: str label: str aliases: tuple[str, ...] category: str source: str unit: str value_type: str precision: int aggregation: str expected_frequency: str overlay_group: str baseline_rule: dict[str, Any] reference_policy: str search_terms: tuple[str, ...] drill_down_target: str correlation_enabled: bool allowed_lags: tuple[int, ...] privacy_class: str source_identifier: str parser_contract: str panel_id: str | None = None @dataclass(frozen=True) class AppleSourceSpec: metric_ids: tuple[str, ...] unit: str aggregation: str parser_contract: str = "apple_health_analytics_v2" @dataclass(frozen=True) class LabCatalogSpec: metric_id: str aliases: tuple[str, ...] unit: str APPLE_SOURCE_SPECS: dict[str, AppleSourceSpec] = { "sleep_analysis": AppleSourceSpec( ( "apple.sleep", "apple.sleep.core", "apple.sleep.deep", "apple.sleep.rem", "apple.sleep.awake", ), "h", "sum", "apple_health_sleep_payload_v1", ), "resting_heart_rate": AppleSourceSpec(("apple.resting_heart_rate",), "bpm", "avg"), "heart_rate_variability": AppleSourceSpec(("apple.hrv",), "ms", "avg"), "step_count": AppleSourceSpec(("apple.steps",), "Schritte", "sum"), "walking_running_distance": AppleSourceSpec(("apple.distance",), "km", "sum"), "active_energy": AppleSourceSpec(("apple.active_energy",), "kcal", "sum"), "apple_exercise_time": AppleSourceSpec(("apple.exercise_time",), "min", "sum"), "blood_oxygen_saturation": AppleSourceSpec(("apple.oxygen_saturation",), "%", "avg"), "respiratory_rate": AppleSourceSpec(("apple.respiratory_rate",), "Atemzüge/min", "avg"), "physical_effort": AppleSourceSpec(("apple.physical_effort",), "kcal/h/kg", "avg"), "weight_body_mass": AppleSourceSpec(("apple.weight",), "kg", "last"), "body_mass_index": AppleSourceSpec(("apple.bmi",), "kg/m²", "last"), "blood_pressure": AppleSourceSpec( ("panel.blood_pressure", "apple.blood_pressure.systolic", "apple.blood_pressure.diastolic"), "mmHg", "last", "apple_health_blood_pressure_panel_v1", ), } LAB_CATALOG_SPECS: dict[str, LabCatalogSpec] = { "CRP": LabCatalogSpec("lab.crp", ("crp", "c-reaktives protein"), "mg/L"), "D-Dimer": LabCatalogSpec("lab.d_dimer", ("d-dimer", "ddimer"), "µg/L"), "Fibrinogen": LabCatalogSpec("lab.fibrinogen", ("fibrinogen",), "g/L"), "Faktor VIII": LabCatalogSpec("lab.factor_viii", ("faktor viii", "faktor 8"), "%"), "Thrombozyten": LabCatalogSpec("lab.platelets", ("thrombozyten", "blutplättchen"), "Tsd/µL"), "Leukozyten": LabCatalogSpec("lab.leukocytes", ("leukozyten", "weisse blutkörperchen"), "Tsd/µL"), "Ferritin": LabCatalogSpec("lab.ferritin", ("ferritin", "eisenspeicher"), "ng/mL"), } MAX_INVENTORY_IDENTIFIERS = 1000 MAX_BLOOD_PRESSURE_SOURCE_ROWS = 10000 MAX_APPLE_METRIC_SOURCE_ROWS = 100000 MAX_DAILY_SOURCE_ROWS = 30000 LOCAL_TZ = ZoneInfo("Europe/Zurich") APPLE_ANALYTICS_UNITS = { "sleep_analysis": "hr", "resting_heart_rate": "bpm", "heart_rate_variability": "ms", "step_count": "count", "walking_running_distance": "km", "active_energy": "kcal", "apple_exercise_time": "min", "blood_oxygen_saturation": "%", "respiratory_rate": "breaths/min", "physical_effort": "kcal/h/kg", "weight_body_mass": "kg", "body_mass_index": "kg/m²", } class SourceInventoryLimitError(ValueError): """Raised instead of silently truncating source inventory evidence.""" def _baseline(window: int, minimum: int = 3) -> dict[str, Any]: return { "method": "rolling_median_prior_observations", "window_days": window, "minimum_observations": minimum, "future_values": "excluded", } def _metric( metric_id: str, label: str, aliases: tuple[str, ...], category: str, source: str, unit: str, value_type: str, precision: int, aggregation: str, frequency: str, overlay: str, baseline: dict[str, Any], reference: str, search_terms: tuple[str, ...], drill_down: str, correlation: bool, lags: tuple[int, ...], source_identifier: str, parser: str, panel_id: str | None = None, ) -> MetricV2: return MetricV2( metric_id, label, aliases, category, source, unit, value_type, precision, aggregation, frequency, overlay, baseline, reference, search_terms, drill_down, correlation, lags, "health_sensitive", source_identifier, parser, panel_id, ) METRICS_V2: tuple[MetricV2, ...] = ( _metric("panel.blood_pressure", "Blutdruck", ("Blutdruck", "BD", "RR", "Vitalwerte"), "vital_signs", "apple_health", "mmHg", "panel", 0, "panel", "intermittent", "blood_pressure_panel", {}, "paired_panel", ("systolisch", "diastolisch", "vitalwerte"), "explorer", False, (), "blood_pressure", "apple_health_blood_pressure_panel_v1"), _metric("apple.blood_pressure.systolic", "Blutdruck systolisch", ("systolisch", "Vitalwerte"), "vital_signs", "apple_health", "mmHg", "number", 0, "last", "intermittent", "blood_pressure", _baseline(30), "paired_panel", ("blutdruck", "bd", "rr", "vitalwerte"), "explorer", False, (), "blood_pressure", "apple_health_blood_pressure_panel_v1", "panel.blood_pressure"), _metric("apple.blood_pressure.diastolic", "Blutdruck diastolisch", ("diastolisch", "Vitalwerte"), "vital_signs", "apple_health", "mmHg", "number", 0, "last", "intermittent", "blood_pressure", _baseline(30), "paired_panel", ("blutdruck", "bd", "rr", "vitalwerte"), "explorer", False, (), "blood_pressure", "apple_health_blood_pressure_panel_v1", "panel.blood_pressure"), _metric("apple.sleep", "Schlaf", ("Schlaf", "Schlafdauer"), "sleep", "apple_health", "h", "number", 2, "sum", "daily", "duration_hours", _baseline(14), "personal_baseline", ("erholung",), "explorer", True, (0, 1, 2, 3), "sleep_analysis", "apple_health_sleep_payload_v1"), _metric("apple.sleep.core", "Kernschlaf", ("Kernschlaf",), "sleep", "apple_health", "h", "number", 2, "sum", "daily", "duration_hours", _baseline(14), "personal_baseline", ("schlafphase", "erholung"), "explorer", False, (), "sleep_analysis", "apple_health_sleep_payload_v1"), _metric("apple.sleep.deep", "Tiefschlaf", ("Tiefschlaf",), "sleep", "apple_health", "h", "number", 2, "sum", "daily", "duration_hours", _baseline(14), "personal_baseline", ("schlafphase", "erholung"), "explorer", False, (), "sleep_analysis", "apple_health_sleep_payload_v1"), _metric("apple.sleep.rem", "REM-Schlaf", ("REM", "REM Schlaf"), "sleep", "apple_health", "h", "number", 2, "sum", "daily", "duration_hours", _baseline(14), "personal_baseline", ("schlafphase", "erholung"), "explorer", False, (), "sleep_analysis", "apple_health_sleep_payload_v1"), _metric("apple.sleep.awake", "Wachzeit im Schlafzeitraum", ("Wachzeit",), "sleep", "apple_health", "h", "number", 2, "sum", "daily", "duration_hours", _baseline(14), "personal_baseline", ("schlafphase", "erholung"), "explorer", False, (), "sleep_analysis", "apple_health_sleep_payload_v1"), _metric("apple.resting_heart_rate", "Ruhepuls", ("Ruhepuls", "Puls", "Herzfrequenz"), "cardiovascular", "apple_health", "bpm", "number", 1, "avg", "daily", "heart_rate_bpm", _baseline(30), "personal_baseline", ("ruheherzfrequenz",), "explorer", True, (0, 1, 2, 3), "resting_heart_rate", "apple_health_analytics_v2"), _metric("apple.hrv", "HRV", ("HRV", "Herzratenvariabilität"), "cardiovascular", "apple_health", "ms", "number", 1, "avg", "daily", "duration_ms", _baseline(30), "personal_baseline", ("heart rate variability",), "explorer", True, (0, 1, 2, 3), "heart_rate_variability", "apple_health_analytics_v2"), _metric("apple.steps", "Schritte", ("Schritte",), "activity", "apple_health", "Schritte", "integer", 0, "sum", "daily", "count_steps", _baseline(14), "personal_baseline", ("gehen", "aktivität"), "explorer", True, (0, 1, 2, 3), "step_count", "apple_health_analytics_v2"), _metric("apple.distance", "Geh- und Laufdistanz", ("Distanz", "Gehdistanz", "Laufdistanz"), "activity", "apple_health", "km", "number", 2, "sum", "daily", "distance_km", _baseline(14), "personal_baseline", ("walking running distance", "aktivität"), "explorer", True, (0, 1, 2, 3), "walking_running_distance", "apple_health_analytics_v2"), _metric("apple.active_energy", "Aktivenergie", ("Aktivenergie", "aktive Energie"), "activity", "apple_health", "kcal", "number", 1, "sum", "daily", "energy_kcal", _baseline(14), "personal_baseline", ("kalorien", "aktivität"), "explorer", True, (0, 1, 2, 3), "active_energy", "apple_health_analytics_v2"), _metric("apple.exercise_time", "Trainingsminuten", ("Trainingsminuten", "Apple Trainingszeit"), "activity", "apple_health", "min", "number", 0, "sum", "daily", "duration_minutes", _baseline(14), "personal_baseline", ("training", "aktivität"), "explorer", True, (0, 1, 2, 3), "apple_exercise_time", "apple_health_analytics_v2"), _metric("apple.oxygen_saturation", "Sauerstoffsättigung", ("Sauerstoffsättigung", "SpO2", "Blutsauerstoff", "Vitalwerte"), "respiratory", "apple_health", "%", "number", 1, "avg", "daily", "percentage", _baseline(30), "personal_baseline", ("oxygen saturation", "vitalwerte"), "explorer", False, (), "blood_oxygen_saturation", "apple_health_analytics_v2"), _metric("apple.respiratory_rate", "Atemfrequenz", ("Atemfrequenz", "Atemzüge", "Vitalwerte"), "respiratory", "apple_health", "Atemzüge/min", "number", 1, "avg", "daily", "respiratory_rate", _baseline(30), "personal_baseline", ("respiratory rate", "vitalwerte"), "explorer", False, (), "respiratory_rate", "apple_health_analytics_v2"), _metric("apple.physical_effort", "Körperliche Anstrengung", ("körperliche Anstrengung", "Anstrengung"), "activity", "apple_health", "kcal/h/kg", "number", 2, "avg", "daily", "physical_effort", _baseline(14), "personal_baseline", ("physical effort",), "explorer", False, (), "physical_effort", "apple_health_analytics_v2"), _metric("apple.weight", "Gewicht", ("Gewicht", "Körpergewicht", "Vitalwerte"), "body_measurement", "apple_health", "kg", "number", 2, "last", "intermittent", "body_mass", _baseline(90), "personal_baseline", ("body mass", "vitalwerte"), "explorer", False, (), "weight_body_mass", "apple_health_analytics_v2"), _metric("apple.bmi", "BMI", ("BMI", "Body-Mass-Index", "Vitalwerte"), "body_measurement", "apple_health", "kg/m²", "number", 1, "last", "intermittent", "body_mass_index", _baseline(90), "personal_baseline", ("körpermasseindex", "vitalwerte"), "explorer", False, (), "body_mass_index", "apple_health_analytics_v2"), _metric("symptom.total", "Symptome total", ("Symptome", "Beschwerden"), "symptoms", "symptom_log", "0–21", "integer", 0, "sum", "daily", "symptom_score", _baseline(14), "none", ("beschwerdescore",), "day", True, (0, 1, 2, 3), "daily_quick_score", "complete_seven_dimension_symptom_contract_v1"), _metric("symptom.aphthae", "Aphten/Aphthen", ("Aphten", "Aphthen", "Mundulzera"), "symptoms", "symptom_log", "0–3", "integer", 0, "avg", "daily", "symptom_dimension", _baseline(14), "none", ("aphthen mundulzera",), "day", False, (), "Aphthen/Mundulzera", "daily_quick_score_dimension_v1"), *tuple( _metric( contract.metric_id, contract.label, (contract.label, "Ernährung"), "nutrition", "nutrition_daily_summary_v2", contract.unit, "number", contract.precision, "sum", "daily", f"nutrition_{contract.unit}", {}, "documented_source_only", ("yazio", "nährstoff"), "nutrition", True, (0, 1, 2, 3), contract.key, "yazio_nutrients_v1", ) for contract in NUTRIENT_CONTRACTS.values() ), _metric("nutrition.histamine", "Histamin-Zuordnungsindex", ("Histamin", "SIGHi-Zuordnungsindex"), "nutrition", "nutrition_daily_summary_v2", "Index", "number", 1, "sum", "daily", "histamine_assignment_index", {}, "none", ("ernährung", "sighi"), "nutrition", True, (0, 1, 2, 3), "histamine_score", "complete_sighi_assignment_summary_v2"), _metric("nutrition.sighi_max", "Höchster SIGHi-Wert", ("SIGHi Maximum", "Histamin Maximum"), "nutrition", "nutrition_daily_summary_v2", "SIGHi 0–3", "number", 0, "avg", "daily", "sighi_max", {}, "none", ("ernährung", "histamin"), "nutrition", True, (0, 1, 2, 3), "histamine_max", "documented_sighi_max_v1"), _metric("nutrition.mapping_coverage", "Ernährungs-Mappingabdeckung", ("Mappingabdeckung", "Zuordnungsabdeckung"), "nutrition", "nutrition_daily_summary_v2", "%", "number", 1, "avg", "daily", "percentage", {}, "none", ("ernährung", "histamin"), "nutrition", True, (0, 1, 2, 3), "mapping_coverage", "nutrition_mapping_coverage_v1"), *tuple( _metric(spec.metric_id, parameter, spec.aliases, "laboratory", "laborwerte", spec.unit, "number", 2, "observation", "intermittent", f"laboratory_{spec.unit}", {}, "observation_specific_verified_original", spec.aliases, "doctor_labs", False, (), parameter, "verified_original_lab_allowlist_v1") for parameter, spec in LAB_CATALOG_SPECS.items() ), ) BY_ID_V2 = {metric.id: metric for metric in METRICS_V2} def _normalize(value: str) -> str: decomposed = unicodedata.normalize("NFKD", value.casefold()) asciiish = "".join(char for char in decomposed if not unicodedata.combining(char)) asciiish = asciiish.replace("ß", "ss") return " ".join(re.sub(r"[^a-z0-9]+", " ", asciiish).split()) def _table_exists(connection: sqlite3.Connection, table: str) -> bool: return connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) ).fetchone() is not None def _table_columns(connection: sqlite3.Connection, table: str) -> set[str]: """Return allowlisted schema metadata without assuming the newest migration.""" if not _table_exists(connection, table): return set() return {str(row[1]) for row in connection.execute(f'PRAGMA table_info("{table}")')} def _observed_day(value: Any) -> date | None: try: timestamp = datetime.fromisoformat(str(value or "").replace("Z", "+00:00")) except ValueError: try: return date.fromisoformat(str(value or "")) except ValueError: return None if timestamp.tzinfo is None: timestamp = timestamp.replace(tzinfo=LOCAL_TZ) return timestamp.astimezone(LOCAL_TZ).date() def _today() -> date: return datetime.now(LOCAL_TZ).date() def _availability(connection: sqlite3.Connection, metric: MetricV2) -> dict[str, Any]: count = 0 today = _today() if metric.source == "apple_health" and _table_exists(connection, "apple_health_records"): if metric.source_identifier == "blood_pressure": rows = list(connection.execute( """SELECT unit,raw_json,start_date FROM apple_health_records WHERE metric='blood_pressure' LIMIT ?""", (MAX_BLOOD_PRESSURE_SOURCE_ROWS + 1,), )) if len(rows) > MAX_BLOOD_PRESSURE_SOURCE_ROWS: raise SourceInventoryLimitError("blood pressure source limit exceeded") for unit, raw_json, start_date in rows: try: payload = json.loads(str(raw_json or "")) component = ( "systolic" if metric.id.endswith(".systolic") else "diastolic" if metric.id.endswith(".diastolic") else None ) required = (component,) if component else ("systolic", "diastolic") values = tuple(float(payload[key]) for key in required) except (ValueError, TypeError, KeyError, json.JSONDecodeError): continue normalized_unit = re.sub(r"[^a-z]", "", str(unit or "").casefold()) observed = _observed_day(start_date) if ( normalized_unit == "mmhg" and all(math.isfinite(value) for value in values) and observed is not None and observed <= today ): count += 1 elif metric.id.startswith("apple.sleep."): phase = metric.id.rsplit(".", 1)[-1] rows = list( connection.execute( """SELECT raw_json,start_date FROM apple_health_records WHERE metric='sleep_analysis' LIMIT ?""", (MAX_APPLE_METRIC_SOURCE_ROWS + 1,), ) ) if len(rows) > MAX_APPLE_METRIC_SOURCE_ROWS: raise SourceInventoryLimitError("apple sleep source limit exceeded") for raw_json, start_date in rows: try: value = float(json.loads(str(raw_json or ""))[phase]) except (ValueError, TypeError, KeyError, json.JSONDecodeError): continue observed = _observed_day(start_date) if math.isfinite(value) and observed is not None and observed <= today: count += 1 else: rows = list(connection.execute( """SELECT value,unit,start_date FROM apple_health_records WHERE metric=? AND value IS NOT NULL LIMIT ?""", (metric.source_identifier, MAX_APPLE_METRIC_SOURCE_ROWS + 1), )) if len(rows) > MAX_APPLE_METRIC_SOURCE_ROWS: raise SourceInventoryLimitError("apple metric source limit exceeded") expected_unit = APPLE_ANALYTICS_UNITS[metric.source_identifier] for raw_value, raw_unit, start_date in rows: try: value = float(raw_value) canonical_value, canonical_unit = normalize_value( metric.source_identifier, value, raw_unit ) except (TypeError, ValueError): continue observed = _observed_day(start_date) if ( math.isfinite(canonical_value) and canonical_unit == expected_unit and observed is not None and observed <= today ): count += 1 elif metric.source == "laborwerte" and _table_exists(connection, "laborwerte"): pairs = [key for key, canonical in LAB_ALLOWLIST.items() if canonical[0] == metric.source_identifier] rows = list(connection.execute( """SELECT parameter_name,einheit,wert,abnahme_datum,befund_datum, reference_min,reference_max FROM laborwerte WHERE lower(trim(COALESCE(validierungsstatus,'')))='validiert' AND verified_against_original=1 AND reference_range_source='scanned_original' LIMIT ?""", (MAX_BLOOD_PRESSURE_SOURCE_ROWS + 1,), )) if len(rows) > MAX_BLOOD_PRESSURE_SOURCE_ROWS: raise SourceInventoryLimitError("laboratory source limit exceeded") valid_lab_groups: dict[tuple[str, str, date], int] = {} for row in rows: raw_parameter = _normalize(row[0]).replace(" ", "_") raw_unit = _normalize( str(row[1]).replace("µ", "u").replace("%", " percent ") ).replace(" ", "_") raw_value = ("" if row[2] is None else str(row[2])).strip() observed = _observed_day(row[3]) or _observed_day(row[4]) try: numeric_value = float(raw_value.replace(",", ".")) except ValueError: numeric_value = math.nan references = [ ("" if value is None else str(value)).strip() for value in (row[5], row[6]) ] valid_reference = any( EXACT_LAB_NUMBER.fullmatch(value) and math.isfinite(float(value.replace(",", "."))) for value in references ) if ( (raw_parameter, raw_unit) in pairs and EXACT_LAB_NUMBER.fullmatch(raw_value) and math.isfinite(numeric_value) and observed is not None and observed <= today and valid_reference ): key = (metric.source_identifier, metric.unit, observed) valid_lab_groups[key] = valid_lab_groups.get(key, 0) + 1 count = sum(1 for observations in valid_lab_groups.values() if observations == 1) elif metric.source == "symptom_log" and _table_exists(connection, "symptom_log"): rows = list(connection.execute( """SELECT datum,symptom,schwergrad FROM symptom_log WHERE kontext='daily_quick_score' AND datum<=? LIMIT ?""", (today.isoformat(), MAX_DAILY_SOURCE_ROWS + 1), )) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise SourceInventoryLimitError("symptom source limit exceeded") grouped: dict[str, dict[str, list[Any]]] = {} for raw_day, symptom, raw_score in rows: observed = _observed_day(raw_day) if observed is None or observed > today: continue grouped.setdefault(observed.isoformat(), {}).setdefault(str(symptom), []).append(raw_score) if metric.id == "symptom.aphthae": count = sum( 1 for values in grouped.values() if len(values.get(metric.source_identifier, [])) == 1 and parse_score(values[metric.source_identifier][0]) is not None ) else: count = sum( 1 for values in grouped.values() if set(values) == SYMPTOM_LABELS and all(len(values[label]) == 1 and parse_score(values[label][0]) is not None for label in SYMPTOM_LABELS) ) elif metric.source == "nutrition_daily_summary_v2" and _table_exists( connection, "nutrition_daily_summary_v2" ): if metric.source_identifier in NUTRIENT_CONTRACTS and _table_exists( connection, "nutrition_item_nutrients" ): rows = list( connection.execute( """SELECT i.datum,n.value,n.unit FROM nutrition_item_nutrients n JOIN nutrition_items i ON i.id=n.item_id WHERE n.nutrient_key=? AND i.datum<=? LIMIT ?""", ( metric.source_identifier, today.isoformat(), MAX_DAILY_SOURCE_ROWS + 1, ), ) ) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise SourceInventoryLimitError("nutrition nutrient source limit exceeded") count = len( { str(row[0]) for row in rows if _observed_day(row[0]) is not None and normalize_nutrient(metric.source_identifier, row[1], row[2]) is not None } ) else: column = { "histamine_score": "histamine_score", "histamine_max": "histamine_max", "mapping_coverage": "item_count", }.get(metric.source_identifier) if column is not None: required_columns = { "datum", column, "item_count", "histamine_unknown_count", "histamine_label", } if not required_columns.issubset( _table_columns(connection, "nutrition_daily_summary_v2") ): return {"status": "supported_no_data", "observations": 0} sql = f"SELECT datum,{column},item_count,histamine_unknown_count,histamine_label FROM nutrition_daily_summary_v2 WHERE item_count>0 AND datum<=? LIMIT ?" rows = list( connection.execute( sql, (today.isoformat(), MAX_DAILY_SOURCE_ROWS + 1) ) ) if len(rows) > MAX_DAILY_SOURCE_ROWS: raise SourceInventoryLimitError("nutrition source limit exceeded") for raw_day, raw_value, item_count, unknown, label in rows: observed = _observed_day(raw_day) try: value = float(raw_value) except (TypeError, ValueError): continue complete = int(item_count or 0) > 0 and int(unknown or 0) == 0 if metric.source_identifier == "histamine_score" and ( not complete or label not in {"classified", "green", "yellow", "orange", "red"} ): continue if metric.source_identifier == "histamine_max" and int(item_count or 0) <= int(unknown or 0): continue if observed is not None and math.isfinite(value): count += 1 return {"status": "available" if count else "supported_no_data", "observations": count} def public_metric(metric: MetricV2, availability: dict[str, Any] | None = None) -> dict[str, Any]: item = asdict(metric) item["aliases"] = list(metric.aliases) item["search_terms"] = list(metric.search_terms) item["allowed_lags"] = list(metric.allowed_lags) if availability is not None: item["availability"] = availability return item def public_catalog(connection: sqlite3.Connection | None = None) -> list[dict[str, Any]]: return [ public_metric(metric, _availability(connection, metric) if connection is not None else None) for metric in METRICS_V2 ] def catalog_search(connection: sqlite3.Connection, query: str) -> list[dict[str, Any]]: needle = _normalize(query) if len(needle) < 2: return [] ranked: list[tuple[int, int, MetricV2]] = [] for index, metric in enumerate(METRICS_V2): terms = [_normalize(metric.label), *(_normalize(value) for value in metric.aliases), *(_normalize(value) for value in metric.search_terms)] if needle in terms: rank = 0 elif any(term.startswith(needle) for term in terms): rank = 1 elif any(needle in term for term in terms): rank = 2 else: continue ranked.append((rank, index, metric)) return [public_metric(metric, _availability(connection, metric)) for _, _, metric in sorted(ranked)] def inventory_sources(connection: sqlite3.Connection) -> dict[str, list[dict[str, Any]]]: apple: list[dict[str, Any]] = [] if _table_exists(connection, "apple_health_records"): apple_rows = list(connection.execute( """SELECT metric,COUNT(*) observations,COUNT(value) numeric_observations FROM apple_health_records GROUP BY metric ORDER BY metric LIMIT ?""", (MAX_INVENTORY_IDENTIFIERS + 1,), )) if len(apple_rows) > MAX_INVENTORY_IDENTIFIERS: raise SourceInventoryLimitError("apple inventory limit exceeded") for row in apple_rows: identifier = str(row["metric"]) released = list(APPLE_SOURCE_SPECS.get(identifier, AppleSourceSpec((), "", "avg")).metric_ids) units = [ str(item[0] or "") for item in connection.execute( """SELECT DISTINCT unit FROM apple_health_records WHERE metric=? ORDER BY unit LIMIT 100""", (identifier,), ) ] apple.append({ "identifier": identifier, "units": units, "observations": int(row["observations"]), "numeric_observations": int(row["numeric_observations"]), "released_metric_ids": released, "status": "released" if released else "observed_not_released", }) laboratory_counts: dict[tuple[str, str], int] = {} if _table_exists(connection, "laborwerte"): lab_rows = list(connection.execute( """SELECT parameter_name,einheit,COUNT(*) observations FROM laborwerte WHERE lower(trim(COALESCE(validierungsstatus,'')))='validiert' AND verified_against_original=1 AND reference_range_source='scanned_original' GROUP BY parameter_name,einheit LIMIT ?""", (MAX_INVENTORY_IDENTIFIERS + 1,), )) if len(lab_rows) > MAX_INVENTORY_IDENTIFIERS: raise SourceInventoryLimitError("laboratory inventory limit exceeded") for row in lab_rows: raw_key = ( _normalize(str(row["parameter_name"])).replace(" ", "_"), _normalize(str(row["einheit"] or "").replace("µ", "u").replace("%", " percent ")).replace(" ", "_"), ) canonical = LAB_ALLOWLIST.get(raw_key) if canonical: laboratory_counts[canonical] = laboratory_counts.get(canonical, 0) + int(row["observations"]) laboratory = [ { "canonical_parameter": parameter, "unit": unit, "observations": count, "released_metric_ids": [LAB_CATALOG_SPECS[parameter].metric_id], "status": "verified_canonical", } for (parameter, unit), count in sorted(laboratory_counts.items()) ] vital: list[dict[str, Any]] = [] if _table_exists(connection, "vitalzeichen"): vital_rows = list(connection.execute( """SELECT parameter,COUNT(*) observations FROM vitalzeichen GROUP BY parameter ORDER BY parameter LIMIT ?""", (MAX_INVENTORY_IDENTIFIERS + 1,), )) if len(vital_rows) > MAX_INVENTORY_IDENTIFIERS: raise SourceInventoryLimitError("vital-sign inventory limit exceeded") for row in vital_rows: units = [ str(item[0]) for item in connection.execute( """SELECT DISTINCT einheit FROM vitalzeichen WHERE parameter=? AND einheit IS NOT NULL ORDER BY einheit LIMIT 100""", (row["parameter"],), ) ] vital.append({ "identifier": str(row["parameter"]), "units": units, "observations": int(row["observations"]), "released_metric_ids": [], "status": "observed_not_released", }) return {"apple_health": apple, "laboratory": laboratory, "vital_signs": vital}