Spaces:

Shwy1001
/

BioPatch

Sleeping

BioPatch / backend /tests /test_jarvis.py

Jerry

BioPatch backend (clean state, history rewritten)

10795db 4 months ago

11.2 kB

"""Tests for the BioPatch Jarvis explainability + RAG layer."""

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

import pytest

from ai.features import PatientFactors  # noqa: E402
from ai.jarvis import (  # noqa: E402
    AutonomousMonitor,
    ClinicalBrain,
    JarvisPredictor,
    PatientSummary,
    ReasonerInputs,
    active_risk_factors,
    cite,
    get_brain,
    get_monitor,
    get_rag,
    get_reasoner,
    get_store,
    ontology_snapshot,
)
from ai.jarvis.knowledge_store import KnowledgeStore  # noqa: E402
from ai.jarvis.medical_rag import MedicalRAG, expand_query  # noqa: E402
from ai.jarvis.recommendations import build_recommendations  # noqa: E402


# ---------------------------------------------------------------------------
# Ontology
# ---------------------------------------------------------------------------


def test_ontology_snapshot_has_three_sensors() -> None:
    snap = ontology_snapshot()
    assert set(snap["sensors"].keys()) == {"ph", "temperature_c", "humidity"}
    for sensor in snap["sensors"].values():
        assert sensor["bands"], "every sensor must declare bands"
        for band in sensor["bands"]:
            assert band["severity"] in {"stable", "watch", "warning", "critical"}
            assert band["rationale"]
            assert band["citations"], "every band must point at a citation"


def test_active_risk_factors_diabetic_elderly() -> None:
    rfs = active_risk_factors(age=82, diabetes=True, smoker=False)
    keys = {rf.key for rf in rfs}
    assert keys == {"diabetes", "elderly_80"}


def test_cite_returns_known_metadata() -> None:
    out = cite("WHO_SSI_2018")
    assert out and out[0]["publisher"] == "World Health Organization"


def test_cite_skips_unknown_keys() -> None:
    assert cite("does-not-exist") == []


# ---------------------------------------------------------------------------
# Knowledge store + RAG
# ---------------------------------------------------------------------------


def test_store_bootstrapped_from_corpus() -> None:
    store = get_store()
    assert store.count() >= 25
    assert "ph" in store.topics()


def test_search_returns_relevant_passages() -> None:
    rag = get_rag()
    hits = rag.search("Why is alkaline pH a sign of infection?", top_k=3)
    assert hits, "expected at least one hit"
    # Top result should be ph-related
    assert hits[0].passage.topic == "ph"


def test_search_supports_language_filter() -> None:
    rag = get_rag()
    hits_fr = rag.search("humidite cicatrisation", top_k=3, lang="fr")
    assert any(h.passage.lang == "fr" for h in hits_fr)


def test_expand_query_adds_synonyms() -> None:
    expanded = expand_query("pH infection")
    assert "alkaline" in expanded
    assert "SSI" in expanded


def test_writable_upsert(tmp_path: Path) -> None:
    store = KnowledgeStore(store_path=tmp_path / "jarvis.sqlite")
    passage = store.upsert(
        id="custom-test-1",
        topic="custom",
        tags=["unit-test"],
        lang="en",
        text="Unit-test snippet describing arbitrary clinical content for retrieval.",
        citations=["WHO_SSI_2018"],
    )
    assert store.get_by_id("custom-test-1") is not None
    rag = MedicalRAG(store)
    hits = rag.search("arbitrary clinical content", top_k=1, expand=False)
    assert hits and hits[0].passage.id == passage.id


def test_explain_threshold_returns_context() -> None:
    rag = get_rag()
    ctx = rag.explain_threshold("ph", lang="en")
    assert ctx.hits, "expected at least one passage"
    assert ctx.citation_keys


# ---------------------------------------------------------------------------
# Explainable predictor
# ---------------------------------------------------------------------------


def test_predictor_marginal_attribution_signs() -> None:
    from core.ai_engine import get_clinical_model

    predictor = JarvisPredictor(get_clinical_model())
    pred = predictor.predict(
        temperature_c=38.4,
        ph=8.0,
        humidity=88.0,
        velocity={"temp_delta_6h": 1.4, "ph_delta_6h": 0.7, "humidity_volatility_6h": 4.0},
        factors=PatientFactors(age=70, diabetes=True, smoker=False),
    )
    assert pred.base.severity in ("warning", "critical")
    top = pred.top_features(3)
    assert len(top) == 3
    # At least one of the temperature / pH features should push the score up
    assert any(a.delta_score > 0 for a in top if a.feature in {"ph", "temperature_c"})


