""" 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 # --------------------------------------------------------------------------- @dataclass 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()}")