Spaces:
Running
Running
| """One MappingStore: each mapping file loaded once, every usability rule owned here. | |
| The on-disk shape is the unified record (key / key_kind / target / match / | |
| derived / source / label / lineage). The store normalizes each row into the | |
| small in-memory view its consumers read, so codesearch and graph never touch | |
| the storage shape. The rules, stated once: | |
| - Lab rows are runtime-usable only when they assert a single LOINC and carry | |
| no loinc_status flag (inactive_flagged rows have a retired target; | |
| multi_target rows assert none). | |
| - Phecode entries exist only when at least one map version (v1.2 or phecodeX) | |
| assigns something; derived rollups keep their lineage as the marker the UI | |
| renders. | |
| - CPT groups: the validated CMS RBCS row wins over the embedding-derived | |
| match, which fills in behind it flagged as derived. | |
| - Medications resolve validated-before-derived inside each key kind; the | |
| router consults the kinds in its fixed order. | |
| Consumers take the store instance; `get_store()` is the process singleton. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from functools import lru_cache | |
| from pathlib import Path | |
| from typing import Iterable | |
| from . import paths | |
| def _load_jsonl(path: Path): | |
| if path.exists(): | |
| with path.open(encoding="utf-8") as handle: | |
| for line in handle: | |
| if line.strip(): | |
| yield json.loads(line) | |
| # The `source` field is one of a handful of registry ids repeated across | |
| # hundreds of thousands of rows, and json.loads builds a fresh string for each | |
| # one. Sharing a single object per id keeps the views the size they were. | |
| _SOURCE_IDS: dict[str, str] = {} | |
| def _source_id(row: dict) -> str | None: | |
| src = row.get("source") | |
| return _SOURCE_IDS.setdefault(src, src) if src else None | |
| def _lab_view(row: dict) -> dict: | |
| # Absent fields stay absent so consumer .get() defaults behave exactly | |
| # as they did over the legacy rows. The row's free-text `label` is not | |
| # carried: provenance is rendered from `source` and `derived`, which is | |
| # what the row was actually built from (the labels claim the Dim_Labs | |
| # crosswalk for every tier, including the OMOP one). | |
| target = row.get("target") or {} | |
| view = {"loinc": target.get("loinc"), | |
| "match": row.get("match"), | |
| "derived": bool(row.get("derived")), | |
| "source": _source_id(row)} | |
| if row.get("lab_name") is not None: | |
| view["lab_name"] = row["lab_name"] | |
| if "loinc_version" in target: | |
| view["loinc_version"] = target["loinc_version"] | |
| return view | |
| def _med_view(row: dict) -> dict: | |
| view = {"med_code": row["key"], "kind": row.get("key_kind"), | |
| **(row.get("target") or {}), | |
| "match": row.get("match"), "source": _source_id(row)} | |
| if row.get("derived"): | |
| view["derived"] = True | |
| return view | |
| class MappingStore: | |
| def __init__(self, lab_loinc_records: Iterable[dict] | None = None): | |
| # -- lab: VA LabChemTestSID -> LOINC (usable rows only) ----------- | |
| rows = (lab_loinc_records if lab_loinc_records is not None | |
| else _load_jsonl(paths.LAB_LOINC_MAP)) | |
| self._lab_loinc: dict[str, dict] = {} | |
| for r in rows: | |
| target = r.get("target") or {} | |
| if target.get("loinc") and not r.get("loinc_status"): | |
| self._lab_loinc[str(r["key"]).strip()] = _lab_view(r) | |
| # -- diagnosis: (ICD version, dotted code) -> phecode assignments - | |
| version_of = {"icd10": "ICD10", "icd9": "ICD9"} | |
| self._phecode: dict[tuple[str, str], dict] = {} | |
| for r in _load_jsonl(paths.ICD_PHECODE_MAP): | |
| target = r.get("target") or {} | |
| entry = {} | |
| if target.get("v12"): | |
| entry["v12"] = target["v12"] | |
| if target.get("x"): | |
| entry["x"] = target["x"] | |
| if entry: | |
| if r.get("derived"): | |
| entry["derived"] = r["lineage"] | |
| version = version_of.get(r["key_kind"], r["key_kind"]) | |
| self._phecode[(version, r["key"])] = entry | |
| # -- procedure: CPT/HCPCS -> CMS RBCS group ----------------------- | |
| self._rbcs_validated: dict[str, dict] = {} | |
| self._cpt_embedding: dict[str, dict] = {} | |
| for r in _load_jsonl(paths.CPT_RBCS_MAP): | |
| target = r.get("target") or {} | |
| if r.get("derived"): | |
| view = {"code": r["key"], "group": target["family"], "derived": True, | |
| "match": r.get("match"), "source": _source_id(r)} | |
| for field in ("category", "description"): | |
| if field in target: | |
| view[field] = target[field] | |
| self._cpt_embedding[r["key"]] = view | |
| else: | |
| self._rbcs_validated[r["key"]] = {"code": r["key"], **target, | |
| "source": _source_id(r)} | |
| # Merged view for result stamping: validated wins, derived fills in. | |
| self._rbcs_merged = dict(self._rbcs_validated) | |
| for code, view in self._cpt_embedding.items(): | |
| self._rbcs_merged.setdefault(code, { | |
| "code": code, "category": view.get("category", ""), | |
| "subcategory": "", "family": view["group"], | |
| "major": "", "derived": True, | |
| "match": view.get("match"), "source": view.get("source")}) | |
| # Family sizes are not counted here. This file holds every code CMS | |
| # assigned, which is a wider set than the CPT index serves, and two | |
| # counts of one family is what testing complained about. The count | |
| # lives in CodeSearchEngine.rbcs_family_sizes(), over the served codes. | |
| # -- medication: three key kinds in one file ---------------------- | |
| self._code_ingredient: dict[str, dict] = {} | |
| self._med_by_name: dict[str, dict] = {} | |
| self._rxcui: dict[str, dict] = {} | |
| for r in _load_jsonl(paths.MED_RXNORM_MAP): | |
| kind = r.get("key_kind") | |
| if kind in ("ndc", "drug_sid"): | |
| self._code_ingredient[str(r["key"]).upper()] = _med_view(r) | |
| elif kind == "drug_name": | |
| self._med_by_name[str(r["key"]).upper()] = _med_view(r) | |
| elif kind == "rxcui": | |
| self._rxcui[str(r["key"])] = _med_view(r) | |
| # -- lab --------------------------------------------------------------- | |
| def loinc_for(self, sids: Iterable[str]) -> list[dict]: | |
| """Usable mapping views for the given SIDs, in input order.""" | |
| found = (self._lab_loinc.get(str(sid).strip()) for sid in sids) | |
| return [row for row in found if row] | |
| def loinc_of(self, sid: str) -> dict | None: | |
| """The mapping view for one SID, or None. Callers that need to know | |
| which SID carried the mapping use this instead of loinc_for, which | |
| drops the unmapped ones and so loses the pairing.""" | |
| return self._lab_loinc.get(str(sid).strip()) | |
| def has_lab_sid(self, sids: Iterable[str]) -> bool: | |
| return any(str(sid).strip() in self._lab_loinc for sid in sids) | |
| # -- diagnosis --------------------------------------------------------- | |
| def phecodes_for(self, code: str, versions: Iterable[str]) -> dict | None: | |
| """First version's assignments for one dotted ICD code, else None.""" | |
| code = code.strip() | |
| for version in versions: | |
| entry = self._phecode.get((version, code)) | |
| if entry: | |
| return entry | |
| return None | |
| # -- procedure --------------------------------------------------------- | |
| def rbcs_for(self, code: str) -> dict | None: | |
| """Validated-or-derived group for result stamping (validated wins).""" | |
| return self._rbcs_merged.get(code.strip().upper()) | |
| def rbcs_validated_for(self, code: str) -> dict | None: | |
| return self._rbcs_validated.get(code.upper()) | |
| def cpt_embedding_for(self, code: str) -> dict | None: | |
| return self._cpt_embedding.get(code) | |
| def has_cpt_embedding(self, code: str) -> bool: | |
| return code in self._cpt_embedding | |
| # -- medication -------------------------------------------------------- | |
| def ingredients_for(self, code: str) -> dict | None: | |
| """Precomputed ingredient mapping for an NDC or VA drug SID.""" | |
| return self._code_ingredient.get(code.upper()) | |
| def ingredients_by_name(self, name: str) -> dict | None: | |
| return self._med_by_name.get(name.upper()) | |
| def ingredients_by_rxcui(self, rxcui: str) -> dict | None: | |
| return self._rxcui.get(rxcui) | |
| def get_store() -> MappingStore: | |
| return MappingStore() | |