def test_predictor_returns_band_hits_in_order() -> None:
    from core.ai_engine import get_clinical_model

    predictor = JarvisPredictor(get_clinical_model())
    pred = predictor.predict(
        temperature_c=34.0,
        ph=6.0,
        humidity=78.0,
        velocity={"temp_delta_6h": 0.0, "ph_delta_6h": 0.0, "humidity_volatility_6h": 1.0},
        factors=PatientFactors(),
    )
    params = [bh.parameter for bh in pred.band_hits]
    assert params == ["temperature_c", "ph", "humidity"]
    assert all(bh.band.severity == "stable" for bh in pred.band_hits)


# ---------------------------------------------------------------------------
# Recommendations
# ---------------------------------------------------------------------------


def test_recommendations_for_critical_include_culture() -> None:
    from core.ai_engine import get_clinical_model

    predictor = JarvisPredictor(get_clinical_model())
    pred = predictor.predict(
        temperature_c=38.8,
        ph=8.2,
        humidity=90.0,
        velocity={"temp_delta_6h": 1.6, "ph_delta_6h": 0.8, "humidity_volatility_6h": 5.0},
        factors=PatientFactors(age=72, diabetes=True, smoker=False),
    )
    recs = build_recommendations(prediction=pred, bacterial_signature="Pseudomonas aeruginosa (suspected)")
    assert any("culture" in r.text_en.lower() for r in recs)
    assert any(r.tier == "immediate" for r in recs)
    assert any("piperacillin" in r.text_en.lower() or "pseudomonas" in r.text_en.lower() for r in recs)


def test_recommendations_for_stable_are_empty_or_routine() -> None:
    from core.ai_engine import get_clinical_model

    predictor = JarvisPredictor(get_clinical_model())
    pred = predictor.predict(
        temperature_c=33.8,
        ph=6.0,
        humidity=78.0,
        velocity={"temp_delta_6h": 0.0, "ph_delta_6h": 0.0, "humidity_volatility_6h": 0.5},
        factors=PatientFactors(),
    )
    recs = build_recommendations(prediction=pred)
    assert all(r.tier in {"monitoring", "prevention"} for r in recs)
    assert not any(r.tier == "immediate" for r in recs)


# ---------------------------------------------------------------------------
# Clinical brain
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("lang", ["en", "fr", "pt"])
def test_clinical_note_languages(lang: str) -> None:
    from core.ai_engine import get_clinical_model

    predictor = JarvisPredictor(get_clinical_model())
    pred = predictor.predict(
        temperature_c=38.0,
        ph=7.7,
        humidity=89.0,
        velocity={"temp_delta_6h": 1.0, "ph_delta_6h": 0.5, "humidity_volatility_6h": 4.0},
        factors=PatientFactors(age=68, diabetes=True, smoker=False),
    )
    recs = build_recommendations(prediction=pred, bacterial_signature="Staphylococcus aureus (suspected)")
    brain = get_brain()
    note = brain.generate_note(
        patient=PatientSummary(
            name="Test Patient",
            age=68,
            sex="F",
            surgery_type="colectomy",
            days_post_op=3,
            diabetes=True,
            smoker=False,
        ),
        prediction=pred,
        recommendations=recs,
        rag_context=None,
        bacterial_signature="Staphylococcus aureus (suspected)",
        signature_confidence=0.72,
        lang=lang,
    )
    assert note.lang == lang
    assert note.headline
    assert len(note.body) > 200
    assert note.citations
    assert note.summary_bullets


# ---------------------------------------------------------------------------
# Reasoner orchestration
# ---------------------------------------------------------------------------


