mapping-to-snomed

Mapping OpenMed spans to SNOMED CT

Ground clinical concept spans that OpenMed extracts — disorders, findings, procedures, body structures, substances — to SNOMED CT, the comprehensive clinical reference terminology. The atom is the SCTID (a SNOMED CT concept identifier), organized into a description-logic hierarchy you can query with ECL (Expression Constraint Language).

Hard licensing boundary — read first. SNOMED CT is license-restricted. OpenMed and this skill never bundle, ship, cache, or redistribute any SNOMED CT content. All mapping happens out-of-process against a terminology server the user supplies and is licensed for — their own Ontoserver, Snowstorm, the NLM's UTS/UMLS FHIR endpoint, or a national release server. SNOMED International requires an Affiliate License (free in member territories like the US via the NLM; check your country). Your code receives a base URL + credentials from the user; it must work with any compliant FHIR terminology server and store nothing but the returned codes.

When to use

• You need rich, hierarchy-aware clinical codes (more granular than ICD-10) for problems, procedures, or body sites.
• You want to translate an existing code (ICD-10-CM, local code) to SNOMED CT via a ConceptMap/$translate.
• You need subsumption/ECL queries ("is this a descendant of Diabetes mellitus?") for cohorting or decision support.

For billing codes use coding-icd10; for drugs normalizing-rxnorm; for labs mapping-loinc. SNOMED CT is the clinical-meaning layer.

Quick start (user-supplied FHIR terminology server)

Configuration is injected, never hardcoded. The operations are standard FHIR R4.

import os, requests

# Provided by the USER — their licensed server. Nothing bundled.
TX = os.environ["FHIR_TX_URL"]              # e.g. https://snowstorm.example.org/fhir
TOKEN = os.environ.get("FHIR_TX_TOKEN")     # if the server requires auth
SNOMED = "http://snomed.info/sct"
HDRS = {"Accept": "application/fhir+json"}
if TOKEN:
    HDRS["Authorization"] = f"Bearer {TOKEN}"

def lookup(code: str) -> dict:
    """$lookup: fully specified name + properties for an SCTID."""
    r = requests.get(f"{TX}/CodeSystem/$lookup",
                     params={"system": SNOMED, "code": code},
                     headers=HDRS, timeout=15)
    r.raise_for_status()
    return r.json()

def find_concepts(text: str, ecl: str = "<<404684003", count: int = 10):
    """Text search constrained by ECL (default: descendants of Clinical finding)."""
    vs = f"{SNOMED}?fhir_vs=ecl/{ecl}"
    r = requests.get(f"{TX}/ValueSet/$expand",
                     params={"url": vs, "filter": text, "count": count},
                     headers=HDRS, timeout=20)
    r.raise_for_status()
    return r.json().get("expansion", {}).get("contains", [])

def translate(code: str, source_system: str, conceptmap_url: str):
    """$translate an existing code to SNOMED CT via a ConceptMap."""
    r = requests.get(f"{TX}/ConceptMap/$translate",
                     params={"url": conceptmap_url, "system": source_system,
                             "code": code, "targetsystem": SNOMED},
                     headers=HDRS, timeout=20)
    r.raise_for_status()
    return r.json()

# ECL examples: 64572001=disease, 71388002=procedure, 123037004=body structure
print(find_concepts("type 2 diabetes", ecl="<<64572001"))