FHIR Integration

import requests

resp = requests.post(
    "https://api.omophub.com/v1/fhir/resolve",
    headers={"Authorization": "Bearer oh_your_api_key"},
    json={
        "system": "http://loinc.org",
        "code": "2339-0",
        "display": "Blood Glucose",
        "resource_type": "Observation",
        "include_recommendations": False,
        "include_quality": True,
    },
)
result = resp.json()["data"]["resolution"]

result["standard_concept"]["concept_id"]    # 3004501
result["standard_concept"]["concept_name"]  # "Glucose [Mass/volume] in Serum or Plasma"
result["target_table"]                      # "measurement"
result["mapping_type"]                      # "direct"
result["domain_resource_alignment"]         # "aligned"
result["mapping_quality"]                   # "high"

resp = requests.post(
    "https://api.omophub.com/v1/fhir/resolve/codeable-concept",
    headers={"Authorization": "Bearer oh_your_api_key"},
    json={
        "coding": [
            {"system": "http://snomed.info/sct", "code": "44054006"},
            {"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "E11.9"},
        ],
        "text": "Type 2 diabetes mellitus",
        "resource_type": "Condition",
    },
)
best = resp.json()["data"]["best_match"]["resolution"]

best["source_concept"]["vocabulary_id"]   # "SNOMED"  (preferred over ICD10CM for Condition)
best["standard_concept"]["concept_id"]    # 201826
best["target_table"]                      # "condition_occurrence"

import omophub
client = omophub.OMOPHub()

# 1. Find a concept by code
concept = client.search.basic("44054006", vocabulary_ids=["SNOMED"])["concepts"][0]

# 2. Walk its Maps-to relationships
mappings = client.mappings.get(concept["concept_id"], target_vocabulary="SNOMED")

# 3. Explore hierarchy (ancestors / descendants)
ancestors = client.hierarchy.ancestors(concept["concept_id"], max_levels=5)

# 4. Non-hierarchy relationships (e.g. "Has ingredient" for drugs)
rels = client.concepts.relationships(concept["concept_id"])

import requests, json

fhir_obs = {
    "resourceType": "Observation",
    "status": "final",
    "code": {
        "coding": [{
            "system": "http://loinc.org",
            "code": "2339-0",
            "display": "Glucose [Mass/volume] in Blood",
        }],
        "text": "Blood Glucose",
    },
    "valueQuantity": {"value": 120, "unit": "mg/dL"},
    "effectiveDateTime": "2023-10-26T09:30:00Z",
}

coding = fhir_obs["code"]["coding"][0]
resp = requests.post(
    "https://api.omophub.com/v1/fhir/resolve",
    headers={"Authorization": "Bearer oh_your_api_key"},
    json={
        "system": coding["system"],
        "code": coding["code"],
        "display": fhir_obs["code"].get("text"),
        "resource_type": "Observation",
    },
)
result = resp.json()["data"]["resolution"]

omop_measurement = {
    "measurement_concept_id": result["standard_concept"]["concept_id"],
    "measurement_source_value": coding.get("display", ""),
    "value_as_number": fhir_obs["valueQuantity"]["value"],
    "unit_source_value": fhir_obs["valueQuantity"].get("unit"),
    "measurement_date": fhir_obs["effectiveDateTime"][:10],
}
print(json.dumps(omop_measurement, indent=2))

import requests, omophub

client = omophub.OMOPHub()

# Step 1: NDC → standard RxNorm via the Resolver
resp = requests.post(
    "https://api.omophub.com/v1/fhir/resolve",
    headers={"Authorization": "Bearer oh_your_api_key"},
    json={
        "system": "http://hl7.org/fhir/sid/ndc",
        "code": "0071-0155-01",
        "display": "Lipitor 10 MG Oral Tablet",
        "resource_type": "MedicationStatement",
    },
)
rxnorm = resp.json()["data"]["resolution"]["standard_concept"]
print(rxnorm["concept_id"], rxnorm["concept_name"])
# 40165252 "atorvastatin 10 MG Oral Tablet [Lipitor]"

# Step 2: Walk "Has ingredient" to find the active ingredient concept
rels = client.concepts.relationships(rxnorm["concept_id"])
ingredients = [
    r for r in rels.get("relationships", [])
    if "ingredient" in r.get("relationship_id", "").lower()
    or r.get("concept_class_id") == "Ingredient"
]
if ingredients:
    ing = ingredients[0]
    print(ing["concept_id"], ing["concept_name"])
    # 1545958 "atorvastatin"

# US Core detailed race extension on a Patient resource
patient_extension = {
    "url": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-race",
    "extension": [{
        "url": "ombCategory",
        "valueCoding": {
            "system": "urn:oid:2.16.840.1.113883.6.238",
            "code": "2106-3",
            "display": "White",
        }
    }]
}

# Resolve the Coding inside the extension just like any other Coding
coding = patient_extension["extension"][0]["valueCoding"]
resp = requests.post(
    "https://api.omophub.com/v1/fhir/resolve",
    headers={"Authorization": "Bearer oh_your_api_key"},
    json={
        "system": coding["system"],
        "code": coding["code"],
        "display": coding["display"],
        "resource_type": "Patient",
    },
)
# Write the result to person.race_concept_id