def test_reasoner_full_pipeline_serialisable() -> None:
    reasoner = get_reasoner()
    inputs = ReasonerInputs(
        patient=PatientSummary(
            name="Maria Silva",
            age=66,
            sex="F",
            surgery_type="appendectomy",
            days_post_op=2,
            diabetes=True,
            smoker=False,
        ),
        temperature_c=37.9,
        ph=7.7,
        humidity=88.0,
        velocity={"temp_delta_6h": 1.2, "ph_delta_6h": 0.6, "humidity_volatility_6h": 4.5},
        factors=PatientFactors(age=66, diabetes=True, smoker=False),
        bacterial_signature="Staphylococcus aureus (suspected)",
        signature_confidence=0.72,
        tissue_state="inflammatory",
        tissue_confidence=0.81,
        lang="en",
    )
    analysis = reasoner.analyze(inputs)
    payload = analysis.to_dict()
    # All nested dataclasses serialise cleanly
    assert payload["prediction"]["severity"] in {"warning", "critical"}
    assert payload["clinical_note"]["body"]
    assert payload["recommendations"]
    assert payload["rag_context"]["passages"]
    # The serialised dict should be JSON-roundtrippable
    import json

    json.loads(json.dumps(payload))


# ---------------------------------------------------------------------------
# Autonomous monitor
# ---------------------------------------------------------------------------


def _synthetic_history(values: list[tuple[float, float, float]], start: datetime | None = None):
    start = start or datetime.now(timezone.utc) - timedelta(hours=5)
    rows = []
    for i, (t, p, h) in enumerate(values):
        rows.append(
            {
                "timestamp": start + timedelta(minutes=i * 5),
                "temperature_c": t,
                "ph": p,
                "humidity": h,
            }
        )
    return rows


def test_autonomous_monitor_flags_coupled_climb() -> None:
    monitor = get_monitor()
    values = []
    for i in range(20):
        values.append((33.0 + i * 0.1, 6.5 + i * 0.05, 78.0))
    report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
    assert report.findings, "expected at least one finding"
    kinds = {f.kind for f in report.findings}
    assert "coupled_climb" in kinds


def test_autonomous_monitor_stable_baseline_quiet() -> None:
    monitor = AutonomousMonitor()
    values = [(34.0, 6.2, 80.0)] * 30
    report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
    assert report.composite_score < 0.2
    assert report.severity_hint == "stable"


