FHIR Integration

from rubric import Rubric
from rubric.integrations import FHIRClient

client = Rubric()

# Configure FHIR server connection
fhir = client.integrations.fhir.configure(
    name="hospital_fhir",

    # FHIR server endpoint
    base_url="https://fhir.hospital.org/r4",

    # Authentication
    auth_type="smart",  # or "basic", "bearer", "oauth2"
    client_id="your-client-id",
    client_secret="your-client-secret",

    # Token endpoint (for OAuth)
    token_endpoint="https://fhir.hospital.org/auth/token",

    # FHIR version
    fhir_version="R4",

    # Capabilities
    supports_batch=True,
    supports_search=True
)

print(f"FHIR connection configured: {fhir.id}")

# Test connection
capabilities = fhir.capability_statement()
print(f"Server: {capabilities.software.name}")
print(f"FHIR Version: {capabilities.fhirVersion}")

from rubric import Rubric

client = Rubric()

# Get FHIR client
fhir = client.integrations.fhir.get("hospital_fhir")

# Fetch a patient
patient = fhir.read("Patient", "12345")
print(f"Patient: {patient.name[0].given[0]} {patient.name[0].family}")

# Search for observations
observations = fhir.search(
    "Observation",
    params={
        "patient": "12345",
        "category": "vital-signs",
        "_sort": "-date",
        "_count": 10
    }
)

for obs in observations.entry:
    resource = obs.resource
    print(f"{resource.code.text}: {resource.valueQuantity.value} {resource.valueQuantity.unit}")

# Fetch patient's recent encounters
encounters = fhir.search(
    "Encounter",
    params={
        "patient": "12345",
        "date": "ge2024-01-01",
        "_include": "Encounter:diagnosis"
    }
)

from rubric import Rubric

client = Rubric()

fhir = client.integrations.fhir.get("hospital_fhir")

# Initiate bulk export
export_job = fhir.bulk_export(
    # Export type
    export_type="patient",  # or "group", "system"

    # Resource types to export
    types=[
        "Patient",
        "Encounter",
        "Observation",
        "Condition",
        "MedicationRequest"
    ],

    # Time filter
    since="2024-01-01T00:00:00Z",

    # Output format
    output_format="ndjson"
)

print(f"Bulk export started: {export_job.id}")

# Monitor progress
while export_job.status != "completed":
    export_job = fhir.bulk_export_status(export_job.id)
    print(f"Status: {export_job.status}")
    time.sleep(30)

# Download results
for output_file in export_job.output:
    print(f"Downloading: {output_file.type}")
    data = fhir.download_export(output_file.url)

    # Import into Rubric dataset
    client.datasets.import_fhir(
        dataset="fhir_bulk_q1_2024",
        fhir_data=data,
        resource_type=output_file.type
    )

from rubric import Rubric
from rubric.integrations.fhir import FHIRMapper

client = Rubric()

# Define mapping from FHIR to evaluation schema
mapper = FHIRMapper(
    mappings={
        # Patient demographics
        "patient_id": "Patient.id",
        "age": {
            "source": "Patient.birthDate",
            "transform": "calculate_age"
        },
        "sex": "Patient.gender",

        # Encounter context
        "encounter_id": "Encounter.id",
        "chief_complaint": "Encounter.reasonCode[0].text",
        "encounter_class": "Encounter.class.code",

        # Clinical data
        "vitals": {
            "source": "Observation.where(category='vital-signs')",
            "transform": "extract_vitals_dict"
        },
        "medications": {
            "source": "MedicationRequest",
            "transform": "extract_medication_list"
        },
        "conditions": {
            "source": "Condition.where(clinicalStatus='active')",
            "transform": "extract_condition_list"
        },

        # Clinical notes
        "clinical_note": {
            "source": "DocumentReference.where(type.coding.code='11506-3')",
            "transform": "extract_document_text"
        }
    }
)

# Apply mapping to create evaluation samples
def create_evaluation_sample(patient_id, encounter_id):
    # Fetch FHIR bundle
    bundle = fhir.fetch_bundle(
        patient_id=patient_id,
        encounter_id=encounter_id
    )

    # Map to evaluation format
    sample_data = mapper.map(bundle)

    # Create sample in Rubric
    client.samples.create(
        project="triage-evaluation",
        dataset="fhir_samples",
        **sample_data
    )

from rubric import Rubric
from rubric.integrations.fhir import FHIRValidator

client = Rubric()

validator = FHIRValidator(
    # Validation profiles
    profiles=[
        "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient",
        "http://hl7.org/fhir/us/core/StructureDefinition/us-core-encounter"
    ],

    # Validation level
    level="error"  # or "warning", "information"
)

# Validate a resource
result = validator.validate(patient_resource)

if result.valid:
    print("Resource is valid")
else:
    for issue in result.issues:
        print(f"{issue.severity}: {issue.diagnostics}")
        print(f"  Location: {issue.location}")