Spaces:
Running
Running
| """RxNAV client for medication graph mappings. | |
| RxNAV is queried after the local mappings. Name lookups use RxNAV's exact or | |
| normalized mode; approximate matches are excluded. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| from functools import lru_cache | |
| from typing import Callable | |
| from urllib.error import HTTPError, URLError | |
| from urllib.parse import urlencode | |
| from urllib.request import Request, urlopen | |
| RXNAV_BASE = "https://rxnav.nlm.nih.gov" | |
| _NDC = re.compile(r"^\d{10,11}$") | |
| _NAME_STRIP = re.compile(r"[%\"']") | |
| _MAX_RELATED_PRODUCTS = 26 | |
| FetchJson = Callable[[str, dict[str, str]], dict | None] | |
| class RxNavClient: | |
| """Resolve NDCs or unique medication-name matches into RxNorm ingredients. | |
| Network failures and unknown/ambiguous responses return ``None``. The | |
| caller reports that no mapping is available. Public lookups are cached for | |
| the life of the API process. | |
| """ | |
| def __init__(self, fetch_json: FetchJson | None = None, base_url: str = RXNAV_BASE, | |
| timeout: float = 4.0): | |
| self.fetch_json = fetch_json | |
| self.base_url = base_url.rstrip("/") | |
| self.timeout = timeout | |
| def supports(self, code: str, code_type: str | None) -> bool: | |
| kind = (code_type or "").upper() | |
| return "NDC" in kind or "MED" in kind or "DRUG" in kind or "RXNORM" in kind | |
| def resolve(self, code: str, code_type: str | None, drug_name: str | None = None) -> dict | None: | |
| kind = (code_type or "").upper() | |
| if "NDC" in kind: | |
| return self.resolve_ndc(code) | |
| if "RXNORM" in kind and code.strip().isdigit(): | |
| return self.resolve_rxcui(code.strip()) | |
| if "MED" in kind or "DRUG" in kind: | |
| return self.resolve_name(drug_name or code) | |
| return None | |
| def resolve_ndc(self, code: str) -> dict | None: | |
| for ndc in self._ndc_candidates(code): | |
| data = self._request("/REST/ndcstatus.json", { | |
| "ndc": ndc, | |
| "history": "1", | |
| "altpkg": "1", | |
| }) | |
| status = (data or {}).get("ndcStatus") or {} | |
| rxcui = str(status.get("rxcui") or "").strip() | |
| if not rxcui.isdigit(): | |
| continue | |
| mapping = self.resolve_rxcui(rxcui) | |
| if mapping: | |
| return {**mapping, "matched_ndc": ndc, | |
| "query_name": status.get("conceptName") or mapping["name"]} | |
| return None | |
| def resolve_name(self, name: str) -> dict | None: | |
| cleaned = _NAME_STRIP.sub("", (name or "")).strip() | |
| if not cleaned or len(cleaned) > 240: | |
| return None | |
| data = self._request("/REST/Prescribe/rxcui.json", { | |
| "name": cleaned, | |
| "search": "2", # exact; then RxNAV's normalized match when needed | |
| }) | |
| ids = [str(value) for value in ((data or {}).get("idGroup") or {}).get("rxnormId") or [] | |
| if str(value).isdigit()] | |
| # The endpoint does not rank multiple exact/normalized concepts. | |
| if len(ids) != 1: | |
| return None | |
| mapping = self.resolve_rxcui(ids[0]) | |
| return {**mapping, "query_name": cleaned} if mapping else None | |
| def resolve_rxcui(self, rxcui: str) -> dict | None: | |
| if not rxcui.isdigit(): | |
| return None | |
| properties = self._request(f"/REST/rxcui/{rxcui}/properties.json", {}) | |
| props = (properties or {}).get("properties") or {} | |
| name = str(props.get("name") or rxcui) | |
| ingredients = self._related(rxcui, "IN") | |
| if str(props.get("tty") or "").upper() == "IN" and not ingredients: | |
| ingredients = [{"rxcui": rxcui, "name": name}] | |
| if not ingredients: | |
| return None | |
| child_drugs: dict[str, list[str]] = {} | |
| product_count = 0 | |
| for ingredient in ingredients: | |
| products = self._related(ingredient["rxcui"], "SCD SBD") | |
| names = sorted({p["name"] for p in products if p["name"].lower() != name.lower()}, | |
| key=str.casefold) | |
| product_count += len(names) | |
| child_drugs[ingredient["name"]] = names[:_MAX_RELATED_PRODUCTS] | |
| shown_count = sum(len(products) for products in child_drugs.values()) | |
| return { | |
| "name": name, | |
| "matched_rxcuis": [rxcui], | |
| "ingredients": ingredients, | |
| "child_drugs": child_drugs, | |
| # Not a stored mapping: RxNAV answered this one at request time. | |
| "source": "rxnav_live", | |
| "related_product_count": product_count, | |
| "related_products_shown": shown_count, | |
| } | |
| def _related(self, rxcui: str, tty: str) -> list[dict[str, str]]: | |
| data = self._request(f"/REST/Prescribe/rxcui/{rxcui}/related.json", {"tty": tty}) | |
| values: list[dict[str, str]] = [] | |
| for group in ((data or {}).get("relatedGroup") or {}).get("conceptGroup") or []: | |
| for concept in group.get("conceptProperties") or []: | |
| related_id = str(concept.get("rxcui") or "") | |
| related_name = str(concept.get("name") or "") | |
| if related_id.isdigit() and related_name: | |
| values.append({"rxcui": related_id, "name": related_name}) | |
| return values | |
| def _ndc_candidates(code: str) -> list[str]: | |
| candidates = [] | |
| for value in (code or "").split(","): | |
| digits = re.sub(r"\D", "", value) | |
| if _NDC.fullmatch(digits) and digits not in candidates: | |
| candidates.append(digits) | |
| return candidates | |
| def ndcs_for_rxcui(self, rxcui: str) -> tuple[str, ...]: | |
| """Package NDCs RxNorm lists for one concept, empty on any failure.""" | |
| data = self._request(f"/REST/rxcui/{rxcui}/ndcs.json", {}) | |
| ndcs = (((data or {}).get("ndcGroup") or {}).get("ndcList") or {}).get("ndc") or [] | |
| return tuple(str(n) for n in ndcs) | |
| def _request(self, path: str, params: dict[str, str]) -> dict | None: | |
| if self.fetch_json: | |
| return self.fetch_json(path, params) | |
| url = f"{self.base_url}{path}" | |
| if params: | |
| url += "?" + urlencode(params) | |
| request = Request(url, headers={"Accept": "application/json", "User-Agent": "ENCODE/1.0"}) | |
| try: | |
| with urlopen(request, timeout=self.timeout) as response: | |
| return json.load(response) | |
| except (HTTPError, URLError, OSError, ValueError): | |
| return None | |