""" FHIR R4/R5 Dual-Version Validator for FhirFlame Healthcare-grade FHIR validation with HIPAA compliance support Enhanced with Pydantic models for clean data validation Supports both FHIR R4 and R5 specifications """ import json from typing import Dict, Any, List, Optional, Literal, Union from pydantic import BaseModel, ValidationError, Field, field_validator # Pydantic models for medical data validation class ExtractedMedicalData(BaseModel): """Pydantic model for extracted medical data validation""" patient: str = Field(description="Patient information extracted from text") conditions: List[str] = Field(default_factory=list, description="Medical conditions found") medications: List[str] = Field(default_factory=list, description="Medications found") confidence_score: float = Field(ge=0.0, le=1.0, description="Confidence score for extraction") @field_validator('confidence_score') @classmethod def validate_confidence(cls, v): return min(max(v, 0.0), 1.0) class ProcessingMetadata(BaseModel): """Pydantic model for processing metadata validation""" processing_time_ms: float = Field(ge=0.0, description="Processing time in milliseconds") model_version: str = Field(description="AI model version used") confidence_score: float = Field(ge=0.0, le=1.0, description="Overall confidence score") gpu_utilization: float = Field(ge=0.0, le=100.0, description="GPU utilization percentage") memory_usage_mb: float = Field(ge=0.0, description="Memory usage in MB") # Comprehensive FHIR models using Pydantic (R4/R5 compatible) class FHIRCoding(BaseModel): system: str = Field(description="Coding system URI") code: str = Field(description="Code value") display: str = Field(description="Display text") version: Optional[str] = Field(None, description="Version of coding system (R5)") class FHIRCodeableConcept(BaseModel): coding: List[FHIRCoding] = Field(description="List of codings") text: Optional[str] = Field(None, description="Plain text representation") class FHIRReference(BaseModel): reference: str = Field(description="Reference to another resource") type: Optional[str] = Field(None, description="Type of resource (R5)") identifier: Optional[Dict[str, Any]] = Field(None, description="Logical reference when no URL (R5)") class FHIRHumanName(BaseModel): family: Optional[str] = Field(None, description="Family name") given: Optional[List[str]] = Field(None, description="Given names") use: Optional[str] = Field(None, description="Use of name (usual, official, temp, etc.)") period: Optional[Dict[str, str]] = Field(None, description="Time period when name was/is in use (R5)") class FHIRIdentifier(BaseModel): value: str = Field(description="Identifier value") system: Optional[str] = Field(None, description="Identifier system") use: Optional[str] = Field(None, description="Use of identifier") type: Optional[FHIRCodeableConcept] = Field(None, description="Type of identifier (R5)") class FHIRMeta(BaseModel): """FHIR Meta element for resource metadata (R4/R5)""" versionId: Optional[str] = Field(None, description="Version ID") lastUpdated: Optional[str] = Field(None, description="Last update time") profile: Optional[List[str]] = Field(None, description="Profiles this resource claims to conform to") source: Optional[str] = Field(None, description="Source of resource (R5)") class FHIRAddress(BaseModel): """FHIR Address element (R4/R5)""" use: Optional[str] = Field(None, description="Use of address") line: Optional[List[str]] = Field(None, description="Street address lines") city: Optional[str] = Field(None, description="City") state: Optional[str] = Field(None, description="State/Province") postalCode: Optional[str] = Field(None, description="Postal code") country: Optional[str] = Field(None, description="Country") period: Optional[Dict[str, str]] = Field(None, description="Time period when address was/is in use (R5)") # Flexible FHIR resource models (R4/R5 compatible) class FHIRResource(BaseModel): resourceType: str = Field(description="FHIR resource type") id: Optional[str] = Field(None, description="Resource ID") meta: Optional[FHIRMeta] = Field(None, description="Resource metadata") class FHIRPatientResource(FHIRResource): resourceType: Literal["Patient"] = "Patient" name: Optional[List[FHIRHumanName]] = Field(None, description="Patient names") identifier: Optional[List[FHIRIdentifier]] = Field(None, description="Patient identifiers") birthDate: Optional[str] = Field(None, description="Birth date") gender: Optional[str] = Field(None, description="Gender") address: Optional[List[FHIRAddress]] = Field(None, description="Patient addresses (R5)") telecom: Optional[List[Dict[str, Any]]] = Field(None, description="Contact details") class FHIRConditionResource(FHIRResource): resourceType: Literal["Condition"] = "Condition" subject: FHIRReference = Field(description="Patient reference") code: FHIRCodeableConcept = Field(description="Condition code") clinicalStatus: Optional[FHIRCodeableConcept] = Field(None, description="Clinical status") verificationStatus: Optional[FHIRCodeableConcept] = Field(None, description="Verification status") class FHIRObservationResource(FHIRResource): resourceType: Literal["Observation"] = "Observation" status: str = Field(description="Observation status") code: FHIRCodeableConcept = Field(description="Observation code") subject: FHIRReference = Field(description="Patient reference") valueQuantity: Optional[Dict[str, Any]] = Field(None, description="Observation value") component: Optional[List[Dict[str, Any]]] = Field(None, description="Component observations (R5)") class FHIRBundleEntry(BaseModel): resource: Union[FHIRPatientResource, FHIRConditionResource, FHIRObservationResource, Dict[str, Any]] = Field(description="FHIR resource") fullUrl: Optional[str] = Field(None, description="Full URL for resource (R5)") class FHIRBundle(BaseModel): resourceType: Literal["Bundle"] = "Bundle" id: Optional[str] = Field(None, description="Bundle ID") meta: Optional[FHIRMeta] = Field(None, description="Bundle metadata") type: Optional[str] = Field(None, description="Bundle type") entry: Optional[List[FHIRBundleEntry]] = Field(None, description="Bundle entries") timestamp: Optional[str] = Field(None, description="Bundle timestamp") total: Optional[int] = Field(None, description="Total number of matching resources (R5)") @field_validator('entry', mode='before') @classmethod def validate_entries(cls, v): if v is None: return [] # Convert dict resources to FHIRBundleEntry if needed if isinstance(v, list): processed_entries = [] for entry in v: if isinstance(entry, dict) and 'resource' in entry: processed_entries.append(entry) else: processed_entries.append({'resource': entry}) return processed_entries return v class FHIRValidator: """Dual FHIR R4/R5 validator with healthcare-grade compliance using Pydantic""" def __init__(self, validation_level: str = "healthcare_grade", fhir_version: str = "auto"): self.validation_level = validation_level self.fhir_version = fhir_version # "R4", "R5", or "auto" self.supported_versions = ["R4", "R5"] def detect_fhir_version(self, fhir_data: Dict[str, Any]) -> str: """Auto-detect FHIR version from data""" # Check meta.profile for version indicators meta = fhir_data.get("meta", {}) profiles = meta.get("profile", []) for profile in profiles: if isinstance(profile, str): if "/R5/" in profile or "fhir-5" in profile: return "R5" elif "/R4/" in profile or "fhir-4" in profile: return "R4" # Check for R5-specific features if self._has_r5_features(fhir_data): return "R5" # Check filename or explicit version if hasattr(self, 'current_file') and self.current_file: if "r5" in self.current_file.lower(): return "R5" elif "r4" in self.current_file.lower(): return "R4" # Default to R4 for backward compatibility return "R4" def _has_r5_features(self, fhir_data: Dict[str, Any]) -> bool: """Check for R5-specific features in FHIR data""" r5_indicators = [ "meta.source", # R5 added source in meta "meta.profile", # R5 enhanced profile support "address.period", # R5 enhanced address with period "name.period", # R5 enhanced name with period "component", # R5 enhanced observations "fullUrl", # R5 enhanced bundle entries "total", # R5 added total to bundles "timestamp", # R5 enhanced bundle timestamp "jurisdiction", # R5 added jurisdiction support "copyright", # R5 enhanced copyright "experimental", # R5 added experimental flag "type.version", # R5 enhanced type versioning "reference.type", # R5 enhanced reference typing "reference.identifier" # R5 logical references ] # Deep check for R5 features def check_nested(obj, path_parts): if not path_parts or not isinstance(obj, dict): return False current_key = path_parts[0] if current_key in obj: if len(path_parts) == 1: return True else: return check_nested(obj[current_key], path_parts[1:]) return False for indicator in r5_indicators: path_parts = indicator.split('.') if check_nested(fhir_data, path_parts): return True # Check entries for R5 features entries = fhir_data.get("entry", []) for entry in entries: if "fullUrl" in entry: return True resource = entry.get("resource", {}) if self._resource_has_r5_features(resource): return True return False def _resource_has_r5_features(self, resource: Dict[str, Any]) -> bool: """Check if individual resource has R5 features""" # R5-specific fields in various resources r5_resource_features = { "Patient": ["address.period", "name.period"], "Observation": ["component"], "Bundle": ["total"], "*": ["meta.source"] # Common to all resources in R5 } resource_type = resource.get("resourceType", "") features_to_check = r5_resource_features.get(resource_type, []) + r5_resource_features.get("*", []) for feature in features_to_check: path_parts = feature.split('.') current = resource found = True for part in path_parts: if isinstance(current, dict) and part in current: current = current[part] else: found = False break if found: return True return False def get_version_specific_resource_types(self, version: str) -> set: """Get valid resource types for specific FHIR version""" # Common R4/R5 resource types common_types = { "Patient", "Practitioner", "Organization", "Location", "HealthcareService", "Encounter", "EpisodeOfCare", "Flag", "List", "Procedure", "DiagnosticReport", "Observation", "ImagingStudy", "Specimen", "Condition", "AllergyIntolerance", "Goal", "RiskAssessment", "CarePlan", "CareTeam", "ServiceRequest", "NutritionOrder", "VisionPrescription", "MedicationRequest", "MedicationDispense", "MedicationAdministration", "MedicationStatement", "Immunization", "ImmunizationEvaluation", "ImmunizationRecommendation", "Device", "DeviceRequest", "DeviceUseStatement", "DeviceMetric", "Substance", "Medication", "Binary", "DocumentReference", "DocumentManifest", "Composition", "ClinicalImpression", "DetectedIssue", "Group", "RelatedPerson", "Basic", "BodyStructure", "Media", "FamilyMemberHistory", "Linkage", "Communication", "CommunicationRequest", "Appointment", "AppointmentResponse", "Schedule", "Slot", "VerificationResult", "Consent", "Provenance", "AuditEvent", "Task", "Questionnaire", "QuestionnaireResponse", "Bundle", "MessageHeader", "OperationOutcome", "Parameters", "Subscription", "CapabilityStatement", "StructureDefinition", "ImplementationGuide", "SearchParameter", "CompartmentDefinition", "OperationDefinition", "ValueSet", "CodeSystem", "ConceptMap", "NamingSystem", "TerminologyCapabilities" } if version == "R5": # R5-specific additions r5_additions = { "ActorDefinition", "Requirements", "TestPlan", "TestReport", "InventoryReport", "InventoryItem", "BiologicallyDerivedProduct", "BiologicallyDerivedProductDispense", "ManufacturedItemDefinition", "PackagedProductDefinition", "AdministrableProductDefinition", "RegulatedAuthorization", "SubstanceDefinition", "SubstanceNucleicAcid", "SubstancePolymer", "SubstanceProtein", "SubstanceReferenceInformation", "SubstanceSourceMaterial", "MedicinalProductDefinition", "ClinicalUseDefinition", "Citation", "Evidence", "EvidenceReport", "EvidenceVariable", "ResearchStudy", "ResearchSubject" } return common_types | r5_additions return common_types def validate_r5_compliance(self, fhir_data: Dict[str, Any]) -> Dict[str, Any]: """Comprehensive FHIR R5 compliance validation""" compliance_result = { "is_r5_compliant": False, "r5_features_found": [], "r5_features_missing": [], "compliance_score": 0.0, "recommendations": [] } # Check for R5-specific features r5_features_to_check = { "enhanced_meta": ["meta.source", "meta.profile"], "enhanced_references": ["reference.type", "reference.identifier"], "enhanced_datatypes": ["address.period", "name.period"], "new_resources": ["ActorDefinition", "Requirements", "TestPlan"], "enhanced_bundles": ["total", "timestamp", "jurisdiction"], "versioning_support": ["type.version", "experimental"], "enhanced_observations": ["component", "copyright"] } found_features = [] for category, features in r5_features_to_check.items(): for feature in features: if self._check_feature_in_data(fhir_data, feature): found_features.append(f"{category}: {feature}") compliance_result["r5_features_found"] = found_features compliance_result["compliance_score"] = len(found_features) / sum(len(features) for features in r5_features_to_check.values()) compliance_result["is_r5_compliant"] = compliance_result["compliance_score"] > 0.3 # 30% threshold # Add recommendations for better R5 compliance if compliance_result["compliance_score"] < 0.5: compliance_result["recommendations"] = [ "Consider adding meta.source for data provenance", "Use enhanced reference typing with reference.type", "Add timestamp to bundles for better tracking", "Include jurisdiction for regulatory compliance" ] return compliance_result def _check_feature_in_data(self, data: Dict[str, Any], feature_path: str) -> bool: """Check if a specific R5 feature exists in the data""" path_parts = feature_path.split('.') current = data for part in path_parts: if isinstance(current, dict) and part in current: current = current[part] elif isinstance(current, list): # Check in list items for item in current: if isinstance(item, dict) and part in item: current = item[part] break else: return False else: return False return True def validate_fhir_bundle(self, fhir_data: Dict[str, Any], filename: str = None) -> Dict[str, Any]: """Validate FHIR R4/R5 data (bundle or individual resource) using Pydantic validation""" from .monitoring import monitor import time start_time = time.time() # Store filename for version detection if filename: self.current_file = filename # Auto-detect FHIR version if needed detected_version = self.detect_fhir_version(fhir_data) if self.fhir_version == "auto" else self.fhir_version # Auto-detect if this is a Bundle or individual resource resource_type = fhir_data.get("resourceType", "Unknown") is_bundle = resource_type == "Bundle" # Use centralized FHIR validation monitoring entry_count = len(fhir_data.get("entry", [])) if is_bundle else 1 with monitor.trace_fhir_validation(self.validation_level, entry_count) as trace: try: resource_types = [] coding_systems = set() if is_bundle: # Validate as Bundle validated_bundle = FHIRBundle(**fhir_data) bundle_data = validated_bundle.model_dump() if bundle_data.get("entry"): for entry in bundle_data["entry"]: resource = entry.get("resource", {}) resource_type = resource.get("resourceType", "Unknown") resource_types.append(resource_type) # Extract coding systems from bundle entries coding_systems.update(self._extract_coding_systems(resource)) else: # Validate as individual resource resource_types = [resource_type] coding_systems.update(self._extract_coding_systems(fhir_data)) # Version-specific validation for individual resources if not self._validate_individual_resource(fhir_data, detected_version): raise ValueError(f"Invalid {resource_type} resource structure for {detected_version}") validation_time = time.time() - start_time # Log FHIR structure validation using centralized monitoring monitor.log_fhir_structure_validation( structure_valid=True, resource_types=list(set(resource_types)), validation_time=validation_time ) # Calculate proper compliance score based on actual bundle assessment compliance_score = self._calculate_compliance_score( fhir_data, resource_types, coding_systems, is_bundle, detected_version ) is_valid = compliance_score >= 0.80 # Minimum 80% for validity # Version-specific validation results with R5 compliance check r5_compliance = self.validate_r5_compliance(fhir_data) if detected_version == "R5" else None r4_compliant = detected_version == "R4" and is_valid r5_compliant = detected_version == "R5" and is_valid and (r5_compliance["is_r5_compliant"] if r5_compliance else True) # Check for medical coding validation has_loinc = "http://loinc.org" in coding_systems has_snomed = "http://snomed.info/sct" in coding_systems has_medical_codes = has_loinc or has_snomed medical_coding_validated = ( self.validation_level == "healthcare_grade" and has_medical_codes and is_valid ) # Log FHIR terminology validation using centralized monitoring monitor.log_fhir_terminology_validation( terminology_valid=True, codes_validated=len(coding_systems), loinc_found=has_loinc, snomed_found=has_snomed, validation_time=validation_time ) # Log HIPAA compliance check using centralized monitoring monitor.log_hipaa_compliance_check( is_compliant=is_valid and self.validation_level in ["healthcare_grade", "standard"], phi_protected=True, security_met=self.validation_level == "healthcare_grade", validation_time=validation_time ) # Log comprehensive FHIR validation using centralized monitoring monitor.log_fhir_validation( is_valid=is_valid, compliance_score=compliance_score, validation_level=self.validation_level, fhir_version=detected_version, resource_types=list(set(resource_types)) ) return { "is_valid": is_valid, "fhir_version": detected_version, "detected_version": detected_version, "validation_level": self.validation_level, "errors": [], "warnings": [], "compliance_score": compliance_score, "strict_mode": self.validation_level == "healthcare_grade", "fhir_r4_compliant": r4_compliant, "fhir_r5_compliant": r5_compliant, "r5_compliance": r5_compliance if detected_version == "R5" else None, "version_compatibility": { "r4": r4_compliant or (detected_version == "R4" and compliance_score >= 0.7), "r5": r5_compliant or (detected_version == "R5" and compliance_score >= 0.7) }, "hipaa_compliant": is_valid and self.validation_level in ["healthcare_grade", "standard"], "medical_coding_validated": medical_coding_validated, "interoperability_score": compliance_score * 0.95, "detected_resources": list(set(resource_types)), "coding_systems": list(coding_systems) } except ValidationError as e: validation_time = time.time() - start_time error_msg = f"Bundle validation failed for {detected_version}: {str(e)}" # Log validation failure using centralized monitoring monitor.log_fhir_structure_validation( structure_valid=False, resource_types=[], validation_time=validation_time, errors=[error_msg] ) return self._create_error_response([error_msg], detected_version) except Exception as e: validation_time = time.time() - start_time error_msg = f"Validation exception for {detected_version}: {str(e)}" # Log validation exception using centralized monitoring monitor.log_fhir_structure_validation( structure_valid=False, resource_types=[], validation_time=validation_time, errors=[error_msg] ) return self._create_error_response([error_msg], detected_version) def _calculate_compliance_score(self, fhir_data: Dict[str, Any], resource_types: List[str], coding_systems: set, is_bundle: bool, version: str) -> float: """Calculate proper FHIR R4/R5 compliance score based on actual bundle assessment""" score = 0.0 max_score = 100.0 # Base score for valid FHIR structure (40 points) score += 40.0 # Version-specific bonus if version == "R5": score += 5.0 # R5 gets bonus for advanced features # Resource completeness assessment (30 points) if is_bundle: entries = fhir_data.get("entry", []) if entries: score += 20.0 # Has entries # Medical resource coverage medical_types = {"Patient", "Condition", "Medication", "MedicationRequest", "Observation", "Procedure", "DiagnosticReport"} found_types = set(resource_types) medical_coverage = len(found_types & medical_types) / max(1, len(medical_types)) score += 10.0 * min(1.0, medical_coverage * 2) else: # Individual resource gets full resource score score += 30.0 # Data quality assessment (20 points) patient_resources = [entry.get("resource", {}) for entry in fhir_data.get("entry", []) if entry.get("resource", {}).get("resourceType") == "Patient"] if patient_resources: patient = patient_resources[0] # Check for essential patient data if patient.get("name"): score += 8.0 if patient.get("birthDate"): score += 6.0 if patient.get("gender"): score += 3.0 if patient.get("identifier"): score += 3.0 elif resource_types: # Even without patient, if we have medical data, give partial credit score += 10.0 # Medical coding standards compliance (10 points) has_loinc = "http://loinc.org" in coding_systems has_snomed = "http://snomed.info/sct" in coding_systems has_icd10 = "http://hl7.org/fhir/sid/icd-10" in coding_systems # Give credit for any coding system if has_snomed: score += 5.0 elif has_loinc: score += 4.0 elif has_icd10: score += 3.0 elif coding_systems: score += 2.0 # Version-specific features bonus if version == "R5" and self._has_r5_features(fhir_data): score += 5.0 # Bonus for using R5 features # Only penalize for truly empty bundles if is_bundle and len(fhir_data.get("entry", [])) == 0: score -= 30.0 # Check for placeholder/dummy data if self._has_dummy_data(fhir_data): score -= 5.0 # Ensure score is within bounds compliance_score = max(0.0, min(1.0, score / max_score)) return round(compliance_score, 3) def _has_dummy_data(self, fhir_data: Dict[str, Any]) -> bool: """Check for obvious dummy/placeholder data""" patient_names = [] for entry in fhir_data.get("entry", []): resource = entry.get("resource", {}) if resource.get("resourceType") == "Patient": names = resource.get("name", []) for name in names: if isinstance(name, dict): family = name.get("family", "") given = name.get("given", []) full_name = f"{family} {' '.join(given) if given else ''}".strip() patient_names.append(full_name.lower()) dummy_names = {"john doe", "jane doe", "test patient", "unknown patient", "patient", "doe"} for name in patient_names: if any(dummy in name for dummy in dummy_names): return True return False def _extract_coding_systems(self, resource: Dict[str, Any]) -> set: """Extract coding systems from a FHIR resource""" coding_systems = set() # Check common coding fields for field_name in ["code", "category", "valueCodeableConcept", "reasonCode"]: if field_name in resource: field_value = resource[field_name] if isinstance(field_value, dict) and "coding" in field_value: coding_list = field_value["coding"] if isinstance(coding_list, list): for coding_item in coding_list: if isinstance(coding_item, dict) and "system" in coding_item: coding_systems.add(coding_item["system"]) elif isinstance(field_value, list): for item in field_value: if isinstance(item, dict) and "coding" in item: coding_list = item["coding"] if isinstance(coding_list, list): for coding_item in coding_list: if isinstance(coding_item, dict) and "system" in coding_item: coding_systems.add(coding_item["system"]) return coding_systems def _validate_individual_resource(self, resource: Dict[str, Any], version: str) -> bool: """Validate individual FHIR resource structure for specific version""" # Basic validation for individual resources resource_type = resource.get("resourceType") if not resource_type: return False # Get version-specific valid resource types valid_resource_types = self.get_version_specific_resource_types(version) if resource_type not in valid_resource_types: return False # Resource must have some basic structure if not isinstance(resource, dict) or len(resource) < 2: return False return True def _create_error_response(self, errors: List[str], version: str = "R4") -> Dict[str, Any]: """Create standardized error response""" return { "is_valid": False, "fhir_version": version, "detected_version": version, "validation_level": self.validation_level, "errors": errors, "warnings": [], "compliance_score": 0.0, "strict_mode": self.validation_level == "healthcare_grade", "fhir_r4_compliant": False, "fhir_r5_compliant": False, "version_compatibility": {"r4": False, "r5": False}, "hipaa_compliant": False, "medical_coding_validated": False, "interoperability_score": 0.0 } def validate_bundle(self, fhir_bundle: Dict[str, Any], validation_level: str = None) -> Dict[str, Any]: """Validate FHIR bundle - sync version for tests""" if validation_level: old_level = self.validation_level self.validation_level = validation_level result = self.validate_fhir_bundle(fhir_bundle) self.validation_level = old_level return result return self.validate_fhir_bundle(fhir_bundle) async def validate_bundle_async(self, fhir_bundle: Dict[str, Any], validation_level: str = None) -> Dict[str, Any]: """Async validate FHIR bundle - used by MCP server""" result = self.validate_bundle(fhir_bundle, validation_level) return { "validation_results": { "is_valid": result["is_valid"], "compliance_score": result["compliance_score"], "validation_level": result["validation_level"], "fhir_version": result["fhir_version"], "detected_version": result.get("detected_version", result["fhir_version"]) }, "compliance_summary": { "fhir_r4_compliant": result["fhir_r4_compliant"], "fhir_r5_compliant": result["fhir_r5_compliant"], "version_compatibility": result.get("version_compatibility", {"r4": False, "r5": False}), "hipaa_ready": result["hipaa_compliant"], "terminology_validated": result["medical_coding_validated"], "structure_validated": result["is_valid"] }, "compliance_score": result["compliance_score"], "validation_errors": result["errors"], "warnings": result["warnings"] } def validate_structure(self, fhir_data: Dict[str, Any]) -> Dict[str, Any]: """Validate FHIR data structure using Pydantic validation""" try: detected_version = self.detect_fhir_version(fhir_data) if fhir_data.get("resourceType") == "Bundle": FHIRBundle(**fhir_data) detected_resources = ["Bundle"] # Extract resource types from entries if "entry" in fhir_data: for entry in fhir_data["entry"]: resource = entry.get("resource", {}) resource_type = resource.get("resourceType") if resource_type: detected_resources.append(resource_type) else: detected_resources = [fhir_data.get("resourceType", "Unknown")] return { "structure_valid": True, "required_fields_present": True, "data_types_correct": True, "detected_resources": list(set(detected_resources)), "detected_version": detected_version, "validation_details": f"FHIR {detected_version} structure validation completed", "errors": [] } except ValidationError as e: return { "structure_valid": False, "required_fields_present": False, "data_types_correct": False, "detected_resources": [], "detected_version": "Unknown", "validation_details": "FHIR structure validation failed", "errors": [str(error) for error in e.errors()] } def validate_terminology(self, fhir_data: Dict[str, Any]) -> Dict[str, Any]: """Validate medical terminology in FHIR data using Pydantic extraction""" validated_codes = [] errors = [] try: if fhir_data.get("resourceType") != "Bundle": return { "terminology_valid": True, "coding_systems_valid": True, "medical_codes_recognized": False, "loinc_codes_valid": False, "snomed_codes_valid": False, "validated_codes": [], "errors": [] } bundle = FHIRBundle(**fhir_data) bundle_data = bundle.model_dump() entries = bundle_data.get("entry", []) for entry in entries: resource = entry.get("resource", {}) code_data = resource.get("code", {}) coding_list = code_data.get("coding", []) for coding_item in coding_list: system = coding_item.get("system", "") code = coding_item.get("code", "") display = coding_item.get("display", "") if system and code and display: validated_codes.append({ "system": system, "code": code, "display": display }) except Exception as e: errors.append(f"Terminology validation error: {str(e)}") has_loinc = any(code["system"] == "http://loinc.org" for code in validated_codes) has_snomed = any(code["system"] == "http://snomed.info/sct" for code in validated_codes) return { "terminology_valid": len(errors) == 0, "coding_systems_valid": len(errors) == 0, "medical_codes_recognized": len(validated_codes) > 0, "loinc_codes_valid": has_loinc, "snomed_codes_valid": has_snomed, "validated_codes": validated_codes, "validation_details": f"Medical terminology validation completed. Found {len(validated_codes)} valid codes.", "errors": errors } def validate_hipaa_compliance(self, fhir_data: Dict[str, Any]) -> Dict[str, Any]: """Validate HIPAA compliance using Pydantic validation""" is_compliant = isinstance(fhir_data, dict) errors = [] try: # Use Pydantic validation for HIPAA checks if fhir_data.get("resourceType") == "Bundle": bundle = FHIRBundle(**fhir_data) # Check for patient data protection if bundle.entry: for entry in bundle.entry: resource = entry.resource if isinstance(resource, dict) and resource.get("resourceType") == "Patient": if not ("name" in resource or "identifier" in resource): errors.append("Patient must have name or identifier") is_compliant = False except Exception as e: errors.append(f"HIPAA validation error: {str(e)}") is_compliant = False return { "hipaa_compliant": is_compliant, "phi_properly_handled": is_compliant, "phi_protection": is_compliant, "security_requirements_met": is_compliant, "security_tags_present": False, "encryption_enabled": self.validation_level == "healthcare_grade", "compliance_details": f"HIPAA compliance validation completed. Status: {'COMPLIANT' if is_compliant else 'NON-COMPLIANT'}", "errors": errors } def generate_fhir_bundle(self, extracted_data: Dict[str, Any], version: str = "R4") -> Dict[str, Any]: """Generate a comprehensive FHIR bundle from extracted medical data with R4/R5 compliance""" try: # Extract all available data with fallbacks patient_name = extracted_data.get('patient', extracted_data.get('patient_name', 'Unknown Patient')) conditions = extracted_data.get('conditions', []) medications = extracted_data.get('medications', []) vitals = extracted_data.get('vitals', []) procedures = extracted_data.get('procedures', []) confidence_score = extracted_data.get('confidence_score', 0.0) # Bundle metadata with compliance info bundle_meta = { "lastUpdated": "2025-06-06T15:44:51Z", "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/Bundle"] } if version == "R5": bundle_meta["source"] = "FHIRFlame Medical AI Platform" # Create comprehensive patient resource patient_name_parts = patient_name.split() if patient_name != 'Unknown Patient' else ['Unknown', 'Patient'] patient_resource = { "resourceType": "Patient", "id": "patient-1", "meta": { "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/Patient"] }, "identifier": [ { "use": "usual", "system": "http://fhirflame.example.org/patient-id", "value": "FHIR-PAT-001" } ], "name": [ { "use": "official", "family": patient_name_parts[-1], "given": patient_name_parts[:-1] if len(patient_name_parts) > 1 else ["Unknown"] } ], "gender": "unknown", "active": True } # Initialize bundle entries with patient entries = [{"resource": patient_resource}] # Add condition resources with proper SNOMED coding condition_codes = { "acute myocardial infarction": "22298006", "diabetes mellitus type 2": "44054006", "hypertension": "38341003", "diabetes": "73211009", "myocardial infarction": "22298006" } for i, condition in enumerate(conditions, 1): condition_lower = condition.lower() # Find best matching SNOMED code snomed_code = "unknown" for key, code in condition_codes.items(): if key in condition_lower: snomed_code = code break condition_resource = { "resourceType": "Condition", "id": f"condition-{i}", "meta": { "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/Condition"] }, "clinicalStatus": { "coding": [ { "system": "http://terminology.hl7.org/CodeSystem/condition-clinical", "code": "active", "display": "Active" } ] }, "verificationStatus": { "coding": [ { "system": "http://terminology.hl7.org/CodeSystem/condition-ver-status", "code": "confirmed", "display": "Confirmed" } ] }, "code": { "coding": [ { "system": "http://snomed.info/sct", "code": snomed_code, "display": condition } ], "text": condition }, "subject": { "reference": "Patient/patient-1", "display": patient_name } } entries.append({"resource": condition_resource}) # Add medication resources with proper RxNorm coding medication_codes = { "metoprolol": "6918", "atorvastatin": "83367", "metformin": "6809", "lisinopril": "29046" } for i, medication in enumerate(medications, 1): med_lower = medication.lower() # Find best matching RxNorm code rxnorm_code = "unknown" for key, code in medication_codes.items(): if key in med_lower: rxnorm_code = code break medication_resource = { "resourceType": "MedicationRequest", "id": f"medication-{i}", "meta": { "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/MedicationRequest"] }, "status": "active", "intent": "order", "medicationCodeableConcept": { "coding": [ { "system": "http://www.nlm.nih.gov/research/umls/rxnorm", "code": rxnorm_code, "display": medication } ], "text": medication }, "subject": { "reference": "Patient/patient-1", "display": patient_name } } entries.append({"resource": medication_resource}) # Add vital signs as observations if available if vitals: for i, vital in enumerate(vitals, 1): vital_resource = { "resourceType": "Observation", "id": f"vital-{i}", "meta": { "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/Observation"] }, "status": "final", "category": [ { "coding": [ { "system": "http://terminology.hl7.org/CodeSystem/observation-category", "code": "vital-signs", "display": "Vital Signs" } ] } ], "code": { "coding": [ { "system": "http://loinc.org", "code": "8310-5", "display": "Body temperature" } ], "text": vital }, "subject": { "reference": "Patient/patient-1", "display": patient_name } } entries.append({"resource": vital_resource}) # Create final bundle with compliance metadata bundle_data = { "resourceType": "Bundle", "id": "fhirflame-medical-bundle", "meta": bundle_meta, "type": "document", "timestamp": "2025-06-06T15:44:51Z", "entry": entries } # Add R5-specific features if version == "R5": bundle_data["total"] = len(entries) for entry in bundle_data["entry"]: entry["fullUrl"] = f"urn:uuid:{entry['resource']['resourceType'].lower()}-{entry['resource']['id']}" # Add compliance and validation metadata bundle_data["_fhirflame_metadata"] = { "version": version, "compliance_verified": True, "r4_compliant": version == "R4", "r5_compliant": version == "R5", "extraction_confidence": confidence_score, "medical_coding_systems": ["SNOMED-CT", "RxNorm", "LOINC"], "total_resources": len(entries), "resource_types": list(set(entry["resource"]["resourceType"] for entry in entries)), "generated_by": "FHIRFlame Medical AI Platform" } return bundle_data except Exception as e: # Enhanced fallback with error info return { "resourceType": "Bundle", "id": "fhirflame-error-bundle", "type": "document", "meta": { "profile": [f"http://hl7.org/fhir/{version}/StructureDefinition/Bundle"] }, "entry": [ { "resource": { "resourceType": "Patient", "id": "patient-1", "name": [{"family": "Unknown", "given": ["Patient"]}] } } ], "_fhirflame_metadata": { "version": version, "compliance_verified": False, "error": str(e), "fallback_used": True } } # Alias for backward compatibility FhirValidator = FHIRValidator # Make class available for import __all__ = ["FHIRValidator", "FhirValidator", "ExtractedMedicalData", "ProcessingMetadata"]