Spaces:
Runtime error
Runtime error
| """ | |
| validation_module.py | |
| -------------------- | |
| Clinical validation module for radiology reports. | |
| Performs: | |
| 1. Structural completeness check | |
| 2. Medical terminology validation | |
| 3. Logical consistency analysis (findings → impression) | |
| 4. Confidence score computation | |
| 5. Detailed validation feedback | |
| """ | |
| import logging | |
| import re | |
| from dataclasses import dataclass, field | |
| from typing import Dict, List, Optional, Tuple | |
| logger = logging.getLogger(__name__) | |
| # --------------------------------------------------------------------------- | |
| # Constants — Medical Terminology Dictionaries | |
| # --------------------------------------------------------------------------- | |
| # Positive radiological finding terms (suggest pathology is mentioned) | |
| FINDING_TERMS = { | |
| # Pulmonary | |
| "opacity", "consolidation", "infiltrate", "effusion", "pneumothorax", | |
| "atelectasis", "nodule", "mass", "lesion", "cardiomegaly", "pneumonia", | |
| "edema", "fibrosis", "emphysema", "pleural", "cavitation", "calcification", | |
| # Neuro | |
| "hemorrhage", "infarct", "midline shift", "herniation", "hydrocephalus", | |
| "atrophy", "gliosis", "white matter", "leukoencephalopathy", | |
| # Abdominal | |
| "hepatomegaly", "splenomegaly", "ascites", "obstruction", "perforation", | |
| "ileus", "dilatation", "gallstone", "calculi", "adenopathy", | |
| # Musculoskeletal | |
| "fracture", "dislocation", "lytic", "sclerotic", "arthritis", "erosion", | |
| "osteoporosis", "osteopenia", "hardware", "implant", "arthroplasty", | |
| "healed", "healing", "surgical", "fixation", "screw", "prosthesis", | |
| "joint", "bone", "density", "cortical", "trabecular", "alignment", | |
| # Vascular | |
| "stenosis", "occlusion", "aneurysm", "thrombosis", "calcified", | |
| # Soft tissue | |
| "swelling", "inflammation", "fluid", "collection", "thickening", | |
| # Normal descriptors | |
| "normal", "unremarkable", "no acute", "within normal limits", "clear", | |
| "intact", "preserved", "well-healed", "satisfactory", "stable", | |
| "demonstrates", "appears", "shows", "noted", "visible", "evident", | |
| } | |
| # Impression must include at least one of these conclusion terms | |
| IMPRESSION_CONCLUSION_TERMS = { | |
| "consistent with", "suggestive of", "findings indicate", "impression", | |
| "likely", "possible", "probable", "no evidence", "cannot exclude", | |
| "differential", "consider", "normal", "unremarkable", "recommend", | |
| "represents", "compatible with", "appears to", "evidence of", | |
| "healed", "healing", "demonstrates", "shows", "indicates", | |
| "no significant", "well-healed", "satisfactory", "stable", | |
| "there is", "there are", "findings are", | |
| } | |
| # Recommendation terms | |
| RECOMMENDATION_TERMS = { | |
| "correlation", "follow-up", "repeat", "ct scan", "mri", "biopsy", | |
| "consult", "refer", "further", "additional", "clinical", "evaluation", | |
| "monitoring", "treatment", "management", "urgent", "emergent", | |
| "recommended", "suggested", "advised", "periodic", "regular", | |
| "orthopedic", "specialist", "physiotherapy", "physical therapy", | |
| "bone density", "dexa", "rheumatologist", "scan", "imaging", | |
| } | |
| # Required report sections | |
| REQUIRED_SECTIONS = [ | |
| "examination_details", | |
| "clinical_information", | |
| "findings", | |
| "impression", | |
| "recommendations", | |
| ] | |
| # Minimum word counts per section | |
| MINIMUM_WORDS = { | |
| "examination_details": 5, | |
| "clinical_information": 8, | |
| "findings": 10, | |
| "impression": 5, | |
| "recommendations": 5, | |
| } | |
| # --------------------------------------------------------------------------- | |
| # Data Models | |
| # --------------------------------------------------------------------------- | |
| class ValidationResult: | |
| is_valid: bool = True | |
| confidence_score: float = 0.50 | |
| section_scores: Dict[str, float] = field(default_factory=dict) | |
| issues: List[str] = field(default_factory=list) | |
| warnings: List[str] = field(default_factory=list) | |
| suggestions: List[str] = field(default_factory=list) | |
| def to_dict(self) -> dict: | |
| return { | |
| "is_valid": self.is_valid, | |
| "confidence_score": round(self.confidence_score, 3), | |
| "section_scores": {k: round(v, 3) for k, v in self.section_scores.items()}, | |
| "issues": self.issues, | |
| "warnings": self.warnings, | |
| "suggestions": self.suggestions, | |
| } | |
| def summary(self) -> str: | |
| status = "✓ VALID" if self.is_valid else "✗ INVALID" | |
| lines = [ | |
| f"Validation Status : {status}", | |
| f"Confidence Score : {self.confidence_score:.2f}", | |
| ] | |
| if self.issues: | |
| lines.append("Issues:") | |
| lines.extend(f" • {i}" for i in self.issues) | |
| if self.warnings: | |
| lines.append("Warnings:") | |
| lines.extend(f" ⚠ {w}" for w in self.warnings) | |
| if self.suggestions: | |
| lines.append("Suggestions:") | |
| lines.extend(f" → {s}" for s in self.suggestions) | |
| return "\n".join(lines) | |
| # --------------------------------------------------------------------------- | |
| # Validator | |
| # --------------------------------------------------------------------------- | |
| class ClinicalValidator: | |
| """ | |
| Multi-stage clinical validation pipeline for radiology reports. | |
| """ | |
| WEIGHTS = { | |
| "completeness": 0.30, | |
| "content_length": 0.15, | |
| "terminology": 0.25, | |
| "logical_consistency": 0.20, | |
| "recommendation_quality": 0.10, | |
| } | |
| def validate(self, report) -> ValidationResult: | |
| result = ValidationResult() | |
| scores: Dict[str, float] = {} | |
| sections = { | |
| "examination_details": report.examination_details or "", | |
| "clinical_information": report.clinical_information or "", | |
| "findings": report.findings or "", | |
| "impression": report.impression or "", | |
| "recommendations": report.recommendations or "", | |
| } | |
| scores["completeness"] = self._check_completeness(sections, result) | |
| scores["content_length"] = self._check_content_length(sections, result) | |
| scores["terminology"] = self._check_terminology(sections, result) | |
| # ✅ UPDATED LOGIC USED HERE | |
| scores["logical_consistency"] = self._check_logical_consistency(sections, result) | |
| scores["recommendation_quality"] = self._check_recommendations(sections, result) | |
| weighted = sum(scores[k] * self.WEIGHTS[k] for k in self.WEIGHTS) | |
| result.confidence_score = round(min(max(weighted, 0.30), 0.88), 3) | |
| result.section_scores = scores | |
| result.is_valid = ( | |
| result.confidence_score >= 0.55 | |
| and len(result.issues) == 0 | |
| ) | |
| return result | |
| # ------------------------------------------------------------------ | |
| def _check_completeness(self, sections: Dict[str, str], result: ValidationResult) -> float: | |
| missing = [] | |
| for sec in REQUIRED_SECTIONS: | |
| if not sections.get(sec, "").strip(): | |
| missing.append(sec) | |
| if missing: | |
| result.issues.append( | |
| f"Missing required sections: {', '.join(missing)}" | |
| ) | |
| return 1.0 - (len(missing) / len(REQUIRED_SECTIONS)) | |
| return 1.0 | |
| # ------------------------------------------------------------------ | |
| def _check_content_length(self, sections: Dict[str, str], result: ValidationResult) -> float: | |
| total, passing = 0, 0 | |
| for sec, min_words in MINIMUM_WORDS.items(): | |
| total += 1 | |
| word_count = len(sections.get(sec, "").split()) | |
| if word_count >= min_words: | |
| passing += 1 | |
| else: | |
| result.warnings.append( | |
| f"Section '{sec}' is brief ({word_count} words; minimum recommended: {min_words})." | |
| ) | |
| return passing / total if total > 0 else 0.0 | |
| # ------------------------------------------------------------------ | |
| def _check_terminology(self, sections: Dict[str, str], result: ValidationResult) -> float: | |
| findings_text = sections.get("findings", "").lower() | |
| impression_text = sections.get("impression", "").lower() | |
| combined = findings_text + " " + impression_text | |
| matched_terms = FINDING_TERMS.intersection( | |
| set(re.findall(r"\b\w+\b", combined)) | |
| ) | |
| for term in FINDING_TERMS: | |
| if " " in term and term in combined: | |
| matched_terms.add(term) | |
| score = min(len(matched_terms) / 2.0, 1.0) | |
| if len(matched_terms) == 0: | |
| result.issues.append( | |
| "No recognised radiological terminology found in findings/impression." | |
| ) | |
| elif len(matched_terms) < 2: | |
| result.warnings.append( | |
| "Limited radiological terminology detected. Report may lack specificity." | |
| ) | |
| else: | |
| result.suggestions.append( | |
| f"Detected radiological terms: {', '.join(sorted(matched_terms)[:6])}." | |
| ) | |
| return score | |
| # ------------------------------------------------------------------ | |
| # ✅ FIXED FUNCTION (MAIN CHANGE) | |
| # ------------------------------------------------------------------ | |
| def _check_logical_consistency( | |
| self, sections: Dict[str, str], result: ValidationResult | |
| ) -> float: | |
| findings = sections.get("findings", "").lower() | |
| impression = sections.get("impression", "").lower() | |
| if not findings or not impression: | |
| return 0.5 | |
| score = 1.0 | |
| has_conclusion = any( | |
| term in impression for term in IMPRESSION_CONCLUSION_TERMS | |
| ) | |
| if not has_conclusion: | |
| result.warnings.append( | |
| "Impression does not contain a clear diagnostic conclusion." | |
| ) | |
| score -= 0.3 | |
| # ✅ NEW CLEAN LOGIC | |
| ABNORMAL_TERMS = { | |
| "opacity", "opacities", "ground glass", "consolidation", | |
| "lesion", "mass", "nodule", "effusion", "infiltrate", | |
| "atelectasis", "fibrosis", "infection" | |
| } | |
| NORMAL_TERMS = { | |
| "normal", "unremarkable", "no acute", "within normal limits", | |
| "clear", "intact", "preserved" | |
| } | |
| findings_has_pathology = any(t in findings for t in ABNORMAL_TERMS) | |
| impression_says_normal = any(t in impression for t in NORMAL_TERMS) | |
| # 🚨 STRONG ERROR DETECTION | |
| if findings_has_pathology and impression_says_normal: | |
| result.issues.append( | |
| "Critical error: Findings indicate abnormal pathology but impression states normal." | |
| ) | |
| result.is_valid = False | |
| score -= 0.7 | |
| # ⚠ Mixed statement detection | |
| if "no" in findings and findings_has_pathology: | |
| result.warnings.append( | |
| "Findings contain both abnormal and negative descriptors; verify interpretation." | |
| ) | |
| return max(score, 0.0) | |
| # ------------------------------------------------------------------ | |
| def _check_recommendations(self, sections: Dict[str, str], result: ValidationResult) -> float: | |
| recs = sections.get("recommendations", "").lower() | |
| if not recs: | |
| result.warnings.append("Recommendations section is empty.") | |
| return 0.0 | |
| matched = RECOMMENDATION_TERMS.intersection( | |
| set(re.findall(r"\b\w+\b", recs)) | |
| ) | |
| for term in RECOMMENDATION_TERMS: | |
| if " " in term and term in recs: | |
| matched.add(term) | |
| score = min(len(matched) / 2.0, 1.0) | |
| if len(matched) == 0: | |
| result.warnings.append( | |
| "Recommendations section does not contain specific clinical guidance." | |
| ) | |
| return score | |
| # --------------------------------------------------------------------------- | |
| # Convenience Function | |
| # --------------------------------------------------------------------------- | |
| def validate_report(report) -> ValidationResult: | |
| """Top-level convenience function.""" | |
| validator = ClinicalValidator() | |
| return validator.validate(report) | |
| # --------------------------------------------------------------------------- | |
| # Smoke Test | |
| # --------------------------------------------------------------------------- | |
| if __name__ == "__main__": | |
| import sys | |
| sys.path.insert(0, ".") | |
| from backend.report_generator import PatientInfo, ImagingInfo, RadiologyReport | |
| mock_report = RadiologyReport( | |
| examination_details="Chest X-ray, posteroanterior view.", | |
| clinical_information="52-year-old male presenting with persistent cough and fever for 3 days.", | |
| findings=( | |
| "Patchy consolidation in the right lower lobe with air bronchograms. " | |
| "No pleural effusion. Cardiac silhouette within normal limits. " | |
| "Costophrenic angles are sharp bilaterally." | |
| ), | |
| impression=( | |
| "Findings are consistent with right lower lobe pneumonia. " | |
| "No evidence of pleural complication." | |
| ), | |
| recommendations=( | |
| "Clinical correlation recommended. " | |
| "Consider repeat chest X-ray in 4 to 6 weeks to confirm resolution. " | |
| "If symptoms persist, CT chest may be warranted." | |
| ), | |
| ) | |
| result = validate_report(mock_report) | |
| print(result.summary()) | |
| print(f"\nFull JSON: {result.to_dict()}") | |