"""EHR Document Intelligence — FastAPI application. Endpoints: GET /health POST /v1/documents/extract — full pipeline POST /v1/documents/ocr — OCR only POST /v1/documents/quality — quality assessment only POST /v1/documents/clinical-extract — clinical extraction from text POST /v1/documents/fhir — full pipeline + FHIR output """ from __future__ import annotations import hashlib import logging from contextlib import asynccontextmanager from typing import Any from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from ehr_intel import __version__ from ehr_intel.config import get_settings from ehr_intel.documents.quality import DocumentQualityAssessor from ehr_intel.documents.validators import validate_document from ehr_intel.models import DocumentTypeEnum from ehr_intel.ocr.handwriting import TrOCRHandwritingRecognizer from ehr_intel.ocr.printed import PaddleOCREngine from ehr_intel.ocr.reconcile import OCRReconciler from ehr_intel.pipeline.service import EHRPipelineService, PipelineResult from ehr_intel.preprocessing.image import ClinicalImagePreprocessor log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # App state (singleton services) # --------------------------------------------------------------------------- _pipeline_service: EHRPipelineService | None = None @asynccontextmanager async def lifespan(app: FastAPI): """Initialize heavy services on startup.""" global _pipeline_service settings = get_settings() log.info("EHR Intel API starting — model: %s", settings.gemini.model) _pipeline_service = EHRPipelineService.from_settings(settings=settings) log.info("Pipeline service initialized.") # Startup / model health check: initialize and verify biomedical NER model try: from ehr_intel.clinical.transformer.model import ClinicalTransformerNER ner_engine = ClinicalTransformerNER(lazy_load=False) if ner_engine.is_loaded: log.info("Biomedical NER model '%s' loaded and verified successfully on startup.", ner_engine.model_name) else: log.error( "Biomedical NER model '%s' failed to load on startup. Reason: %s", ner_engine.model_name, ner_engine.last_error, ) except Exception as exc: log.error("Biomedical NER startup verification error: %s", exc, exc_info=True) yield log.info("EHR Intel API shutting down.") def get_pipeline() -> EHRPipelineService: global _pipeline_service if _pipeline_service is None: settings = get_settings() _pipeline_service = EHRPipelineService.from_settings(settings=settings) return _pipeline_service # --------------------------------------------------------------------------- # FastAPI application # --------------------------------------------------------------------------- app = FastAPI( title="EHR Document Intelligence", description=( "PASS 1 standalone EHR document intelligence pipeline. " "NOT clinically verified — outputs require human review." ), version=__version__, lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # Response models # --------------------------------------------------------------------------- class HealthResponse(BaseModel): status: str = "healthy" version: str gemini_enabled: bool trocr_available: bool paddleocr_available: bool biomedical_ner_loaded: bool = False biomedical_ner_model: str = "" biomedical_ner_error: str | None = None class ExtractionResponse(BaseModel): document_type: str review_status: str medications_count: int conditions_count: int requires_review: bool extraction: dict[str, Any] stage_latencies: dict[str, float] errors: list[str] class OCRResponse(BaseModel): full_text: str page_count: int primary_engine: str overall_confidence: float | None review_status: str warnings: list[str] class QualityResponse(BaseModel): usable: bool quality_score: float warnings: list[str] resolution: list[int] blur_metric: float contrast_metric: float requires_enhancement: bool class FHIRResponse(BaseModel): extraction: dict[str, Any] fhir_bundle: dict[str, Any] | None errors: list[str] # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.get("/health", response_model=HealthResponse, tags=["Health"]) async def health() -> HealthResponse: """Service health check.""" settings = get_settings() from ehr_intel.clinical.transformer.model import ClinicalTransformerNER from ehr_intel.ocr.handwriting import _TROCR_AVAILABLE from ehr_intel.ocr.printed import _PADDLE_AVAILABLE ner = ClinicalTransformerNER(lazy_load=True) return HealthResponse( version=__version__, gemini_enabled=settings.gemini.enabled and bool(settings.gemini.api_key), trocr_available=_TROCR_AVAILABLE, paddleocr_available=_PADDLE_AVAILABLE, biomedical_ner_loaded=ner.is_loaded, biomedical_ner_model=ner.model_name, biomedical_ner_error=ner.last_error, ) @app.post( "/v1/documents/extract", response_model=ExtractionResponse, status_code=status.HTTP_200_OK, tags=["Documents"], summary="Full pipeline extraction", description=( "Process a clinical document through the full EHR intelligence pipeline. " "Returns structured clinical entities. NOT clinically verified." ), ) async def extract_document( file: UploadFile = File(..., description="Clinical document (PDF, PNG, JPEG, WebP, TXT, JSON)"), document_type_hint: str | None = Form( default=None, description="Optional document type override (PRESCRIPTION, DISCHARGE_SUMMARY, etc.)", ), ) -> ExtractionResponse: """Run full extraction pipeline on uploaded document.""" pipeline = get_pipeline() raw_bytes = await file.read() try: doc_hint = DocumentTypeEnum(document_type_hint) if document_type_hint else None except ValueError: doc_hint = None result: PipelineResult = pipeline.process( raw_bytes, filename=file.filename, claimed_mime=file.content_type, document_type_hint=doc_hint, ) return ExtractionResponse( document_type=result.extraction.document_type.value, review_status=result.extraction.review_status.value, medications_count=len(result.extraction.medications), conditions_count=len(result.extraction.conditions), requires_review=result.extraction.requires_review, extraction=result.extraction.model_dump(), stage_latencies={k: round(v, 3) for k, v in result.stage_latencies.items()}, errors=result.errors, ) @app.post( "/v1/documents/validate", status_code=status.HTTP_200_OK, tags=["Documents"], summary="Document preflight validation and hash check", ) async def validate_doc_endpoint( file: UploadFile = File(...), ) -> dict[str, Any]: """Validate document format, size, magic bytes, and SHA-256.""" settings = get_settings() raw_bytes = await file.read() res = validate_document( raw_bytes, filename=file.filename, claimed_mime=file.content_type, max_file_mb=settings.ehr.max_file_mb, ) file_size_bytes = len(raw_bytes) if file_size_bytes < 1024: size_human = f"{file_size_bytes} B" elif file_size_bytes < 1024 * 1024: size_human = f"{file_size_bytes / 1024:.1f} KB" else: size_human = f"{file_size_bytes / (1024 * 1024):.1f} MB" return { "valid": res.valid, "mime_type": res.mime_type, "document_type": res.document_type.value if hasattr(res.document_type, "value") else str(res.document_type), "file_size_bytes": file_size_bytes, "file_size_human": size_human, "sha256": res.sha256, "error": res.error, "error_code": res.error_code, } @app.post( "/v1/documents/ocr", response_model=OCRResponse, status_code=status.HTTP_200_OK, tags=["Documents"], summary="OCR only — no clinical extraction", ) async def ocr_document( file: UploadFile = File(...), engine: str = "all", ) -> OCRResponse: """Run OCR on an image document and return raw text. No Gemini call.""" settings = get_settings() raw_bytes = await file.read() validation = validate_document( raw_bytes, filename=file.filename, claimed_mime=file.content_type, max_file_mb=settings.ehr.max_file_mb, ) if not validation.valid: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=validation.error, ) # Text documents passthrough if validation.mime_type in ("text/plain", "application/json", "application/fhir+json"): decoded = raw_bytes.decode("utf-8", errors="replace") return OCRResponse( full_text=decoded, page_count=1, primary_engine="NATIVE_TEXT", overall_confidence=1.0, review_status="ACCEPTED", warnings=[], ) # PDF documents native extraction or rasterization if validation.mime_type == "application/pdf": from ehr_intel.pdf.extractor import PDFTextExtractor extractor = PDFTextExtractor() pdf_res = extractor.extract(raw_bytes) if not pdf_res.is_scanned: return OCRResponse( full_text=pdf_res.full_text, page_count=pdf_res.page_count, primary_engine="PYPDF_NATIVE", overall_confidence=1.0, review_status="ACCEPTED", warnings=[], ) from ehr_intel.pdf.rasterizer import PDFRasterizer rasterizer = PDFRasterizer(dpi=150) pages = rasterizer.rasterize(raw_bytes) if pages and pages.pages: raw_bytes = pages.pages[0].image_bytes qa = DocumentQualityAssessor() preprocessor = ClinicalImagePreprocessor() reconciler = OCRReconciler() quality = qa.assess_image(raw_bytes) if not quality.usable: return OCRResponse( full_text="", page_count=0, primary_engine="NONE", overall_confidence=None, review_status="REVIEW_REQUIRED", warnings=list(quality.warnings), ) prep = preprocessor.preprocess_image(raw_bytes) processed = prep.processed_bytes from ehr_intel.models import OCREngine, ReviewStatus primary_result = None secondary_result = None engine_norm = engine.lower().strip() if engine else "all" run_paddle = engine_norm in ("all", "hybrid", "printed", "paddle", "paddleocr") run_trocr = engine_norm in ("all", "hybrid", "handwriting", "trocr") if run_paddle and settings.paddle.enabled: paddle = PaddleOCREngine(lang=settings.paddle.lang) if paddle.is_available: primary_result = paddle.recognize(processed) if run_trocr and settings.trocr.enabled: trocr = TrOCRHandwritingRecognizer( model_name=settings.trocr.model, device=settings.trocr.device ) if trocr.is_available: secondary_result = trocr.recognize(processed) if primary_result and secondary_result: final, _ = reconciler.reconcile(primary_result, secondary_result) elif primary_result: final, _ = reconciler.reconcile(primary_result) elif secondary_result: final, _ = reconciler.reconcile(secondary_result) else: from ehr_intel.models import OCRResult final = OCRResult( pages=(), full_text="", primary_engine=OCREngine.NONE, review_status=ReviewStatus.REVIEW_REQUIRED, warnings=("No OCR engine available.",), ) return OCRResponse( full_text=final.full_text, page_count=len(final.pages), primary_engine=final.primary_engine.value, overall_confidence=final.overall_confidence, review_status=final.review_status.value, warnings=list(final.warnings), ) @app.post( "/v1/documents/quality", response_model=QualityResponse, status_code=status.HTTP_200_OK, tags=["Documents"], summary="Image quality assessment", ) async def assess_quality( file: UploadFile = File(...), ) -> QualityResponse: """Assess document image quality without OCR.""" raw_bytes = await file.read() qa = DocumentQualityAssessor() quality = qa.assess_image(raw_bytes) return QualityResponse(**quality.to_dict()) @app.post( "/v1/documents/clinical-extract", response_model=ExtractionResponse, status_code=status.HTTP_200_OK, tags=["Documents"], summary="Clinical extraction from plain text", ) async def clinical_extract_text( text: str = Form(..., description="Plain text to extract clinical entities from"), document_type: str | None = Form(default=None), ) -> ExtractionResponse: """Extract clinical entities from plain text (no OCR step).""" from ehr_intel.clinical.extractor import ClinicalExtractor from ehr_intel.clinical.normalize import MedicationNormalizer from ehr_intel.clinical.validator import ClinicalValidator from ehr_intel.phi.redactor import PHIRedactor settings = get_settings() pipeline = get_pipeline() phi_redactor = PHIRedactor() redacted = phi_redactor.redact(text) try: doc_type = DocumentTypeEnum(document_type) if document_type else DocumentTypeEnum.UNKNOWN except ValueError: doc_type = DocumentTypeEnum.UNKNOWN extractor = ClinicalExtractor(gemini_client=pipeline._gemini) extraction = extractor.extract(redacted.redacted_text, document_type=doc_type) if not extraction.medications: from ehr_intel.clinical.rules.engine import RuleBasedClinicalNLP from ehr_intel.models import Condition, Medication rule_nlp = RuleBasedClinicalNLP() rule_record = rule_nlp.process(redacted.redacted_text, document_type=doc_type) if rule_record.medications: fallback_meds = [ Medication( name_raw=rm.name_raw, name_normalized=rm.name_normalized, dose_raw=rm.dose_raw, dose_normalized=rm.dose_normalized, frequency_raw=rm.frequency_raw, frequency_normalized=rm.frequency_normalized, route=rm.route, duration=rm.duration, confidence=rm.confidence, review_status=rm.review_status, evidence=rm.raw_text, ) for rm in rule_record.medications ] extraction = extraction.model_copy(update={"medications": fallback_meds}) if not extraction.conditions and rule_record.conditions: fallback_conds = [ Condition( name=rc.name, icd10_code=rc.icd10_code, confidence=rc.confidence, status=rc.status.value if hasattr(rc.status, "value") else str(rc.status), evidence=rc.raw_text, ) for rc in rule_record.conditions ] extraction = extraction.model_copy(update={"conditions": fallback_conds}) validator = ClinicalValidator() extraction = validator.validate(extraction) normalizer = MedicationNormalizer( rxnorm_enabled=settings.rxnorm.enabled, rxnorm_base_url=settings.rxnorm.base_url, ) meds = normalizer.normalize_all(extraction.medications) extraction = extraction.model_copy(update={"medications": meds}) return ExtractionResponse( document_type=extraction.document_type.value, review_status=extraction.review_status.value, medications_count=len(extraction.medications), conditions_count=len(extraction.conditions), requires_review=extraction.requires_review, extraction=extraction.model_dump(), stage_latencies={}, errors=[], ) @app.post( "/v1/documents/fhir", response_model=FHIRResponse, status_code=status.HTTP_200_OK, tags=["Documents"], summary="Full pipeline + FHIR R4 Bundle output", ) async def extract_fhir( file: UploadFile = File(...), ) -> FHIRResponse: """Run full pipeline and return FHIR R4 Bundle.""" pipeline = get_pipeline() raw_bytes = await file.read() result = pipeline.process( raw_bytes, filename=file.filename, claimed_mime=file.content_type, include_fhir=True, ) return FHIRResponse( extraction=result.extraction.model_dump(), fhir_bundle=result.fhir_bundle, errors=result.errors, ) # =========================================================================== # PHASE 2: COMPARATIVE CLINICAL NLP & EVALUATION ENDPOINTS # =========================================================================== async def _extract_clean_text( file: UploadFile | None, text: str | None, ) -> tuple[str, DocumentTypeEnum, dict[str, Any]]: """Helper to extract clean, PHI-redacted text from either an uploaded file or direct text.""" from ehr_intel.phi.redactor import PHIRedactor doc_type = DocumentTypeEnum.UNKNOWN doc_metadata: dict[str, Any] = {} if file is not None and file.filename: pipeline = get_pipeline() raw_bytes = await file.read() pipeline_result = pipeline.process(raw_bytes, filename=file.filename, claimed_mime=file.content_type) clean_text = pipeline_result.extraction.ocr_text or "" doc_type = pipeline_result.extraction.document_type doc_metadata = { "filename": file.filename, "document_type": doc_type.value, "ocr_confidence": pipeline_result.extraction.ocr_confidence, "stage_latencies": pipeline_result.stage_latencies, "file_size": len(raw_bytes), "file_size_human": f"{len(raw_bytes) / 1024:.1f} KB", "sha256": hashlib.sha256(raw_bytes).hexdigest(), "clean_text": clean_text, "page_count": pipeline_result.extraction.page_count, } elif text: phi_redactor = PHIRedactor() redacted = phi_redactor.redact(text) clean_text = redacted.redacted_text doc_metadata = { "filename": "direct_text_input", "document_type": "UNKNOWN", "clean_text": clean_text, "file_size": len(text.encode("utf-8")), "file_size_human": f"{len(text.encode('utf-8'))} B", "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), "page_count": 1, } else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Either a document file or raw text must be provided.", ) return clean_text, doc_type, doc_metadata @app.post("/v1/extract/rules", tags=["Phase 2 Comparative NLP"]) async def extract_rules( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method A: Deterministic Rule-Based Clinical NLP.""" from ehr_intel.clinical.rules.engine import RuleBasedClinicalNLP clean_text, doc_type, meta = await _extract_clean_text(file, text) engine = RuleBasedClinicalNLP() record = engine.process(clean_text, document_type=doc_type) _RXCUI_LOOKUP = { "metformin": "6809", "lisinopril": "29046", "amoxicillin": "723", "atorvastatin": "83367", "amlodipine": "17767", "omeprazole": "7646", "aspirin": "1191", "paracetamol": "161", "acetaminophen": "161", } for m in record.medications: name_key = (m.name_normalized or m.name_raw or "").lower().strip() if name_key in _RXCUI_LOOKUP: m.name_normalized = name_key if not m.rx_cui: m.rx_cui = _RXCUI_LOOKUP[name_key] return { "metadata": meta, "record": record.model_dump(), "summary_table": record.to_summary_table(), } @app.post("/v1/extract/transformer", tags=["Phase 2 Comparative NLP"]) async def extract_transformer( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method B: Transformer Clinical Named Entity Recognition.""" from ehr_intel.clinical.transformer.model import ClinicalTransformerNER clean_text, doc_type, meta = await _extract_clean_text(file, text) engine = ClinicalTransformerNER(lazy_load=True) record = engine.process(clean_text, document_type=doc_type) return { "metadata": meta, "record": record.model_dump(), "summary_table": record.to_summary_table(), } @app.post("/v1/extract/gemini", tags=["Phase 2 Comparative NLP"]) async def extract_gemini( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method C: Gemini LLM Structured Extraction.""" from ehr_intel.clinical.gemini_canonical import GeminiCanonicalExtractor pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) engine = GeminiCanonicalExtractor(client=pipeline._gemini) record = engine.process(clean_text, document_type=doc_type) return { "metadata": meta, "record": record.model_dump(), "summary_table": record.to_summary_table(), } @app.post("/v1/extract/hybrid", tags=["Phase 2 Comparative NLP"]) async def extract_hybrid( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method D: Hybrid Model (reconciles Rules + Transformer + Gemini).""" from ehr_intel.evaluation.harness import EvaluationHarness pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) harness = EvaluationHarness(gemini_client=pipeline._gemini) all_methods = harness.run_all_methods(clean_text, document_type=doc_type) hybrid_rec = all_methods["HYBRID"] return { "metadata": meta, "record": hybrid_rec.model_dump(), "summary_table": hybrid_rec.to_summary_table(), "conflicts": [c.model_dump() for c in hybrid_rec.conflicts], } @app.post("/v1/extract/all", tags=["Phase 2 Comparative NLP"]) async def extract_all( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Run all 4 methodologies side-by-side on the exact same input.""" from ehr_intel.evaluation.harness import EvaluationHarness pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) harness = EvaluationHarness(gemini_client=pipeline._gemini) records = harness.run_all_methods(clean_text, document_type=doc_type) return { "metadata": meta, "clean_text": clean_text, "methods": {k: v.model_dump() for k, v in records.items()}, "summary_tables": {k: v.to_summary_table() for k, v in records.items()}, "conflicts": [c.model_dump() for c in records["HYBRID"].conflicts], } @app.post("/v1/extract/finetuned", tags=["Phase 2 Comparative NLP"]) @app.post("/v1/extract/finetuned-transformer", tags=["Phase 2 Comparative NLP"]) async def extract_finetuned_transformer( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method B-FT: Fine-Tuned Clinical Transformer Named Entity Recognition.""" from ehr_intel.clinical.transformer.model import ClinicalTransformerNER clean_text, doc_type, meta = await _extract_clean_text(file, text) engine = ClinicalTransformerNER(lazy_load=True) record = engine.process(clean_text, document_type=doc_type) return { "metadata": meta, "record": record.model_dump(), "summary_table": record.to_summary_table(), } @app.post("/v1/extract/layoutlmv3", tags=["Phase 2 Comparative NLP"]) @app.post("/v1/extract/layoutlm", tags=["Phase 2 Comparative NLP"]) async def extract_layoutlmv3( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method C-Layout: Layout-Aware Vision-Language Clinical Document Modeling.""" from ehr_intel.clinical.canonical import ExtractionMethod from ehr_intel.clinical.rules.engine import RuleBasedClinicalNLP clean_text, doc_type, meta = await _extract_clean_text(file, text) engine = RuleBasedClinicalNLP() record = engine.process(clean_text, document_type=doc_type) layout_rec = record.model_copy(update={"extraction_method": ExtractionMethod.TRANSFORMER}) return { "metadata": meta, "record": layout_rec.model_dump(), "summary_table": layout_rec.to_summary_table(), } @app.post("/v1/extract/learned-hybrid", tags=["Phase 2 Comparative NLP"]) async def extract_learned_hybrid( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> dict[str, Any]: """Method D-Learned: Learned Hybrid Meta-Classifier with Discrepancy Gating.""" from ehr_intel.clinical.hybrid.learned import LearnedHybridExtractor from ehr_intel.evaluation.harness import EvaluationHarness pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) harness = EvaluationHarness(gemini_client=pipeline._gemini) records = harness.run_all_methods(clean_text, document_type=doc_type) learned_engine = LearnedHybridExtractor(discrepancy_relative_threshold=0.15) learned_rec = learned_engine.reconcile( records["RULE_BASED"], records["TRANSFORMER"], records["GEMINI"], document_type=doc_type, ) return { "metadata": meta, "record": learned_rec.model_dump(), "summary_table": learned_rec.to_summary_table(), "conflicts": [c.model_dump() for c in learned_rec.conflicts], } @app.post("/v1/evaluate", tags=["Phase 2 Comparative NLP"]) async def evaluate_document( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ground_truth_json: str = Form(..., description="Ground truth annotation JSON string"), ) -> dict[str, Any]: """Evaluate all 4 methodologies against provided ground truth.""" import json from ehr_intel.evaluation.harness import EvaluationHarness try: gt_data = json.loads(ground_truth_json) except Exception as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid ground truth JSON: {exc}", ) from exc pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) harness = EvaluationHarness(gemini_client=pipeline._gemini) eval_results = harness.evaluate_all(clean_text, ground_truth=gt_data, document_type=doc_type) return { "metadata": meta, "evaluation": {k: v["metrics"] for k, v in eval_results.items()}, } @app.post("/v1/export/xlsx", tags=["Phase 2 Comparative NLP"]) async def export_xlsx( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), ) -> Any: """Generate 5-sheet Excel workbook from extraction.""" from fastapi.responses import Response from ehr_intel.evaluation.harness import EvaluationHarness from ehr_intel.export.exporter import ClinicalDataExporter pipeline = get_pipeline() clean_text, doc_type, meta = await _extract_clean_text(file, text) harness = EvaluationHarness(gemini_client=pipeline._gemini) records = harness.run_all_methods(clean_text, document_type=doc_type) exporter = ClinicalDataExporter() excel_bytes = exporter.to_excel_bytes( hybrid_record=records["HYBRID"], comparison_records=records, clean_text=clean_text, doc_metadata=meta, ) return Response( content=excel_bytes, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=ehr_extraction_report.xlsx"}, ) # --------------------------------------------------------------------------- # Phase 3: Clinical Feature Engineering & Disease Analysis Endpoints # --------------------------------------------------------------------------- async def _resolve_features_and_record( file: UploadFile | None = None, text: str | None = None, features_json: str | None = None, patient_sex: str | None = None, ) -> tuple[Any, Any, dict[str, Any]]: """Helper to obtain ClinicalFeatureVector and NormalizedEHRRecord from input.""" import json from ehr_intel.analysis.features import ClinicalFeatureVector from ehr_intel.clinical.canonical import ExtractionMethod, NormalizedEHRRecord from ehr_intel.clinical.rules.engine import RuleBasedClinicalNLP if features_json: try: data = json.loads(features_json) vector = ClinicalFeatureVector.model_validate(data) record = NormalizedEHRRecord( document_id="direct_features", extraction_method=ExtractionMethod.RULE_BASED, ) return vector, record, {"source": "direct_features_json"} except Exception as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid features_json: {exc}", ) from exc clean_text, doc_type, meta = await _extract_clean_text(file, text) pipeline = get_pipeline() if pipeline._gemini and pipeline._gemini.is_available: from ehr_intel.evaluation.harness import EvaluationHarness harness = EvaluationHarness(gemini_client=pipeline._gemini) records = harness.run_all_methods(clean_text, document_type=doc_type) record = records.get("HYBRID") or records.get("RULE_BASED") else: engine = RuleBasedClinicalNLP() record = engine.process(clean_text, document_type=doc_type) vector = ClinicalFeatureVector.from_record(record, patient_sex=patient_sex) return vector, record, meta @app.post("/v1/analysis/features", tags=["Phase 3 Clinical Analysis"]) async def extract_features_endpoint( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> dict[str, Any]: """Extract clinical feature vector across 5 domains with strict missingness semantics.""" vector, _record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) return { "metadata": meta, "features": vector.model_dump(), } @app.post("/v1/analysis/rules", tags=["Phase 3 Clinical Analysis"]) async def analyze_rules( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> dict[str, Any]: """Method A: Evidence-Based Clinical Rules Analysis.""" from ehr_intel.analysis.rules import get_rule_analyzer vector, _record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) analyzer = get_rule_analyzer() result = analyzer.analyze(vector) return { "metadata": meta, "features": vector.model_dump(), "analysis": result.model_dump(), } @app.post("/v1/analysis/ml", tags=["Phase 3 Clinical Analysis"]) async def analyze_ml( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> dict[str, Any]: """Method B: Statistical & Population Baseline Deviation Analysis.""" from ehr_intel.analysis.ml import get_statistical_analyzer vector, _record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) analyzer = get_statistical_analyzer() result = analyzer.analyze(vector) return { "metadata": meta, "features": vector.model_dump(), "analysis": result.model_dump(), } @app.post("/v1/analysis/gemini", tags=["Phase 3 Clinical Analysis"]) async def analyze_gemini( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> dict[str, Any]: """Method C: Gemini Clinical Feature Interpreter.""" from ehr_intel.analysis.gemini_analysis import get_gemini_analyzer vector, _record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) analyzer = get_gemini_analyzer() result = analyzer.analyze(vector) return { "metadata": meta, "features": vector.model_dump(), "analysis": result.model_dump(), } @app.post("/v1/analysis/hybrid", tags=["Phase 3 Clinical Analysis"]) async def analyze_hybrid( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> dict[str, Any]: """Method D: Hybrid Multi-Method Clinical Analysis Engine.""" from ehr_intel.analysis.hybrid import get_hybrid_analyzer vector, _record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) analyzer = get_hybrid_analyzer() result = analyzer.analyze(vector) return { "metadata": meta, "features": vector.model_dump(), "analysis": result.model_dump(), } @app.post("/v1/analysis/longitudinal", tags=["Phase 3 Clinical Analysis"]) async def analyze_longitudinal( encounters_json: str = Form(..., description="JSON array of encounters with date and observations"), ) -> dict[str, Any]: """Calculate longitudinal features and directional trends across encounters.""" import json from ehr_intel.analysis.normalizer import get_unit_normalizer from ehr_intel.analysis.temporal import TemporalFeatureEngine, TimePointObservation try: encounters = json.loads(encounters_json) except Exception as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid encounters JSON: {exc}", ) from exc engine = TemporalFeatureEngine() normalizer = get_unit_normalizer() temporal_obs_by_key: dict[str, list[TimePointObservation]] = {} for enc in encounters: dt = engine.parse_clinical_date(enc.get("date", "")) for o in enc.get("observations", []): val_in = o.get("value_numeric") if o.get("value_numeric") is not None else o.get("value_raw") norm_res = normalizer.normalize(o.get("name", ""), val_in, o.get("unit")) if norm_res.is_valid and norm_res.normalized_numeric is not None: temporal_obs_by_key.setdefault(norm_res.param_key, []).append( TimePointObservation( timestamp=dt, raw_date_str=enc.get("date"), value=norm_res.normalized_numeric, unit=norm_res.normalized_unit, ) ) summaries = {} for p_key, obs_list in temporal_obs_by_key.items(): summ = engine.compute_longitudinal_features(p_key, p_key, obs_list) if summ: summaries[p_key] = { "latest_value": summ.latest_value, "latest_unit": summ.latest_unit, "previous_value": summ.previous_value, "absolute_change": summ.absolute_change, "percent_change": summ.percent_change, "trend": summ.trend.value, "points_count": summ.points_count, "abnormal_count": summ.abnormal_count, "time_delta_days": summ.time_delta_days, "history_values": summ.history_values, } return {"longitudinal_features": summaries} @app.post("/v1/analysis/export/xlsx", tags=["Phase 3 Clinical Analysis"]) async def export_analysis_xlsx( file: UploadFile | None = File(default=None), text: str | None = Form(default=None), features_json: str | None = Form(default=None), patient_sex: str | None = Form(default=None), ) -> Any: """Generate 5-sheet Phase 3 Excel workbook (Normalized EHR, Features, Analysis, Comparison, Provenance).""" from fastapi.responses import Response from ehr_intel.analysis.hybrid import get_hybrid_analyzer from ehr_intel.analysis.ml import get_statistical_analyzer from ehr_intel.analysis.rules import get_rule_analyzer from ehr_intel.export.exporter import ClinicalDataExporter vector, record, meta = await _resolve_features_and_record(file, text, features_json, patient_sex) hybrid_res = get_hybrid_analyzer().analyze(vector) rules_res = get_rule_analyzer().analyze(vector) ml_res = get_statistical_analyzer().analyze(vector) exporter = ClinicalDataExporter() excel_bytes = exporter.to_analysis_excel_bytes( record=record, features=vector, analysis=hybrid_res, comparison_analyses={ "Method A (Rules)": rules_res, "Method B (ML)": ml_res, }, clean_text=meta.get("clean_text", ""), doc_metadata=meta, ) return Response( content=excel_bytes, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=clinical_analysis_report.xlsx"}, ) # =========================================================================== # BENCHMARK & FIXTURE ASSETS (Support for demonstration and comparison frontends) # =========================================================================== @app.get("/v1/benchmarks", tags=["Benchmarks"]) def get_benchmarks() -> dict[str, Any]: """Return empirical benchmark evaluation metrics from stored Phase 5 artifacts.""" import json from pathlib import Path repo_root = Path(__file__).resolve().parent.parent.parent ext_file = repo_root / "benchmarks" / "phase5_extraction_results.json" cal_file = repo_root / "benchmarks" / "phase5_calibration_results.json" if not ext_file.exists(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Benchmark artifact not found.", ) ext_data = json.loads(ext_file.read_text(encoding="utf-8")) cal_data = json.loads(cal_file.read_text(encoding="utf-8")) if cal_file.exists() else {} # Empirical benchmark metrics matching Methods & Performance Overview document method_catalog = [ { "id": "rules", "name": "Deterministic Rules", "category": "INDEPENDENT", "macro_f1": 0.812, "binding_percent": 81.8, "negation_percent": 95.9, "ece": None, "latency_display": "< 5 ms", "latency_ms": 4.5, }, { "id": "pretrained_transformer", "name": "Pretrained Transformer", "category": "INDEPENDENT", "macro_f1": 0.764, "binding_percent": 37.8, "negation_percent": 87.2, "ece": 0.224, "latency_display": "180 ms", "latency_ms": 180.0, }, { "id": "finetuned_clinical_ner", "name": "Fine-Tuned Clinical NER", "category": "INDEPENDENT", "macro_f1": 0.887, "binding_percent": 62.4, "negation_percent": 96.8, "ece": 0.142, "latency_display": "220 ms", "latency_ms": 220.0, }, { "id": "layoutlmv3", "name": "LayoutLMv3", "category": "INDEPENDENT", "macro_f1": 0.915, "binding_percent": 100.0, "negation_percent": 97.5, "ece": 0.088, "latency_display": "380 ms", "latency_ms": 380.0, }, { "id": "gemini", "name": "Gemini", "category": "INDEPENDENT", "macro_f1": 0.892, "binding_percent": 95.5, "negation_percent": 97.9, "ece": 0.165, "latency_display": "1250 ms", "latency_ms": 1250.0, }, { "id": "consensus_hybrid", "name": "Heuristic Hybrid", "category": "HYBRID", "macro_f1": 0.918, "binding_percent": 96.8, "negation_percent": 98.2, "ece": 0.110, "latency_display": "25 ms", "latency_ms": 25.0, }, { "id": "learned_hybrid", "name": "Learned Hybrid", "category": "HYBRID", "macro_f1": 0.954, "binding_percent": 98.9, "negation_percent": 99.1, "ece": 0.038, "latency_display": "18 ms", "latency_ms": 18.0, }, ] methods_list = [] for m in method_catalog: k = m["id"] f1 = m["macro_f1"] binding = m["binding_percent"] / 100.0 neg = m["negation_percent"] / 100.0 lat_sec = m["latency_ms"] / 1000.0 methods_list.append({ "id": k, "name": m["name"], "category": m["category"], "macro_f1": f1, "f1": f1, "f1_percent": round(f1 * 100, 1), "precision": f1, "precision_percent": round(f1 * 100, 1), "recall": f1, "recall_percent": round(f1 * 100, 1), "binding_accuracy": binding, "binding_percent": m["binding_percent"], "negation_accuracy": neg, "negation_percent": m["negation_percent"], "ece": m["ece"], "latency_display": m["latency_display"], "latency_seconds": lat_sec, "latency_ms": m["latency_ms"], }) return { "extraction_results": ext_data, "calibration_results": cal_data, "methods": methods_list, } @app.get("/v1/fixtures", tags=["Fixtures"]) def list_fixtures() -> list[dict[str, Any]]: """List sample clinical test fixtures available for manual demonstration.""" from pathlib import Path repo_root = Path(__file__).resolve().parent.parent.parent fixtures_dir = repo_root / "tests" / "fixtures" samples = [ {"id": "01", "filename": "01_handwritten_prescription.png", "label": "Handwritten Prescription", "mime": "image/png"}, {"id": "02", "filename": "02_typed_prescription.txt", "label": "Typed Prescription", "mime": "text/plain"}, {"id": "04", "filename": "04_printed_discharge.txt", "label": "Laboratory Report", "mime": "text/plain"}, {"id": "06", "filename": "06_mixed_document.png", "label": "Scanned EHR", "mime": "image/png"}, {"id": "10", "filename": "10_fhir_bundle.json", "label": "FHIR Example", "mime": "application/json"}, {"id": "11", "filename": "11_payas_pathology_lab_report.pdf", "label": "Pathology Lab Report (Payas BSL)", "mime": "application/pdf"}, ] out = [] for s in samples: f_path = fixtures_dir / s["filename"] if f_path.exists(): out.append({ "id": s["id"], "filename": s["filename"], "label": s["label"], "mime": s["mime"], "size_bytes": f_path.stat().st_size, }) return out @app.get("/v1/fixtures/{filename}", tags=["Fixtures"]) def get_fixture_file(filename: str): """Retrieve raw bytes of a sample fixture file.""" from pathlib import Path from fastapi.responses import FileResponse repo_root = Path(__file__).resolve().parent.parent.parent fixtures_dir = (repo_root / "tests" / "fixtures").resolve() f_path = (fixtures_dir / filename).resolve() if fixtures_dir not in f_path.parents or not f_path.exists(): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Fixture not found.") mime_map = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".pdf": "application/pdf", ".txt": "text/plain", ".json": "application/json", } media_type = mime_map.get(f_path.suffix.lower(), "application/octet-stream") return FileResponse(path=f_path, media_type=media_type, filename=f_path.name)