|
|
| """
|
| FHIR R4 DiagnosticReport export for MediAgent.
|
| Converts a FinalReport into a standards-compliant HL7 FHIR R4 resource
|
| suitable for import into any EMR system (Epic, Cerner, etc.).
|
| """
|
|
|
| import base64
|
| import uuid
|
| from datetime import datetime
|
| from typing import Any, Dict, List
|
|
|
| from core.models import FinalReport
|
|
|
|
|
| def to_fhir_diagnostic_report(report: FinalReport) -> Dict[str, Any]:
|
| """
|
| Convert a MediAgent FinalReport into a FHIR R4 DiagnosticReport resource.
|
|
|
| Conforms to:
|
| http://hl7.org/fhir/R4/diagnosticreport.html
|
| LOINC 18748-4 (Diagnostic imaging study)
|
| """
|
| meta = report.patient_metadata or {}
|
| sections = report.sections
|
| ts = report.generation_timestamp.strftime("%Y-%m-%dT%H:%M:%SZ")
|
| patient_uid = f"patient-{uuid.uuid4().hex[:8]}"
|
|
|
|
|
| patient: Dict[str, Any] = {
|
| "resourceType": "Patient",
|
| "id": patient_uid,
|
| }
|
| if meta.get("sex"):
|
| patient["gender"] = _map_sex(str(meta["sex"]))
|
| age = meta.get("age")
|
| if age:
|
| birth_year = datetime.now().year - int(age)
|
| patient["birthDate"] = str(birth_year)
|
| if meta.get("patient_name"):
|
| patient["name"] = [{"text": str(meta["patient_name"])}]
|
|
|
|
|
| observations: List[Dict[str, Any]] = []
|
| finding_text = sections.findings or ""
|
| if finding_text:
|
| obs_id = f"obs-{uuid.uuid4().hex[:8]}"
|
| observations.append({
|
| "resourceType": "Observation",
|
| "id": obs_id,
|
| "status": "final",
|
| "category": [{
|
| "coding": [{
|
| "system": "http://terminology.hl7.org/CodeSystem/observation-category",
|
| "code": "imaging",
|
| "display": "Imaging"
|
| }]
|
| }],
|
| "code": {
|
| "coding": [{
|
| "system": "http://loinc.org",
|
| "code": "59776-5",
|
| "display": "Findings"
|
| }]
|
| },
|
| "subject": {"reference": f"#{patient_uid}"},
|
| "effectiveDateTime": ts,
|
| "valueString": finding_text[:2000]
|
| })
|
|
|
|
|
| contained: List[Dict] = [patient] + observations
|
| obs_refs = [{"reference": f"#{o['id']}"} for o in observations]
|
|
|
|
|
| severity_val = report.overall_severity.value if hasattr(report.overall_severity, "value") else str(report.overall_severity)
|
| conclusion_codes = _severity_to_snomed(severity_val)
|
|
|
|
|
| modality_code = _extract_modality_code(report.vision_summary)
|
|
|
|
|
| report_text = (
|
| f"CLINICAL HISTORY\n{sections.clinical_history}\n\n"
|
| f"TECHNIQUE\n{sections.technique}\n\n"
|
| f"FINDINGS\n{sections.findings}\n\n"
|
| f"IMPRESSION\n{sections.impression}\n\n"
|
| f"RECOMMENDATIONS\n{sections.recommendations}\n\n"
|
| f"DISCLAIMER\n{sections.disclaimer}"
|
| )
|
|
|
|
|
| pipeline_str = ", ".join(
|
| f"{k}:{v.value}" for k, v in report.agent_pipeline_status.items()
|
| )
|
|
|
| resource: Dict[str, Any] = {
|
| "resourceType": "DiagnosticReport",
|
| "id": report.report_id,
|
| "meta": {
|
| "profile": [
|
| "http://hl7.org/fhir/StructureDefinition/DiagnosticReport"
|
| ],
|
| "lastUpdated": ts,
|
| "tag": [{
|
| "system": "https://mediagent.ai/tags",
|
| "code": "ai-generated",
|
| "display": "AI Generated — Requires Radiologist Review"
|
| }]
|
| },
|
| "contained": contained,
|
| "status": "final",
|
| "category": [{
|
| "coding": [{
|
| "system": "http://terminology.hl7.org/CodeSystem/v2-0074",
|
| "code": "RAD",
|
| "display": "Radiology"
|
| }]
|
| }],
|
| "code": {
|
| "coding": [
|
| {
|
| "system": "http://loinc.org",
|
| "code": "18748-4",
|
| "display": "Diagnostic imaging study"
|
| },
|
| modality_code
|
| ],
|
| "text": "Medical Imaging Analysis — AI-Assisted Radiology Report"
|
| },
|
| "subject": {
|
| "reference": f"#{patient_uid}"
|
| },
|
| "effectiveDateTime": ts,
|
| "issued": ts,
|
| "performer": [{
|
| "display": "MediAgent AI System (AMD Instinct MI300X)"
|
| }],
|
| "result": obs_refs,
|
| "conclusion": sections.impression,
|
| "conclusionCode": conclusion_codes,
|
| "presentedForm": [{
|
| "contentType": "text/plain; charset=utf-8",
|
| "data": base64.b64encode(report_text.encode("utf-8")).decode("utf-8"),
|
| "title": f"Radiology Report {report.report_id}",
|
| "creation": ts
|
| }],
|
| "extension": [
|
| {
|
| "url": "https://mediagent.ai/fhir/StructureDefinition/ai-quality-score",
|
| "valueInteger": _extract_qa_score(sections.recommendations)
|
| },
|
| {
|
| "url": "https://mediagent.ai/fhir/StructureDefinition/overall-severity",
|
| "valueCode": severity_val
|
| },
|
| {
|
| "url": "https://mediagent.ai/fhir/StructureDefinition/pipeline-status",
|
| "valueString": pipeline_str
|
| },
|
| {
|
| "url": "https://mediagent.ai/fhir/StructureDefinition/inference-platform",
|
| "valueString": "AMD Instinct MI300X / ROCm / vLLM / Qwen"
|
| }
|
| ]
|
| }
|
|
|
| return resource
|
|
|
|
|
|
|
|
|
| def _map_sex(sex: str) -> str:
|
| return {"M": "male", "F": "female", "O": "other"}.get(sex.upper(), "unknown")
|
|
|
|
|
| def _severity_to_snomed(severity: str) -> List[Dict]:
|
| snomed = {
|
| "NORMAL": ("260313008", "Normal"),
|
| "INCIDENTAL": ("102483000", "Incidental finding"),
|
| "SIGNIFICANT": ("404684003", "Clinical finding"),
|
| "CRITICAL": ("399625000", "Critical finding"),
|
| }
|
| code, display = snomed.get(severity.upper(), ("404684003", "Clinical finding"))
|
| return [{
|
| "coding": [{
|
| "system": "http://snomed.info/sct",
|
| "code": code,
|
| "display": display
|
| }]
|
| }]
|
|
|
|
|
| def _extract_modality_code(vision_summary: str) -> Dict:
|
| """Map detected modality to DICOM/RadLex LOINC codes."""
|
| loinc_map = {
|
| "X-RAY": ("24627-2", "Chest X-ray"),
|
| "CT": ("18747-6", "CT study"),
|
| "MRI": ("18755-9", "MRI study"),
|
| }
|
| for key, (code, display) in loinc_map.items():
|
| if key in vision_summary.upper():
|
| return {"system": "http://loinc.org", "code": code, "display": display}
|
| return {"system": "http://loinc.org", "code": "18748-4", "display": "Diagnostic imaging study"}
|
|
|
|
|
| def _extract_qa_score(recommendations: str) -> int:
|
| import re
|
| m = re.search(r"Score[:\s]+(\d+)", recommendations or "")
|
| return int(m.group(1)) if m else 85
|
|
|
