"""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) @lru_cache(maxsize=1) def get_store() -> MappingStore: return MappingStore()