def test_autonomous_monitor_detects_humidity_drift() -> None:
    monitor = AutonomousMonitor()
    values = []
    for i in range(24):
        values.append((34.0, 6.2, 75.0 + i * 0.7))
    report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
    kinds = {f.kind for f in report.findings}
    assert "humidity_drift" in kinds

	"""Tests for the BioPatch Jarvis explainability + RAG layer."""

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

	import pytest

	from ai.features import PatientFactors # noqa: E402
	from ai.jarvis import ( # noqa: E402
	AutonomousMonitor,
	ClinicalBrain,
	JarvisPredictor,
	PatientSummary,
	ReasonerInputs,
	active_risk_factors,
	cite,
	get_brain,
	get_monitor,
	get_rag,
	get_reasoner,
	get_store,
	ontology_snapshot,
	)
	from ai.jarvis.knowledge_store import KnowledgeStore # noqa: E402
	from ai.jarvis.medical_rag import MedicalRAG, expand_query # noqa: E402
	from ai.jarvis.recommendations import build_recommendations # noqa: E402


	# ---------------------------------------------------------------------------
	# Ontology
	# ---------------------------------------------------------------------------


	def test_ontology_snapshot_has_three_sensors() -> None:
	snap = ontology_snapshot()
	assert set(snap["sensors"].keys()) == {"ph", "temperature_c", "humidity"}
	for sensor in snap["sensors"].values():
	assert sensor["bands"], "every sensor must declare bands"
	for band in sensor["bands"]:
	assert band["severity"] in {"stable", "watch", "warning", "critical"}
	assert band["rationale"]
	assert band["citations"], "every band must point at a citation"


	def test_active_risk_factors_diabetic_elderly() -> None:
	rfs = active_risk_factors(age=82, diabetes=True, smoker=False)
	keys = {rf.key for rf in rfs}
	assert keys == {"diabetes", "elderly_80"}


	def test_cite_returns_known_metadata() -> None:
	out = cite("WHO_SSI_2018")
	assert out and out[0]["publisher"] == "World Health Organization"


	def test_cite_skips_unknown_keys() -> None:
	assert cite("does-not-exist") == []


	# ---------------------------------------------------------------------------
	# Knowledge store + RAG
	# ---------------------------------------------------------------------------


	def test_store_bootstrapped_from_corpus() -> None:
	store = get_store()
	assert store.count() >= 25
	assert "ph" in store.topics()


	def test_search_returns_relevant_passages() -> None:
	rag = get_rag()
	hits = rag.search("Why is alkaline pH a sign of infection?", top_k=3)
	assert hits, "expected at least one hit"
	# Top result should be ph-related
	assert hits[0].passage.topic == "ph"


	def test_search_supports_language_filter() -> None:
	rag = get_rag()
	hits_fr = rag.search("humidite cicatrisation", top_k=3, lang="fr")
	assert any(h.passage.lang == "fr" for h in hits_fr)


	def test_expand_query_adds_synonyms() -> None:
	expanded = expand_query("pH infection")
	assert "alkaline" in expanded
	assert "SSI" in expanded


	def test_writable_upsert(tmp_path: Path) -> None:
	store = KnowledgeStore(store_path=tmp_path / "jarvis.sqlite")
	passage = store.upsert(
	id="custom-test-1",
	topic="custom",
	tags=["unit-test"],
	lang="en",
	text="Unit-test snippet describing arbitrary clinical content for retrieval.",
	citations=["WHO_SSI_2018"],
	)
	assert store.get_by_id("custom-test-1") is not None
	rag = MedicalRAG(store)
	hits = rag.search("arbitrary clinical content", top_k=1, expand=False)
	assert hits and hits[0].passage.id == passage.id


	def test_explain_threshold_returns_context() -> None:
	rag = get_rag()
	ctx = rag.explain_threshold("ph", lang="en")
	assert ctx.hits, "expected at least one passage"
	assert ctx.citation_keys


	# ---------------------------------------------------------------------------
	# Explainable predictor
	# ---------------------------------------------------------------------------


	def test_predictor_marginal_attribution_signs() -> None:
	from core.ai_engine import get_clinical_model

	predictor = JarvisPredictor(get_clinical_model())
	pred = predictor.predict(
	temperature_c=38.4,
	ph=8.0,
	humidity=88.0,
	velocity={"temp_delta_6h": 1.4, "ph_delta_6h": 0.7, "humidity_volatility_6h": 4.0},
	factors=PatientFactors(age=70, diabetes=True, smoker=False),
	)
	assert pred.base.severity in ("warning", "critical")
	top = pred.top_features(3)
	assert len(top) == 3
	# At least one of the temperature / pH features should push the score up
	assert any(a.delta_score > 0 for a in top if a.feature in {"ph", "temperature_c"})


	def test_predictor_returns_band_hits_in_order() -> None:
	from core.ai_engine import get_clinical_model

	predictor = JarvisPredictor(get_clinical_model())
	pred = predictor.predict(
	temperature_c=34.0,
	ph=6.0,
	humidity=78.0,
	velocity={"temp_delta_6h": 0.0, "ph_delta_6h": 0.0, "humidity_volatility_6h": 1.0},
	factors=PatientFactors(),
	)
	params = [bh.parameter for bh in pred.band_hits]
	assert params == ["temperature_c", "ph", "humidity"]
	assert all(bh.band.severity == "stable" for bh in pred.band_hits)


	# ---------------------------------------------------------------------------
	# Recommendations
	# ---------------------------------------------------------------------------


	def test_recommendations_for_critical_include_culture() -> None:
	from core.ai_engine import get_clinical_model

	predictor = JarvisPredictor(get_clinical_model())
	pred = predictor.predict(
	temperature_c=38.8,
	ph=8.2,
	humidity=90.0,
	velocity={"temp_delta_6h": 1.6, "ph_delta_6h": 0.8, "humidity_volatility_6h": 5.0},
	factors=PatientFactors(age=72, diabetes=True, smoker=False),
	)
	recs = build_recommendations(prediction=pred, bacterial_signature="Pseudomonas aeruginosa (suspected)")
	assert any("culture" in r.text_en.lower() for r in recs)
	assert any(r.tier == "immediate" for r in recs)
	assert any("piperacillin" in r.text_en.lower() or "pseudomonas" in r.text_en.lower() for r in recs)


	def test_recommendations_for_stable_are_empty_or_routine() -> None:
	from core.ai_engine import get_clinical_model

	predictor = JarvisPredictor(get_clinical_model())
	pred = predictor.predict(
	temperature_c=33.8,
	ph=6.0,
	humidity=78.0,
	velocity={"temp_delta_6h": 0.0, "ph_delta_6h": 0.0, "humidity_volatility_6h": 0.5},
	factors=PatientFactors(),
	)
	recs = build_recommendations(prediction=pred)
	assert all(r.tier in {"monitoring", "prevention"} for r in recs)
	assert not any(r.tier == "immediate" for r in recs)


	# ---------------------------------------------------------------------------
	# Clinical brain
	# ---------------------------------------------------------------------------


	@pytest.mark.parametrize("lang", ["en", "fr", "pt"])
	def test_clinical_note_languages(lang: str) -> None:
	from core.ai_engine import get_clinical_model

	predictor = JarvisPredictor(get_clinical_model())
	pred = predictor.predict(
	temperature_c=38.0,
	ph=7.7,
	humidity=89.0,
	velocity={"temp_delta_6h": 1.0, "ph_delta_6h": 0.5, "humidity_volatility_6h": 4.0},
	factors=PatientFactors(age=68, diabetes=True, smoker=False),
	)
	recs = build_recommendations(prediction=pred, bacterial_signature="Staphylococcus aureus (suspected)")
	brain = get_brain()
	note = brain.generate_note(
	patient=PatientSummary(
	name="Test Patient",
	age=68,
	sex="F",
	surgery_type="colectomy",
	days_post_op=3,
	diabetes=True,
	smoker=False,
	),
	prediction=pred,
	recommendations=recs,
	rag_context=None,
	bacterial_signature="Staphylococcus aureus (suspected)",
	signature_confidence=0.72,
	lang=lang,
	)
	assert note.lang == lang
	assert note.headline
	assert len(note.body) > 200
	assert note.citations
	assert note.summary_bullets


	# ---------------------------------------------------------------------------
	# Reasoner orchestration
	# ---------------------------------------------------------------------------


	def test_reasoner_full_pipeline_serialisable() -> None:
	reasoner = get_reasoner()
	inputs = ReasonerInputs(
	patient=PatientSummary(
	name="Maria Silva",
	age=66,
	sex="F",
	surgery_type="appendectomy",
	days_post_op=2,
	diabetes=True,
	smoker=False,
	),
	temperature_c=37.9,
	ph=7.7,
	humidity=88.0,
	velocity={"temp_delta_6h": 1.2, "ph_delta_6h": 0.6, "humidity_volatility_6h": 4.5},
	factors=PatientFactors(age=66, diabetes=True, smoker=False),
	bacterial_signature="Staphylococcus aureus (suspected)",
	signature_confidence=0.72,
	tissue_state="inflammatory",
	tissue_confidence=0.81,
	lang="en",
	)
	analysis = reasoner.analyze(inputs)
	payload = analysis.to_dict()
	# All nested dataclasses serialise cleanly
	assert payload["prediction"]["severity"] in {"warning", "critical"}
	assert payload["clinical_note"]["body"]
	assert payload["recommendations"]
	assert payload["rag_context"]["passages"]
	# The serialised dict should be JSON-roundtrippable
	import json

	json.loads(json.dumps(payload))


	# ---------------------------------------------------------------------------
	# Autonomous monitor
	# ---------------------------------------------------------------------------


	def _synthetic_history(values: list[tuple[float, float, float]], start: datetime \| None = None):
	start = start or datetime.now(timezone.utc) - timedelta(hours=5)
	rows = []
	for i, (t, p, h) in enumerate(values):
	rows.append(
	{
	"timestamp": start + timedelta(minutes=i * 5),
	"temperature_c": t,
	"ph": p,
	"humidity": h,
	}
	)
	return rows


	def test_autonomous_monitor_flags_coupled_climb() -> None:
	monitor = get_monitor()
	values = []
	for i in range(20):
	values.append((33.0 + i * 0.1, 6.5 + i * 0.05, 78.0))
	report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
	assert report.findings, "expected at least one finding"
	kinds = {f.kind for f in report.findings}
	assert "coupled_climb" in kinds


	def test_autonomous_monitor_stable_baseline_quiet() -> None:
	monitor = AutonomousMonitor()
	values = [(34.0, 6.2, 80.0)] * 30
	report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
	assert report.composite_score < 0.2
	assert report.severity_hint == "stable"


	def test_autonomous_monitor_detects_humidity_drift() -> None:
	monitor = AutonomousMonitor()
	values = []
	for i in range(24):
	values.append((34.0, 6.2, 75.0 + i * 0.7))
	report = monitor.analyze(_synthetic_history(values), factors=PatientFactors())
	kinds = {f.kind for f in report.findings}
	assert "humidity_drift" in kinds