"""FHIR R4 resource generator for medication data.""" import json import logging from datetime import datetime from typing import Any from uuid import uuid4 from fhir.resources.medicationrequest import MedicationRequest from fhir.resources.medicationstatement import MedicationStatement from fhir.resources.codeableconcept import CodeableConcept from fhir.resources.dosage import Dosage from fhir.resources.reference import Reference logger = logging.getLogger(__name__) class FhirGenerator: """Generator for FHIR R4 medication resources.""" def generate_medication_statement( self, medication_data: dict[str, Any] ) -> str: """ Generate a FHIR MedicationStatement resource. Args: medication_data: Dictionary with medication info - drug_name: Name of the medication - dosage: Dosage amount and unit - frequency: How often to take - route: Route of administration (optional) - notes: Additional instructions (optional) Returns: JSON string of FHIR MedicationStatement """ try: # Ensure drug_name is non-empty drug_name = medication_data.get("drug_name", "").strip() if not drug_name: drug_name = "Medication (see prescription)" # Build dosage text dosage_text = self._format_dosage_text(medication_data) if not dosage_text or dosage_text == "As directed": dosage_text = "See prescription for details" # Build dosage with optional route route_text = medication_data.get("route", "").strip() dosage_kwargs = {"text": dosage_text} if route_text: dosage_kwargs["route"] = CodeableConcept(text=route_text) statement = MedicationStatement( id=str(uuid4()), status="active", medicationCodeableConcept=CodeableConcept(text=drug_name), subject=Reference(reference="Patient/example"), effectiveDateTime=datetime.now().isoformat(), dosage=[Dosage(**dosage_kwargs)], ) return statement.json(indent=2) except Exception as e: logger.error(f"Failed to generate MedicationStatement: {e}") raise ValueError("Could not generate FHIR MedicationStatement.") def generate_medication_request( self, medication_data: dict[str, Any] ) -> str: """ Generate a FHIR MedicationRequest resource. Args: medication_data: Dictionary with medication info - drug_name: Name of the medication - dosage: Dosage amount and unit - frequency: How often to take - duration: How long to take - route: Route of administration (optional) - notes: Additional instructions (optional) Returns: JSON string of FHIR MedicationRequest """ try: # Ensure drug_name is non-empty drug_name = medication_data.get("drug_name", "").strip() if not drug_name: drug_name = "Medication (see prescription)" # Build dosage text dosage_text = self._format_dosage_text(medication_data) if not dosage_text or dosage_text == "As directed": dosage_text = "See prescription for details" # Build dosage with optional route route_text = medication_data.get("route", "").strip() dosage_kwargs = {"text": dosage_text} if route_text: dosage_kwargs["route"] = CodeableConcept(text=route_text) request = MedicationRequest( id=str(uuid4()), status="active", intent="order", medicationCodeableConcept=CodeableConcept(text=drug_name), subject=Reference(reference="Patient/example"), authoredOn=datetime.now().isoformat(), dosageInstruction=[Dosage(**dosage_kwargs)], ) return request.json(indent=2) except Exception as e: logger.error(f"Failed to generate MedicationRequest: {e}") raise ValueError("Could not generate FHIR MedicationRequest.") def _format_dosage_text(self, medication_data: dict[str, Any]) -> str: """Format dosage information as human-readable text.""" parts = [] if dosage := medication_data.get("dosage"): parts.append(dosage) if frequency := medication_data.get("frequency"): parts.append(frequency) if duration := medication_data.get("duration"): parts.append(f"for {duration}") if notes := medication_data.get("notes"): parts.append(f"({notes})") return " ".join(parts) if parts else "As directed" def parse_medications_to_dict(extraction_text: str) -> list[dict[str, Any]]: """ Parse extracted medication text into structured dictionaries. Args: extraction_text: Raw text from MedGemma extraction Returns: List of medication dictionaries """ # Simple parsing - in production, would use more robust NLP medications = [] lines = extraction_text.strip().split("\n") current_med = {} for line in lines: line = line.strip() if not line: if current_med: medications.append(current_med) current_med = {} continue line_lower = line.lower() if "drug" in line_lower or "medication" in line_lower or "name:" in line_lower: if current_med: medications.append(current_med) current_med = {"drug_name": line.split(":", 1)[-1].strip()} elif "dosage" in line_lower or "dose:" in line_lower: current_med["dosage"] = line.split(":", 1)[-1].strip() elif "frequency" in line_lower or "times" in line_lower: current_med["frequency"] = line.split(":", 1)[-1].strip() elif "duration" in line_lower or "days" in line_lower: current_med["duration"] = line.split(":", 1)[-1].strip() elif "route" in line_lower: current_med["route"] = line.split(":", 1)[-1].strip() elif "instruction" in line_lower or "note" in line_lower: current_med["notes"] = line.split(":", 1)[-1].strip() if current_med: medications.append(current_med) return medications if medications else [{"drug_name": "See prescription details", "notes": extraction_text}]