"""BioPatch AI Engine — ML-backed scoring + tissue + bacterial classifier. The engine orchestrates three layers: 1. **Risk scoring** is delegated to the Scikit-Learn Random Forest in :mod:`ai.model` — a 9-feature classifier trained on 5 000 literature- calibrated snapshots that emits per-class probabilities, an expected severity score, per-sensor XAI contributions, and a context-aware severity bucketing (sensitivity rises for diabetic / smoker / elderly patients). 2. **Tissue state** is a deterministic rule layer on top of the live sensor triple — granulação / epitelização / inflamatório / desvitalização / transição — kept clinical-explainable. 3. **Bacterial signature** is a rule classifier keyed on the metabolic fingerprint of common SSI pathogens, gated by the ML score and tissue state so we never cry wolf on a healing wound. Public API: ``SepsisRiskEngine.assess(reading, factors)`` returns a :class:`RiskAssessment` whose ``to_dict`` is broadcast verbatim over the WebSocket. """ from __future__ import annotations import math from collections import deque from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from statistics import pstdev from typing import Deque, Literal from ai.features import PatientFactors, compute_velocity from ai.model import ClinicalRiskModel, RiskPrediction # ---- Clinical thresholds (unchanged) -------------------------------------- TEMP_HEALTHY_MAX = 35.5 TEMP_FEBRILE = 37.5 TEMP_HYPERPYREXIA = 38.5 PH_HEALING_MAX = 6.5 PH_INFECTED = 7.4 PH_STRONG_ALKALINE = 8.5 HUMIDITY_OPTIMAL_LOW = 65.0 HUMIDITY_OPTIMAL_HIGH = 85.0 HUMIDITY_VOLATILITY_ERRATIC = 5.0 Severity = Literal["stable", "watch", "warning", "critical"] TissueState = Literal[ "granulation", "epithelialization", "inflammatory", "devitalized", "transitional", ] TISSUE_LABEL_PT = { "granulation": "Tecido de Granulação", "epithelialization": "Tecido de Epitelização", "inflammatory": "Tecido Inflamatório", "devitalized": "Tecido de Desvitalização", "transitional": "Tecido em Transição", } ARTIFACT_PATH = Path(__file__).resolve().parent.parent / "ai" / "artifacts" / "rf_v1.joblib" # Lazily-initialised singleton — the joblib is ~5 MB so we only want one # copy in memory, shared across patient engines. _clinical_model: ClinicalRiskModel | None = None def get_clinical_model() -> ClinicalRiskModel: global _clinical_model if _clinical_model is None: _clinical_model = ClinicalRiskModel.load(ARTIFACT_PATH) return _clinical_model # ---- Data carriers -------------------------------------------------------- @dataclass(frozen=True) class Reading: temperature_c: float ph: float humidity: float timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @dataclass(frozen=True) class RiskAssessment: score: float severity: Severity bacterial_signature: str | None signature_confidence: float drivers: list[str] tissue_state: TissueState tissue_confidence: float humidity_volatility: float fusion_quality: float # ML / XAI fields sensor_contributions: dict[str, float] feature_importance: dict[str, float] class_probabilities: dict[str, float] velocity_6h: dict[str, float] context_factor: float model_version: str def to_dict(self) -> dict: return { "score": round(self.score, 1), "severity": self.severity, "bacterial_signature": self.bacterial_signature, "signature_confidence": round(self.signature_confidence, 2), "drivers": self.drivers, "tissue_state": self.tissue_state, "tissue_state_label_pt": TISSUE_LABEL_PT[self.tissue_state], "tissue_confidence": round(self.tissue_confidence, 2), "humidity_volatility": round(self.humidity_volatility, 2), "fusion_quality": round(self.fusion_quality, 2), "sensor_contributions": {k: round(v, 3) for k, v in self.sensor_contributions.items()}, "feature_importance": {k: round(v, 3) for k, v in self.feature_importance.items()}, "class_probabilities": {k: round(v, 3) for k, v in self.class_probabilities.items()}, "velocity_6h": {k: round(v, 3) for k, v in self.velocity_6h.items()}, "context_factor": round(self.context_factor, 2), "model_version": self.model_version, } # ---- Engine --------------------------------------------------------------- class SepsisRiskEngine: """Stateful per-stream engine. Holds a rolling history long enough to compute the model's 6h velocity features: 3 600 entries ≈ 6 h at the 6 s simulator cadence (real sensors will populate the same buffer regardless of cadence). """ def __init__(self, history_window: int = 3600) -> None: self._history: Deque[Reading] = deque(maxlen=history_window) # ----- public API ------------------------------------------------------ def assess( self, reading: Reading, factors: PatientFactors | None = None, ) -> RiskAssessment: factors = factors or PatientFactors() self._history.append(reading) velocity = compute_velocity(list(self._history)) humidity_volatility = velocity["humidity_volatility_6h"] # 1. ML scoring ----------------------------------------------------- model = get_clinical_model() prediction = model.predict( temperature_c=reading.temperature_c, ph=reading.ph, humidity=reading.humidity, velocity=velocity, factors=factors, ) # 2. Tissue classification (rule layer) ----------------------------- tissue_state, tissue_conf = self._classify_tissue( reading, humidity_volatility ) # 3. Bacterial signature (rule layer) ------------------------------- signature, sig_conf = self._classify_signature( reading, prediction.score, tissue_state ) # 4. Drivers — assembled from XAI + threshold checks ---------------- drivers = self._build_drivers(reading, velocity, prediction) # 5. Fusion quality — entropy of the three sensor contributions fusion_quality = self._fusion_quality(prediction.sensor_contributions) return RiskAssessment( score=prediction.score, severity=prediction.severity, bacterial_signature=signature, signature_confidence=sig_conf, drivers=drivers, tissue_state=tissue_state, tissue_confidence=tissue_conf, humidity_volatility=humidity_volatility, fusion_quality=fusion_quality, sensor_contributions=prediction.sensor_contributions, feature_importance=prediction.feature_importance, class_probabilities=prediction.probabilities, velocity_6h=velocity, context_factor=prediction.context_factor, model_version=prediction.model_version, ) def reset(self) -> None: self._history.clear() # ----- tissue classifier (rule layer) --------------------------------- @staticmethod def _classify_tissue( r: Reading, humidity_volatility: float ) -> tuple[TissueState, float]: """Triple-axis tissue classifier — requires Temp + pH + Humidity together; no single axis can decide on its own. Devitalization here follows the user's spec: acidic pH (<6.5), ischemic (low) local temperature, erratic humidity — physiologic signs of failing perfusion + necrotic acidosis. """ t, p, h = r.temperature_c, r.ph, r.humidity # --- Granulação: healthy red-pink healing --- if 70.0 <= h <= 90.0 and 7.0 <= p <= 7.4 and 33.0 <= t <= 36.0: center_score = ( (1.0 - abs(h - 80.0) / 10.0) * (1.0 - abs(p - 7.2) / 0.2) * (1.0 - abs(t - 34.5) / 1.5) ) return "granulation", max(0.65, min(0.92, 0.65 + center_score * 0.25)) # --- Epitelização: late-stage healing --- if 55.0 <= h <= 90.0 and 5.5 <= p <= 7.0 and 32.5 <= t <= 36.0: return "epithelialization", 0.75 # --- Inflamatório: hot + alkaline --- if t >= 37.0 and p >= 7.4: sev_kick = min(1.0, (t - 37.0) / 1.5 + (p - 7.4) / 1.0) return "inflammatory", min(0.92, 0.65 + sev_kick * 0.2) # --- Desvitalização: ischemic acidosis --- # Strong signature: low local temperature + acidic pH (<6.5). # Adding erratic humidity raises confidence further. erratic = humidity_volatility >= HUMIDITY_VOLATILITY_ERRATIC ischemic = t < 33.0 very_ischemic = t < 31.5 acidic = p < 6.5 very_acidic = p < 5.8 if very_ischemic and acidic: return "devitalized", 0.92 if ischemic and acidic and erratic: return "devitalized", 0.86 if ischemic and acidic: return "devitalized", 0.78 if ischemic and erratic: return "devitalized", 0.70 if very_ischemic: return "devitalized", 0.66 if very_acidic and erratic: return "devitalized", 0.65 return "transitional", 0.55 # ----- bacterial signature (rule layer) ------------------------------- @staticmethod def _classify_signature( r: Reading, score: float, tissue_state: TissueState ) -> tuple[str | None, float]: """Heuristic rule-based signature keyed on the metabolic fingerprint of common SSI pathogens. Suppressed when the wound is healing. """ if tissue_state in {"granulation", "epithelialization"} and score < 50: return None, 0.0 if score < 35: return None, 0.0 t, p, h = r.temperature_c, r.ph, r.humidity if p >= 8.0 and h >= 75.0 and t >= 36.5: return "Pseudomonas aeruginosa (suspected)", min(0.92, 0.60 + (p - 8.0) * 0.15) if 7.4 <= p < 8.0 and t >= 37.8: label = "MRSA (suspected)" if t >= 38.5 else "Staphylococcus aureus (suspected)" return label, min(0.88, 0.55 + (t - 37.8) * 0.12) if t >= 38.0 and p < 7.4: return "Streptococcus pyogenes (suspected)", 0.62 if p >= 7.6 and h >= 85.0 and t < 37.5: return "Anaerobic colonisation (suspected)", 0.58 if 7.2 <= p < 7.8 and 36.8 <= t < 38.0: return "Gram-negative enteric (suspected)", 0.52 return "Unspecified polymicrobial pattern", 0.40 # ----- drivers -------------------------------------------------------- @staticmethod def _build_drivers( r: Reading, velocity: dict, prediction: RiskPrediction ) -> list[str]: """Synthesize the top-N risk drivers from the XAI breakdown + the absolute threshold checks the clinical literature describes.""" drivers: list[tuple[float, str]] = [] contribs = prediction.sensor_contributions sensor_labels = { "temperature": "Temperature", "ph": "Wound pH", "humidity": "Humidity", } if contribs: top_axis, top_pct = max(contribs.items(), key=lambda kv: kv[1]) if top_pct >= 0.45: drivers.append( ( 100 * top_pct, f"{sensor_labels[top_axis]} dominates the AI signal ({top_pct * 100:.0f}%)", ) ) # Absolute thresholds (clinically explicit) if r.temperature_c >= TEMP_HYPERPYREXIA: drivers.append((90, f"Hyperpyrexia at wound site ({r.temperature_c:.1f} C)")) elif r.temperature_c >= TEMP_FEBRILE: drivers.append((70, f"Febrile thermal pattern ({r.temperature_c:.1f} C)")) elif r.temperature_c < 32.0: drivers.append((60, f"Ischemic local temperature ({r.temperature_c:.1f} C)")) if r.ph >= PH_STRONG_ALKALINE: drivers.append((85, f"Strong alkaline shift (pH {r.ph:.2f})")) elif r.ph >= PH_INFECTED: drivers.append((65, f"Alkaline pH suggests infection (pH {r.ph:.2f})")) elif r.ph < 6.0: drivers.append((55, f"Acidic pH consistent with ischemia (pH {r.ph:.2f})")) # Velocity drivers if velocity.get("temp_delta_6h", 0.0) >= 1.0: drivers.append((75, f"Temperature rising rapidly (+{velocity['temp_delta_6h']:.1f} °C / 6h)")) if velocity.get("ph_delta_6h", 0.0) >= 0.5: drivers.append((70, f"Alkaline pH shift (+{velocity['ph_delta_6h']:.2f} / 6h)")) if velocity.get("humidity_volatility_6h", 0.0) >= HUMIDITY_VOLATILITY_ERRATIC: drivers.append((50, f"Erratic humidity (σ = {velocity['humidity_volatility_6h']:.1f}%)")) # Synergy if r.temperature_c >= TEMP_FEBRILE and r.ph >= PH_INFECTED: drivers.append((95, "Coupled febrile + alkaline pattern (classic SSI signature)")) # Context modifier if prediction.context_factor > 1.0: modifiers = [] if prediction.context_factor >= 1.3: modifiers.append("high-risk patient context") elif prediction.context_factor >= 1.2: modifiers.append("elevated patient sensitivity") if modifiers: drivers.append((30, f"Thresholds tightened for {modifiers[0]} (×{prediction.context_factor:.2f})")) # Dedupe + keep the highest-weighted three seen = set() ordered: list[str] = [] for _, txt in sorted(drivers, key=lambda x: -x[0]): if txt in seen: continue seen.add(txt) ordered.append(txt) if len(ordered) >= 4: break return ordered # ----- fusion quality from contributions ------------------------------ @staticmethod def _fusion_quality(contribs: dict[str, float]) -> float: """0..1 entropy of the three-axis contribution mix. 1 = every axis contributing roughly equally (full triple-sensor fusion); 0 = one axis carrying the whole signal alone.""" if not contribs: return 1.0 values = [v for v in contribs.values() if v > 0] if not values: return 1.0 # Normalise (defensive — the model returns ratios already) total = sum(values) ps = [v / total for v in values] entropy = -sum(p * math.log(p) for p in ps if p > 0) max_entropy = math.log(len(ps)) return float(entropy / max_entropy) if max_entropy > 0 else 1.0 # ---- Module-level convenience API ----------------------------------------- _default_engine = SepsisRiskEngine() def assess_reading( temperature_c: float, ph: float, humidity: float, *, engine: SepsisRiskEngine | None = None, factors: PatientFactors | None = None, ) -> dict: """One-shot assessment. Pass an ``engine`` to retain velocity context.""" eng = engine or SepsisRiskEngine() return eng.assess(Reading(temperature_c, ph, humidity), factors).to_dict()