medical-agent / test.py
Nourhenem's picture
Upload folder using huggingface_hub
1eb76aa verified
import re
import logging
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass
from enum import Enum
logger = logging.getLogger(__name__)
class FieldType(Enum):
"""Types de champs dans le template"""
CHECKBOX = "checkbox" # &x cases à cocher
TEXT = "text" # &x texte libre
MEASUREMENT = "measurement" # &x valeurs numériques
@dataclass
class TemplateField:
"""Définition d'un champ du template"""
placeholder: str # &x dans le template
field_type: FieldType
source_field: str # Champ correspondant dans ExtractedData
default_value: str = ""
validation_pattern: Optional[str] = None
transformation_func: Optional[callable] = None
context_identifier: Optional[str] = None # Pour différencier gauche/droite
@dataclass
class MappingResult:
"""Résultat du mapping"""
filled_template: str
mapped_fields: Dict[str, str]
unmapped_placeholders: List[str]
mapping_confidence: float
errors: List[str]
class MedicalTemplateMapper:
"""Moteur de mapping des données extraites vers le template médical"""
def __init__(self):
self.template = self._load_template()
self.field_mappings = self._define_field_mappings()
self.checkbox_logic = self._define_checkbox_logic()
def _load_template(self) -> str:
"""Template médical de base avec placeholders &x"""
return """L'utérus est &x antéversé, &x rétroversé, &x intermédiaire, &x rétrofléchi, &x antéfléchi, &x fixe de taille normale (&x x &x x &x cm).
Hystérométrie : distance orifice externe du col - fond de la cavité utérine : &x mm.
L'endomètre : mesuré à &x mm.
Myometre : pas de myome.
Zone jonctionnelle : Atteinte de la zone de jonction : &x non &x oui
Adénomyose associée : &x non &x oui : &x diffuse &x focale &x interne &x externe
Col utérin: pas de kyste de Naboth. Absence de pathologies échographiquement décelable à son niveau.
Cavité utérine en 3D: morphologie triangulaire.
&xKISSING OVARIES
L'ovaire droit mesure &x x &x mm, &x est de dimensions supérieures à la normale il mesure &x x &x mm, &x folliculaire CFA &x follicules: (&x mm). &x Absence d'endométriome. &x Présence d'une formation kystique hypoéchogène, uniloculaire, non vascularisé, à contenu ground glass mesurée à &x mm d'allure endométriome.
Accessibilité : &x rétro-utérin &x fixe &x aisée.
L'ovaire gauche mesure &x x &x mm, &x est de dimensions supérieures à la normale il mesure &x x &x mm, &x folliculaire CFA &x follicules: (&x mm). &x Absence d'endométriome. &x Présence d'une formation kystique hypoéchogène, uniloculaire, non vascularisé, à contenu ground glass mesurée à &x mm d'allure endométriome.
Accessibilité : &x rétro-utérin &x fixe &x aisée.
&x Présence de micro-calcifications sous thécales &x bilatérales &x droites &x gauches pouvant témoigner d'implants endométriosiques superficiels.
L'échostructure des deux ovaires apparait normale, avec une vascularisation artério-veineuse normale au Doppler, sans formation ou image kystique pathologique échographiquement décelable à leur niveau.
Cavité péritonéale
&x- Pas d'épanchement liquidien dans le cul du sac du Douglas. Pas de douleur à l'écho-palpation.
&x- Faible épanchement corpusculé dans le cul du sac du Douglas qui silhouette des adhérences (soft marqueur d'endométriose?). Pas de douleur à l'écho-palpation.
- &xVessie vide pendant l'examen. &x Vessie en semi-réplétion pendant l'examen.
- &x Absence de dilatation pyélo-calicielle.
- Artère utérine : IP : &x - IR : 0,&x - Spectre : type 2 avec notch protodiastolique.
- Pas d'image d'hydrosalpinx visible à ce jour.
RECHERCHE ENDOMETRIOSE PELVIENNE
A-Compartiment antérieur (vessie en semi-réplétion)
- Signe du glissement (sliding) : &xprésent &xdiminué &xabsent
- Présence d'un nodule : &xnon &xoui
- Uretères dans la partie pelvienne vus non dilatés.
B-Compartiment postérieur
- Signe du glissement (sliding) :
- Espace recto-vaginal : &xprésent &xdiminué &xabsent
- Plan sus-péritonéal : &xprésent &xdiminué &xabsent
- Aspect du torus : &x normal &x épaissi
- Aspect des ligaments utéro-sacrés :
- Ligament utéro- sacré droit : &x normal &x épaissi
- Ligament utéro-sacré gauche : &x normal &x épaissi
- Présence d'un nodule hypoéchogène : &x non
- Infiltration digestive: &x non &x oui : &x bas rectum &x moyen rectum &x haut rectum &x jonction recto-sigmoïde
Conclusions
Utérus de taille et de morphologie normales.
Endomètre mesuré à &x mm.
CFA : &x+&x follicules.
Ovaires sans formation ou image kystique pathologique échographiquement décelable à leur niveau.
&x Absence d'image d'endométriose visible ce jour, à confronter éventuellement à une IRM.
&x Endométriose &x superficielle &x et profonde.
Absence d'anomalie échographiquement décelable au niveau des trompes.
--> L'ensemble de ces aspects reste à confronter au contexte clinico-thérapeutique.
(qui contient des trous représentés par &x)"""
def _define_field_mappings(self) -> Dict[str, TemplateField]:
"""Définit les mappings entre données extraites et placeholders template"""
return {
# Position utérus - checkboxes
"uterus_position_antéversé": TemplateField(
placeholder="&x antéversé",
field_type=FieldType.CHECKBOX,
source_field="uterus_position",
transformation_func=lambda x: "X" if x and "antéversé" in x.lower() else ""
),
"uterus_position_rétroversé": TemplateField(
placeholder="&x rétroversé",
field_type=FieldType.CHECKBOX,
source_field="uterus_position",
transformation_func=lambda x: "X" if x and "rétroversé" in x.lower() else ""
),
# Taille utérus - dimensions
"uterus_size_length": TemplateField(
placeholder="normale (&x x",
field_type=FieldType.MEASUREMENT,
source_field="uterus_size",
transformation_func=self._extract_first_dimension
),
"uterus_size_width": TemplateField(
placeholder="x &x x",
field_type=FieldType.MEASUREMENT,
source_field="uterus_size",
transformation_func=self._extract_second_dimension
),
"uterus_size_height": TemplateField(
placeholder="x &x cm)",
field_type=FieldType.MEASUREMENT,
source_field="uterus_size",
transformation_func=self._extract_third_dimension
),
# Hystérométrie
"hysterometry_value": TemplateField(
placeholder="fond de la cavité utérine : &x mm",
field_type=FieldType.MEASUREMENT,
source_field="hysterometry",
transformation_func=self._clean_numeric_value
),
# Endomètre
"endometrium_thickness": TemplateField(
placeholder="L'endomètre : mesuré à &x mm",
field_type=FieldType.MEASUREMENT,
source_field="endometrium_thickness",
transformation_func=self._clean_numeric_value
),
# Zone jonctionnelle
"junctional_zone_non": TemplateField(
placeholder="Atteinte de la zone de jonction : &x non",
field_type=FieldType.CHECKBOX,
source_field="junctional_zone_status",
transformation_func=lambda x: "X" if not x or x.lower() in ["normale", "normal"] else ""
),
"junctional_zone_oui": TemplateField(
placeholder="&x oui",
field_type=FieldType.CHECKBOX,
source_field="junctional_zone_status",
transformation_func=lambda x: "X" if x and x.lower() in ["épaissie", "épaisse", "atteinte"] else ""
),
# Adénomyose
"adenomyosis_non": TemplateField(
placeholder="Adénomyose associée : &x non",
field_type=FieldType.CHECKBOX,
source_field="adenomyosis_type",
transformation_func=lambda x: "X" if not x or x.lower() in ["absente", "non"] else ""
),
"adenomyosis_oui": TemplateField(
placeholder="&x oui :",
field_type=FieldType.CHECKBOX,
source_field="adenomyosis_type",
transformation_func=lambda x: "X" if x and x.lower() in ["diffuse", "focale"] else ""
),
"adenomyosis_diffuse": TemplateField(
placeholder="&x diffuse",
field_type=FieldType.CHECKBOX,
source_field="adenomyosis_type",
transformation_func=lambda x: "X" if x and "diffuse" in x.lower() else ""
),
# Doppler
"doppler_ip": TemplateField(
placeholder="IP : &x",
field_type=FieldType.MEASUREMENT,
source_field="doppler_ip",
transformation_func=self._clean_numeric_value
),
"doppler_ir": TemplateField(
placeholder="IR : 0,&x",
field_type=FieldType.MEASUREMENT,
source_field="doppler_ir",
transformation_func=self._format_doppler_ir
),
# Conclusion - endomètre
"conclusion_endometrium": TemplateField(
placeholder="Endomètre mesuré à &x mm",
field_type=FieldType.MEASUREMENT,
source_field="endometrium_thickness",
transformation_func=self._clean_numeric_value
),
# Conclusions - CFA total
"conclusion_cfa_right": TemplateField(
placeholder="CFA : &x+",
field_type=FieldType.MEASUREMENT,
source_field="right_ovary_cfa",
transformation_func=self._clean_cfa_value
),
"conclusion_cfa_left": TemplateField(
placeholder="+&x follicules",
field_type=FieldType.MEASUREMENT,
source_field="left_ovary_cfa",
transformation_func=self._clean_cfa_value
),
}
def _define_checkbox_logic(self) -> Dict[str, List[str]]:
"""Définit la logique des checkboxes mutuellement exclusives"""
return {
"uterus_position": ["antéversé", "rétroversé", "intermédiaire", "rétrofléchi", "antéfléchi"],
"adenomyosis": ["non", "oui"],
"adenomyosis_type": ["diffuse", "focale", "interne", "externe"],
"ovary_accessibility": ["rétro-utérin", "fixe", "aisée"]
}
def map_extracted_data_to_template(self, extracted_data) -> MappingResult:
"""
Fonction principale de mapping des données extraites vers le template
"""
logger.info("🔄 Début du mapping vers le template médical")
filled_template = self.template
mapped_fields = {}
unmapped_placeholders = []
errors = []
# Étape 1: Identifier tous les placeholders &x dans le template
all_placeholders = self._find_all_placeholders(filled_template)
logger.info(f"📍 {len(all_placeholders)} placeholders trouvés dans le template")
# Étape 2: Appliquer les mappings définis
for mapping_key, template_field in self.field_mappings.items():
try:
# Récupérer la valeur source
source_value = getattr(extracted_data, template_field.source_field, None)
if source_value:
# Appliquer la transformation
if template_field.transformation_func:
mapped_value = template_field.transformation_func(source_value)
else:
mapped_value = str(source_value)
# Remplacer dans le template
if mapped_value and mapped_value.strip():
filled_template = self._replace_placeholder_in_context(
filled_template, template_field.placeholder, mapped_value
)
mapped_fields[mapping_key] = mapped_value
logger.debug(f"✅ {mapping_key}: {mapped_value}")
except Exception as e:
error_msg = f"Erreur mapping {mapping_key}: {e}"
errors.append(error_msg)
logger.error(error_msg)
# Étape 3: Traitement spécial pour les ovaires
filled_template = self._handle_ovary_section(filled_template, extracted_data)
# Étape 4: Application des règles de logique métier
filled_template = self._apply_business_logic(filled_template, extracted_data)
# Étape 5: Gestion des placeholders non mappés
remaining_placeholders = self._find_all_placeholders(filled_template)
unmapped_placeholders = [p for p in remaining_placeholders if "&x" in p]
# Étape 6: Calcul du score de mapping
mapping_confidence = self._calculate_mapping_confidence(
len(mapped_fields), len(all_placeholders), len(errors)
)
logger.info(f"✅ Mapping terminé - {len(mapped_fields)} champs mappés, {len(unmapped_placeholders)} non mappés")
return MappingResult(
filled_template=filled_template,
mapped_fields=mapped_fields,
unmapped_placeholders=unmapped_placeholders,
mapping_confidence=mapping_confidence,
errors=errors
)
def _handle_ovary_section(self, template: str, extracted_data) -> str:
"""Traite spécifiquement la section des ovaires"""
# Traitement ovaire droit
if hasattr(extracted_data, 'right_ovary_dimensions') and extracted_data.right_ovary_dimensions:
dimensions = self._parse_dimensions(extracted_data.right_ovary_dimensions)
if len(dimensions) >= 2:
# Remplacer les dimensions de l'ovaire droit
template = self._replace_ovary_dimensions(template, "droit", dimensions[0], dimensions[1])
# CFA ovaire droit
if hasattr(extracted_data, 'right_ovary_cfa') and extracted_data.right_ovary_cfa:
cfa_value = self._clean_cfa_value(extracted_data.right_ovary_cfa)
template = self._replace_ovary_cfa(template, "droit", cfa_value)
# Accessibilité ovaire droit
if hasattr(extracted_data, 'right_ovary_accessibility') and extracted_data.right_ovary_accessibility:
template = self._replace_ovary_accessibility(template, "droit", extracted_data.right_ovary_accessibility)
# Traitement ovaire gauche
if hasattr(extracted_data, 'left_ovary_dimensions') and extracted_data.left_ovary_dimensions:
dimensions = self._parse_dimensions(extracted_data.left_ovary_dimensions)
if len(dimensions) >= 2:
# Remplacer les dimensions de l'ovaire gauche
template = self._replace_ovary_dimensions(template, "gauche", dimensions[0], dimensions[1])
# CFA ovaire gauche
if hasattr(extracted_data, 'left_ovary_cfa') and extracted_data.left_ovary_cfa:
cfa_value = self._clean_cfa_value(extracted_data.left_ovary_cfa)
template = self._replace_ovary_cfa(template, "gauche", cfa_value)
# Accessibilité ovaire gauche
if hasattr(extracted_data, 'left_ovary_accessibility') and extracted_data.left_ovary_accessibility:
template = self._replace_ovary_accessibility(template, "gauche", extracted_data.left_ovary_accessibility)
return template
def _replace_ovary_dimensions(self, template: str, side: str, dim1: str, dim2: str) -> str:
"""Remplace les dimensions d'un ovaire spécifique"""
lines = template.split('\n')
for i, line in enumerate(lines):
if f"ovaire {side} mesure" in line.lower():
# Remplacer les deux premiers &x pour les dimensions principales
if "&x x &x mm" in line:
line = line.replace("&x x &x mm", f"{dim1} x {dim2} mm", 1)
lines[i] = line
break
return '\n'.join(lines)
def _replace_ovary_cfa(self, template: str, side: str, cfa_value: str) -> str:
"""Remplace la valeur CFA d'un ovaire spécifique"""
lines = template.split('\n')
for i, line in enumerate(lines):
if f"ovaire {side}" in line.lower() and i < len(lines) - 1:
# Chercher la ligne avec CFA dans les lignes suivantes
for j in range(i, min(i+3, len(lines))):
if "folliculaire CFA &x follicules" in lines[j]:
lines[j] = lines[j].replace("&x folliculaire CFA &x follicules", f"{cfa_value} folliculaire CFA")
break
break
return '\n'.join(lines)
def _replace_ovary_accessibility(self, template: str, side: str, accessibility: str) -> str:
"""Remplace l'accessibilité d'un ovaire spécifique"""
lines = template.split('\n')
in_ovary_section = False
for i, line in enumerate(lines):
if f"ovaire {side}" in line.lower():
in_ovary_section = True
elif in_ovary_section and "Accessibilité" in line:
# Déterminer quelle case cocher
if "rétro" in accessibility.lower():
line = line.replace("Accessibilité : &x rétro-utérin", "Accessibilité : X rétro-utérin")
elif "fixe" in accessibility.lower():
line = line.replace("&x fixe", "X fixe")
else: # normale ou aisée
line = line.replace("&x aisée", "X aisée")
lines[i] = line
in_ovary_section = False
break
return '\n'.join(lines)
def _parse_dimensions(self, dimensions_str: str) -> List[str]:
"""Parse les dimensions à partir d'une chaîne"""
if not dimensions_str:
return []
# Extraire tous les nombres
matches = re.findall(r'(\d+(?:[.,]\d+)?)', dimensions_str)
return [match.replace(',', '.') for match in matches]
def _find_all_placeholders(self, template: str) -> List[str]:
"""Trouve tous les placeholders &x dans le template"""
pattern = r'[^.]*&x[^.]*'
matches = re.findall(pattern, template)
return matches
def _replace_placeholder_in_context(self, template: str, context_pattern: str, value: str) -> str:
"""Remplace &x dans un contexte spécifique"""
escaped_pattern = re.escape(context_pattern).replace(r'\&x', r'&x')
def replace_func(match):
return match.group(0).replace('&x', value, 1)
return re.sub(escaped_pattern, replace_func, template)
def _apply_business_logic(self, template: str, extracted_data) -> str:
"""Applique la logique métier spécifique au domaine médical"""
# Logique par défaut pour les examens standard
template = template.replace("- &xVessie vide pendant l'examen", "- XVessie vide pendant l'examen")
template = template.replace("&x Absence de dilatation pyélo-calicielle", "X Absence de dilatation pyélo-calicielle")
template = template.replace("&x Absence d'image d'endométriose visible ce jour", "X Absence d'image d'endométriose visible ce jour")
return template
def _calculate_mapping_confidence(self, mapped_count: int, total_placeholders: int, error_count: int) -> float:
"""Calcule le score de confiance du mapping"""
if total_placeholders == 0:
return 1.0
base_confidence = mapped_count / total_placeholders
error_penalty = min(error_count * 0.1, 0.3)
return max(0.0, base_confidence - error_penalty)
# Fonctions de transformation des données
def _clean_numeric_value(self, value: str) -> str:
"""Nettoie les valeurs numériques"""
if not value:
return ""
cleaned = re.sub(r'\s*(mm|cm)\s*(mm|cm)', r' \1', str(value))
cleaned = re.sub(r'\s*(mm|cm).*', r'', cleaned)
cleaned = cleaned.replace(',', '.').strip()
return cleaned
def _clean_cfa_value(self, value: str) -> str:
"""Nettoie les valeurs CFA"""
if not value:
return ""
cleaned = str(value).replace(' follicules', '').replace(' follicules follicules', '').strip()
match = re.search(r'(\d+)', cleaned)
return match.group(1) if match else cleaned
def _extract_first_dimension(self, dimensions: str) -> str:
"""Extrait la première dimension"""
if not dimensions:
return ""
match = re.search(r'(\d+(?:[.,]\d+)?)', dimensions)
return match.group(1).replace(',', '.') if match else ""
def _extract_second_dimension(self, dimensions: str) -> str:
"""Extrait la deuxième dimension"""
if not dimensions:
return ""
matches = re.findall(r'(\d+(?:[.,]\d+)?)', dimensions)
return matches[1].replace(',', '.') if len(matches) > 1 else ""
def _extract_third_dimension(self, dimensions: str) -> str:
"""Extrait la troisième dimension"""
if not dimensions:
return ""
matches = re.findall(r'(\d+(?:[.,]\d+)?)', dimensions)
return matches[2].replace(',', '.') if len(matches) > 2 else ""
def _format_doppler_ir(self, ir_value: str) -> str:
"""Formate la valeur IR pour le template"""
if not ir_value:
return ""
cleaned = self._clean_numeric_value(ir_value)
if cleaned.startswith('0.'):
return cleaned[2:]
elif '.' in cleaned:
return cleaned.split('.')[1]
return cleaned
def print_mapping_report(self, result: MappingResult) -> str:
"""Génère un rapport de mapping formaté"""
report = "🔄 RAPPORT DE MAPPING TEMPLATE\n"
report += "=" * 50 + "\n\n"
report += f"📊 STATISTIQUES:\n"
report += f" Champs mappés: {len(result.mapped_fields)}\n"
report += f" Placeholders non mappés: {len(result.unmapped_placeholders)}\n"
report += f" Score de confiance: {result.mapping_confidence:.1%}\n"
report += f" Erreurs: {len(result.errors)}\n\n"
if result.mapped_fields:
report += "✅ CHAMPS MAPPÉS:\n"
for field, value in result.mapped_fields.items():
report += f" {field}: {value}\n"
report += "\n"
if result.unmapped_placeholders:
report += "❌ PLACEHOLDERS NON MAPPÉS:\n"
for placeholder in result.unmapped_placeholders[:10]:
report += f" {placeholder[:50]}...\n"
if len(result.unmapped_placeholders) > 10:
report += f" ... et {len(result.unmapped_placeholders) - 10} autres\n"
report += "\n"
if result.errors:
report += "⚠️ ERREURS:\n"
for error in result.errors:
report += f" {error}\n"
return report
# Classe exemple pour les données extraites avec vos données
class ExtractedData:
"""Classe exemple pour les données extraites"""
def __init__(self):
self.uterus_position = "antéversé"
self.uterus_size = "7,8 cm"
self.hysterometry = "60 mm"
self.endometrium_thickness = "3,7 mm"
self.junctional_zone_status = "épaissie"
self.adenomyosis_type = "diffuse"
# Données ovaires avec vos valeurs exactes
self.right_ovary_dimensions = "26 x 20 mm"
self.right_ovary_cfa = "5 follicules"
self.right_ovary_accessibility = "normale"
self.left_ovary_dimensions = "25 x 19 mm"
self.left_ovary_cfa = "22 follicules"
self.left_ovary_accessibility = "difficile rétro-utérine"
# Données Doppler
self.doppler_ip = "3,24"
self.doppler_ir = "0,91"
def test_corrected_ovary_mapping():
"""Test du mapping corrigé des ovaires"""
data = ExtractedData()
mapper = MedicalTemplateMapper()
result = mapper.map_extracted_data_to_template(data)
print("🔧 TEST DU MAPPING OVAIRES CORRIGÉ")
print("=" * 40)
print(mapper.print_mapping_report(result))
print("\n🔍 SECTION OVAIRES DANS LE RÉSULTAT:")
print("-" * 40)
# Extraire et afficher la section ovaires
lines = result.filled_template.split('\n')
ovary_section = []
in_ovary_section = False
for line in lines:
if "KISSING OVARIES" in line:
in_ovary_section = True
elif in_ovary_section and line.strip() == "":
if ovary_section: # Si on a déjà collecté des lignes
break
if in_ovary_section:
ovary_section.append(line)
if "L'échostructure des deux ovaires" in line:
break
print('\n'.join(ovary_section))
return result.filled_template
if __name__ == "__main__":
filled_report = test_corrected_ovary_mapping()