Spaces:

Shwy1001
/

BioPatch

Sleeping

Jerry

BioPatch backend (clean state, history rewritten)

10795db 4 months ago

15.4 kB

"""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()

	"""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()