da34bdc8d7
- DPPoolCandidate model (terme, section, preuve, score_initial) - build_dp_candidate_pool() with filters (_is_pool_excluded, _dedup_pool) - Pool exclusion: admin noise, bio values, vague symptoms, place names - DP_POOL_RANK template for LLM-based ranking among pool candidates - llm_dp_pool_rank() with guardrails (GF-1 evidence, GF-3 confidence) - benchmark_quality.py: --dp-candidates, --use-dp-pool-rank flags - 41 new tests (pool, exclusion, dedup, pool rank, synthese) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1587 lines
62 KiB
Python
1587 lines
62 KiB
Python
"""Tests pour le module de scoring DP (Diagnostic Principal)."""
|
|
|
|
import pytest
|
|
|
|
from src.config import (
|
|
ActeCCAM,
|
|
DossierMedical,
|
|
Diagnostic,
|
|
DPCandidate,
|
|
DPPoolCandidate,
|
|
DPSelection,
|
|
DP_SCORING_WEIGHTS,
|
|
DP_REVIEW_THRESHOLD,
|
|
PreuveSynthese,
|
|
SynthesePMSI,
|
|
Sejour,
|
|
)
|
|
from src.medical.dp_scoring import (
|
|
build_dp_shortlist,
|
|
build_dp_candidate_pool,
|
|
score_candidates,
|
|
select_dp,
|
|
generate_synthese_pmsi,
|
|
llm_dp_pool_rank,
|
|
_format_pool_for_prompt,
|
|
_build_clinical_context,
|
|
_get_context_window,
|
|
_is_z_code_whitelisted,
|
|
_is_comorbidity_code,
|
|
_has_explicit_pec_proof,
|
|
_dedup_by_code,
|
|
_dedup_pool,
|
|
_is_pool_excluded,
|
|
_normalize_evidence_section,
|
|
_is_comorbidite_banale,
|
|
_has_pec_marker,
|
|
_has_acute_problem,
|
|
_postprocess_synthese,
|
|
_build_motif,
|
|
)
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
def _make_parsed(sections: dict | None = None, diagnostics: list | None = None) -> dict:
|
|
return {
|
|
"type": "crh",
|
|
"patient": {"sexe": "M"},
|
|
"sejour": {},
|
|
"diagnostics": diagnostics or [],
|
|
"sections": sections or {},
|
|
}
|
|
|
|
|
|
def _make_candidate(
|
|
code: str = "K85.1",
|
|
label: str = "Pancréatite aiguë biliaire",
|
|
source_section: str = "diag_sortie",
|
|
**kwargs,
|
|
) -> DPCandidate:
|
|
return DPCandidate(code=code, label=label, source_section=source_section, **kwargs)
|
|
|
|
|
|
# === Tests build_dp_shortlist ===
|
|
|
|
class TestBuildDPShortlist:
|
|
def test_from_diag_sortie_with_cim10_code(self):
|
|
parsed = _make_parsed(sections={
|
|
"diag_sortie": "Pancréatite aiguë biliaire K85.1",
|
|
})
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", None, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "K85.1" in codes
|
|
|
|
def test_from_diag_principal_section(self):
|
|
parsed = _make_parsed(sections={
|
|
"diag_principal": "Embolie pulmonaire I26.9",
|
|
})
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", None, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "I26.9" in codes
|
|
|
|
def test_from_conclusion_via_cim10_map(self):
|
|
parsed = _make_parsed(sections={
|
|
"conclusion": "pancréatite aiguë biliaire, bonne évolution",
|
|
})
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", None, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "K85.1" in codes
|
|
|
|
def test_from_regex_fallback(self):
|
|
parsed = _make_parsed(sections={})
|
|
text = "Au total : pancréatite aiguë biliaire.\nDevenir : retour."
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, text, None, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "K85.1" in codes
|
|
|
|
def test_from_edsnlp(self):
|
|
from src.medical.edsnlp_pipeline import EdsnlpResult, CIM10Entity
|
|
|
|
parsed = _make_parsed(sections={})
|
|
edsnlp = EdsnlpResult(cim10_entities=[
|
|
CIM10Entity(texte="douleur abdominale", code="R10.4", negation=False),
|
|
])
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", edsnlp, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "R10.4" in codes
|
|
|
|
def test_edsnlp_negated_excluded(self):
|
|
from src.medical.edsnlp_pipeline import EdsnlpResult, CIM10Entity
|
|
|
|
parsed = _make_parsed(sections={})
|
|
edsnlp = EdsnlpResult(cim10_entities=[
|
|
CIM10Entity(texte="fièvre", code="R50.9", negation=True),
|
|
])
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", edsnlp, dossier)
|
|
codes = [c.code for c in candidates]
|
|
assert "R50.9" not in codes
|
|
|
|
def test_dedup_keeps_strongest_section(self):
|
|
"""Si le même code vient de diag_sortie et conclusion, garder diag_sortie."""
|
|
parsed = _make_parsed(sections={
|
|
"diag_sortie": "Pancréatite K85.1",
|
|
"conclusion": "pancréatite K85.1 bonne évolution",
|
|
})
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "", None, dossier)
|
|
k85_candidates = [c for c in candidates if c.code == "K85.1"]
|
|
assert len(k85_candidates) == 1
|
|
assert k85_candidates[0].source_section == "diag_sortie"
|
|
|
|
def test_empty_sections_returns_empty(self):
|
|
parsed = _make_parsed(sections={})
|
|
dossier = DossierMedical()
|
|
candidates = build_dp_shortlist(parsed, "Patient en bon état.", None, dossier)
|
|
assert candidates == []
|
|
|
|
|
|
# === Tests score_candidates ===
|
|
|
|
class TestScoreCandidates:
|
|
def test_section_bonus_diag_sortie(self):
|
|
c = _make_candidate(source_section="diag_sortie")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_diag_sortie"]
|
|
|
|
def test_section_bonus_conclusion(self):
|
|
c = _make_candidate(source_section="conclusion")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_conclusion"]
|
|
|
|
def test_section_bonus_edsnlp(self):
|
|
c = _make_candidate(source_section="edsnlp")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("section") == DP_SCORING_WEIGHTS["section_edsnlp"]
|
|
|
|
def test_proof_excerpt_bonus(self):
|
|
c = _make_candidate(source_excerpt="Pancréatite aiguë biliaire confirmée au scanner")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("proof_excerpt") == DP_SCORING_WEIGHTS["proof_excerpt"]
|
|
|
|
def test_no_proof_bonus_without_excerpt(self):
|
|
c = _make_candidate(source_excerpt=None)
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert "proof_excerpt" not in scored[0].score_details
|
|
|
|
def test_negation_penalty(self):
|
|
c = _make_candidate(label="Fièvre")
|
|
text = "Pas de fièvre constatée."
|
|
scored = score_candidates([c], DossierMedical(), full_text=text)
|
|
assert scored[0].is_negated is True
|
|
assert scored[0].score_details.get("negation") == DP_SCORING_WEIGHTS["negation"]
|
|
|
|
def test_conditional_penalty(self):
|
|
c = _make_candidate(label="Embolie pulmonaire", code="I26.9")
|
|
text = "Embolie pulmonaire suspectée, à confirmer par angioscanner."
|
|
scored = score_candidates([c], DossierMedical(), full_text=text)
|
|
assert scored[0].is_conditional is True
|
|
assert scored[0].score_details.get("conditional") == DP_SCORING_WEIGHTS["conditional"]
|
|
|
|
def test_z_code_penalty(self):
|
|
c = _make_candidate(code="Z76.0", label="Bilan de santé", source_section="conclusion")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("z_code_dp") == DP_SCORING_WEIGHTS["z_code_dp"]
|
|
|
|
def test_z_code_whitelist_no_penalty(self):
|
|
c = _make_candidate(code="Z51.1", label="Chimiothérapie", source_section="conclusion")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert "z_code_dp" not in scored[0].score_details
|
|
|
|
def test_r_code_penalty(self):
|
|
c = _make_candidate(code="R10.4", label="Douleur abdominale", source_section="edsnlp")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert scored[0].score_details.get("r_code_dp") == DP_SCORING_WEIGHTS["r_code_dp"]
|
|
|
|
def test_sort_by_score_descending(self):
|
|
c1 = _make_candidate(code="K85.1", source_section="diag_sortie")
|
|
c2 = _make_candidate(code="R10.4", label="Douleur", source_section="edsnlp")
|
|
scored = score_candidates([c2, c1], DossierMedical())
|
|
assert scored[0].code == "K85.1" # diag_sortie score > edsnlp
|
|
|
|
def test_combined_scoring(self):
|
|
"""Score = section bonus + proof - negation penalties."""
|
|
c = _make_candidate(
|
|
code="K85.1",
|
|
source_section="diag_sortie",
|
|
source_excerpt="Pancréatite aiguë",
|
|
)
|
|
scored = score_candidates([c], DossierMedical())
|
|
expected = DP_SCORING_WEIGHTS["section_diag_sortie"] + DP_SCORING_WEIGHTS["proof_excerpt"]
|
|
assert scored[0].score == expected
|
|
|
|
|
|
# === Tests select_dp ===
|
|
|
|
class TestSelectDP:
|
|
def test_no_candidates_returns_review(self):
|
|
sel = select_dp([], DossierMedical())
|
|
assert sel.verdict == "review"
|
|
|
|
def test_single_candidate_confirmed(self):
|
|
c = _make_candidate()
|
|
c.score = 6
|
|
sel = select_dp([c], DossierMedical())
|
|
assert sel.verdict == "confirmed"
|
|
assert sel.winner_reason == "candidat unique"
|
|
|
|
def test_clear_winner_confirmed(self):
|
|
c1 = _make_candidate(code="K85.1")
|
|
c1.score = 6
|
|
c2 = _make_candidate(code="R10.4", label="Douleur", source_section="edsnlp")
|
|
c2.score = 1
|
|
sel = select_dp([c1, c2], DossierMedical())
|
|
assert sel.verdict == "confirmed"
|
|
assert "delta" in sel.winner_reason
|
|
|
|
def test_close_scores_returns_review(self):
|
|
c1 = _make_candidate(code="K85.1")
|
|
c1.score = 3
|
|
c2 = _make_candidate(code="K80.5", label="Lithiase", source_section="conclusion")
|
|
c2.score = 2
|
|
sel = select_dp([c1, c2], DossierMedical())
|
|
assert sel.verdict == "review"
|
|
|
|
def test_review_returns_top3(self):
|
|
candidates = [
|
|
_make_candidate(code=f"K8{i}.{i}", label=f"Diag {i}")
|
|
for i in range(5)
|
|
]
|
|
for i, c in enumerate(candidates):
|
|
c.score = 5 - i
|
|
# delta between top1 and top2 = 1, < DP_REVIEW_THRESHOLD
|
|
sel = select_dp(candidates, DossierMedical())
|
|
assert sel.verdict == "review"
|
|
assert len(sel.candidates) <= 3
|
|
|
|
|
|
# === Tests utilitaires ===
|
|
|
|
class TestContextWindow:
|
|
def test_finds_label_in_text(self):
|
|
text = "Patient admis pour pancréatite aiguë biliaire confirmée."
|
|
window = _get_context_window(text, "pancréatite aiguë", radius=50)
|
|
assert "pancréatite" in window.lower()
|
|
|
|
def test_returns_empty_when_not_found(self):
|
|
text = "Patient en bon état."
|
|
window = _get_context_window(text, "embolie pulmonaire")
|
|
assert window == ""
|
|
|
|
|
|
class TestZCodeWhitelist:
|
|
def test_z51_1_whitelisted(self):
|
|
assert _is_z_code_whitelisted("Z51.1") is True
|
|
|
|
def test_z45_prefix_whitelisted(self):
|
|
assert _is_z_code_whitelisted("Z45.80") is True
|
|
|
|
def test_z76_not_whitelisted(self):
|
|
assert _is_z_code_whitelisted("Z76.0") is False
|
|
|
|
|
|
class TestDedupByCode:
|
|
def test_dedup_same_code_keeps_strongest(self):
|
|
c1 = _make_candidate(code="K85.1", source_section="conclusion")
|
|
c2 = _make_candidate(code="K85.1", source_section="diag_sortie")
|
|
priority = ["diag_sortie", "diag_principal", "motif_hospitalisation", "conclusion", "synthese"]
|
|
result = _dedup_by_code([c1, c2], priority)
|
|
assert len(result) == 1
|
|
assert result[0].source_section == "diag_sortie"
|
|
|
|
def test_dedup_different_codes_kept(self):
|
|
c1 = _make_candidate(code="K85.1")
|
|
c2 = _make_candidate(code="K80.5", label="Lithiase")
|
|
priority = ["diag_sortie"]
|
|
result = _dedup_by_code([c1, c2], priority)
|
|
assert len(result) == 2
|
|
|
|
|
|
# === Tests intégration légère ===
|
|
|
|
class TestDPScoringIntegration:
|
|
def test_crh_with_diag_sortie_section(self):
|
|
"""Un CRH avec section 'Diagnostic de sortie' produit un dp_selection."""
|
|
from src.medical.cim10_extractor import extract_medical_info
|
|
|
|
parsed = {
|
|
"type": "crh",
|
|
"patient": {"sexe": "M"},
|
|
"sejour": {},
|
|
"diagnostics": [],
|
|
"sections": {
|
|
"diag_sortie": "Pancréatite aiguë biliaire K85.1",
|
|
},
|
|
}
|
|
text = "Diagnostic de sortie :\nPancréatite aiguë biliaire K85.1\n\nTraitement de sortie :\nParacétamol"
|
|
|
|
dossier = extract_medical_info(parsed, text)
|
|
assert dossier.diagnostic_principal is not None
|
|
assert dossier.diagnostic_principal.cim10_suggestion == "K85.1"
|
|
assert dossier.dp_selection is not None
|
|
assert dossier.dp_selection.verdict == "confirmed"
|
|
|
|
def test_llm_fallback_confirmed_high_strong_section(self):
|
|
"""LLM one-shot CONFIRMED : high confidence + section forte."""
|
|
from unittest.mock import patch
|
|
from src.medical.cim10_extractor import extract_medical_info
|
|
|
|
parsed = {
|
|
"type": "crh",
|
|
"patient": {"sexe": "M"},
|
|
"sejour": {},
|
|
"diagnostics": [],
|
|
"sections": {
|
|
"conclusion": "Pancréatite aiguë biliaire avec HTA connue.",
|
|
},
|
|
}
|
|
text = "Conclusion : Pancréatite aiguë biliaire avec HTA connue."
|
|
|
|
mock_result = {
|
|
"dp_code": "K85.1",
|
|
"dp_label": "Pancréatite aiguë biliaire",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "Pancréatite aiguë biliaire",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
dossier = extract_medical_info(parsed, text, use_rag=True)
|
|
|
|
assert dossier.dp_selection is not None
|
|
assert dossier.dp_selection.verdict == "confirmed"
|
|
assert dossier.diagnostic_principal is not None
|
|
assert dossier.diagnostic_principal.cim10_suggestion == "K85.1"
|
|
|
|
def test_llm_fallback_confirmed_conclusion_section(self):
|
|
"""LLM one-shot CONFIRMED : conclusion est section forte."""
|
|
from unittest.mock import patch
|
|
from src.medical.cim10_extractor import extract_medical_info
|
|
|
|
parsed = {
|
|
"type": "crh",
|
|
"patient": {"sexe": "M"},
|
|
"sejour": {},
|
|
"diagnostics": [],
|
|
"sections": {"conclusion": "Pneumopathie avec insuffisance rénale aiguë."},
|
|
}
|
|
text = "Conclusion : Pneumopathie avec insuffisance rénale aiguë."
|
|
|
|
mock_result = {
|
|
"dp_code": "J18.9",
|
|
"dp_label": "Pneumopathie, sans précision",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "Pneumopathie avec insuffisance rénale aiguë",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
dossier = extract_medical_info(parsed, text, use_rag=True)
|
|
|
|
assert dossier.dp_selection is not None
|
|
assert dossier.dp_selection.verdict == "confirmed"
|
|
assert dossier.diagnostic_principal is not None
|
|
|
|
def test_llm_fallback_review_weak_section(self):
|
|
"""LLM one-shot REVIEW : evidence de histoire_maladie (section faible) → guardrail."""
|
|
from unittest.mock import patch
|
|
from src.medical.dp_scoring import llm_dp_fallback
|
|
from src.config import DossierMedical, DPCandidate
|
|
|
|
parsed = {"type": "crh", "sections": {"histoire_maladie": "Dyspnée aiguë."}}
|
|
text = "Histoire de la maladie : Dyspnée aiguë."
|
|
dossier = DossierMedical()
|
|
dp_candidates = [DPCandidate(code="R06.0", label="Dyspnée", source_section="edsnlp")]
|
|
|
|
mock_result = {
|
|
"dp_code": "R06.0",
|
|
"dp_label": "Dyspnée",
|
|
"evidence_section": "histoire_maladie",
|
|
"evidence_excerpt": "Dyspnée aiguë",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates)
|
|
|
|
assert selection.verdict == "review"
|
|
assert len(selection.candidates) >= 1
|
|
|
|
def test_llm_fallback_review_low_confidence(self):
|
|
"""LLM one-shot REVIEW : confidence=medium → guardrail."""
|
|
from unittest.mock import patch
|
|
from src.medical.dp_scoring import llm_dp_fallback
|
|
from src.config import DossierMedical, DPCandidate
|
|
|
|
parsed = {"type": "crh", "sections": {"conclusion": "HTA connue, diabète équilibré."}}
|
|
text = "Conclusion : HTA connue, diabète équilibré."
|
|
dossier = DossierMedical()
|
|
dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")]
|
|
|
|
mock_result = {
|
|
"dp_code": "I10",
|
|
"dp_label": "Hypertension essentielle",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "HTA connue",
|
|
"confidence": "medium",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates)
|
|
|
|
assert selection.verdict == "review"
|
|
assert "confidence medium" in selection.winner_reason
|
|
|
|
def test_llm_fallback_guardrail_no_evidence(self):
|
|
"""Garde-fou : LLM renvoie evidence vide → REVIEW."""
|
|
from unittest.mock import patch
|
|
from src.medical.dp_scoring import llm_dp_fallback
|
|
from src.config import DossierMedical, DPCandidate
|
|
|
|
parsed = {"type": "crh", "sections": {"conclusion": "Pancréatite."}}
|
|
text = "Conclusion : Pancréatite."
|
|
dossier = DossierMedical()
|
|
dp_candidates = [DPCandidate(code="K85.9", label="Pancréatite", source_section="edsnlp")]
|
|
|
|
mock_result = {
|
|
"dp_code": "K85.9",
|
|
"dp_label": "Pancréatite aiguë",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates)
|
|
|
|
assert selection.verdict == "review"
|
|
|
|
def test_llm_fallback_guardrail_comorbidity_weak_section(self):
|
|
"""Garde-fou : HTA en section non-forte → REVIEW."""
|
|
from unittest.mock import patch
|
|
from src.medical.dp_scoring import llm_dp_fallback
|
|
from src.config import DossierMedical, DPCandidate
|
|
|
|
parsed = {"type": "crh", "sections": {"histoire_maladie": "Patient hypertendu."}}
|
|
text = "Histoire de la maladie : Patient hypertendu."
|
|
dossier = DossierMedical()
|
|
dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")]
|
|
|
|
mock_result = {
|
|
"dp_code": "I10",
|
|
"dp_label": "Hypertension essentielle",
|
|
"evidence_section": "histoire_maladie",
|
|
"evidence_excerpt": "Patient hypertendu",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates)
|
|
|
|
assert selection.verdict == "review"
|
|
|
|
def test_llm_fallback_comorbidity_in_strong_section(self):
|
|
"""I10 en section forte + high confidence → CONFIRMED (garde-fou GF-2 ne bloque pas)."""
|
|
from unittest.mock import patch
|
|
from src.medical.dp_scoring import llm_dp_fallback
|
|
from src.config import DossierMedical, DPCandidate
|
|
|
|
parsed = {"type": "crh", "sections": {"motif_hospitalisation": "HTA maligne."}}
|
|
text = "Motif d'hospitalisation : HTA maligne."
|
|
dossier = DossierMedical()
|
|
dp_candidates = [DPCandidate(code="I10", label="HTA", source_section="edsnlp")]
|
|
|
|
mock_result = {
|
|
"dp_code": "I10",
|
|
"dp_label": "Hypertension essentielle",
|
|
"evidence_section": "motif_hospitalisation",
|
|
"evidence_excerpt": "HTA maligne",
|
|
"confidence": "high",
|
|
}
|
|
with patch("src.medical.ollama_client.call_ollama", return_value=mock_result):
|
|
selection = llm_dp_fallback(parsed, text, dossier, dp_candidates=dp_candidates)
|
|
|
|
assert selection.verdict == "confirmed"
|
|
assert selection.candidates[0].code == "I10"
|
|
|
|
def test_no_llm_fallback_without_use_rag(self):
|
|
"""Sans use_rag, le fallback LLM ne se déclenche PAS."""
|
|
from src.medical.cim10_extractor import extract_medical_info
|
|
|
|
parsed = {
|
|
"type": "crh",
|
|
"patient": {"sexe": "M"},
|
|
"sejour": {},
|
|
"diagnostics": [],
|
|
"sections": {"conclusion": "Bonne évolution."},
|
|
}
|
|
text = "Conclusion : Bonne évolution."
|
|
|
|
dossier = extract_medical_info(parsed, text, use_rag=False)
|
|
# Sans use_rag → pas de fallback LLM → verdict review
|
|
assert dossier.dp_selection is not None
|
|
assert dossier.dp_selection.verdict == "review"
|
|
|
|
def test_trackare_dp_bypasses_scoring(self):
|
|
"""Un Trackare avec DP codé ne déclenche PAS le scoring."""
|
|
from src.medical.cim10_extractor import extract_medical_info
|
|
|
|
parsed = {
|
|
"type": "trackare",
|
|
"patient": {"sexe": "F"},
|
|
"sejour": {"date_entree": "01/01/2024", "date_sortie": "05/01/2024"},
|
|
"diagnostics": [
|
|
{"type": "Principal", "code_cim10": "K80.5", "libelle": "Calcul des canaux biliaires"},
|
|
],
|
|
}
|
|
text = "Calcul des canaux biliaires."
|
|
|
|
dossier = extract_medical_info(parsed, text)
|
|
assert dossier.diagnostic_principal is not None
|
|
assert dossier.diagnostic_principal.cim10_suggestion == "K80.5"
|
|
assert dossier.dp_selection is None # Trackare DP, pas de scoring
|
|
|
|
|
|
# === Tests comorbidité-banale DP ===
|
|
|
|
class TestComorbidityGuard:
|
|
"""Règle comorbidité-banale : I10/E66.x/E78.x/E11.x/D64.9 en DP → REVIEW
|
|
sauf preuve explicite de PEC principale."""
|
|
|
|
def test_is_comorbidity_expanded(self):
|
|
"""La liste élargie couvre I10, E66.*, E78.*, E11.*, D64.9."""
|
|
assert _is_comorbidity_code("I10") is True
|
|
assert _is_comorbidity_code("E66.0") is True
|
|
assert _is_comorbidity_code("E66.9") is True
|
|
assert _is_comorbidity_code("E78.0") is True
|
|
assert _is_comorbidity_code("E11.9") is True
|
|
assert _is_comorbidity_code("E11.0") is True
|
|
assert _is_comorbidity_code("D64.9") is True
|
|
# Pas comorbidité
|
|
assert _is_comorbidity_code("D64.0") is False
|
|
assert _is_comorbidity_code("E10.9") is False
|
|
assert _is_comorbidity_code("K85.1") is False
|
|
|
|
def test_sole_comorbidity_review(self):
|
|
"""Candidat unique comorbidité → REVIEW (même section forte)."""
|
|
c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion")
|
|
c.score = 4
|
|
c.score_details = {"section": 2, "proof_excerpt": 2, "comorbidity_weak": -3}
|
|
sel = select_dp([c], DossierMedical())
|
|
assert sel.verdict == "review"
|
|
assert "comorbidité banale" in sel.winner_reason
|
|
|
|
def test_comorbidity_top1_multi_review(self):
|
|
"""Comorbidité top1 parmi plusieurs → REVIEW."""
|
|
c1 = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation")
|
|
c1.score = 3
|
|
c1.score_details = {"section": 3, "comorbidity_weak": -3}
|
|
c2 = _make_candidate(code="K85.1", label="Pancréatite", source_section="edsnlp")
|
|
c2.score = 1
|
|
sel = select_dp([c1, c2], DossierMedical())
|
|
assert sel.verdict == "review"
|
|
assert "comorbidité banale" in sel.winner_reason
|
|
|
|
def test_comorbidity_with_pec_proof_confirmed(self):
|
|
"""Comorbidité + preuve PEC → CONFIRMED."""
|
|
c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation")
|
|
c.score = 3
|
|
c.score_details = {"section": 3, "comorbidity_weak": -3, "comorbidity_pec_proof": 3}
|
|
sel = select_dp([c], DossierMedical())
|
|
assert sel.verdict == "confirmed"
|
|
assert sel.winner_reason == "candidat unique"
|
|
|
|
def test_non_comorbidity_sole_confirmed(self):
|
|
"""Candidat unique non-comorbidité → CONFIRMED (pas affecté)."""
|
|
c = _make_candidate(code="K85.1", label="Pancréatite", source_section="conclusion")
|
|
c.score = 4
|
|
sel = select_dp([c], DossierMedical())
|
|
assert sel.verdict == "confirmed"
|
|
|
|
def test_score_comorbidity_penalty_strong_section(self):
|
|
"""Comorbidité pénalisée même en section forte (conclusion)."""
|
|
c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert "comorbidity_weak" in scored[0].score_details
|
|
assert scored[0].score_details["comorbidity_weak"] == DP_SCORING_WEIGHTS["comorbidity_weak"]
|
|
|
|
def test_score_comorbidity_penalty_motif(self):
|
|
"""Comorbidité pénalisée en motif_hospitalisation."""
|
|
c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation")
|
|
scored = score_candidates([c], DossierMedical())
|
|
assert "comorbidity_weak" in scored[0].score_details
|
|
|
|
def test_pec_proof_detected(self):
|
|
"""PEC proof détectée dans le texte → bonus dans score_details."""
|
|
c = _make_candidate(code="I10", label="Hta", source_section="motif_hospitalisation")
|
|
text = "Patient hospitalisé pour hta maligne résistante au traitement."
|
|
scored = score_candidates([c], DossierMedical(), full_text=text)
|
|
assert "comorbidity_pec_proof" in scored[0].score_details
|
|
assert scored[0].score_details["comorbidity_pec_proof"] > 0
|
|
|
|
def test_pec_proof_not_found(self):
|
|
"""Pas de PEC proof → pas de bonus."""
|
|
c = _make_candidate(code="E66.0", label="Obésité", source_section="conclusion")
|
|
text = "Patient obèse, pneumopathie communautaire."
|
|
scored = score_candidates([c], DossierMedical(), full_text=text)
|
|
assert "comorbidity_pec_proof" not in scored[0].score_details
|
|
|
|
def test_has_explicit_pec_proof_hospitalized(self):
|
|
"""Détection 'hospitalisé pour' + label."""
|
|
assert _has_explicit_pec_proof("hta", "Patient hospitalisé pour HTA maligne.") is True
|
|
|
|
def test_has_explicit_pec_proof_prise_en_charge(self):
|
|
"""Détection 'prise en charge' + label."""
|
|
assert _has_explicit_pec_proof("obésité", "Prise en charge de l'obésité morbide.") is True
|
|
|
|
def test_has_explicit_pec_proof_absent(self):
|
|
"""Pas de PEC proof pour un label non mentionné."""
|
|
assert _has_explicit_pec_proof("hta", "Patient admis pour douleur thoracique.") is False
|
|
|
|
def test_has_explicit_pec_proof_admission(self):
|
|
"""Détection 'admission pour' + label."""
|
|
assert _has_explicit_pec_proof("diabète", "Admission pour diabète déséquilibré.") is True
|
|
|
|
|
|
class TestSectionNormalization:
|
|
"""Tests pour _normalize_evidence_section — normalisation robuste."""
|
|
|
|
# --- Correspondances exactes existantes ---
|
|
|
|
def test_exact_conclusion(self):
|
|
assert _normalize_evidence_section("conclusion") == "conclusion"
|
|
|
|
def test_exact_synthese(self):
|
|
assert _normalize_evidence_section("synthèse") == "synthese"
|
|
|
|
def test_exact_motif_hospitalisation(self):
|
|
assert _normalize_evidence_section("motif_hospitalisation") == "motif_hospitalisation"
|
|
|
|
# --- Nouveaux alias exacts ---
|
|
|
|
def test_synthese_du_sejour(self):
|
|
assert _normalize_evidence_section("synthèse du séjour") == "synthese"
|
|
|
|
def test_synthese_du_sejour_ascii(self):
|
|
assert _normalize_evidence_section("synthese du sejour") == "synthese"
|
|
|
|
def test_conclusions_pluriel(self):
|
|
assert _normalize_evidence_section("conclusions") == "conclusion"
|
|
|
|
def test_secretariat_to_autres(self):
|
|
assert _normalize_evidence_section("secrétariat") == "autres"
|
|
|
|
def test_medecine_interne_to_autres(self):
|
|
assert _normalize_evidence_section("médecine interne") == "autres"
|
|
|
|
def test_sections_cliniques_to_autres(self):
|
|
assert _normalize_evidence_section("sections cliniques") == "autres"
|
|
|
|
# --- Nettoyage crochets/guillemets ---
|
|
|
|
def test_brackets_conclusion(self):
|
|
assert _normalize_evidence_section("[conclusion]") == "conclusion"
|
|
|
|
def test_brackets_motif(self):
|
|
assert _normalize_evidence_section("[motif_hospitalisation]") == "motif_hospitalisation"
|
|
|
|
def test_colon_conclusion(self):
|
|
assert _normalize_evidence_section("conclusion:") == "conclusion"
|
|
|
|
def test_quotes_synthese(self):
|
|
assert _normalize_evidence_section('"synthèse"') == "synthese"
|
|
|
|
# --- Fallback par mots-clés ---
|
|
|
|
def test_keyword_conclusion_du_sejour(self):
|
|
assert _normalize_evidence_section("conclusion du séjour") == "conclusion"
|
|
|
|
def test_keyword_synthese_medicale(self):
|
|
assert _normalize_evidence_section("synthèse médicale du dossier") == "synthese"
|
|
|
|
def test_keyword_diagnostic_de_sortie_variant(self):
|
|
assert _normalize_evidence_section("diagnostic(s) de sortie") == "diag_sortie"
|
|
|
|
def test_keyword_diagnostic_retenu_variant(self):
|
|
assert _normalize_evidence_section("diagnostics retenus à la sortie") == "diagnostics_retenus"
|
|
|
|
def test_keyword_motif_admission(self):
|
|
assert _normalize_evidence_section("motif d'admission aux urgences") == "motif_hospitalisation"
|
|
|
|
# --- Cas limites ---
|
|
|
|
def test_empty_string(self):
|
|
assert _normalize_evidence_section("") == ""
|
|
|
|
def test_none_like_empty(self):
|
|
assert _normalize_evidence_section(" ") == ""
|
|
|
|
def test_unknown_section_passthrough(self):
|
|
"""Section inconnue sans mot-clé → passthrough nettoyé."""
|
|
result = _normalize_evidence_section("biologie")
|
|
assert result == "biologie"
|
|
|
|
def test_sections_fortes_du_dossier(self):
|
|
"""Alias administratif observé en benchmark."""
|
|
assert _normalize_evidence_section("sections fortes du dossier") == "autres"
|
|
|
|
|
|
class TestSynthesePMSI:
|
|
"""Tests pour generate_synthese_pmsi()."""
|
|
|
|
def test_returns_synthese_on_valid_response(self, monkeypatch):
|
|
"""Réponse LLM valide → SynthesePMSI complète."""
|
|
mock_response = {
|
|
"motif_admission": "Douleur abdominale aiguë",
|
|
"probleme_pris_en_charge": "Pancréatite aiguë biliaire",
|
|
"diagnostic_retenu": "Pancréatite aiguë d'origine biliaire",
|
|
"actes_ou_traitements_majeurs": ["Scanner abdominal", "Mise à jeun"],
|
|
"complications": [],
|
|
"terrain_comorbidites": ["HTA traitée", "Diabète type 2"],
|
|
"preuves": [
|
|
{"section": "motif_hospitalisation", "excerpt": "douleur abdominale intense"},
|
|
{"section": "conclusion", "excerpt": "pancréatite aiguë biliaire confirmée"},
|
|
],
|
|
}
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return mock_response
|
|
|
|
import src.medical.dp_scoring as mod
|
|
monkeypatch.setattr(mod, "call_ollama", mock_call_ollama, raising=False)
|
|
# Forcer l'import inline à utiliser notre mock
|
|
import src.medical.ollama_client as oc_mod
|
|
monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed(sections={"conclusion": "pancréatite aiguë biliaire confirmée"})
|
|
dossier = DossierMedical()
|
|
|
|
result = generate_synthese_pmsi(parsed, "texte complet", dossier)
|
|
|
|
assert result is not None
|
|
assert isinstance(result, SynthesePMSI)
|
|
assert result.probleme_pris_en_charge == "Pancréatite aiguë biliaire"
|
|
assert result.motif_admission == "Douleur abdominale aiguë"
|
|
assert "Scanner abdominal" in result.actes_ou_traitements_majeurs
|
|
assert len(result.terrain_comorbidites) == 2
|
|
assert result.complications == []
|
|
assert len(result.preuves) == 2
|
|
assert result.preuves[0].section == "motif_hospitalisation"
|
|
|
|
def test_returns_none_on_invalid_response(self, monkeypatch):
|
|
"""Réponse LLM non-dict → None."""
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return "texte brut"
|
|
|
|
import src.medical.ollama_client as oc_mod
|
|
monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
result = generate_synthese_pmsi(parsed, "texte", dossier)
|
|
assert result is None
|
|
|
|
def test_returns_none_on_exception(self, monkeypatch):
|
|
"""Exception LLM → None."""
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
raise ConnectionError("Ollama down")
|
|
|
|
import src.medical.ollama_client as oc_mod
|
|
monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
result = generate_synthese_pmsi(parsed, "texte", dossier)
|
|
assert result is None
|
|
|
|
def test_robust_to_string_lists(self, monkeypatch):
|
|
"""Le LLM renvoie des strings au lieu de listes → toléré."""
|
|
mock_response = {
|
|
"motif_admission": "Fièvre",
|
|
"probleme_pris_en_charge": "Pneumopathie",
|
|
"diagnostic_retenu": "Pneumopathie bactérienne",
|
|
"actes_ou_traitements_majeurs": "Antibiothérapie IV", # string
|
|
"complications": "Insuffisance respiratoire", # string
|
|
"terrain_comorbidites": "BPCO", # string
|
|
"preuves": [],
|
|
}
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return mock_response
|
|
|
|
import src.medical.ollama_client as oc_mod
|
|
monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
result = generate_synthese_pmsi(parsed, "texte", dossier)
|
|
|
|
assert result is not None
|
|
assert result.actes_ou_traitements_majeurs == ["Antibiothérapie IV"]
|
|
assert result.complications == ["Insuffisance respiratoire"]
|
|
assert result.terrain_comorbidites == ["BPCO"]
|
|
|
|
def test_preuves_malformed_skipped(self, monkeypatch):
|
|
"""Preuves sans section/excerpt → ignorées."""
|
|
mock_response = {
|
|
"motif_admission": "Test",
|
|
"probleme_pris_en_charge": "Test",
|
|
"diagnostic_retenu": "Test",
|
|
"preuves": [
|
|
{"section": "conclusion", "excerpt": "valide"},
|
|
{"section": "", "excerpt": "section vide"},
|
|
{"no_section": True},
|
|
"pas un dict",
|
|
],
|
|
}
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return mock_response
|
|
|
|
import src.medical.ollama_client as oc_mod
|
|
monkeypatch.setattr(oc_mod, "call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
result = generate_synthese_pmsi(parsed, "texte", dossier)
|
|
|
|
assert result is not None
|
|
assert len(result.preuves) == 1
|
|
assert result.preuves[0].section == "conclusion"
|
|
|
|
def test_serialization_round_trip(self):
|
|
"""SynthesePMSI se sérialise/désérialise correctement."""
|
|
syn = SynthesePMSI(
|
|
motif_admission="Douleur thoracique",
|
|
probleme_pris_en_charge="Infarctus du myocarde",
|
|
diagnostic_retenu="IDM ST+ antérieur",
|
|
actes_ou_traitements_majeurs=["Coronarographie", "Angioplastie"],
|
|
complications=["Insuffisance cardiaque"],
|
|
terrain_comorbidites=["HTA", "Tabagisme"],
|
|
preuves=[PreuveSynthese(section="conclusion", excerpt="IDM confirmé")],
|
|
)
|
|
data = syn.model_dump()
|
|
restored = SynthesePMSI(**data)
|
|
assert restored.probleme_pris_en_charge == "Infarctus du myocarde"
|
|
assert len(restored.preuves) == 1
|
|
assert restored.preuves[0].section == "conclusion"
|
|
|
|
def test_dossier_medical_field(self):
|
|
"""Le champ synthese_pmsi est disponible sur DossierMedical."""
|
|
dossier = DossierMedical()
|
|
assert dossier.synthese_pmsi is None
|
|
dossier.synthese_pmsi = SynthesePMSI(
|
|
probleme_pris_en_charge="Test",
|
|
)
|
|
assert dossier.synthese_pmsi.probleme_pris_en_charge == "Test"
|
|
data = dossier.model_dump(exclude_none=True)
|
|
assert "synthese_pmsi" in data
|
|
|
|
|
|
# ===========================================================================
|
|
# DP Candidate Pool
|
|
# ===========================================================================
|
|
|
|
class TestDPPoolCandidate:
|
|
"""Tests du modèle DPPoolCandidate."""
|
|
|
|
def test_basic_creation(self):
|
|
c = DPPoolCandidate(terme="Pancréatite aiguë", section="conclusion")
|
|
assert c.terme == "Pancréatite aiguë"
|
|
assert c.section == "conclusion"
|
|
assert c.score_initial == 0.0
|
|
assert c.preuve == ""
|
|
|
|
def test_serialization(self):
|
|
c = DPPoolCandidate(
|
|
terme="Cholécystite aiguë",
|
|
section="diag_sortie",
|
|
preuve="cholécystite aiguë lithiasique",
|
|
score_initial=0.9,
|
|
)
|
|
data = c.model_dump()
|
|
restored = DPPoolCandidate(**data)
|
|
assert restored.terme == "Cholécystite aiguë"
|
|
assert restored.score_initial == 0.9
|
|
|
|
|
|
class TestIsPoolExcluded:
|
|
"""Tests du filtrage des candidats pool."""
|
|
|
|
def test_bio_value_excluded(self):
|
|
assert _is_pool_excluded("CRP 180 mg/L") is True
|
|
|
|
def test_bio_term_with_number_excluded(self):
|
|
assert _is_pool_excluded("Hémoglobine 7.2 g/dL") is True
|
|
|
|
def test_vague_symptom_excluded(self):
|
|
assert _is_pool_excluded("douleur") is True
|
|
assert _is_pool_excluded("fièvre") is True
|
|
|
|
def test_vague_symptom_with_context_kept(self):
|
|
"""Symptôme qualifié (multi-mots) → conservé."""
|
|
assert _is_pool_excluded("douleur abdominale aiguë") is False
|
|
|
|
def test_medical_diagnosis_kept(self):
|
|
assert _is_pool_excluded("Pancréatite aiguë biliaire") is False
|
|
|
|
def test_numeric_value_excluded(self):
|
|
assert _is_pool_excluded("12.5 g/dL") is True
|
|
|
|
|
|
class TestDedupPool:
|
|
"""Tests de la déduplication du pool."""
|
|
|
|
def test_dedup_keeps_highest_score(self):
|
|
candidates = [
|
|
DPPoolCandidate(terme="Pancréatite aiguë", section="conclusion", score_initial=0.7),
|
|
DPPoolCandidate(terme="Pancréatite aiguë", section="diag_sortie", score_initial=1.0),
|
|
]
|
|
result = _dedup_pool(candidates)
|
|
assert len(result) == 1
|
|
assert result[0].score_initial == 1.0
|
|
assert result[0].section == "diag_sortie"
|
|
|
|
def test_dedup_normalizes_text(self):
|
|
"""Variantes d'accents/espaces → même clé."""
|
|
candidates = [
|
|
DPPoolCandidate(terme="Pancréatite aiguë", section="a", score_initial=0.5),
|
|
DPPoolCandidate(terme="pancreatite aigue", section="b", score_initial=0.8),
|
|
]
|
|
result = _dedup_pool(candidates)
|
|
assert len(result) == 1
|
|
|
|
def test_distinct_terms_kept(self):
|
|
candidates = [
|
|
DPPoolCandidate(terme="Pancréatite aiguë", section="a", score_initial=0.7),
|
|
DPPoolCandidate(terme="Cholécystite aiguë", section="b", score_initial=0.9),
|
|
]
|
|
result = _dedup_pool(candidates)
|
|
assert len(result) == 2
|
|
|
|
|
|
class TestBuildDPCandidatePool:
|
|
"""Tests d'intégration de build_dp_candidate_pool()."""
|
|
|
|
def test_indicative_phrase_extraction(self):
|
|
"""Les phrases indicatives sont extraites du texte."""
|
|
text = "Le patient a été hospitalisé pour pancréatite aiguë biliaire. Suivi habituel."
|
|
parsed = _make_parsed(sections={"conclusion": "Pancréatite aiguë biliaire confirmée."})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, text, None, dossier)
|
|
|
|
termes = [c.terme.lower() for c in pool]
|
|
assert any("pancréatite" in t or "pancreatite" in t for t in termes)
|
|
|
|
def test_sections_fortes_extraction(self):
|
|
"""Les diagnostics des sections fortes apparaissent dans le pool."""
|
|
parsed = _make_parsed(sections={
|
|
"diag_sortie": "Cholécystite aiguë lithiasique",
|
|
"conclusion": "Évolution favorable après cholécystectomie",
|
|
})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte complet", None, dossier)
|
|
|
|
termes = [c.terme.lower() for c in pool]
|
|
assert any("cholécystite" in t or "cholecystite" in t for t in termes)
|
|
|
|
def test_edsnlp_entities_included(self):
|
|
"""Les entités edsnlp non-niées apparaissent dans le pool."""
|
|
from dataclasses import dataclass
|
|
|
|
@dataclass
|
|
class MockEntity:
|
|
texte: str
|
|
code: str
|
|
negation: bool = False
|
|
hypothese: bool = False
|
|
|
|
@dataclass
|
|
class MockResult:
|
|
cim10_entities: list
|
|
|
|
edsnlp = MockResult(cim10_entities=[
|
|
MockEntity(texte="pneumopathie", code="J18.9"),
|
|
MockEntity(texte="HTA", code="I10", negation=True), # exclu
|
|
])
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", edsnlp, dossier)
|
|
|
|
termes = [c.terme.lower() for c in pool]
|
|
assert any("pneumopathie" in t for t in termes)
|
|
# HTA niée ne doit pas apparaître
|
|
assert not any(t == "hta" for t in termes)
|
|
|
|
def test_actes_included(self):
|
|
"""Les actes CCAM du dossier apparaissent comme candidats."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
dossier.actes_ccam = [
|
|
ActeCCAM(texte="Cholécystectomie", code_ccam_suggestion="HMFC004"),
|
|
]
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", None, dossier)
|
|
|
|
termes = [c.terme.lower() for c in pool]
|
|
assert any("cholécystectomie" in t or "cholecystectomie" in t for t in termes)
|
|
|
|
def test_cim10_map_matches(self):
|
|
"""Les termes CIM10_MAP matchés dans les sections fortes sont inclus."""
|
|
parsed = _make_parsed(sections={
|
|
"conclusion": "Patient avec pancréatite aiguë biliaire sévère.",
|
|
})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", None, dossier)
|
|
|
|
sections = [c.section for c in pool]
|
|
assert "cim10_map" in sections
|
|
|
|
def test_bio_values_excluded(self):
|
|
"""Les valeurs biologiques ne polluent pas le pool."""
|
|
parsed = _make_parsed(sections={
|
|
"conclusion": "CRP 180 mg/L. Hémoglobine 7.2 g/dL. Pancréatite aiguë.",
|
|
})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", None, dossier)
|
|
|
|
termes = [c.terme.lower() for c in pool]
|
|
assert not any("crp" in t and "mg" in t for t in termes)
|
|
|
|
def test_dedup_across_sources(self):
|
|
"""Un même terme de 2 sources → 1 seule entrée (meilleur score)."""
|
|
parsed = _make_parsed(sections={
|
|
"conclusion": "Pancréatite aiguë biliaire confirmée.",
|
|
"motif_hospitalisation": "Pancréatite aiguë biliaire.",
|
|
})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", None, dossier)
|
|
|
|
# Compter les variantes "pancréatite aiguë biliaire"
|
|
from src.medical.cim10_dict import normalize_text
|
|
keys = [normalize_text(c.terme) for c in pool]
|
|
pancreatite_keys = [k for k in keys if "pancreatite" in k and "biliaire" in k]
|
|
# Après dedup, devrait être au plus 1-2 (phrase complète vs segment)
|
|
assert len(pancreatite_keys) <= 2
|
|
|
|
def test_cap_at_30(self):
|
|
"""Le pool est plafonné à 30 candidats."""
|
|
# Créer un texte avec beaucoup de diagnostics
|
|
diagnostics = [f"diagnostic numéro {i}" for i in range(50)]
|
|
section_text = ". ".join(diagnostics) + "."
|
|
parsed = _make_parsed(sections={"conclusion": section_text})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, section_text, None, dossier)
|
|
|
|
assert len(pool) <= 30
|
|
|
|
def test_empty_input(self):
|
|
"""Entrée vide → pool vide."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "", None, dossier)
|
|
|
|
assert isinstance(pool, list)
|
|
|
|
def test_score_ordering(self):
|
|
"""Le pool est trié par score_initial décroissant."""
|
|
parsed = _make_parsed(sections={
|
|
"diag_sortie": "Cholécystite aiguë",
|
|
"conclusion": "Angiocholite associée",
|
|
})
|
|
dossier = DossierMedical()
|
|
|
|
pool = build_dp_candidate_pool(parsed, "texte", None, dossier)
|
|
|
|
if len(pool) >= 2:
|
|
scores = [c.score_initial for c in pool]
|
|
assert scores == sorted(scores, reverse=True)
|
|
|
|
|
|
# ===========================================================================
|
|
# Anti-comorbidité SynthesePMSI
|
|
# ===========================================================================
|
|
|
|
class TestIsComorbBanale:
|
|
"""Tests de détection des comorbidités banales."""
|
|
|
|
def test_hta(self):
|
|
assert _is_comorbidite_banale("HTA") == "HTA"
|
|
assert _is_comorbidite_banale("hypertension artérielle") == "HTA"
|
|
|
|
def test_diabete_stable(self):
|
|
assert _is_comorbidite_banale("diabète de type 2") == "diabète"
|
|
assert _is_comorbidite_banale("diabète") == "diabète"
|
|
|
|
def test_diabete_decompense_not_banal(self):
|
|
"""Diabète déséquilibré ne doit PAS être considéré banal."""
|
|
assert _is_comorbidite_banale("diabète déséquilibré") is None
|
|
assert _is_comorbidite_banale("diabète décompensé") is None
|
|
assert _is_comorbidite_banale("acidocétose diabétique") is None
|
|
|
|
def test_obesite(self):
|
|
assert _is_comorbidite_banale("obésité") == "obésité"
|
|
|
|
def test_anemie_chronique(self):
|
|
assert _is_comorbidite_banale("anémie chronique") == "anémie"
|
|
|
|
def test_anemie_severe_not_banal(self):
|
|
"""Anémie sévère ne doit PAS être banale."""
|
|
assert _is_comorbidite_banale("anémie sévère") is None
|
|
assert _is_comorbidite_banale("anémie aiguë") is None
|
|
|
|
def test_bpco_stable(self):
|
|
assert _is_comorbidite_banale("BPCO") == "BPCO"
|
|
|
|
def test_bpco_exacerbation_not_banal(self):
|
|
"""BPCO exacerbée ne doit PAS être banale."""
|
|
assert _is_comorbidite_banale("BPCO exacerbée") is None
|
|
assert _is_comorbidite_banale("BPCO surinfectée") is None
|
|
|
|
def test_non_comorbidite(self):
|
|
assert _is_comorbidite_banale("pneumothorax") is None
|
|
assert _is_comorbidite_banale("cholécystite aiguë") is None
|
|
assert _is_comorbidite_banale("méningite à entérovirus") is None
|
|
|
|
|
|
class TestHasPecMarker:
|
|
"""Tests des marqueurs de PEC principale."""
|
|
|
|
def test_hospitalise_pour(self):
|
|
assert _has_pec_marker("diabète", "hospitalisé pour diabète déséquilibré") is True
|
|
|
|
def test_desequilibre(self):
|
|
assert _has_pec_marker("diabète déséquilibré", "") is True
|
|
|
|
def test_acidocetose(self):
|
|
assert _has_pec_marker("", "acidocétose diabétique") is True
|
|
|
|
def test_transfusion(self):
|
|
assert _has_pec_marker("anémie", "transfusion de 2 CGR") is True
|
|
|
|
def test_no_marker(self):
|
|
assert _has_pec_marker("diabète", "diabète type 2 équilibré") is False
|
|
|
|
def test_hta_maligne(self):
|
|
assert _has_pec_marker("HTA maligne", "") is True
|
|
|
|
|
|
class TestHasAcuteProblem:
|
|
"""Tests de détection de problème aigu."""
|
|
|
|
def test_with_diagnostic_retenu(self):
|
|
result = {
|
|
"diagnostic_retenu": "pneumothorax spontané",
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": [],
|
|
}
|
|
assert _has_acute_problem(result) is True
|
|
|
|
def test_with_complications(self):
|
|
result = {
|
|
"diagnostic_retenu": "",
|
|
"complications": ["insuffisance rénale aiguë"],
|
|
"actes_ou_traitements_majeurs": [],
|
|
}
|
|
assert _has_acute_problem(result) is True
|
|
|
|
def test_with_surgical_acte(self):
|
|
result = {
|
|
"diagnostic_retenu": "",
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": ["cholécystectomie"],
|
|
}
|
|
assert _has_acute_problem(result) is True
|
|
|
|
def test_only_surveillance(self):
|
|
result = {
|
|
"diagnostic_retenu": "",
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": ["surveillance", "bilan biologique"],
|
|
}
|
|
assert _has_acute_problem(result) is False
|
|
|
|
def test_diag_retenu_is_comorb(self):
|
|
"""Si diagnostic_retenu est aussi une comorbidité banale, pas de problème aigu via ce champ."""
|
|
result = {
|
|
"diagnostic_retenu": "diabète",
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": [],
|
|
}
|
|
assert _has_acute_problem(result) is False
|
|
|
|
|
|
class TestPostprocessSynthese:
|
|
"""Tests du post-traitement anti-comorbidité."""
|
|
|
|
def test_non_comorbidite_untouched(self):
|
|
"""Un problème non-banal ne doit pas être modifié."""
|
|
result = {
|
|
"probleme_pris_en_charge": "pneumothorax spontané",
|
|
"diagnostic_retenu": "pneumothorax spontané",
|
|
"terrain_comorbidites": ["HTA"],
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": ["drainage"],
|
|
"preuves": [],
|
|
}
|
|
processed = _postprocess_synthese(result, "texte")
|
|
assert processed["probleme_pris_en_charge"] == "pneumothorax spontané"
|
|
|
|
def test_comorbidite_with_acute_problem_promoted(self):
|
|
"""Comorbidité banale + problème aigu → diagnostic retenu promu."""
|
|
result = {
|
|
"probleme_pris_en_charge": "diabète",
|
|
"diagnostic_retenu": "décompensation cardiaque globale",
|
|
"terrain_comorbidites": [],
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": ["diurétiques IV"],
|
|
"preuves": [],
|
|
}
|
|
processed = _postprocess_synthese(result, "texte complet")
|
|
assert processed["probleme_pris_en_charge"] == "décompensation cardiaque globale"
|
|
assert "diabète" in processed["terrain_comorbidites"]
|
|
|
|
def test_comorbidite_with_pec_marker_kept(self):
|
|
"""Comorbidité banale avec marqueur PEC → conservée."""
|
|
result = {
|
|
"probleme_pris_en_charge": "diabète",
|
|
"diagnostic_retenu": "diabète déséquilibré",
|
|
"terrain_comorbidites": [],
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": [],
|
|
"preuves": [],
|
|
}
|
|
context = "hospitalisé pour diabète déséquilibré avec insulinothérapie IV"
|
|
processed = _postprocess_synthese(result, context)
|
|
# Marqueur "hospitalisé pour" + "déséquilibré" trouvé → conservé
|
|
assert processed["probleme_pris_en_charge"] == "diabète"
|
|
|
|
def test_comorbidite_no_acute_indeterminate(self):
|
|
"""Comorbidité banale sans aigu ni marqueur → indéterminé."""
|
|
result = {
|
|
"probleme_pris_en_charge": "HTA",
|
|
"diagnostic_retenu": "",
|
|
"terrain_comorbidites": [],
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": [],
|
|
"preuves": [],
|
|
}
|
|
processed = _postprocess_synthese(result, "texte")
|
|
assert "indéterminé" in processed["probleme_pris_en_charge"]
|
|
assert "HTA" in processed["terrain_comorbidites"]
|
|
|
|
def test_proof_added_on_correction(self):
|
|
"""Une preuve de post-traitement est ajoutée lors de correction."""
|
|
result = {
|
|
"probleme_pris_en_charge": "diabète",
|
|
"diagnostic_retenu": "pneumopathie bactérienne",
|
|
"terrain_comorbidites": [],
|
|
"complications": [],
|
|
"actes_ou_traitements_majeurs": ["antibiothérapie IV"],
|
|
"preuves": [],
|
|
}
|
|
processed = _postprocess_synthese(result, "texte")
|
|
sections = [p["section"] for p in processed["preuves"]]
|
|
assert "post-traitement" in sections
|
|
|
|
|
|
class TestBuildMotifFallback:
|
|
"""Tests du fallback motif admission."""
|
|
|
|
def test_mode_entree_priority(self):
|
|
"""Le mode_entree du séjour a priorité."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
dossier.sejour = Sejour(mode_entree="Urgences")
|
|
assert _build_motif(parsed, dossier) == "Urgences"
|
|
|
|
def test_section_motif_hospitalisation(self):
|
|
"""Section motif_hospitalisation utilisée si mode_entree vide."""
|
|
parsed = _make_parsed(sections={"motif_hospitalisation": "Douleur thoracique"})
|
|
dossier = DossierMedical()
|
|
assert _build_motif(parsed, dossier) == "Douleur thoracique"
|
|
|
|
def test_fallback_lexical_conclusion(self):
|
|
"""Fallback lexical sur la conclusion."""
|
|
parsed = _make_parsed(sections={
|
|
"conclusion": "Patient hospitalisé pour pneumothorax spontané."
|
|
})
|
|
dossier = DossierMedical()
|
|
result = _build_motif(parsed, dossier)
|
|
assert "pneumothorax" in result.lower()
|
|
|
|
def test_fallback_lexical_full_text(self):
|
|
"""Fallback lexical sur le texte complet."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
text = "Compte-rendu\nMotif d'hospitalisation : ictère choléstatique.\nExamen..."
|
|
result = _build_motif(parsed, dossier, full_text=text)
|
|
assert "ictère" in result.lower()
|
|
|
|
def test_non_renseigne_when_nothing(self):
|
|
"""Pas de motif trouvé → 'Non renseigné'."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
assert _build_motif(parsed, dossier) == "Non renseigné"
|
|
|
|
|
|
# ===================================================================
|
|
# Tests DP Pool Rank
|
|
# ===================================================================
|
|
|
|
class TestFormatPoolForPrompt:
|
|
"""Tests pour _format_pool_for_prompt()."""
|
|
|
|
def test_basic_formatting(self):
|
|
"""Vérifie le format des candidats pour le prompt."""
|
|
pool = [
|
|
DPPoolCandidate(terme="Pneumopathie", section="conclusion",
|
|
preuve="Au total : pneumopathie", score_initial=0.7),
|
|
DPPoolCandidate(terme="Embolie pulmonaire", section="diag_sortie",
|
|
preuve="Diagnostic de sortie", score_initial=1.0),
|
|
]
|
|
text = _format_pool_for_prompt(pool)
|
|
assert "[0]" in text
|
|
assert "[1]" in text
|
|
assert "Pneumopathie" in text
|
|
assert "Embolie pulmonaire" in text
|
|
assert "conclusion" in text
|
|
assert "diag_sortie" in text
|
|
|
|
def test_max_items_cap(self):
|
|
"""Vérifie que max_items est respecté."""
|
|
pool = [
|
|
DPPoolCandidate(terme=f"Diag_{i}", section="conclusion", score_initial=0.5)
|
|
for i in range(10)
|
|
]
|
|
text = _format_pool_for_prompt(pool, max_items=3)
|
|
assert "[0]" in text
|
|
assert "[2]" in text
|
|
assert "[3]" not in text
|
|
|
|
def test_empty_pool(self):
|
|
"""Pool vide → texte vide."""
|
|
assert _format_pool_for_prompt([]) == ""
|
|
|
|
|
|
class TestBuildClinicalContext:
|
|
"""Tests pour _build_clinical_context()."""
|
|
|
|
def test_with_synthese(self):
|
|
"""Avec SynthesePMSI disponible."""
|
|
synthese = SynthesePMSI(
|
|
motif_admission="Douleur thoracique",
|
|
probleme_pris_en_charge="Embolie pulmonaire",
|
|
diagnostic_retenu="Embolie pulmonaire bilatérale",
|
|
)
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
ctx = _build_clinical_context(parsed, dossier, "", synthese)
|
|
assert "Embolie pulmonaire" in ctx
|
|
assert "Douleur thoracique" in ctx
|
|
|
|
def test_without_synthese(self):
|
|
"""Sans SynthesePMSI → fallback sections fortes."""
|
|
parsed = _make_parsed(sections={"conclusion": "Pneumopathie traitée"})
|
|
dossier = DossierMedical()
|
|
ctx = _build_clinical_context(parsed, dossier, "texte complet", None)
|
|
assert "Pneumopathie traitée" in ctx
|
|
assert "Motif" in ctx
|
|
|
|
|
|
class TestLlmDpPoolRank:
|
|
"""Tests unitaires pour llm_dp_pool_rank() — sans appel LLM réel."""
|
|
|
|
def test_empty_pool_fallback_off(self):
|
|
"""Pool vide + fallback OFF → REVIEW."""
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=[],
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "review"
|
|
assert "pool vide" in selection.winner_reason
|
|
|
|
def test_empty_pool_fallback_on(self, monkeypatch):
|
|
"""Pool vide + fallback ON → tente llm_dp_fallback."""
|
|
# Mock llm_dp_fallback pour retourner un résultat connu
|
|
from src.medical import dp_scoring
|
|
mock_selection = DPSelection(
|
|
verdict="review",
|
|
winner_reason="fallback activé",
|
|
)
|
|
monkeypatch.setattr(dp_scoring, "llm_dp_fallback", lambda *a, **kw: mock_selection)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=[],
|
|
fallback_oneshot=True,
|
|
)
|
|
assert selection.verdict == "review"
|
|
assert "fallback" in selection.winner_reason
|
|
|
|
def test_valid_llm_response_high_confidence(self, monkeypatch):
|
|
"""Réponse LLM valide avec confidence high → CONFIRMED."""
|
|
pool = [
|
|
DPPoolCandidate(terme="Embolie pulmonaire", section="conclusion",
|
|
preuve="Au total : embolie pulmonaire", score_initial=0.7),
|
|
DPPoolCandidate(terme="HTA", section="conclusion",
|
|
preuve="terrain HTA", score_initial=0.3),
|
|
]
|
|
|
|
# Mock call_ollama
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return {
|
|
"chosen_index": 0,
|
|
"chosen_terme": "Embolie pulmonaire",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "Au total : embolie pulmonaire",
|
|
"confidence": "high",
|
|
"reason": "pathologie aiguë traitée",
|
|
}
|
|
|
|
from src.medical import dp_scoring
|
|
monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed(sections={"conclusion": "Au total : embolie pulmonaire"})
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=pool,
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "confirmed"
|
|
assert len(selection.candidates) == 1
|
|
assert selection.candidates[0].label == "Embolie pulmonaire"
|
|
assert selection.candidates[0].source_section == "llm_pool_rank (conclusion)"
|
|
assert selection.candidates[0].code is None # pas de code CIM-10, sera codé en aval
|
|
|
|
def test_valid_llm_response_medium_confidence(self, monkeypatch):
|
|
"""Réponse LLM avec confidence medium → REVIEW."""
|
|
pool = [
|
|
DPPoolCandidate(terme="Insuffisance cardiaque", section="conclusion",
|
|
preuve="insuffisance cardiaque", score_initial=0.7),
|
|
]
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return {
|
|
"chosen_index": 0,
|
|
"chosen_terme": "Insuffisance cardiaque",
|
|
"evidence_section": "conclusion",
|
|
"evidence_excerpt": "insuffisance cardiaque globale",
|
|
"confidence": "medium",
|
|
"reason": "diagnostic probable",
|
|
}
|
|
|
|
monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=pool,
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "review"
|
|
assert "confidence medium" in selection.winner_reason
|
|
|
|
def test_chosen_index_minus_one_fallback_off(self, monkeypatch):
|
|
"""chosen_index=-1 + fallback OFF → REVIEW."""
|
|
pool = [
|
|
DPPoolCandidate(terme="HTA", section="conclusion",
|
|
preuve="HTA", score_initial=0.3),
|
|
]
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return {
|
|
"chosen_index": -1,
|
|
"chosen_terme": "",
|
|
"confidence": "low",
|
|
"reason": "aucun candidat solide",
|
|
}
|
|
|
|
monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=pool,
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "review"
|
|
assert "aucun candidat retenu" in selection.winner_reason
|
|
|
|
def test_index_out_of_range_fallback_off(self, monkeypatch):
|
|
"""Index hors plage → REVIEW."""
|
|
pool = [
|
|
DPPoolCandidate(terme="Pneumopathie", section="conclusion",
|
|
preuve="...", score_initial=0.7),
|
|
]
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return {
|
|
"chosen_index": 5,
|
|
"chosen_terme": "Fantôme",
|
|
"confidence": "high",
|
|
}
|
|
|
|
monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=pool,
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "review"
|
|
|
|
def test_score_details_contain_pool_info(self, monkeypatch):
|
|
"""Les score_details du candidat contiennent les infos pool."""
|
|
pool = [
|
|
DPPoolCandidate(terme="Cholécystite aiguë", section="diag_sortie",
|
|
preuve="cholécystite aiguë lithiasique", score_initial=0.9),
|
|
]
|
|
|
|
def mock_call_ollama(prompt, **kwargs):
|
|
return {
|
|
"chosen_index": 0,
|
|
"chosen_terme": "Cholécystite aiguë",
|
|
"evidence_section": "diag_sortie",
|
|
"evidence_excerpt": "cholécystite aiguë lithiasique",
|
|
"confidence": "high",
|
|
"reason": "diagnostic chirurgical aigu",
|
|
}
|
|
|
|
monkeypatch.setattr("src.medical.ollama_client.call_ollama", mock_call_ollama)
|
|
|
|
parsed = _make_parsed()
|
|
dossier = DossierMedical()
|
|
selection = llm_dp_pool_rank(
|
|
parsed, "texte", dossier,
|
|
pool_candidates=pool,
|
|
fallback_oneshot=False,
|
|
)
|
|
assert selection.verdict == "confirmed"
|
|
details = selection.candidates[0].score_details
|
|
assert "pool_score" in details
|
|
assert "pool_index" in details
|
|
assert details["pool_index"] == 0
|