#!/usr/bin/env python3 """Nutrition insight computation for JARVIS HealthManager. Builds review queues, lag correlations, conservative food insights, Hyrimoz phase summaries, and next-action recommendations from already-normalized YAZIO data. No raw food lists are printed by default. """ from __future__ import annotations import argparse import json import math import re import sqlite3 from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path from statistics import mean from typing import Any BASE = Path('/home/agent/.hermes/assets/Gesundheit') DB = BASE / 'health_data.db' SEVERITY_HINTS = [ ('keine', 0), ('none', 0), ('mild', 1), ('leicht', 1), ('klein', 1), ('mittel', 2), ('moderat', 2), ('nicht angegeben', 1), ('schwer', 3), ('stark', 3), ('hoch', 3), ] def conn() -> sqlite3.Connection: c = sqlite3.connect(DB) c.row_factory = sqlite3.Row return c def normalize_text(s: str) -> str: s = (s or '').lower() repl = str.maketrans({'ä': 'ae', 'ö': 'oe', 'ü': 'ue', 'é': 'e', 'è': 'e', 'à': 'a', 'ß': 'ss'}) s = s.translate(repl) s = re.sub(r'[^a-z0-9]+', ' ', s) return re.sub(r'\s+', ' ', s).strip() def fnum(x: Any) -> float: try: if x is None or x == '': return 0.0 return float(x) except Exception: return 0.0 def severity_score(text: str | None) -> float: t = (text or '').lower() for needle, score in SEVERITY_HINTS: if needle in t: return float(score) return 1.0 if t or text is None else 0.0 def date_add(day: str, lag: int) -> str: return (datetime.strptime(day[:10], '%Y-%m-%d').date() + timedelta(days=lag)).isoformat() def pearson(xs: list[float], ys: list[float]) -> float | None: n = len(xs) if n < 3: return None mx, my = mean(xs), mean(ys) vx = sum((x - mx) ** 2 for x in xs) vy = sum((y - my) ** 2 for y in ys) if vx == 0 or vy == 0: return None return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / math.sqrt(vx * vy) def interpretation(r: float | None, n: int) -> str: if r is None: return 'zu wenig Varianz/Daten' a = abs(r) prefix = 'sehr niedrige Datenbasis' if n < 7 else 'niedrige Datenbasis' if n < 14 else 'brauchbare Datenbasis' if a >= 0.7: strength = 'stark' elif a >= 0.4: strength = 'moderat' elif a >= 0.2: strength = 'schwach' else: strength = 'kein klares Signal' direction = 'positiv' if (r or 0) > 0 else 'negativ' return f'{prefix}; {strength} {direction}' def ensure_schema(c: sqlite3.Connection) -> None: c.executescript(''' CREATE TABLE IF NOT EXISTS personal_food_tolerance ( canonical_food TEXT PRIMARY KEY, personal_status TEXT NOT NULL DEFAULT 'unknown' CHECK(personal_status IN ('safe','problematic','unclear','unknown')), evidence_level TEXT DEFAULT 'low' CHECK(evidence_level IN ('low','medium','high')), notes TEXT, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS nutrition_review_queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, normalized_name TEXT NOT NULL UNIQUE, example_name TEXT NOT NULL, occurrence_count INTEGER DEFAULT 0, first_seen TEXT, last_seen TEXT, suggested_canonical_food TEXT, suggested_score INTEGER, reason TEXT, status TEXT DEFAULT 'open' CHECK(status IN ('open','mapped','ignored')), updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS nutrition_correlation_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, metric TEXT NOT NULL, target TEXT NOT NULL, lag_days INTEGER NOT NULL, n INTEGER NOT NULL, correlation REAL, method TEXT DEFAULT 'pearson', interpretation TEXT, computed_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(metric,target,lag_days) ); CREATE TABLE IF NOT EXISTS nutrition_food_insights ( id INTEGER PRIMARY KEY AUTOINCREMENT, canonical_food TEXT NOT NULL, insight_type TEXT NOT NULL CHECK(insight_type IN ('trigger_candidate','safe_candidate','insufficient_data')), days_seen INTEGER DEFAULT 0, avg_same_day_symptom REAL, avg_next_day_symptom REAL, avg_followup_symptom REAL, histamine_avg REAL, confidence TEXT DEFAULT 'low', notes TEXT, computed_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(canonical_food, insight_type) ); CREATE TABLE IF NOT EXISTS nutrition_treatment_phase_summary ( phase TEXT PRIMARY KEY, start_date TEXT, end_date TEXT, nutrition_days INTEGER DEFAULT 0, avg_histamine_score REAL, avg_symptom_score REAL, avg_kcal REAL, avg_protein_g REAL, notes TEXT, computed_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS nutrition_action_recommendations ( id INTEGER PRIMARY KEY AUTOINCREMENT, priority INTEGER NOT NULL, recommendation_type TEXT NOT NULL, title TEXT NOT NULL, rationale TEXT, next_step TEXT, status TEXT DEFAULT 'open' CHECK(status IN ('open','done','dismissed')), computed_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(recommendation_type,title) ); ''') def seed_more_aliases(c: sqlite3.Connection) -> None: rules = [ ('Pizza/Prosciutto', 3, 'Fertiggericht', 'processed;pork;aged_cheese_possible', 'Pizza mit Schinken/Käse: histaminologisch meist kritisch/unklar.', 'low'), ('Fast Food Burger', 2, 'Fertiggericht', 'processed;additives', 'Stark verarbeitet; individuelle Toleranz prüfen.', 'low'), ('Pommes/Frittiertes', 1, 'Fertiggericht', 'fried', 'Nicht primär Histamin, aber Entzündungs-/GI-Kontext möglich.', 'low'), ('Kürbiskerne', 1, 'Samen/Nüsse', 'seed;liberator_possible', 'Individuell testen.', 'low'), ('Aprikose', 1, 'Obst', 'stonefruit', 'Individuell testen.', 'low'), ('Mango', 1, 'Obst', 'dried_fruit_possible', 'Getrocknet/Zusätze individuell prüfen.', 'low'), ('Weisskohl', 0, 'Gemüse', '', 'Frisch meist eher unkritisch; fermentiert anders bewerten.', 'low'), ('Weizentortilla', 1, 'Getreide', 'gluten', 'Nicht primär Histamin; GI/Gluten-Kontext prüfen.', 'low'), ('Olivenöl', 0, 'Fett', '', 'Meist verträglich.', 'low'), ('Süsssauer Sauce', 2, 'Sauce', 'additives;vinegar_possible', 'Zucker/Essig/Zusätze möglich.', 'low'), ('Rahm/Sahne', 0, 'Milchprodukt', '', 'Frisch meist eher unkritisch.', 'low'), ('Dinkelpasta', 1, 'Getreide', 'gluten', 'Nicht primär Histamin; GI/Gluten-Kontext prüfen.', 'low'), ('Kohlrabi', 0, 'Gemüse', '', 'Meist verträglich.', 'low'), ('Rind Steak frisch', 0, 'Fleisch', 'freshness_sensitive', 'Frische/Resten relevant.', 'low'), ('Randen/Rote Bete', 0, 'Gemüse', '', 'Meist verträglich.', 'low'), ] c.executemany('''INSERT INTO histamine_food_rules(canonical_food,sighi_score,category,tags,notes,confidence) VALUES(?,?,?,?,?,?) ON CONFLICT(canonical_food) DO UPDATE SET sighi_score=excluded.sighi_score,category=excluded.category,tags=excluded.tags,notes=excluded.notes,confidence=excluded.confidence,updated_at=CURRENT_TIMESTAMP''', rules) aliases = { 'randensalat': 'Randen/Rote Bete', 'randen': 'Randen/Rote Bete', 'rote bete': 'Randen/Rote Bete', 'kuerbiskerne': 'Kürbiskerne', 'kürbiskerne': 'Kürbiskerne', 'pumpkin seeds': 'Kürbiskerne', 'aprikose': 'Aprikose', 'apricot': 'Aprikose', 'mango': 'Mango', 'mangoschnitze': 'Mango', 'weisskohl': 'Weisskohl', 'weißkohl': 'Weisskohl', 'white cabbage': 'Weisskohl', 'weizentortilla': 'Weizentortilla', 'tortilla': 'Weizentortilla', 'olivenoel': 'Olivenöl', 'olivenöl': 'Olivenöl', 'olive oil': 'Olivenöl', 'big mac': 'Fast Food Burger', 'burger': 'Fast Food Burger', 'suesssauer': 'Süsssauer Sauce', 'süßsauer': 'Süsssauer Sauce', 'sour sauce': 'Süsssauer Sauce', 'pommes': 'Pommes/Frittiertes', 'frites': 'Pommes/Frittiertes', 'french fries': 'Pommes/Frittiertes', 'halbrahm': 'Rahm/Sahne', 'rahm': 'Rahm/Sahne', 'sahne': 'Rahm/Sahne', 'dinkel spirelli': 'Dinkelpasta', 'dinkel': 'Dinkelpasta', 'kohlrabi': 'Kohlrabi', 'entrecote': 'Rind Steak frisch', 'entrecôte': 'Rind Steak frisch', 'pizza prosciutto': 'Pizza/Prosciutto', 'prosciutto': 'Pizza/Prosciutto', } c.executemany('INSERT OR REPLACE INTO histamine_food_aliases(alias, canonical_food) VALUES(?,?)', aliases.items()) def refresh_histamine_scores(c: sqlite3.Connection) -> None: import sys sys.path.insert(0, str(BASE / 'scripts')) import yazio_nutrition_sync as yz # type: ignore for r in c.execute("SELECT id,name FROM nutrition_items WHERE source='yazio_api'").fetchall(): hs = yz.classify_histamine(c, r['name']) c.execute('''INSERT OR REPLACE INTO nutrition_histamine_scores(item_id,canonical_food,sighi_score,traffic_light,tags,confidence,reason,scored_at) VALUES(?,?,?,?,?,?,?,CURRENT_TIMESTAMP)''', (r['id'], hs['canonical_food'], hs['sighi_score'], hs['traffic_light'], hs['tags'], hs['confidence'], hs['reason'])) for r in c.execute("SELECT DISTINCT datum FROM nutrition_items WHERE source='yazio_api'").fetchall(): c.execute('DELETE FROM nutrition_meal_summary WHERE datum=?', (r['datum'],)) c.execute('DELETE FROM nutrition_daily_summary_v2 WHERE datum=?', (r['datum'],)) yz.recompute_summaries(c, r['datum']) def build_review_queue(c: sqlite3.Connection) -> int: c.execute("DELETE FROM nutrition_review_queue WHERE status='open'") unknowns = c.execute(''' SELECT lower(i.name) raw_key, MIN(i.name) example_name, COUNT(*) occurrence_count, MIN(i.datum) first_seen, MAX(i.datum) last_seen FROM nutrition_items i JOIN nutrition_histamine_scores h ON h.item_id=i.id WHERE COALESCE(h.traffic_light,'unknown')='unknown' GROUP BY lower(i.name) ORDER BY occurrence_count DESC, last_seen DESC ''').fetchall() for r in unknowns: norm = normalize_text(r['example_name']) c.execute('''INSERT INTO nutrition_review_queue(normalized_name,example_name,occurrence_count,first_seen,last_seen,reason,status,updated_at) VALUES(?,?,?,?,?,'kein Histamin-/SIGHi-Mapping','open',CURRENT_TIMESTAMP) ON CONFLICT(normalized_name) DO UPDATE SET example_name=excluded.example_name, occurrence_count=excluded.occurrence_count, first_seen=excluded.first_seen, last_seen=excluded.last_seen, reason=excluded.reason, status='open', updated_at=CURRENT_TIMESTAMP ''', (norm, r['example_name'], r['occurrence_count'], r['first_seen'], r['last_seen'])) return len(unknowns) def symptom_series(c: sqlite3.Connection) -> dict[str, float]: out: dict[str, float] = defaultdict(float) for r in c.execute('SELECT datum,schwergrad,notizen FROM symptome'): out[str(r['datum'])[:10]] += severity_score(' '.join([str(r['schwergrad'] or ''), str(r['notizen'] or '')])) for r in c.execute('SELECT datum,schwergrad,notizen,kontext FROM symptom_log'): out[str(r['datum'])[:10]] += severity_score(' '.join([str(r['schwergrad'] or ''), str(r['kontext'] or ''), str(r['notizen'] or '')])) return dict(out) def compute_correlations(c: sqlite3.Connection) -> int: symptoms = symptom_series(c) days = [r['datum'] for r in c.execute('SELECT datum FROM nutrition_daily_summary_v2 ORDER BY datum')] metrics = { 'histamine_load': {r['datum']: fnum(r['histamine_score']) for r in c.execute('SELECT datum,histamine_score FROM nutrition_daily_summary_v2')}, 'kcal': {r['datum']: fnum(r['kcal']) for r in c.execute('SELECT datum,kcal FROM nutrition_daily_summary_v2')}, 'protein_g': {r['datum']: fnum(r['protein_g']) for r in c.execute('SELECT datum,protein_g FROM nutrition_daily_summary_v2')}, 'sugar_g': {}, 'fiber_g': {}, 'saturated_fat_g': {}, } for r in c.execute('SELECT datum,nutrient_json FROM nutrition_daily_summary_v2'): try: n = json.loads(r['nutrient_json'] or '{}') except Exception: n = {} metrics['sugar_g'][r['datum']] = fnum(n.get('nutrient.sugar')) metrics['fiber_g'][r['datum']] = fnum(n.get('nutrient.dietaryfiber') or n.get('nutrient.fiber')) metrics['saturated_fat_g'][r['datum']] = fnum(n.get('nutrient.saturated')) count = 0 for metric, series in metrics.items(): for lag in range(0, 4): xs, ys = [], [] for d in days: if d in series: xs.append(series[d]); ys.append(symptoms.get(date_add(d, lag), 0.0)) r = pearson(xs, ys) c.execute('''INSERT INTO nutrition_correlation_results(metric,target,lag_days,n,correlation,method,interpretation,computed_at) VALUES(?,?,?,?,?,?,?,CURRENT_TIMESTAMP) ON CONFLICT(metric,target,lag_days) DO UPDATE SET n=excluded.n, correlation=excluded.correlation, interpretation=excluded.interpretation, computed_at=CURRENT_TIMESTAMP ''', (metric, 'symptom_score', lag, len(xs), r, 'pearson', interpretation(r, len(xs)))) count += 1 return count def compute_food_insights(c: sqlite3.Connection) -> int: symptoms = symptom_series(c) c.execute('DELETE FROM nutrition_food_insights') foods = c.execute(''' SELECT COALESCE(h.canonical_food,'Unbekannt') canonical_food, i.datum, MAX(COALESCE(h.sighi_score,0)) sighi FROM nutrition_items i LEFT JOIN nutrition_histamine_scores h ON h.item_id=i.id WHERE i.source='yazio_api' AND h.canonical_food IS NOT NULL GROUP BY canonical_food, i.datum ''').fetchall() by_food: dict[str, list[sqlite3.Row]] = defaultdict(list) for r in foods: by_food[r['canonical_food']].append(r) inserted = 0 tolerance = {r['canonical_food']: r for r in c.execute('SELECT * FROM personal_food_tolerance')} for food, rows in by_food.items(): days = sorted({r['datum'] for r in rows}) if len(days) < 2: continue same = [symptoms.get(d, 0.0) for d in days] nxt = [symptoms.get(date_add(d, 1), 0.0) for d in days] follow = [max(symptoms.get(date_add(d, lag), 0.0) for lag in [1, 2, 3]) for d in days] hist = [fnum(r['sighi']) for r in rows] avg_same, avg_next, avg_follow, avg_hist = mean(same), mean(nxt), mean(follow), mean(hist) personal = tolerance.get(food) if personal and personal['personal_status'] == 'safe': itype, confidence, note = 'safe_candidate', personal['evidence_level'], 'Persönlich als safe markiert; weiter beobachten.' elif personal and personal['personal_status'] == 'problematic': itype, confidence, note = 'trigger_candidate', personal['evidence_level'], 'Persönlich als problematisch markiert; Re-Challenge nur bewusst.' elif len(days) < 3: itype, confidence, note = 'insufficient_data', 'low', 'weniger als 3 Tage beobachtet' elif avg_follow >= 1.5 and avg_hist >= 1: itype, confidence, note = 'trigger_candidate', ('low' if len(days) < 5 else 'medium'), 'Follow-up-Symptomscore erhöht; explorativ, keine Kausalität' elif avg_follow <= 0.5 and avg_hist <= 1: itype, confidence, note = 'safe_candidate', ('low' if len(days) < 5 else 'medium'), 'mehrfach gegessen mit niedrigem Follow-up-Symptomscore; weiter beobachten' else: itype, confidence, note = 'insufficient_data', 'low', 'kein klares Muster' c.execute('''INSERT INTO nutrition_food_insights(canonical_food,insight_type,days_seen,avg_same_day_symptom,avg_next_day_symptom,avg_followup_symptom,histamine_avg,confidence,notes,computed_at) VALUES(?,?,?,?,?,?,?,?,?,CURRENT_TIMESTAMP)''', (food, itype, len(days), round(avg_same,2), round(avg_next,2), round(avg_follow,2), round(avg_hist,2), confidence, note)) inserted += 1 return inserted def hyrimoz_start_date(c: sqlite3.Connection) -> str | None: row = c.execute("""SELECT MIN(datum) d FROM medication_administrations WHERE lower(medication_name) LIKE '%hyrimoz%' OR lower(medication_name) LIKE '%adalimumab%'""").fetchone() return row['d'] if row and row['d'] else None def phase_for_day(day: str, start: str | None) -> str: if not start: return 'unknown' d = datetime.strptime(day[:10], '%Y-%m-%d').date() s = datetime.strptime(start[:10], '%Y-%m-%d').date() if d < s: return 'baseline_pre_hyrimoz' if d < s + timedelta(days=56): return 'early_hyrimoz_phase' return 'stable_hyrimoz_phase' def compute_phase_summary(c: sqlite3.Connection) -> int: symptoms = symptom_series(c) start = hyrimoz_start_date(c) c.execute('DELETE FROM nutrition_treatment_phase_summary') rows = list(c.execute('SELECT datum,kcal,protein_g,histamine_score FROM nutrition_daily_summary_v2 ORDER BY datum')) by_phase: dict[str, list[sqlite3.Row]] = defaultdict(list) for r in rows: by_phase[phase_for_day(r['datum'], start)].append(r) labels = { 'baseline_pre_hyrimoz': 'Vor Hyrimoz: Ernährung/Symptome ohne TNF-Blocker; als Baseline interpretieren.', 'early_hyrimoz_phase': 'Frühe Hyrimoz-Phase: Ernährungseffekte nicht vorschnell von Medikamenteneffekt trennen.', 'stable_hyrimoz_phase': 'Stabilere Hyrimoz-Phase: später besser für belastbarere Vergleiche.', 'unknown': 'Kein Hyrimoz-Startdatum gefunden.', } for phase, rs in by_phase.items(): days = [r['datum'] for r in rs] c.execute('''INSERT OR REPLACE INTO nutrition_treatment_phase_summary (phase,start_date,end_date,nutrition_days,avg_histamine_score,avg_symptom_score,avg_kcal,avg_protein_g,notes,computed_at) VALUES(?,?,?,?,?,?,?,?,?,CURRENT_TIMESTAMP)''', ( phase, min(days), max(days), len(days), round(mean([fnum(r['histamine_score']) for r in rs]), 2), round(mean([symptoms.get(d, 0.0) for d in days]), 2), round(mean([fnum(r['kcal']) for r in rs]), 1), round(mean([fnum(r['protein_g']) for r in rs]), 1), labels.get(phase, ''), )) return len(by_phase) def compute_action_recommendations(c: sqlite3.Connection) -> int: c.execute("DELETE FROM nutrition_action_recommendations WHERE status='open'") recs: list[tuple[int, str, str, str, str]] = [] open_unknown = c.execute("SELECT COUNT(*) FROM nutrition_review_queue WHERE status='open'").fetchone()[0] if open_unknown: recs.append((10, 'mapping_review', 'Offene YAZIO-Produkte mappen', f'{open_unknown} Produktnamen haben noch kein Histamin-/SIGHi-Mapping.', 'Mit nutrition_tolerance.py list-open prüfen; häufige Produkte zuerst mappen.')) sym_days = c.execute("SELECT COUNT(DISTINCT datum) FROM (SELECT datum FROM symptome UNION SELECT datum FROM symptom_log)").fetchone()[0] nut_days = c.execute('SELECT COUNT(*) FROM nutrition_daily_summary_v2').fetchone()[0] if sym_days < max(7, nut_days // 2): recs.append((20, 'symptom_logging', 'Symptom-Logging verdichten', f'Nur {sym_days} Symptomtage vs. {nut_days} Ernährungstage; Korrelationen bleiben dadurch schwach.', 'Täglich kurzen Score für Aphthen/GI/Müdigkeit/Haut/Augen/Gelenke erfassen.')) red_days = c.execute("SELECT COUNT(*) FROM nutrition_daily_summary_v2 WHERE histamine_label='red'").fetchone()[0] if red_days: recs.append((30, 'histamine_baseline', 'Low-Histamine-Baseline testbar machen', f'{red_days} rote Histamin-Tage im aktuellen Datenfenster.', 'Für 7 Tage wenige stabile Safe-Foods nutzen und Symptome sauber tracken; danach einzelne Re-Challenges.')) if c.execute('SELECT COUNT(*) FROM nutrition_treatment_phase_summary').fetchone()[0]: recs.append((40, 'hyrimoz_phase', 'Hyrimoz-Phasen getrennt interpretieren', 'Frühe Hyrimoz-Phase ist markiert; Ernährungssignale können medikamentös überlagert sein.', 'Phasenvergleich beobachten; harte Schlüsse erst mit mehr Post-Hyrimoz-Daten.')) for rec in recs: c.execute('''INSERT INTO nutrition_action_recommendations(priority,recommendation_type,title,rationale,next_step,status,computed_at) VALUES(?,?,?,?,?,'open',CURRENT_TIMESTAMP) ON CONFLICT(recommendation_type,title) DO UPDATE SET priority=excluded.priority,rationale=excluded.rationale,next_step=excluded.next_step,status='open',computed_at=CURRENT_TIMESTAMP''', rec) return len(recs) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument('--seed-aliases', action='store_true', default=True) ap.add_argument('--no-seed-aliases', action='store_false', dest='seed_aliases') ap.add_argument('--verbose', action='store_true') args = ap.parse_args() c = conn() try: ensure_schema(c) if args.seed_aliases: seed_more_aliases(c) refresh_histamine_scores(c) review_count = build_review_queue(c) corr_count = compute_correlations(c) insight_count = compute_food_insights(c) phase_count = compute_phase_summary(c) rec_count = compute_action_recommendations(c) c.commit() if args.verbose: print(f'nutrition insights ok: review_open={review_count}; correlations={corr_count}; food_insights={insight_count}; phases={phase_count}; recommendations={rec_count}') finally: c.close() return 0 if __name__ == '__main__': raise SystemExit(main())