Download backend/tests/test_ai_engine.py from Shwy1001/BioPatch: direct link, hf CLI and curl.
- Browser
- Download file 5.8 kB
-
https://huggingface.co/spaces/Shwy1001/BioPatch/resolve/main/backend/tests/test_ai_engine.py
- Command line
-
hf download hf://spaces/Shwy1001/BioPatch/backend/tests/test_ai_engine.py
-
curl -L -o test_ai_engine.py https://huggingface.co/spaces/Shwy1001/BioPatch/resolve/main/backend/tests/test_ai_engine.py
5.8 kB
| """Smoke tests for the BioPatch AI engine. | |
| Run with `python -m pytest backend/tests` once pytest is installed, or | |
| just `python backend/tests/test_ai_engine.py` for a manual sanity check. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from datetime import datetime, timedelta, timezone | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from ai.features import PatientFactors # noqa: E402 | |
| from core.ai_engine import Reading, SepsisRiskEngine, assess_reading # noqa: E402 | |
| def test_stable_healing_reading_is_low_risk() -> None: | |
| result = assess_reading(temperature_c=33.5, ph=6.0, humidity=75.0) | |
| assert result["score"] < 25, result | |
| assert result["severity"] == "stable" | |
| assert result["bacterial_signature"] is None | |
| def test_classic_ssi_pattern_is_critical() -> None: | |
| result = assess_reading(temperature_c=38.6, ph=8.1, humidity=82.0) | |
| assert result["score"] >= 75, result | |
| assert result["severity"] == "critical" | |
| assert result["bacterial_signature"] is not None | |
| def test_pseudomonas_signature_detection() -> None: | |
| result = assess_reading(temperature_c=37.8, ph=8.3, humidity=88.0) | |
| sig = result["bacterial_signature"] or "" | |
| assert "Pseudomonas" in sig, result | |
| def test_velocity_term_flags_rapid_co_deterioration() -> None: | |
| engine = SepsisRiskEngine() | |
| base = datetime.now(timezone.utc) | |
| # Baseline reading (6h earlier, in the velocity window) | |
| engine.assess(Reading(36.8, 7.0, 78.0, timestamp=base - timedelta(hours=5))) | |
| # Now: temp + pH have risen significantly | |
| later = engine.assess(Reading(38.4, 7.9, 80.0, timestamp=base)) | |
| drivers_text = " | ".join(later.drivers) | |
| assert "rising" in drivers_text.lower() or "shift" in drivers_text.lower(), later | |
| def test_scores_are_bounded() -> None: | |
| extreme = assess_reading(temperature_c=42.0, ph=10.0, humidity=99.0) | |
| assert 0.0 <= extreme["score"] <= 100.0 | |
| def test_granulation_classification() -> None: | |
| result = assess_reading(temperature_c=34.5, ph=7.2, humidity=80.0) | |
| assert result["tissue_state"] == "granulation" | |
| assert result["tissue_confidence"] >= 0.65 | |
| assert "Granulação" in result["tissue_state_label_pt"] | |
| def test_devitalized_classification_ischemic_acidic() -> None: | |
| """Spec: low temp + acidic pH (<6.5) -> devitalized.""" | |
| result = assess_reading(temperature_c=30.5, ph=6.0, humidity=60.0) | |
| assert result["tissue_state"] == "devitalized", result | |
| assert "Desvitalização" in result["tissue_state_label_pt"] | |
| def test_inflammatory_tissue_state() -> None: | |
| result = assess_reading(temperature_c=37.8, ph=7.6, humidity=78.0) | |
| assert result["tissue_state"] == "inflammatory" | |
| def test_fusion_quality_penalises_isolated_axis_spike() -> None: | |
| isolated = assess_reading(temperature_c=33.5, ph=8.4, humidity=78.0) | |
| full = assess_reading(temperature_c=38.4, ph=8.4, humidity=88.0) | |
| # Sensor contributions should be more concentrated in the isolated case | |
| assert isolated["fusion_quality"] < full["fusion_quality"], (isolated, full) | |
| assert full["score"] > isolated["score"] | |
| def test_xai_sensor_contributions_sum_to_one() -> None: | |
| result = assess_reading(temperature_c=38.2, ph=7.6, humidity=84.0) | |
| contribs = result["sensor_contributions"] | |
| assert set(contribs) == {"temperature", "ph", "humidity"} | |
| total = sum(contribs.values()) | |
| assert 0.97 <= total <= 1.03, total | |
| def test_context_factor_raises_with_risk_factors() -> None: | |
| base_result = assess_reading( | |
| temperature_c=37.2, ph=7.3, humidity=80.0, | |
| factors=PatientFactors(age=45, diabetes=False, smoker=False), | |
| ) | |
| frail_result = assess_reading( | |
| temperature_c=37.2, ph=7.3, humidity=80.0, | |
| factors=PatientFactors(age=78, diabetes=True, smoker=True), | |
| ) | |
| assert frail_result["context_factor"] > base_result["context_factor"] | |
| # Same sensor numbers, more fragile patient -> at least same severity bucket | |
| sev_rank = {"stable": 0, "watch": 1, "warning": 2, "critical": 3} | |
| assert sev_rank[frail_result["severity"]] >= sev_rank[base_result["severity"]] | |
| def test_model_loaded_and_returns_class_probabilities() -> None: | |
| result = assess_reading(temperature_c=36.0, ph=7.0, humidity=78.0) | |
| probs = result["class_probabilities"] | |
| assert set(probs) == {"stable", "watch", "warning", "critical"} | |
| assert 0.97 <= sum(probs.values()) <= 1.03 | |
| assert result["model_version"].startswith("rf-") | |
| def test_feature_importance_reasonable_after_training() -> None: | |
| result = assess_reading(temperature_c=36.0, ph=7.0, humidity=78.0) | |
| fi = result["feature_importance"] | |
| expected = { | |
| "temperature_c", "ph", "humidity", | |
| "temp_delta_6h", "ph_delta_6h", "humidity_volatility_6h", | |
| "age", "diabetes", "smoker", | |
| } | |
| assert set(fi) == expected | |
| # The three sensors should carry most of the model's signal | |
| sensor_sum = fi["temperature_c"] + fi["ph"] + fi["humidity"] | |
| velocity_sum = fi["temp_delta_6h"] + fi["ph_delta_6h"] + fi["humidity_volatility_6h"] | |
| assert sensor_sum + velocity_sum > 0.7 | |
| if __name__ == "__main__": | |
| test_stable_healing_reading_is_low_risk() | |
| test_classic_ssi_pattern_is_critical() | |
| test_pseudomonas_signature_detection() | |
| test_velocity_term_flags_rapid_co_deterioration() | |
| test_scores_are_bounded() | |
| test_granulation_classification() | |
| test_devitalized_classification_ischemic_acidic() | |
| test_inflammatory_tissue_state() | |
| test_fusion_quality_penalises_isolated_axis_spike() | |
| test_xai_sensor_contributions_sum_to_one() | |
| test_context_factor_raises_with_risk_factors() | |
| test_model_loaded_and_returns_class_probabilities() | |
| test_feature_importance_reasonable_after_training() | |
| print("OK - all ai_engine smoke tests passed") | |