from __future__ import annotations from datetime import datetime from enum import Enum from typing import Literal from pydantic import BaseModel, ConfigDict, Field, model_validator from jarvis_gateway.contracts import ( ActionDescriptor, ActionMode, AttentionItem, DisplayPolicy, KpiItem, ModuleId, ModuleSnapshot, ModuleStatus, Sensitivity, Severity, SourceHealth, SourceType, utcnow, ) DOMAIN_IDS = ( "behcet_activity", "eye_risk", "vascular_thrombosis_risk", "inflammation_lab_trend", "medication_safety", "controls_data_status", ) DomainId = Literal[ "behcet_activity", "eye_risk", "vascular_thrombosis_risk", "inflammation_lab_trend", "medication_safety", "controls_data_status", ] DOMAIN_LABELS = { "behcet_activity": "Behçet-Aktivität", "eye_risk": "Augen-Risiko", "vascular_thrombosis_risk": "Gefässe/Thrombose", "inflammation_lab_trend": "Entzündung/Labortrend", "medication_safety": "Medikamenten-Sicherheit", "controls_data_status": "Kontrollen & Datenstatus", } class RiskStatus(str, Enum): GREEN = "green" YELLOW = "yellow" RED = "red" UNKNOWN = "unknown" class RiskAction(str, Enum): NONE = "none" WATCH = "watch" CONTACT_DOCTOR = "contact_doctor" URGENT = "urgent" class FreshnessStatus(str, Enum): FRESH = "fresh" STALE = "stale" MISSING = "missing" UNKNOWN = "unknown" class DashboardStatus(str, Enum): AVAILABLE = "available" MISSING = "missing" UNKNOWN = "unknown" class StrictBase(BaseModel): model_config = ConfigDict(extra="forbid", use_enum_values=True) class HealthRiskDomain(StrictBase): id: DomainId label: str status: RiskStatus reason: str = Field(max_length=48, pattern=r"^[A-Za-zÄÖÜäöüß0-9 _./:-]+$") action: RiskAction @model_validator(mode="after") def enforce_label(self) -> "HealthRiskDomain": if self.label != DOMAIN_LABELS[str(self.id)]: raise ValueError("domain label must match fixed contract label") return self class HealthRiskDataFreshness(StrictBase): labs: FreshnessStatus duplex: FreshnessStatus eye_check: FreshnessStatus dashboard: DashboardStatus class HealthRiskSnapshot(StrictBase): schema_version: Literal["1.0"] = "1.0" generated_at: datetime overall_status: RiskStatus domains: list[HealthRiskDomain] = Field(min_length=6, max_length=6) data_freshness: HealthRiskDataFreshness @model_validator(mode="after") def enforce_domain_set(self) -> "HealthRiskSnapshot": ids = [str(domain.id) for domain in self.domains] if ids != list(DOMAIN_IDS): raise ValueError("health risk domains must be exactly the fixed ordered domain set") return self class HealthRiskCockpitAdapter: """Builds a safe, highly compressed Health risk snapshot. This adapter does not read HealthManager databases, PDFs, reports, OCR output, symptoms, lab rows, or document contents. In this phase it produces a conservative contract-shaped snapshot from explicit safe flags only. """ def __init__(self, *, dashboard_available: bool = False, emergency_flags: set[str] | None = None, stale_data: bool = True) -> None: self.dashboard_available = dashboard_available self.emergency_flags = emergency_flags or set() self.stale_data = stale_data def build_snapshot(self) -> HealthRiskSnapshot: now = utcnow() emergency_eye = bool({"sudden_vision_loss", "painful_red_eye", "new_floaters"} & self.emergency_flags) emergency_vascular = bool({"chest_pain", "shortness_of_breath", "hemoptysis", "neuro_deficit", "unilateral_leg_swelling"} & self.emergency_flags) emergency_med = bool({"bleeding_signs", "severe_infection"} & self.emergency_flags) domains = [ self._domain("behcet_activity", RiskStatus.UNKNOWN, "keine safe Flags", RiskAction.WATCH), self._domain("eye_risk", RiskStatus.RED if emergency_eye else RiskStatus.UNKNOWN, "Warnzeichen Flag" if emergency_eye else "Augencheck offen", RiskAction.URGENT if emergency_eye else RiskAction.WATCH), self._domain("vascular_thrombosis_risk", RiskStatus.RED if emergency_vascular else RiskStatus.UNKNOWN, "Notfall Flag" if emergency_vascular else "keine safe Daten", RiskAction.URGENT if emergency_vascular else RiskAction.WATCH), self._domain("inflammation_lab_trend", RiskStatus.YELLOW if self.stale_data else RiskStatus.UNKNOWN, "Trend stale" if self.stale_data else "Schwellen fehlen", RiskAction.WATCH), self._domain("medication_safety", RiskStatus.RED if emergency_med else RiskStatus.UNKNOWN, "Warnzeichen Flag" if emergency_med else "Monitoring nötig", RiskAction.URGENT if emergency_med else RiskAction.WATCH), self._domain("controls_data_status", RiskStatus.YELLOW if self.stale_data else RiskStatus.GREEN, "Datenstatus stale" if self.stale_data else "Kontrollen aktuell", RiskAction.WATCH if self.stale_data else RiskAction.NONE), ] overall = self._overall_status([RiskStatus(str(domain.status)) for domain in domains]) return HealthRiskSnapshot( generated_at=now, overall_status=overall, domains=domains, data_freshness=HealthRiskDataFreshness( labs=FreshnessStatus.STALE if self.stale_data else FreshnessStatus.FRESH, duplex=FreshnessStatus.UNKNOWN, eye_check=FreshnessStatus.UNKNOWN, dashboard=DashboardStatus.AVAILABLE if self.dashboard_available else DashboardStatus.MISSING, ), ) def get_snapshot(self) -> ModuleSnapshot: risk = self.build_snapshot() now = risk.generated_at overall_severity = self._severity(RiskStatus(str(risk.overall_status))) vascular = next(domain for domain in risk.domains if domain.id == "vascular_thrombosis_risk") data_status = next(domain for domain in risk.domains if domain.id == "controls_data_status") action = ActionDescriptor(label="Health Risk Cockpit öffnen", href="/health", mode=ActionMode.LINK_ONLY, sensitivity=Sensitivity.CRITICAL, blocked_reason=None, preview_required=False, confirm_required=False, audit_required=False) attention = [] for domain in risk.domains: status = RiskStatus(str(domain.status)) if status in {RiskStatus.RED, RiskStatus.YELLOW}: attention.append(AttentionItem( id=f"att_health_risk_{domain.id}", module_id=ModuleId.HEALTH, title=self._attention_title(domain), message=self._attention_message(domain), severity=self._severity(status), sensitivity=Sensitivity.CRITICAL, created_at=now, action=action, dedupe_key=f"health.risk.{domain.id}", )) return ModuleSnapshot( module_id=ModuleId.HEALTH, title="Health Risk Cockpit", status=self._module_status(RiskStatus(str(risk.overall_status))), sensitivity=Sensitivity.CRITICAL, display_policy=DisplayPolicy.SUMMARY, last_success_at=now, last_attempt_at=now, stale_after_seconds=86400, degraded_reason=None if risk.overall_status == RiskStatus.GREEN else "health_risk_attention", source_health=SourceHealth(reachable=True, latency_ms=None, source_type=SourceType.MOCK, version="health-risk-cockpit-v1", contract_version="health.risk_snapshot.v1", last_success_at=now, last_attempt_at=now, last_error_redacted=None, stale=self.stale_data), kpis=[ KpiItem(key="overall_health_risk", label="Overall Health Risk", value=str(risk.overall_status), unit=None, severity=overall_severity, sensitivity=Sensitivity.CRITICAL, display_policy=DisplayPolicy.SUMMARY, help_text="Ampelstatus ohne Rohdaten."), KpiItem(key="vascular_risk", label="Vascular risk", value=str(vascular.status), unit=None, severity=self._severity(RiskStatus(str(vascular.status))), sensitivity=Sensitivity.CRITICAL, display_policy=DisplayPolicy.SUMMARY, help_text="Keine Detaildaten."), KpiItem(key="data_freshness", label="Data freshness", value=str(data_status.status), unit=None, severity=self._severity(RiskStatus(str(data_status.status))), sensitivity=Sensitivity.CRITICAL, display_policy=DisplayPolicy.SUMMARY, help_text="Kontroll-/Datenstatus."), ], attention_items=attention[:3], links=[action], primary_action=action, contract_version="jarvis.module_snapshot.v1", ) def _domain(self, domain_id: DomainId, status: RiskStatus, reason: str, action: RiskAction) -> HealthRiskDomain: return HealthRiskDomain(id=domain_id, label=DOMAIN_LABELS[domain_id], status=status, reason=reason, action=action) def _overall_status(self, statuses: list[RiskStatus]) -> RiskStatus: if RiskStatus.RED in statuses: return RiskStatus.RED if RiskStatus.YELLOW in statuses: return RiskStatus.YELLOW if all(status == RiskStatus.GREEN for status in statuses): return RiskStatus.GREEN return RiskStatus.UNKNOWN def _module_status(self, status: RiskStatus) -> ModuleStatus: return {RiskStatus.GREEN: ModuleStatus.OK, RiskStatus.YELLOW: ModuleStatus.ATTENTION, RiskStatus.RED: ModuleStatus.ATTENTION, RiskStatus.UNKNOWN: ModuleStatus.DEGRADED}[status] def _severity(self, status: RiskStatus) -> Severity: return {RiskStatus.GREEN: Severity.SUCCESS, RiskStatus.YELLOW: Severity.WARNING, RiskStatus.RED: Severity.CRITICAL, RiskStatus.UNKNOWN: Severity.INFO}[status] def _attention_title(self, domain: HealthRiskDomain) -> str: if domain.id == "eye_risk" and domain.status == RiskStatus.RED: return "Health risk red: eye symptoms" if domain.id == "vascular_thrombosis_risk" and domain.status == RiskStatus.RED: return "Vascular warning signs" if domain.id == "inflammation_lab_trend": return "Lab trend stale" if domain.id == "controls_data_status": return "Health data stale" return f"Health risk {domain.status}: {domain.label}" def _attention_message(self, domain: HealthRiskDomain) -> str: if domain.action == RiskAction.URGENT: return "Notfallzeichen synthetisch markiert; Details im HealthManager prüfen." if domain.action == RiskAction.CONTACT_DOCTOR: return "Arztkontakt prüfen; keine Detaildaten in JARVIS." return "Beobachten oder Termin/Datenstatus prüfen; Details bleiben im HealthManager."