"""Link clinical evidence text to UMLS graph nodes. Primary backend: surface-form dictionary built from the filtered KG (``id_maps.parquet``) plus optional enrichment from ``mrxns_eng`` / ``mrsty`` in the UMLS DuckDB. Optional backend: QuickUMLS (when ``leveldb`` + ``quickumls`` are installed). """ from __future__ import annotations import json import re import unicodedata from pathlib import Path import numpy as np import polars as pl from config.paths import ( CUI_TO_GRAPH_IDX_JSON, GRAPH_ID_MAPS_PARQUET, QUERY_LINKER_CACHE_JSON, UMLS_DUCKDB_PATH, ) # Clinical semantic types (diseases, drugs, symptoms, procedures, anatomy, findings). CLINICAL_SEMTYPES: frozenset[str] = frozenset( { "T047", # Disease or Syndrome "T048", # Mental or Behavioral Dysfunction "T184", # Sign or Symptom "T121", # Pharmacologic Substance "T109", # Organic Chemical "T123", # Biologically Active Substance "T195", # Antibiotic "T200", # Clinical Drug "T023", # Body Part, Organ, or Organ Component "T031", # Body Substance "T033", # Finding "T034", # Laboratory or Test Result "T060", # Diagnostic Procedure "T061", # Therapeutic or Preventive Procedure "T037", # Injury or Poisoning "T046", # Pathologic Function "T191", # Neoplastic Process } ) PREFERRED_SABS: frozenset[str] = frozenset( { "SNOMEDCT_US", "MSH", "RXNORM", "MEDCIN", "NCI", "LNC", "ICD10CM", "ICD10", "ICD9CM", "NDDF", "MMSL", "VANDF", } ) STOPWORDS: frozenset[str] = frozenset( { "a", "an", "the", "and", "or", "of", "in", "on", "at", "to", "for", "with", "without", "by", "from", "is", "are", "was", "were", "be", "been", "being", "has", "have", "had", "not", "no", "due", "associated", "secondary", "primary", "unspecified", "other", "specified", "history", "personal", "patient", "given", "then", "there", "will", "some", "any", "all", "both", "each", "than", "that", "this", "these", "those", "into", "during", "after", "before", "while", "when", "where", "which", "who", "whom", "whose", "if", "as", "but", "so", "such", "via", "per", "also", "only", "just", "very", "more", "most", "less", "least", "new", "old", "acute", "chronic", "stage", "type", "status", "present", "absent", "possible", "probable", "suspected", } ) _SURFACE_RE = re.compile(r"[^a-z0-9]+") def normalize_surface(text: str) -> str: """Lowercase, strip accents, collapse punctuation to spaces.""" text = unicodedata.normalize("NFKD", text) text = "".join(ch for ch in text if not unicodedata.combining(ch)) text = text.lower().strip() text = _SURFACE_RE.sub(" ", text) return " ".join(text.split()) def _tokenize(text: str) -> list[str]: return [t for t in normalize_surface(text).split() if t and t not in STOPWORDS] def build_cui_to_graph_idx( id_maps_parquet: Path = GRAPH_ID_MAPS_PARQUET, max_graph_idx: int | None = None, ) -> dict[str, int]: """Map CUI -> best graph row index (highest degree node per CUI).""" scan = pl.scan_parquet(id_maps_parquet).select("idx", "cui", "degree") if max_graph_idx is not None: scan = scan.filter(pl.col("idx") < max_graph_idx) df = ( scan.filter(pl.col("cui").is_not_null()) .sort(["cui", "degree"], descending=[False, True]) .group_by("cui") .agg(pl.col("idx").first()) .collect() ) return dict(zip(df["cui"].to_list(), df["idx"].to_list())) def build_clinical_cuis( cui_to_idx: dict[str, int], umls_duckdb: Path | None = UMLS_DUCKDB_PATH, ) -> set[str]: """Filter CUIs to clinical semantic types when DuckDB is available.""" if umls_duckdb is None or not Path(umls_duckdb).exists(): return set(cui_to_idx) import duckdb con = duckdb.connect(str(umls_duckdb), read_only=True) con.register("graph_cuis", pl.DataFrame({"cui": list(cui_to_idx)})) sem = ",".join(f"'{s}'" for s in CLINICAL_SEMTYPES) rows = con.execute( f""" SELECT DISTINCT g.cui FROM graph_cuis g JOIN mrsty s ON g.cui = s.CUI WHERE s.STY IN ({sem}) """ ).fetchall() con.close() return {r[0] for r in rows} if rows else set(cui_to_idx) def build_surface_lookup( cui_to_idx: dict[str, int], clinical_cuis: set[str], id_maps_parquet: Path = GRAPH_ID_MAPS_PARQUET, umls_duckdb: Path | None = UMLS_DUCKDB_PATH, max_graph_idx: int | None = None, min_surface_len: int = 3, ) -> dict[str, list[tuple[str, float]]]: """Build normalized surface form -> [(cui, score)] from KG + optional MRXNS.""" lookup: dict[str, dict[str, float]] = {} def add(surface: str, cui: str, score: float) -> None: if cui not in clinical_cuis: return norm = normalize_surface(surface) if len(norm) < min_surface_len: return bucket = lookup.setdefault(norm, {}) bucket[cui] = max(bucket.get(cui, 0.0), score) scan = pl.scan_parquet(id_maps_parquet).select("idx", "cui", "sab", "str") if max_graph_idx is not None: scan = scan.filter(pl.col("idx") < max_graph_idx) df = scan.filter( pl.col("cui").is_not_null() & pl.col("str").is_not_null() & (pl.col("str").str.len_chars() >= min_surface_len) ).collect() for cui, sab, surface in zip( df["cui"].to_list(), df["sab"].to_list(), df["str"].to_list() ): score = 1.0 if sab in PREFERRED_SABS else 0.85 add(surface, cui, score) if umls_duckdb is not None and Path(umls_duckdb).exists(): import duckdb con = duckdb.connect(str(umls_duckdb), read_only=True) con.register("graph_cuis", pl.DataFrame({"cui": list(clinical_cuis)})) sem = ",".join(f"'{s}'" for s in CLINICAL_SEMTYPES) rows = con.execute( f""" SELECT DISTINCT m.CUI, m.NSTR FROM mrxns_eng m JOIN graph_cuis g ON m.CUI = g.cui JOIN mrsty s ON m.CUI = s.CUI WHERE s.STY IN ({sem}) AND length(m.NSTR) >= {int(min_surface_len)} """ ).fetchall() con.close() for cui, nstr in rows: surface = " ".join(nstr.split()) add(surface, cui, 0.9) return {k: sorted(v.items(), key=lambda x: -x[1]) for k, v in lookup.items()} def save_cui_to_graph_idx(path: Path, cui_to_idx: dict[str, int]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(cui_to_idx), encoding="utf-8") def load_cui_to_graph_idx(path: Path = CUI_TO_GRAPH_IDX_JSON) -> dict[str, int]: payload = json.loads(path.read_text(encoding="utf-8")) return {str(k): int(v) for k, v in payload.items()} def ensure_cui_to_graph_idx( path: Path = CUI_TO_GRAPH_IDX_JSON, id_maps_parquet: Path = GRAPH_ID_MAPS_PARQUET, rebuild: bool = False, ) -> dict[str, int]: """Build or load CUI → graph row index (full graph, highest-degree node per CUI).""" if path.exists() and not rebuild: return load_cui_to_graph_idx(path) cui_to_idx = build_cui_to_graph_idx(id_maps_parquet, max_graph_idx=None) save_cui_to_graph_idx(path, cui_to_idx) return cui_to_idx def save_linker_cache( path: Path, cui_to_idx: dict[str, int], surface_lookup: dict[str, list[tuple[str, float]]], ) -> None: path.parent.mkdir(parents=True, exist_ok=True) payload = { "cui_to_idx": cui_to_idx, "surface_lookup": {k: list(v) for k, v in surface_lookup.items()}, } path.write_text(json.dumps(payload), encoding="utf-8") def load_linker_cache( path: Path, ) -> tuple[dict[str, int], dict[str, list[tuple[str, float]]]]: payload = json.loads(path.read_text(encoding="utf-8")) surface_lookup = { k: [(cui, float(score)) for cui, score in v] for k, v in payload["surface_lookup"].items() } return payload["cui_to_idx"], surface_lookup class QueryEntityLinker: """Map query text to graph node indices with confidence scores.""" def __init__( self, *, id_maps_parquet: Path = GRAPH_ID_MAPS_PARQUET, umls_duckdb: Path | None = UMLS_DUCKDB_PATH, cache_path: Path = QUERY_LINKER_CACHE_JSON, quickumls_path: Path | None = None, max_graph_idx: int | None = None, rebuild_cache: bool = False, min_confidence: float = 0.7, max_ngram: int = 6, quickumls_only: bool = False, skip_graph_cui_filter: bool = False, ): self.min_confidence = min_confidence self.max_ngram = max_ngram self.quickumls_only = quickumls_only self.skip_graph_cui_filter = skip_graph_cui_filter self._quickumls = None if quickumls_only and skip_graph_cui_filter: self.cui_to_idx = {} self.surface_lookup = {} elif quickumls_only: self.cui_to_idx = ensure_cui_to_graph_idx( rebuild=rebuild_cache, id_maps_parquet=id_maps_parquet, ) self.surface_lookup = {} elif cache_path.exists() and not rebuild_cache: self.cui_to_idx, self.surface_lookup = load_linker_cache(cache_path) else: # Build full-graph linker cache; OOB embedding rows are filtered at pool time. cui_to_idx = build_cui_to_graph_idx(id_maps_parquet, max_graph_idx=None) clinical_cuis = build_clinical_cuis(cui_to_idx, umls_duckdb) self.surface_lookup = build_surface_lookup( cui_to_idx, clinical_cuis, id_maps_parquet=id_maps_parquet, umls_duckdb=umls_duckdb, max_graph_idx=None, ) self.cui_to_idx = { c: i for c, i in cui_to_idx.items() if c in clinical_cuis } save_linker_cache(cache_path, self.cui_to_idx, self.surface_lookup) if quickumls_path is not None and Path(quickumls_path).exists(): self._quickumls, load_err = self._try_load_quickumls(quickumls_path) if quickumls_only and self._quickumls is None: raise RuntimeError( "quickumls_only=True but QuickUMLS could not be loaded.\n" f"{load_err or 'Unknown error.'}\n" "Fix: run ./setup.sh in the bundle root, then use ./run_export.sh " "(not system python3)." ) @staticmethod def _try_load_quickumls(quickumls_path: Path): import sys try: from quickumls import QuickUMLS except ImportError as exc: return None, ( f"ImportError: {exc}\n" f" python: {sys.executable}\n" " Hint: create the venv with ./setup.sh and run via ./run_export.sh" ) try: matcher = QuickUMLS( str(quickumls_path), threshold=0.7, accepted_semtypes=sorted(CLINICAL_SEMTYPES), ) except Exception as exc: return None, ( f"{type(exc).__name__}: {exc}\n" f" python: {sys.executable}\n" f" quickumls_data: {quickumls_path}" ) return matcher, None def _link_quickumls(self, text: str) -> list[tuple[str, float]]: if self._quickumls is None: return [] out: list[tuple[str, float]] = [] for match_group in self._quickumls.match( text, best_match=True, ignore_syntax=False ): for m in match_group: cui = m.get("cui") sim = float(m.get("similarity", 0.0)) if cui and sim >= self.min_confidence: out.append((cui, sim)) return out def _link_dictionary(self, text: str) -> list[tuple[str, float]]: tokens = _tokenize(text) if not tokens: return [] matches: dict[str, float] = {} n = len(tokens) for start in range(n): for length in range(min(self.max_ngram, n - start), 0, -1): phrase = " ".join(tokens[start : start + length]) hits = self.surface_lookup.get(phrase) if not hits: continue for cui, score in hits: matches[cui] = max(matches.get(cui, 0.0), score) break # longest match at this start position # Also try semicolon / clause segments (common in evidence strings). for clause in re.split(r"[;,\n]+", text): clause = clause.strip() if not clause: continue norm_clause = normalize_surface(clause) if norm_clause in self.surface_lookup: for cui, score in self.surface_lookup[norm_clause]: matches[cui] = max(matches.get(cui, 0.0), score) return sorted(matches.items(), key=lambda x: -x[1]) def link_cuis(self, text: str) -> list[tuple[str, float]]: """Return (cui, confidence) pairs for a query.""" matches: dict[str, float] = {} if self.quickumls_only: for cui, score in self._link_quickumls(text): if self.skip_graph_cui_filter or cui in self.cui_to_idx: matches[cui] = max(matches.get(cui, 0.0), score) else: for cui, score in self._link_dictionary(text): matches[cui] = max(matches.get(cui, 0.0), score) for cui, score in self._link_quickumls(text): if cui in self.cui_to_idx: matches[cui] = max(matches.get(cui, 0.0), score) return sorted(matches.items(), key=lambda x: -x[1]) def link_graph_nodes(self, text: str) -> list[tuple[int, float]]: """Return (graph_row_idx, confidence) pairs for a query.""" out: list[tuple[int, float]] = [] for cui, score in self.link_cuis(text): idx = self.cui_to_idx.get(cui) if idx is not None: out.append((int(idx), score)) return out def coverage_stats(self, texts: list[str]) -> dict[str, float]: """Compute entity-linking coverage over a list of queries.""" n = len(texts) if n == 0: return { "n": 0, "pct_with_match": 0.0, "pct_zero_match": 100.0, "avg_entities": 0.0, } counts = [len(self.link_graph_nodes(t)) for t in texts] with_match = sum(1 for c in counts if c > 0) return { "n": n, "pct_with_match": 100.0 * with_match / n, "pct_zero_match": 100.0 * (n - with_match) / n, "avg_entities": float(np.mean(counts)), "median_entities": float(np.median(counts)), "max_entities": int(max(counts)), }