"""The taxonomy registry engine (T9, #35): storage is convention-neutral, a convention is a projection. One extraction is stored once, against the primitive the report stated, and any convention is computed on demand. Adding a convention is a matter of adding registry rows, never of re-extracting a report. The registry is six normalized CSVs under `data/taxonomy/`: - `coding_systems.csv` the systems a concept can belong to (FIGO 1988/2009/2023, CAP, WHO, ...). - `dimensions.csv` one row per checklist question, tagged `value_kind` (coded, quantity, derived). - `concepts.csv` the coded values, keyed `(system_id, dimension, code)`, because `figo_endo/1988.IB` and `figo_endo/2009.IB` are different concepts that share a code and denote opposite disease extents. - `crosswalk.csv` concept-to-concept mappings with a SKOS predicate plus `conflict` (six values). - `crosswalk_precondition.csv` conditions that gate a mapping. Rows sharing `(mapping_id, dimension)` are OR'd; across dimensions they are AND'd. - `derivations.csv` the `formula`, `threshold`, and `membership` rules. No free-text formula in a cell: a formula or membership row names a `formula_id` this module implements, and a threshold row is data (an operator and a threshold scoped by a system). The public surface: `resolve`, `codes`, `crosswalk`, `derive`, `project`, `derive_stage`, plus `infer_edition` and `determinability_matrix`. `derive_stage` is the per-edition FIGO stage algorithm; it projects from canonical assertions and never from another edition's stage label. The stage rows in `crosswalk.csv` are the algorithm's golden oracle and documentation, not its runtime engine. The two invariants this module serves: 1. **Numbers stay numbers.** A percentage and a category derive from the millimetre quantities by a named rule. `ratio_percent` is registered once (one `formula_id`) and bound twice (two rows), to the myometrial pair and to the cervical pair, so the cervical binding cannot contaminate the myometrial one. A report stating "4 foci" is `substantial` LVSI under FIGO 2009 (>=4) and `focal` under FIGO 2023 (>=5): storing the count keeps both answers reachable, storing the category discards one. 2. **No value without its evidence.** Every registry row cites an `extraction_id` in `data/taxonomy/extractions.csv`, or a `docs/taxonomy.md` section for a rule the registry currently evidences in prose. A citation that resolves to neither fails the loader. ## How B1 (`endopath.validation`, #97) plugs in `endopath.validation` recomputes a derived concept to catch an extraction error. It owns no rule logic: it imports `RECOMPUTE_BY_CONCEPT` from this module, keyed by concept id, positional in dependency-graph order. Every rule that a per-case record can recompute from its graph inputs alone lives there: `ratio_percent` (bound to the myometrial and cervical percents), the myometrial-invasion-category threshold, `grade_binary`, and `histotype_aggressiveness`. Two rules need a projection context the per-case record does not carry and so are not in that table: `lvsi_extent` needs a coding system (the substantial cut-off is 3, 4, or 5 vessels by system), and `figo_stage` needs a FIGO edition (a bare stage code is edition-keyed). A caller projects those through `derive(..., under=...)` and `derive_stage(..., edition)` when the context is supplied. """ from __future__ import annotations import csv import re from collections import defaultdict from dataclasses import dataclass from enum import Enum from functools import lru_cache from pathlib import Path from typing import Any, Callable, Mapping, Optional, Sequence from pydantic import BaseModel, ConfigDict REPO_ROOT = Path(__file__).resolve().parents[2] TAXONOMY_DIR = REPO_ROOT / "data" / "taxonomy" TAXONOMY_DOC_PATH = REPO_ROOT / "docs" / "taxonomy.md" CODING_SYSTEMS_PATH = TAXONOMY_DIR / "coding_systems.csv" DIMENSIONS_PATH = TAXONOMY_DIR / "dimensions.csv" CONCEPTS_PATH = TAXONOMY_DIR / "concepts.csv" CROSSWALK_PATH = TAXONOMY_DIR / "crosswalk.csv" PRECONDITION_PATH = TAXONOMY_DIR / "crosswalk_precondition.csv" DERIVATIONS_PATH = TAXONOMY_DIR / "derivations.csv" EXTRACTIONS_PATH = TAXONOMY_DIR / "extractions.csv" class RegistryError(ValueError): """The registry did not load: an unknown system or dimension, a dangling concept reference, a predicate or rule_kind outside its closed vocabulary, a precondition on an unknown value, or a citation that resolves to neither an extraction nor a docs/taxonomy.md section.""" # --------------------------------------------------------------------------- closed vocabularies class Predicate(str, Enum): """The crosswalk `predicate` column: the five SKOS values plus `conflict` (dec_015, #85). The five are the SKOS vocabulary, not the RDF stack. `conflict` marks a row where two standards each state a rule for the same question and the rules disagree, distinct from `noMatch` (no counterpart is named) and from `relatedMatch` (overlap without contradiction).""" EXACT_MATCH = "exactMatch" BROAD_MATCH = "broadMatch" NARROW_MATCH = "narrowMatch" RELATED_MATCH = "relatedMatch" NO_MATCH = "noMatch" CONFLICT = "conflict" class RuleKind(str, Enum): """The derivation kinds, closed at three so no free-text formula enters a cell. `formula` computes a quantity from quantities via a named formula with typed args. `threshold` computes a coded value from a quantity, scoped by `under_system`. `membership` computes a coded or bool value from coded values, partial by design (a value the source does not determine returns `indeterminate`).""" FORMULA = "formula" THRESHOLD = "threshold" MEMBERSHIP = "membership" class ValueKind(str, Enum): """A dimension's answer kind. `coded` picks from a value set (the concepts), `quantity` is a number, `derived` is computed by a rule.""" CODED = "coded" QUANTITY = "quantity" DERIVED = "derived" class Certainty(str, Enum): CERTAIN = "certain" UNCERTAIN = "uncertain" # --------------------------------------------------------------------------- row models class CodingSystem(BaseModel): model_config = ConfigDict(frozen=True) system_id: str title: str kind: str edition_of: str = "" note: str = "" class Dimension(BaseModel): model_config = ConfigDict(frozen=True) dimension: str value_kind: ValueKind closure: str = "" note: str = "" class Concept(BaseModel): """One coded value, keyed `(system_id, dimension, code)`.""" model_config = ConfigDict(frozen=True) concept_id: str system_id: str dimension: str code: str label: str = "" extraction_id: str = "" note: str = "" @property def key(self) -> tuple[str, str, str]: return (self.system_id, self.dimension, self.code) class CrosswalkMapping(BaseModel): model_config = ConfigDict(frozen=True) mapping_id: str source_concept_id: str target_concept_id: str predicate: Predicate certainty: Certainty = Certainty.CERTAIN rationale: str = "" extraction_id: str = "" class Precondition(BaseModel): model_config = ConfigDict(frozen=True) precondition_id: str mapping_id: str dimension: str operator: str value_code: str class Derivation(BaseModel): model_config = ConfigDict(frozen=True) derivation_id: str rule_kind: RuleKind target_dimension: str target_concept_id: str = "" source_dimension: str = "" operator: str = "" threshold: str = "" unit: str = "" under_system: str = "" formula_id: str = "" arg_numerator: str = "" arg_denominator: str = "" extraction_id: str = "" # --------------------------------------------------------------------------- projection results @dataclass(frozen=True) class Stage: """A determined stage: a code within a coding system, optionally carrying a molecular modifier.""" code: str system_id: str modifier: Optional[str] = None @dataclass(frozen=True) class Constrained: """The asserted values narrow the answer to a set and no further (docs/taxonomy.md section 6.2). The set carries the answer, so it is never collapsed to a single value or to a blank.""" candidates: tuple[str, ...] reason: str dimension: str = "figo_stage" system_id: Optional[str] = None @dataclass(frozen=True) class Indeterminate: """A required assertion is missing, or the report states the value cannot be assessed.""" reason: str dimension: str = "figo_stage" system_id: Optional[str] = None # A sentinel for LVSI present but unquantified: a foci count the report never states. UNQUANTIFIED = "__lvsi_present_unquantified__" # The determinable value set of each derived (or gated coded) dimension. Used to validate a # threshold row's target and a precondition's value_code, and to enumerate a `constrained` set. DERIVED_VALUE_SETS: dict[str, frozenset[str]] = { "myometrial_invasion_category": frozenset({"none", "less_than_half", "half_or_more"}), "lvsi_extent": frozenset({"negative", "focal", "substantial"}), "histotype_aggressiveness": frozenset({"non_aggressive", "aggressive", "indeterminate"}), "grade_binary": frozenset({"low", "high"}), "cervical_stromal_invasion": frozenset({"present", "absent", "cannot_be_assessed"}), } # --------------------------------------------------------------------------- the named rules def ratio_percent( numerator: Any = None, denominator: Any = None, *, depth: Any = None, thickness: Any = None, ) -> Optional[float]: """Depth over thickness as a percentage (docs/taxonomy.md section 2). Registered once and bound twice, to the myometrial and to the cervical quantities. Returns None on a zero or non-numeric denominator, which the caller reports as an unrecomputable derivation rather than a value.""" n = depth if depth is not None else numerator d = thickness if thickness is not None else denominator try: d = float(d) if d == 0: return None return float(n) / d * 100.0 except (TypeError, ValueError): return None def grade_binary(figo_grade: Any) -> Optional[str]: """The membership rule {1,2} low, {3} high (docs/taxonomy.md section 2).""" if figo_grade in ("1", "2", 1, 2): return "low" if figo_grade in ("3", 3): return "high" return None def histotype_aggressiveness(histologic_type: Any, figo_grade: Any = None) -> Optional[str]: """FIGO 2023's aggressiveness membership (ex_figo2023_aggressive_types). Non-aggressive is low-grade (G1/G2) endometrioid; aggressive is every other combination. The rule is partial: endometrioid carcinoma of unknown grade is `indeterminate` rather than aggressive.""" if histologic_type in (None, ""): return None if histologic_type != "endometrioid": return "aggressive" if figo_grade in ("1", "2", 1, 2): return "non_aggressive" if figo_grade in ("3", 3): return "aggressive" return "indeterminate" def cervical_membership(source: Any) -> Optional[str]: """Reduce a source cervical assessment to the exclusive `cervical_stromal_invasion` value set, partial by design. A source that could not assess the cervix (for example CAP's `cannot_assess_supracervical`) determines neither `present` nor `absent`, so the rule returns None and `derive` reports `indeterminate`.""" if source in (None, ""): return None s = str(source).lower() if "cannot" in s or "assess" in s or "supracervical" in s: return None if "present" in s or "invasion" in s or s == "present": return "present" if "absent" in s or "no_" in s or "not" in s or s == "absent": return "absent" return None # formula_id -> the callable a `formula` or `membership` derivation row names. A threshold row carries # no formula_id: its logic is the data in the row (an operator and a threshold), evaluated generically. NAMED_RULES: dict[str, Callable[..., Any]] = { "ratio_percent": ratio_percent, "grade_binary": grade_binary, "histotype_aggressiveness": histotype_aggressiveness, "cervical_stromal_invasion": cervical_membership, } # --------------------------------------------------------------------------- the registry class Registry: """The loaded, cross-validated registry: the six tables plus the indices the engine reads.""" def __init__( self, coding_systems: list[CodingSystem], dimensions: list[Dimension], concepts: list[Concept], crosswalk: list[CrosswalkMapping], preconditions: list[Precondition], derivations: list[Derivation], ) -> None: self.coding_systems = coding_systems self.dimensions = dimensions self.concepts = concepts self.crosswalk = crosswalk self.preconditions = preconditions self.derivations = derivations self.system_by_id = {s.system_id: s for s in coding_systems} self.dimension_by_id = {d.dimension: d for d in dimensions} self.concept_by_id = {c.concept_id: c for c in concepts} self.concept_by_key: dict[tuple[str, str, str], Concept] = {c.key: c for c in concepts} self.mapping_by_id = {m.mapping_id: m for m in crosswalk} self.preconditions_by_mapping: dict[str, list[Precondition]] = defaultdict(list) for pc in preconditions: self.preconditions_by_mapping[pc.mapping_id].append(pc) self._validate() def _validate(self) -> None: extraction_ids = _load_extraction_ids() doc_sections = _load_doc_sections() def cite_ok(cited: str) -> bool: if not cited: return False if cited in extraction_ids: return True m = re.match(r"^docs/taxonomy\.md#(\d+(?:\.\d+)*)$", cited) return bool(m) and m.group(1) in doc_sections # concepts: unique ids and keys, a known system, a resolvable citation. if len(self.concept_by_id) != len(self.concepts): raise RegistryError("concept_id is not unique across the concept table") if len(self.concept_by_key) != len(self.concepts): raise RegistryError("concept key (system_id, dimension, code) is not unique") for c in self.concepts: if c.system_id not in self.system_by_id: raise RegistryError(f"{c.concept_id} names system {c.system_id!r} not in coding_systems.csv") # crosswalk: known concepts, a resolvable citation. Predicate is enum-checked on load. for m in self.crosswalk: if m.source_concept_id not in self.concept_by_id: raise RegistryError(f"{m.mapping_id} source {m.source_concept_id!r} is not a concept") if m.target_concept_id not in self.concept_by_id: raise RegistryError(f"{m.mapping_id} target {m.target_concept_id!r} is not a concept") if not cite_ok(m.extraction_id): raise RegistryError(f"{m.mapping_id} cites {m.extraction_id!r}, which resolves to no evidence") # preconditions: a real mapping, a known dimension, a value in that dimension's set. for pc in self.preconditions: if pc.mapping_id not in self.mapping_by_id: raise RegistryError(f"{pc.precondition_id} gates unknown mapping {pc.mapping_id!r}") if pc.dimension not in self.dimension_by_id: raise RegistryError(f"{pc.precondition_id} tests unknown dimension {pc.dimension!r}") if pc.operator != "=": raise RegistryError(f"{pc.precondition_id} uses unsupported operator {pc.operator!r}") allowed = DERIVED_VALUE_SETS.get(pc.dimension) if allowed is not None and pc.value_code not in allowed: raise RegistryError( f"{pc.precondition_id} tests {pc.dimension}={pc.value_code!r}, not in {sorted(allowed)}" ) # derivations: known dimensions and systems, a named rule for formula/membership, a # threshold that names a valid band, a resolvable citation. for d in self.derivations: if d.target_dimension not in self.dimension_by_id: raise RegistryError(f"{d.derivation_id} targets unknown dimension {d.target_dimension!r}") if d.source_dimension and d.source_dimension not in self.dimension_by_id: raise RegistryError(f"{d.derivation_id} reads unknown dimension {d.source_dimension!r}") if d.under_system and d.under_system not in self.system_by_id: raise RegistryError(f"{d.derivation_id} is scoped by unknown system {d.under_system!r}") if d.rule_kind in (RuleKind.FORMULA, RuleKind.MEMBERSHIP): if d.formula_id not in NAMED_RULES: raise RegistryError( f"{d.derivation_id} is a {d.rule_kind.value} row naming formula_id " f"{d.formula_id!r}, which this module does not implement" ) if d.rule_kind is RuleKind.FORMULA and not (d.arg_numerator and d.arg_denominator): raise RegistryError(f"{d.derivation_id} is a formula row missing its typed args") if d.rule_kind is RuleKind.THRESHOLD: if not d.operator or d.threshold == "": raise RegistryError(f"{d.derivation_id} is a threshold row missing an operator or threshold") allowed = DERIVED_VALUE_SETS.get(d.target_dimension) if allowed is not None and d.target_concept_id not in allowed: raise RegistryError( f"{d.derivation_id} maps to band {d.target_concept_id!r}, not in {sorted(allowed)}" ) if not cite_ok(d.extraction_id): raise RegistryError(f"{d.derivation_id} cites {d.extraction_id!r}, which resolves to no evidence") def _read_rows(path: Path) -> list[dict]: with path.open(newline="", encoding="utf-8") as fh: return list(csv.DictReader(fh)) def _load_extraction_ids() -> frozenset[str]: return frozenset(r["extraction_id"] for r in _read_rows(EXTRACTIONS_PATH)) def _load_doc_sections() -> frozenset[str]: heading = re.compile(r"^#{2,4}\s+(\d+(?:\.\d+)*)\.?\s") sections: set[str] = set() for line in TAXONOMY_DOC_PATH.read_text(encoding="utf-8").splitlines(): m = heading.match(line) if m: sections.add(m.group(1)) return frozenset(sections) @lru_cache(maxsize=1) def load_registry() -> Registry: """Load and cross-validate the six registry tables. Cached: the CSVs do not change at runtime.""" return Registry( coding_systems=[CodingSystem(**r) for r in _read_rows(CODING_SYSTEMS_PATH)], dimensions=[Dimension(**r) for r in _read_rows(DIMENSIONS_PATH)], concepts=[Concept(**r) for r in _read_rows(CONCEPTS_PATH)], crosswalk=[CrosswalkMapping(**r) for r in _read_rows(CROSSWALK_PATH)], preconditions=[Precondition(**r) for r in _read_rows(PRECONDITION_PATH)], derivations=[Derivation(**r) for r in _read_rows(DERIVATIONS_PATH)], ) # --------------------------------------------------------------------------- resolve / codes def resolve(code: str, system_id: str, dimension: Optional[str] = None) -> Concept: """The concept a code names within a coding system. `figo_endo/1988.IB` and `figo_endo/2009.IB` resolve to different concepts, because a stage code has no meaning without its edition (section 4). Pass `dimension` when a system reuses a code across dimensions.""" reg = load_registry() system_id = _system_id(system_id) matches = [ c for c in reg.concepts if c.system_id == system_id and c.code == code and (dimension is None or c.dimension == dimension) ] if not matches: raise LookupError(f"no concept for code {code!r} in {system_id!r}" + (f"/{dimension}" if dimension else "")) if len(matches) > 1: raise LookupError(f"code {code!r} in {system_id!r} is ambiguous across dimensions; pass a dimension") return matches[0] def codes(system_id: str, dimension: Optional[str] = None) -> list[Concept]: """The concepts a coding system carries, optionally within one dimension.""" reg = load_registry() system_id = _system_id(system_id) return [ c for c in reg.concepts if c.system_id == system_id and (dimension is None or c.dimension == dimension) ] # --------------------------------------------------------------------------- derive def _to_float(value: Any) -> Optional[float]: try: return float(value) except (TypeError, ValueError): return None def _cmp(value: float, operator: str, threshold: float) -> bool: if operator == ">=": return value >= threshold if operator == ">": return value > threshold if operator == "<=": return value <= threshold if operator == "<": return value < threshold if operator == "=": return value == threshold raise RegistryError(f"unsupported threshold operator {operator!r}") def _threshold_category(dimension: str, value: Any, under: str) -> Optional[str]: """Evaluate the threshold bands for a dimension under a coding system. Each band is a lower bound (`>` or `>=` a threshold); the value falls into the band with the greatest satisfied bound. A tie at one threshold prefers the strict `>` band (the narrower one), which is how `none` (>= 0) and `less_than_half` (> 0) share the boundary 0 without overlapping.""" fvalue = _to_float(value) if fvalue is None: return None reg = load_registry() under = _system_id(under) if under else "" best_rank: Optional[tuple[float, int]] = None best_band: Optional[str] = None for d in reg.derivations: if d.rule_kind is not RuleKind.THRESHOLD or d.target_dimension != dimension: continue if (d.under_system or "") != under: continue thr = float(d.threshold) if _cmp(fvalue, d.operator, thr): rank = (thr, 1 if d.operator == ">" else 0) if best_rank is None or rank > best_rank: best_rank, best_band = rank, d.target_concept_id return best_band _INPUT_ALIASES: dict[str, str] = { "foci_count": "lvsi_foci_count", "percent": "myometrial_invasion_percent", "depth": "invasion_depth", "thickness": "myometrial_thickness", "source": "cervical_stromal_invasion", } def derive(dimension: str, *, under: Optional[str] = None, **inputs: Any): """Compute a derived dimension by its registered rule. - A `formula` dimension returns the number its `ratio_percent` binding produces. - A `threshold` dimension returns the band code, `Constrained` over its determinable set when the quantity is absent (a count the report never states), or `Indeterminate`. - A `membership` dimension returns the mapped value, or `Indeterminate` when the source does not determine it (partial by design). Inputs are passed by their source-dimension name; a short alias is also accepted (`foci_count` for `lvsi_foci_count`, `depth`/`thickness` for the ratio args, `source` for a membership input). """ reg = load_registry() rows = [d for d in reg.derivations if d.target_dimension == dimension] if not rows: raise RegistryError(f"{dimension!r} is not a derived dimension in derivations.csv") def pull(name: str) -> Any: if name in inputs: return inputs[name] for alias, target in _INPUT_ALIASES.items(): if target == name and alias in inputs: return inputs[alias] return None kind = rows[0].rule_kind if kind is RuleKind.FORMULA: row = rows[0] rule = NAMED_RULES[row.formula_id] return rule(pull(row.arg_numerator), pull(row.arg_denominator)) if kind is RuleKind.THRESHOLD: source_dim = rows[0].source_dimension value = pull(source_dim) if value is None: value = inputs.get("value") if value is None: determinable = tuple(sorted(DERIVED_VALUE_SETS.get(dimension, frozenset()) - {"negative"})) return Constrained(determinable, f"no {source_dim} stated", dimension) band = _threshold_category(dimension, value, under or "") if band is None: return Indeterminate(f"{dimension} could not be evaluated from {source_dim}={value!r}", dimension) return band # membership row = rows[0] rule = NAMED_RULES[row.formula_id] if row.formula_id == "histotype_aggressiveness": result = rule(pull("histologic_type"), pull("figo_grade")) else: value = pull(row.source_dimension) if value is None: value = inputs.get("value") result = rule(value) if result is None: return Indeterminate(f"{dimension} is not determined by the source value", dimension) return result # --------------------------------------------------------------------------- crosswalk + preconditions def _match(value: Any, operator: str, value_code: str) -> bool: if operator == "=": return value == value_code return False def preconditions_satisfied(mapping_id: str, assertions: Mapping[str, Any]) -> bool: """Rows sharing `(mapping_id, dimension)` are OR'd; across dimensions they are AND'd. A mapping with no preconditions is always satisfied. A dimension the assertions do not carry fails its precondition, so a mapping is never fired on a value the case never asserted.""" reg = load_registry() rows = reg.preconditions_by_mapping.get(mapping_id, []) if not rows: return True by_dim: dict[str, list[Precondition]] = defaultdict(list) for pc in rows: by_dim[pc.dimension].append(pc) for dimension, group in by_dim.items(): value = assertions.get(dimension) if not any(_match(value, pc.operator, pc.value_code) for pc in group): return False return True def crosswalk( source_concept_id: str, target_system: Optional[str] = None, assertions: Optional[Mapping[str, Any]] = None, ) -> list[CrosswalkMapping]: """The mappings from a source concept: optionally restricted to a target system, and optionally to those whose preconditions the assertions satisfy. With no assertions, preconditions are ignored and every mapping from the source is returned.""" reg = load_registry() if target_system is not None: target_system = _system_id(target_system) out: list[CrosswalkMapping] = [] for m in reg.crosswalk: if m.source_concept_id != source_concept_id: continue if target_system is not None: target = reg.concept_by_id.get(m.target_concept_id) if target is None or target.system_id != target_system: continue if assertions is not None and not preconditions_satisfied(m.mapping_id, assertions): continue out.append(m) return out # --------------------------------------------------------------------------- edition inference _EDITION_SYSTEMS = { "1988": "figo_endo/1988", "1995": "figo_endo/1988", # the 1995 revision shares the 1988 corpus sub-staging "2009": "figo_endo/2009", "2023": "figo_endo/2023", "figo_endo/1988": "figo_endo/1988", "figo_endo/2009": "figo_endo/2009", "figo_endo/2023": "figo_endo/2023", } def _system_id(edition: Any) -> str: key = str(edition).strip() if key in _EDITION_SYSTEMS: return _EDITION_SYSTEMS[key] if key in load_registry().system_by_id: return key raise RegistryError(f"unknown coding system or edition {edition!r}") def _normalize_stage_code(code: Any) -> Optional[str]: if code in (None, ""): return None c = str(code).strip().upper().replace("P", "").replace("T1", "I") # "T1B" -> after replace P/T1: "1B" -> "IB"; already-figo "IB" untouched. c = c.replace("1A", "IA").replace("1B", "IB").replace("1C", "IC") return c # --------------------------------------------------------------------------- reported-stage normalization # # A reported `pathologic_stage` value is a free-text string a pathologist wrote: "pT1a", "pT1a, pN0 pMX", # "pT2N0", "1B", "IA", "T1a N1". Stored raw, its confirmed-value distribution counts formatting variants # and combined TNM tokens as distinct categories (the dashboard's "Other (N values)" fold, #165). This # parses the string deterministically into its TNM components and, when no FIGO code is written, the FIGO # stage the local T plus any nodal or metastatic upstaging implies. It never guesses: a string that yields # no component leaves every field None (`parsed` is False), so the caller routes it to `indeterminate` # rather than a wrong category. The verbatim string is kept, so normalization derives, never overwrites # (invariant 2). Reading the scanned page into this string is the model's job; parsing the formal codes out # of it is deterministic work. @dataclass(frozen=True) class ReportedStage: """A reported pathologic-stage string parsed into normalized components, the verbatim string kept.""" verbatim: str t_category: Optional[str] # AJCC pT, lower-cased suffix: "T1a", "T1b", "T2", "T3a", "T3b", "T4" n_category: Optional[str] # nodal: "N0", "N1", "N1mi", "N1a", "N2", "N2mi", "N2a", "NX" m_category: Optional[str] # "M0", "M1", "MX" figo_stage: Optional[str] # normalized FIGO code: "IA", "IB", "II", "IIIA", "IIIC1", "IIIC2", "IVB", ... @property def parsed(self) -> bool: """False when nothing was recognized, so the caller codes the value `indeterminate`, not a guess.""" return any((self.t_category, self.n_category, self.m_category, self.figo_stage)) # TNM tokens. A leading "p" (pathologic) is optional; no trailing boundary, so a run-together "pT2N0" # still yields T2 and N0. The patterns are specific enough (T1-4, N0-2, M0/1/X) that a stage string does # not false-match. Case-insensitive. _T_TOKEN = re.compile(r"P?T([1-4][ABC]?)", re.IGNORECASE) _N_TOKEN = re.compile(r"P?N([0-2](?:MI|[AB])?|X)", re.IGNORECASE) _M_TOKEN = re.compile(r"P?M([01X])", re.IGNORECASE) # A written FIGO stage: Roman (IA..IVB, IIIC1/IIIC2), or arabic shorthand ("1B"). Boundaries keep a lone # "I" inside a word from reading as stage I. _FIGO_ROMAN = re.compile(r"(? Optional[str]: """Lower-case a TNM suffix so a distribution groups on one form: "T1A" -> "T1a", "N1MI" -> "N1mi".""" if token is None: return None return token[:2] + token[2:].lower() def _derive_figo_from_tnm(t: Optional[str], n: Optional[str], m: Optional[str]) -> Optional[str]: """The FIGO stage a reported TNM tuple implies, most-advanced component first: a distant metastasis is IVB, a positive node is IIIC (C1 pelvic, C2 para-aortic), otherwise the local pT correspondence.""" if m == "M1": return "IVB" if n and n.upper().startswith("N1"): return "IIIC1" if n and n.upper().startswith("N2"): return "IIIC2" if t: return _T_LOCAL_FIGO.get(t.upper()) return None def parse_reported_stage(raw: Any) -> ReportedStage: """Parse a reported pathologic-stage string into normalized TNM components and a FIGO stage (#165). An explicit FIGO code in the string wins; otherwise the FIGO stage is derived from the TNM tuple by `_derive_figo_from_tnm`. A string with no recognizable component returns an all-None `ReportedStage` (`parsed` is False), never a guessed category. """ verbatim = "" if raw is None else str(raw) s = verbatim.upper() def _first(rx: "re.Pattern[str]", prefix: str) -> Optional[str]: match = rx.search(s) return _canonical_tnm(f"{prefix}{match.group(1)}") if match else None t = _first(_T_TOKEN, "T") n = _first(_N_TOKEN, "N") m = _first(_M_TOKEN, "M") figo: Optional[str] = None roman = _FIGO_ROMAN.search(s) if roman: figo = roman.group(1) + (roman.group(2) or "") + (roman.group(3) or "") else: arabic = _FIGO_ARABIC.search(s) if arabic: figo = _ARABIC_TO_ROMAN[arabic.group(1)] + arabic.group(2) if figo is None: figo = _derive_figo_from_tnm(t, n, m) return ReportedStage(verbatim=verbatim, t_category=t, n_category=n, m_category=m, figo_stage=figo) # A reported nodal string's positive/examined count, e.g. "0/17" or "0 of 16". _NODE_COUNT = re.compile(r"(\d+)\s*(?:/|\bOF\b)\s*\d+") def parse_reported_nodal(raw: Any) -> Optional[str]: """Normalize a reported regional-lymph-node value to a `pn_category` code (#165), or None. An explicit pN code in the string is taken as written. "pNX" or "cannot be assessed" is `pNX`. A zero positive count ("0/17 lymph nodes positive") or prose stating no nodal disease ("No regional lymph node metastasis") is `pN0`. A positive count returns None rather than a category, because `pN1` (pelvic) versus `pN2` (para-aortic) needs the anatomic location the count alone does not carry: the honest failure is `indeterminate`, not an invented category. """ if raw is None: return None s = str(raw).upper() token = _N_TOKEN.search(s) if token: return "p" + str(_canonical_tnm(f"N{token.group(1)}")) if "CANNOT" in s or "NOT ASSESS" in s: return "pNX" count = _NODE_COUNT.search(s) if count: return "pN0" if int(count.group(1)) == 0 else None # Prose that states no nodal disease and no positive node: "All lymph nodes negative", "No regional # lymph node metastasis". "POSITIVE" absent guards against a mixed report ("... 1/3 positive"). if "POSITIVE" not in s and ("NEGATIVE" in s or ("NODE" in s and "NO " in s)): return "pN0" return None def _normalize_extent(extent: Any) -> Optional[str]: if extent in (None, ""): return None e = str(extent).strip().lower() if e in ("less_than_half", "none") or "less than" in e or "less-than" in e or "inner" in e: return "less_than_half" if e in ("half_or_more",) or "half or more" in e or "more than half" in e or "equal to or more" in e or "outer" in e: return "half_or_more" return None def infer_edition(extent: Any = None, code: Any = None) -> Optional[str]: """Infer the FIGO edition of a stage code from the extent phrase beside it (docs section 4, 5.3). `T1b`/`IB` denote opposite extents across editions: a report saying "less than one-half" beside `T1b` is FIGO 1988, because only 1988 labels less-than-half invasion `IB`. `IC` and `IIIC1`/`IIIC2` pin the edition by themselves. Returns a system id, or None when the pair does not determine one.""" c = _normalize_stage_code(code) if c == "IC": return "figo_endo/1988" if c in ("IIIC1", "IIIC2"): return "figo_endo/2009" e = _normalize_extent(extent) if e == "less_than_half": if c == "IB": return "figo_endo/1988" if c == "IA": return "figo_endo/2009" if e == "half_or_more": if c == "IC": return "figo_endo/1988" if c == "IB": return "figo_endo/2009" return None # --------------------------------------------------------------------------- derive_stage _STAGE_I_OR_II = frozenset({"IA", "IA1", "IA2", "IA3", "IB", "IC", "II", "IIA", "IIB", "IIC"}) _ANATOMIC_2023 = {"none": "IA1", "less_than_half": "IA2", "half_or_more": "IB"} def _node_positive(pn: Any) -> bool: return pn in ("pN1mi", "pN1a", "pN2mi", "pN2a") def _iiic(pn: Any) -> str: return "IIIC2" if pn in ("pN2mi", "pN2a") else "IIIC1" def _myo_category(a: Mapping[str, Any]) -> Optional[str]: if a.get("myometrial_invasion_category") is not None: return a["myometrial_invasion_category"] pct = a.get("myometrial_invasion_percent") if pct is None and a.get("invasion_depth") is not None and a.get("myometrial_thickness") is not None: pct = ratio_percent(a["invasion_depth"], a["myometrial_thickness"]) if pct is not None: return _threshold_category("myometrial_invasion_category", pct, "") return None def _lvsi_extent(a: Mapping[str, Any], edition: str) -> Optional[str]: if a.get("lvsi_extent") is not None: return a["lvsi_extent"] foci = a.get("lvsi_foci_count") if foci is not None: return _threshold_category("lvsi_extent", foci, edition) status = a.get("lvsi_status") if status in ("not_identified", "negative", "absent"): return "negative" if status == "present": return UNQUANTIFIED return None # unstated: treated as no/focal LVSI by the caller def _aggressiveness(a: Mapping[str, Any]) -> Optional[str]: if a.get("histotype_aggressiveness") is not None: return a["histotype_aggressiveness"] if a.get("histologic_type") is not None: return histotype_aggressiveness(a["histologic_type"], a.get("figo_grade")) return None def _molecular_modifier(a: Mapping[str, Any]) -> Optional[str]: """The FIGO 2023 molecular modifier, from a confirmed ProMisE class only (dec_019). A raw `tcga_class` of `copy_number_high` does not upstage on its own: the copy-number-high to p53abn surrogate carries a ~5% error, so the p53abn stage is offered as a candidate a reviewer accepts, never applied automatically.""" pc = a.get("promise_class") if pc in ("pole_mutated", "polemut"): return "POLEmut" if pc in ("p53_abnormal", "p53abn"): return "p53abn" return None def _stage_1988_2009(a: Mapping[str, Any], edition: str): pn = a.get("pn_category") if _node_positive(pn): return Stage("IIIC" if edition == "figo_endo/1988" else _iiic(pn), edition) cervical = a.get("cervical_stromal_invasion") if cervical == "cannot_be_assessed": return Indeterminate("cervical stromal invasion cannot be assessed", "figo_stage", edition) if cervical == "present": return Stage("II", edition) cat = _myo_category(a) if cat is None: return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition) if edition == "figo_endo/1988": return Stage({"none": "IA", "less_than_half": "IB", "half_or_more": "IC"}[cat], edition) return Stage("IA" if cat in ("none", "less_than_half") else "IB", edition) def _stage_2023(a: Mapping[str, Any], edition: str): pn = a.get("pn_category") if _node_positive(pn): # The i/ii micrometastasis/macrometastasis split is unmeasurable in this corpus; the base # IIIC1/IIIC2 is determined by which nodal group is involved. return Stage(_iiic(pn), edition) cervical = a.get("cervical_stromal_invasion") if cervical == "cannot_be_assessed": return Indeterminate("cervical stromal invasion cannot be assessed", "figo_stage", edition) aggr = _aggressiveness(a) if aggr is None or aggr == "indeterminate": return Indeterminate("histotype aggressiveness cannot be determined (grade unknown)", "figo_stage", edition) cat = _myo_category(a) if aggr == "aggressive": if cat is None and cervical != "present": return Indeterminate("myometrial invasion presence unknown for an aggressive histotype", "figo_stage", edition) any_mi = cat is not None and cat != "none" base = Stage("IIC" if (any_mi or cervical == "present") else "IC", edition) else: # non-aggressive if cervical == "present": base = Stage("IIA", edition) else: extent = _lvsi_extent(a, edition) if extent == "substantial": base = Stage("IIB", edition) elif extent == UNQUANTIFIED: if cat is None: return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition) anatomic = _ANATOMIC_2023[cat] return Constrained( (anatomic, "IIB"), f"LVSI is present but unquantified: focal keeps {anatomic}, substantial raises to IIB", "figo_stage", edition, ) else: # negative, focal, or unstated (treated as no/focal) if cat is None: return Indeterminate("myometrial invasion category is unknown", "figo_stage", edition) base = Stage(_ANATOMIC_2023[cat], edition) modifier = _molecular_modifier(a) if isinstance(base, Stage) and base.code in _STAGE_I_OR_II and modifier: if modifier == "POLEmut": return Stage("IAmPOLEmut", edition, modifier="POLEmut") if modifier == "p53abn": return Stage("IICmp53abn", edition, modifier="p53abn") return base def derive_stage(assertions: Mapping[str, Any], edition: Any): """Project a FIGO stage from canonical assertions, never from another edition's stage label. Returns a `Stage`, a `Constrained(candidates, reason)` when the assertions narrow the answer to a set and no further (the IB-versus-IIB pair when LVSI is present but unquantified), or an `Indeterminate(reason)` when a required assertion is missing or the report states a value cannot be assessed. An unstated cervical or nodal finding is read as not involved, the standard reading that a pathology report states positive findings; a value the report explicitly marks unassessable is `Indeterminate`, never assumed negative. """ edition = _system_id(edition) if edition == "figo_endo/2023": return _stage_2023(assertions, edition) return _stage_1988_2009(assertions, edition) def project(assertions: Mapping[str, Any], dimension: str = "figo_stage", under: Any = None): """Project a derived dimension into a target convention. For `figo_stage` this is `derive_stage` under the target edition; for any other derived dimension it is `derive` under the target system. Returns a value, `Stage`, `Constrained`, or `Indeterminate`.""" if dimension == "figo_stage": return derive_stage(assertions, under) return derive(dimension, under=(str(under) if under is not None else None), **dict(assertions)) # --------------------------------------------------------------------------- B1 interface def _category_from_percent(percent: Any) -> Optional[str]: if _to_float(percent) is None: return None return _threshold_category("myometrial_invasion_category", percent, "") # concept_id -> the recompute B1 (`endopath.validation`) calls, positional in dependency-graph order. # Only the rules a per-case record can recompute from its graph inputs alone appear here. `lvsi_extent` # (needs a coding system) and `figo_stage` (needs an edition) are projected through `derive`/ # `derive_stage` with that context, not recomputed in the editionless per-case pass. RECOMPUTE_BY_CONCEPT: dict[str, Callable[..., Optional[Any]]] = { "myometrial_invasion_percent": lambda depth, thickness: ratio_percent(depth, thickness), "cervical_wall_percent": lambda depth, thickness: ratio_percent(depth, thickness), "myometrial_invasion_category": _category_from_percent, "grade_binary": grade_binary, "histotype_aggressiveness": histotype_aggressiveness, } def recompute(concept_id: str, inputs: Sequence[Any]) -> Optional[Any]: """B1's clean interface: recompute a derived concept from its dependency-graph inputs, positional. Returns the recomputed value, or None when this module does not recompute the concept in an editionless per-case pass (`lvsi_extent`, `figo_stage`).""" fn = RECOMPUTE_BY_CONCEPT.get(concept_id) if fn is None: return None return fn(*inputs) # --------------------------------------------------------------------------- determinability (section 9) @dataclass(frozen=True) class DeterminabilityRow: code: str inputs_beyond_anatomy: str corpus_availability: str verdict: str # The section-9 matrix. Emitted by scripts/report_determinability.py, not hand-written into a doc. # The verdict is a function of what a FIGO 2023 code needs beyond anatomy and whether the corpus # supplies it: `determined` (anatomy and histology suffice), `determined_by_join` (needs the TCGA # molecular join), `constrained` (the projector enumerates candidates and states why), `unmeasurable`. _DETERMINABILITY: tuple[DeterminabilityRow, ...] = ( DeterminabilityRow("III*/IV* (except IIIC*i/ii)", "none", "high", "determined"), DeterminabilityRow("IIIC1i/ii, IIIC2i/ii", "micro vs macro deposit", "0.4% / 0%", "unmeasurable"), DeterminabilityRow("IA1", "non-aggressive, confined to endometrium/polyp", "grade 82.6%", "determined"), DeterminabilityRow("IA2", "non-aggressive, MI<50%, no/focal LVSI", "MI 14.8%; LVSI 85.5%", "determined_or_constrained"), DeterminabilityRow("IA3", "low-grade endometrioid, uterus and ovary", "ovary/adnexa 96.5%", "determined"), DeterminabilityRow("IB", "non-aggressive, MI>=50%, no/focal LVSI", "as above", "determined_or_constrained"), DeterminabilityRow("IC", "aggressive histotype, no MI", "grade + histotype", "determined"), DeterminabilityRow("IIA", "non-aggressive, cervical stromal invasion", "cervical stroma 11.5%", "determined"), DeterminabilityRow("IIB", "non-aggressive, substantial LVSI", "foci count 0", "constrained"), DeterminabilityRow("IIC", "aggressive, any MI", "grade + histotype", "determined"), DeterminabilityRow("IAmPOLEmut", "POLEmut, stage I/II", "49 cases via join", "determined_by_join"), DeterminabilityRow("IICmp53abn", "p53abn, stage I/II", "162 cases via join", "determined_by_join"), ) def determinability_matrix() -> tuple[DeterminabilityRow, ...]: """The section-9 determinability matrix: one verdict per FIGO 2023 code group. `IB` against `IIB` is the only genuinely constrained pair, and only where LVSI is present but unquantified.""" return _DETERMINABILITY