From 7008153083cacf5f0f8b972e4b592cbdb42b45d9 Mon Sep 17 00:00:00 2001 From: Amrit Krishnan Date: Tue, 25 Aug 2026 08:31:09 -0400 Subject: [PATCH] fix(concepts): complete KDIGO AKI staging with the RRT and urine-output legs The acute_kidney_injury / aki_stage_2 / aki_stage_3 concepts were creatinine-only. KDIGO stages AKI on three legs, and two were missing entirely -- both flagged "not implemented -- see 'still open'" in the concept docstrings. An incomplete gold standard is a correctness problem wherever the label is used, and it is used twice: as the concept bottleneck's supervised target and as the alert-event hazard head's onset definition. 1. Renal replacement therapy is an automatic Stage 3, whatever creatinine and urine output say. Added as one more OR-branch of aki_stage_3, not a separate named concept: the clinical fact is "RRT implies Stage 3", not a concept the bottleneck needs its own head for. The item ids are mimic-code's own rrt.sql dialysis_active = 1 procedureevents set (225441 IHD, 225802 CRRT, 225803 CVVHD, 225805 peritoneal, 225809 CVVHDF, 225955 SCUF) and its two deliberate exclusions are excluded here too (224270 dialysis catheter placement, 225436 CRRT filter change). Matched with the existing CodeOccurrenceRule machinery, whose first-occurrence semantics are exactly what an onset needs, over an alternation anchored to the PROCEDURE//START//{itemid} code shape that SOFA_SOURCE_CONFIG's ventilation ids already use. 2. Urine output stages AKI independently of creatinine. New sofa.urine_output_rate generalizes urine_output_24h to an arbitrary trailing window and to KDIGO's own weight-normalized mL/kg/h form; urine_output_24h is now its 24 h absolute-volume special case, so SOFA's renal component and the oliguria concept are untouched. New DerivedUrineRateRule / CanonicalUrineRate express the three stage criteria declaratively, like every other rule in the file: Stage 1 < 0.5 mL/kg/h over 6 h, Stage 2 < 0.5 over 12 h, Stage 3 < 0.3 over 24 h OR 0 mL over 12 h (true anuria, absolute volume, so it needs no weight at all). Weight is the real constraint and is handled by abstaining, not guessing. Only ~10-17% of subjects in the MIMIC-IV extraction have any weight reading, so each window is asof-joined backward to the most recent daily weight (224639) at or before it, falling back to admission weight (226512) before any daily weight exists -- daily weight is the current weight a mL/kg/h rate means, admission weight is the best available early in a stay. A window with neither, or a non-positive charted weight, is dropped rather than scored against an invented default: it stays unobserved (masked out of supervision) instead of becoming a silent negative, the same observability convention sepsis3 uses. Thresholds follow the existing sign conventions: the rate criteria are strict "below" because KDIGO says "less than 0.5 mL/kg/h" (exactly 0.5 does not stage), and anuria is "at_or_below" 0 mL, mirroring Stage 3's existing at_or_above 4.0 creatinine trigger. No experiment-registry row: this changes the label definition, so it only gets one once something is retrained/rescored against it. Tests: tests/odyssey/data/test_aki_kdigo.py, 29 cases -- each of the six RRT item ids individually, both exclusions, pattern anchoring, first- occurrence onsets, OR composition with the creatinine legs, mL/kg/h arithmetic at all three windows, partial-window exclusion, the weight fallback order (both weights / admission only / daily only / none / later only / non-positive), the unobserved-not-negative contract, threshold boundaries, the weight-free anuria branch, a mixed five-subject cohort, the per-source expansion, and the source-config contract of the two urine-output forms. Coverage on the two touched modules: concepts.py 98%, sofa.py 97% (every uncovered line pre-existing and defensive). Co-Authored-By: Claude Sonnet 5 --- README.md | 2 +- odyssey/data/concepts.py | 234 +++++++++- odyssey/data/sofa.py | 164 ++++++- tests/odyssey/data/test_aki_kdigo.py | 562 +++++++++++++++++++++++ tests/odyssey/data/test_value_binning.py | 33 +- 5 files changed, 948 insertions(+), 47 deletions(-) create mode 100644 tests/odyssey/data/test_aki_kdigo.py diff --git a/README.md b/README.md index f21c47d..05b3f30 100644 --- a/README.md +++ b/README.md @@ -228,7 +228,7 @@ MEDS extraction validated end to end on credentialed MIMIC-IV 3.1 and eICU-CRD 2 ### Known concept-rule limitations -Concept labels are rule-derived, per visit, and evaluated over a visit's whole window (did this happen during the visit), with each concept's first-trigger time also recorded so a running "true as of now" label exists for interventions. Sustained/windowed criteria are used where a single reading over-triggers (`sustained_tachypnea`, KDIGO creatinine windows); GCS-dependent criteria are unavailable on eICU until its nurse-charting table is extracted; urine-output-based AKI staging and full SOFA/NEWS2 are not implemented. +Concept labels are rule-derived, per visit, and evaluated over a visit's whole window (did this happen during the visit), with each concept's first-trigger time also recorded so a running "true as of now" label exists for interventions. Sustained/windowed criteria are used where a single reading over-triggers (`sustained_tachypnea`, KDIGO creatinine windows); GCS-dependent criteria are unavailable on eICU until its nurse-charting table is extracted; full SOFA/NEWS2 as ordinal point-scale concepts are not implemented. AKI staging now covers all three KDIGO legs: creatinine, renal-replacement-therapy initiation (an automatic Stage 3, on mimic-code `rrt.sql`'s active-dialysis itemids), and urine output as a weight-normalized rate (< 0.5 mL/kg/h over 6h / 12h for Stages 1 / 2, < 0.3 mL/kg/h over 24h or 12h of anuria for Stage 3). The rate legs need a charted body weight (daily weight preferred, admission weight as the early-stay fallback) and are left unscored, never defaulted, where none exists -- only ~10-17% of subjects have any weight reading in the MIMIC-IV extraction, so on most patient-positions the rate legs abstain and the creatinine/RRT/anuria legs carry the label. ### GPU notes diff --git a/odyssey/data/concepts.py b/odyssey/data/concepts.py index 0cc476f..8fec653 100644 --- a/odyssey/data/concepts.py +++ b/odyssey/data/concepts.py @@ -45,6 +45,15 @@ readings within ``max_component_gap_minutes`` (GCS components are typically charted together in practice) and summing. +Rule types added since, each alongside the concept that needed it rather +than to that original list: :class:`CodeOccurrenceRule` (the event's +occurrence is the whole signal, no numeric value -- vasopressor +administration, renal-replacement-therapy initiation), +:class:`DerivedSofaSignalRule` and :class:`Sepsis3Rule` (signals derived +by :mod:`odyssey.data.sofa`), and :class:`DerivedUrineRateRule` (KDIGO's +urine-output leg: mL/kg/h over a 6/12/24 h trailing window, or absolute +mL for the anuria branch). + :class:`CompositeConceptDefinition` combines several of the above into an "N of M criteria" concept (SIRS/qSOFA-style), optionally nesting :class:`AnyOf` where one criterion is itself satisfied by any of several @@ -76,6 +85,7 @@ sofa_supported, sofa_timeseries, urine_output_24h, + urine_output_rate, ) @@ -282,6 +292,61 @@ class DerivedSofaSignalRule: source: str +@dataclass(frozen=True) +class DerivedUrineRateRule: + """KDIGO's urine-output leg: trailing urine output crosses a threshold. + + :class:`DerivedSofaSignalRule`'s ``urine_24h`` signal is fixed at + SOFA's own shape (absolute mL, 24 h). KDIGO stages AKI on a + *weight-normalized rate* (mL/kg/h) over three different windows + (6 h, 12 h, 24 h), so the window and the normalization are rule + parameters here, resolved by + :func:`~odyssey.data.sofa.urine_output_rate`: + + - ``weight_normalized=True``: ``value`` is mL/kg/h, and a window is + scored only where a body weight was charted at or before it (daily + weight preferred, admission weight as the early-stay fallback). + Windows with no weight at all are not scored -- see that function + for why defaulting a weight would be worse than abstaining. + - ``weight_normalized=False``: ``value`` is absolute mL over the + window, which is what Stage 3's anuria branch (0 mL over 12 h) + needs -- 0 mL is 0 mL at any body weight, so that branch stays + assessable for the majority of keys that have no weight reading. + + ``source`` is fixed at expansion time for the same reason as + :class:`DerivedSofaSignalRule`: the derivation needs that source's + non-LOINC weight item ids. + """ + + threshold: float + direction: Direction + window_hours: float + source: str + weight_normalized: bool = True + + +# Renal replacement therapy: KDIGO makes RRT initiation an automatic AKI +# Stage 3, whatever creatinine and urine output say. The item ids are +# mimic-code's own ``mimic-iv/concepts/treatment/rrt.sql`` +# ``dialysis_active = 1`` procedureevents set, and its two deliberate +# exclusions are excluded here too: 224270 (Dialysis Catheter -- line +# placement, not therapy) and 225436 (CRRT Filter Change -- maintenance on +# an already-active line, which would double-count an episode already +# caught by its own START row). The codes take the same +# ``PROCEDURE//START//{itemid}`` shape as the ventilation item ids in +# :data:`odyssey.data.sofa.SOFA_SOURCE_CONFIG`; the alternation is anchored +# so a longer item id that merely begins with one of these cannot match. +RRT_ITEMIDS: Tuple[str, ...] = ( + "225441", # Hemodialysis (intermittent, IHD) + "225802", # Dialysis - CRRT + "225803", # Dialysis - CVVHD + "225805", # Peritoneal Dialysis + "225809", # Dialysis - CVVHDF + "225955", # Dialysis - SCUF +) +RRT_CODE_PATTERN = r"^PROCEDURE//START//(" + "|".join(RRT_ITEMIDS) + r")(//|$)" + + @dataclass(frozen=True) class Sepsis3Rule: """Sepsis-3 onset (Singer 2016) as operationalized by mimic-code's ``sepsis3``. @@ -320,6 +385,7 @@ class Sepsis3Rule: CodeOccurrenceRule, Sepsis3Rule, DerivedSofaSignalRule, + DerivedUrineRateRule, ] @@ -476,6 +542,24 @@ class CanonicalSofaSignal: direction: Direction = "below" +@dataclass(frozen=True) +class CanonicalUrineRate: + """Canonical form of :class:`DerivedUrineRateRule` (source-agnostic). + + The urine LOINC itself resolves through the mapping layer inside + :func:`~odyssey.data.sofa.urine_output_rate`; what is source-specific + is only the weight item ids, so this rule -- like + :class:`CanonicalSofaSignal` -- expands wherever + :func:`~odyssey.data.sofa.sofa_supported` holds and is dropped + elsewhere. + """ + + threshold: float + window_hours: float + direction: Direction = "below" + weight_normalized: bool = True + + @dataclass(frozen=True) class CanonicalSepsis3: """Canonical Sepsis-3: expands to :class:`Sepsis3Rule` where SOFA is scorable.""" @@ -491,6 +575,7 @@ class CanonicalSepsis3: CodeOccurrenceRule, CanonicalSepsis3, CanonicalSofaSignal, + CanonicalUrineRate, ] @@ -551,7 +636,7 @@ def _prefix_threshold(rule: LoincThreshold, prefix: str, source: str) -> float: ) -def _expand_non_loinc( +def _expand_non_loinc( # noqa: PLR0911 rule: CanonicalRule, source: str ) -> Optional[List[ComponentRule]]: """Expand the non-LOINC-keyed rules; ``None`` when ``rule`` is LOINC-keyed.""" @@ -568,6 +653,18 @@ def _expand_non_loinc( source=source, ) ] + if isinstance(rule, CanonicalUrineRate): + if not sofa_supported(source): + return [] + return [ + DerivedUrineRateRule( + threshold=rule.threshold, + direction=rule.direction, + window_hours=rule.window_hours, + weight_normalized=rule.weight_normalized, + source=source, + ) + ] if isinstance(rule, CanonicalSepsis3): # Needs vasopressor rates and ventilation intervals (SOFA's # non-LOINC ingredients) and the microbiology sidecar: MIMIC-IV @@ -606,7 +703,13 @@ def _expand_rule(rule: CanonicalRule, source: str) -> List[ComponentRule]: ) ] assert not isinstance( # for mypy - rule, (CodeOccurrenceRule, CanonicalSepsis3, CanonicalSofaSignal) + rule, + ( + CodeOccurrenceRule, + CanonicalSepsis3, + CanonicalSofaSignal, + CanonicalUrineRate, + ), ) prefixes = _loinc_prefixes(rule.loincs, source) if isinstance(rule, LoincThreshold): @@ -871,13 +974,22 @@ def concepts_for_source( LoincBaselineRelative( _CREATININE, ratio=1.5, direction="above", window_hours=168.0 ), + # KDIGO's urine leg: < 0.5 mL/kg/h for 6-12h. Evaluated at the + # 6h lower bound, since every rule here is a single-instant + # check rather than a multi-criteria interval test: a window + # that stays under the rate past 12h has already triggered at + # 6h, and Stage 2 is the >= 12h escalation of the same rate. + CanonicalUrineRate(threshold=0.5, window_hours=6.0), ), - "KDIGO AKI Stage 1 (either trigger): serum creatinine rose by >= 0.3 " + "KDIGO AKI Stage 1 (any trigger): serum creatinine rose by >= 0.3 " "mg/dL within 48 hours, OR rose to >= 1.5x an earlier reading within " - "7 days (168h). Replaces v1's 'creatinine > 1.5 mg/dL' single-value " - "proxy, which ignored a patient's own baseline. See aki_stage_2 and " - "aki_stage_3 for higher severity; urine-output-based staging is not " - "implemented -- see 'still open'.", + "7 days (168h), OR urine output under 0.5 mL/kg/h over a trailing 6 " + "hours. Replaces v1's 'creatinine > 1.5 mg/dL' single-value proxy, " + "which ignored a patient's own baseline. See aki_stage_2 and " + "aki_stage_3 for higher severity. The urine leg needs a charted body " + "weight at or before the window (daily weight preferred, admission " + "weight as the early-stay fallback) and is simply not scored without " + "one, so it adds triggers without ever adding a silent negative.", ), CanonicalConcept( "aki_stage_2", @@ -885,10 +997,12 @@ def concepts_for_source( LoincBaselineRelative( _CREATININE, ratio=2.0, direction="above", window_hours=168.0 ), + CanonicalUrineRate(threshold=0.5, window_hours=12.0), ), - "KDIGO AKI Stage 2: serum creatinine rose to >= 2.0x an earlier " - "reading within 7 days. Urine-output-based staging (<0.5 mL/kg/h for " - ">= 12h) is not implemented -- see 'still open'.", + "KDIGO AKI Stage 2 (either trigger): serum creatinine rose to >= 2.0x " + "an earlier reading within 7 days, OR urine output under 0.5 mL/kg/h " + "over a trailing 12 hours (the same rate as Stage 1, sustained twice " + "as long).", ), CanonicalConcept( "aki_stage_3", @@ -898,12 +1012,30 @@ def concepts_for_source( ), # KDIGO: >= 4.0, inclusive LoincThreshold(_CREATININE, "at_or_above", 4.0), + # RRT initiation is an automatic Stage 3 in KDIGO, whatever + # creatinine and urine output say (mimic-code's rrt.sql item ids; + # see RRT_CODE_PATTERN). First occurrence only, which is what + # CodeOccurrenceRule already reports. + CodeOccurrenceRule(RRT_CODE_PATTERN, observed_families=("PROCEDURE",)), + CanonicalUrineRate(threshold=0.3, window_hours=24.0), + # Anuria: 0 mL over 12h. Absolute volume, not a rate, so it needs + # no body weight (0 mL is 0 mL at any weight) and stays + # assessable for the many keys with no weight charted. + CanonicalUrineRate( + threshold=0.0, + window_hours=12.0, + direction="at_or_below", + weight_normalized=False, + ), ), - "KDIGO AKI Stage 3 (either trigger): serum creatinine rose to " - ">= 3.0x an earlier reading within 7 days, OR any reading >= 4.0 " - "mg/dL. Renal-replacement-therapy initiation and urine-output-based " - "staging (<0.3 mL/kg/h for >= 24h, or anuria for >= 12h) are not " - "implemented -- see 'still open'.", + "KDIGO AKI Stage 3 (any trigger): serum creatinine rose to >= 3.0x an " + "earlier reading within 7 days, OR any reading >= 4.0 mg/dL, OR " + "renal-replacement therapy was initiated (hemodialysis, CRRT, CVVHD, " + "CVVHDF, SCUF or peritoneal dialysis -- an automatic Stage 3 in " + "KDIGO, independent of creatinine and urine output), OR urine output " + "under 0.3 mL/kg/h over a trailing 24 hours, OR anuria (0 mL) over a " + "trailing 12 hours. The rate leg needs a charted body weight and is " + "not scored without one; the anuria and RRT legs need no weight.", ), CanonicalComposite( "sirs", @@ -1380,6 +1512,15 @@ def _component_ids( # noqa: PLR0911 value_col=value_col, time_col=time_col, ) + if isinstance(rule, DerivedUrineRateRule): + return _urine_rate_ids( + events, + rule, + subject_id_col=subject_id_col, + code_col=code_col, + value_col=value_col, + time_col=time_col, + ) if isinstance(rule, Sepsis3Rule): return _sepsis3_ids( events, @@ -1392,6 +1533,28 @@ def _component_ids( # noqa: PLR0911 raise TypeError(f"unknown component rule type: {type(rule)!r}") +def _derived_reading_ids( + readings: pl.DataFrame, + threshold: float, + direction: Direction, + *, + subject_id_col: str, + time_col: str, +) -> Tuple[Set[int], FirstTimes]: + """Observed keys and first-trigger times of a derived (key, time, value) frame. + + Observed = the keys the derivation could actually be evaluated for + (it emits no row where a needed ingredient is missing: no pairable + FiO2, no full trailing window, no body weight), so an absent key is + "not assessable", never a silent negative. + """ + observed = set(readings[subject_id_col].to_list()) + if readings.height == 0: + return observed, {} + fired = readings.filter(_threshold_expr(pl.col("value"), threshold, direction)) + return observed, _first_times(fired, subject_id_col, time_col) + + def _sofa_signal_ids( events: pl.DataFrame, rule: DerivedSofaSignalRule, @@ -1410,13 +1573,42 @@ def _sofa_signal_ids( value_col=value_col, time_col=time_col, ) - observed = set(readings[subject_id_col].to_list()) - if readings.height == 0: - return observed, {} - fired = readings.filter( - _threshold_expr(pl.col("value"), rule.threshold, rule.direction) + return _derived_reading_ids( + readings, + rule.threshold, + rule.direction, + subject_id_col=subject_id_col, + time_col=time_col, + ) + + +def _urine_rate_ids( + events: pl.DataFrame, + rule: DerivedUrineRateRule, + *, + subject_id_col: str, + code_col: str, + value_col: str, + time_col: str, +) -> Tuple[Set[int], FirstTimes]: + """Observed keys and first-trigger times for KDIGO's urine-output leg.""" + readings = urine_output_rate( + events, + source=rule.source, + key=subject_id_col, + code_col=code_col, + value_col=value_col, + time_col=time_col, + window_hours=rule.window_hours, + weight_normalized=rule.weight_normalized, + ) + return _derived_reading_ids( + readings, + rule.threshold, + rule.direction, + subject_id_col=subject_id_col, + time_col=time_col, ) - return observed, _first_times(fired, subject_id_col, time_col) _SIDECAR_WARNED: Set[str] = set() diff --git a/odyssey/data/sofa.py b/odyssey/data/sofa.py index 6a57f4c..2c77b91 100644 --- a/odyssey/data/sofa.py +++ b/odyssey/data/sofa.py @@ -27,10 +27,18 @@ once the key has at least 24 h of record behind it (a partial window would read as oliguria). -Vasopressor item ids and ventilation procedure ids are not LOINC-keyed and -live in a per-source table here (:data:`SOFA_SOURCE_CONFIG`); the -numeric-signal codes resolve through the shared LOINC tables. Only sources -with an entry can be scored (MIMIC-IV today). +Vasopressor item ids, ventilation procedure ids and body-weight item ids +are not LOINC-keyed and live in a per-source table here +(:data:`SOFA_SOURCE_CONFIG`); the numeric-signal codes resolve through the +shared LOINC tables. Only sources with an entry can be scored (MIMIC-IV +today). + +Beyond SOFA itself, :func:`urine_output_rate` generalizes the renal +component's trailing urine sum to an arbitrary window and, optionally, to +KDIGO's own weight-normalized mL/kg/h form -- the shape +:mod:`odyssey.data.concepts`' AKI staging needs. +:func:`urine_output_24h` is now the (unchanged) 24 h absolute-volume +special case of it. """ import functools @@ -67,6 +75,18 @@ class SofaSourceConfig: infusion_end_prefix: str ventilation_start: Tuple[str, ...] ventilation_end: Tuple[str, ...] + daily_weight: Tuple[str, ...] = () + """Code prefixes of the recurring charted body weight, in kg. + + Body weight is only needed by :func:`urine_output_rate` (KDIGO's + mL/kg/h form), never by SOFA itself, so a source may leave both + weight fields empty: the rate criterion is then simply unassessable + there, exactly as it is for a key with no weight charted. + """ + admission_weight: Tuple[str, ...] = () + """Code prefixes of the once-per-stay admission weight, in kg (the + fallback :func:`urine_output_rate` uses before any recurring weight + has been charted).""" SOFA_SOURCE_CONFIG: Dict[str, SofaSourceConfig] = { @@ -79,6 +99,14 @@ class SofaSourceConfig: infusion_end_prefix="INFUSION_END//", ventilation_start=("PROCEDURE//START//225792", "PROCEDURE//START//225794"), ventilation_end=("PROCEDURE//END//225792", "PROCEDURE//END//225794"), + # chartevents 224639 "Daily Weight" / 226512 "Admission Weight (Kg)", + # both charted in kg. Deliberately NOT routed through the LOINC layer: + # both carry the same body-weight LOINC, so a LOINC lookup could not + # keep them apart, and urine_output_rate's fallback order (daily + # first, admission only until a daily weight exists) needs exactly + # that distinction. + daily_weight=("LAB//224639//",), + admission_weight=("LAB//226512//",), ), } @@ -567,7 +595,8 @@ def num(loincs: Sequence[str]) -> pl.DataFrame: return pf.select(key, time_col, "value", "ventilated") -def urine_output_24h( +@_quiet_asof +def urine_output_rate( events: pl.DataFrame, *, source: str = "mimic_iv", @@ -575,15 +604,52 @@ def urine_output_24h( code_col: str = "code", value_col: str = "numeric_value", time_col: str = "time", + window_hours: float = WINDOW_HOURS, + weight_normalized: bool = True, ) -> pl.DataFrame: - """(key, time, value=mL voided in the trailing 24h) per urine-output reading. - - Only at times with at least :data:`WINDOW_HOURS` of record behind them: - a partial window sums less urine simply because less time has passed, - which would read as oliguria. Weight-normalized rates - (mL/kg/h, KDIGO's own form) are not used -- weight is not reliably - present per key in the extractions -- so this is the absolute-volume - form SOFA's renal component uses. + """Trailing urine output per reading: (key, time, value). + + ``value`` is millilitres per kilogram per hour (KDIGO's own form) + when ``weight_normalized``, and plain millilitres over the window + otherwise (SOFA's renal component and the ``oliguria`` concept, and + KDIGO Stage 3's anuria branch, which is "0 mL" regardless of weight + and so must not require one). + + Rows are emitted only at times with at least ``window_hours`` of + record behind them: a partial window sums less urine simply because + less time has passed, which would read as oliguria. Multiple + collection routes (Foley, void, condom cath, suprapubic, + nephrostomy, ureteral stents) are summed, resolved through the LOINC + layer, not hardcoded. + + Weight (``weight_normalized=True`` only) is attached by a backward + ``join_asof``, so each window uses the most recent weight + charted at or before its end instant, never a later one: + + - the most recent **daily** weight if the key has one by then (it is + the current weight, which is what a mL/kg/h rate means clinically); + - otherwise the **admission** weight, the best available estimate + early in a stay before any daily weight has been charted; + - otherwise the window is **dropped**, not scored against a default + or population-average weight. Weight coverage in the real MIMIC-IV + extraction is poor (~10-17% of subjects have any reading at all), + so this criterion is genuinely unassessable for most keys -- + inventing a weight would silently turn "unknown" into a gold- + standard label. A non-positive charted weight (bad data) is + dropped the same way rather than dividing by it. + + Callers therefore must treat the *absence* of a key from this frame + as "not assessable", not as "not oliguric" -- the same observability + convention :func:`assessable_keys` gives the SOFA-derived concepts. + + Known limitation, inherited from the 24 h form and deliberately not + changed here: a window is summed over whatever urine rows it + contains, so *sparse charting* is indistinguishable from low output. + A key charted once a day reads as oliguric (and, if that one row is + 0 mL, as anuric) on the strength of a single row. Guarding it would + need a minimum-readings-per-window rule, which is a modelling + decision affecting the existing ``oliguria`` concept too, not + something to slip in with this change. """ urine = _numeric( events, @@ -595,23 +661,88 @@ def urine_output_24h( ) if urine.height == 0: return urine + window_minutes = int(round(window_hours * 60)) first_time = ( events.filter(pl.col(time_col).is_not_null()) .group_by(key) .agg(pl.col(time_col).min().alias("_first")) ) - return ( + rolled = ( urine.sort([key, time_col]) - .rolling(index_column=time_col, period="24h", group_by=key) + .rolling(index_column=time_col, period=f"{window_minutes}m", group_by=key) .agg(pl.col("value").sum().alias("value")) .join(first_time, on=key, how="left") .filter( - (pl.col(time_col) - pl.col("_first")) >= pl.duration(hours=WINDOW_HOURS) + (pl.col(time_col) - pl.col("_first")) >= pl.duration(minutes=window_minutes) + ) + .select(key, time_col, "value") + ) + if not weight_normalized: + return rolled + + cfg = SOFA_SOURCE_CONFIG[source] + out = rolled.sort([key, time_col]) + for field, alias in ( + (cfg.daily_weight, "_w_daily"), + (cfg.admission_weight, "_w_admission"), + ): + weights = _numeric( + events, + list(field), + key=key, + code_col=code_col, + value_col=value_col, + time_col=time_col, + ).rename({"value": alias}) + if weights.height == 0: + out = out.with_columns(pl.lit(None, dtype=pl.Float64).alias(alias)) + continue + out = out.join_asof( + weights.sort([key, time_col]), + on=time_col, + by=key, + strategy="backward", + ) + return ( + out.with_columns(pl.coalesce("_w_daily", "_w_admission").alias("_weight")) + .filter(pl.col("_weight").is_not_null() & (pl.col("_weight") > 0)) + .with_columns( + (pl.col("value") / pl.col("_weight") / window_hours).alias("value") ) .select(key, time_col, "value") ) +def urine_output_24h( + events: pl.DataFrame, + *, + source: str = "mimic_iv", + key: str = "subject_id", + code_col: str = "code", + value_col: str = "numeric_value", + time_col: str = "time", +) -> pl.DataFrame: + """(key, time, value=mL voided in the trailing 24h) per urine-output reading. + + The absolute-volume, 24 h special case of + :func:`urine_output_rate` -- what SOFA's renal component and the + ``oliguria`` concept score. Weight-normalized rates (mL/kg/h, + KDIGO's own form) are deliberately *not* used here: weight is not + reliably present per key in the extractions, and SOFA's own renal + bands are defined on absolute volume anyway. + """ + return urine_output_rate( + events, + source=source, + key=key, + code_col=code_col, + value_col=value_col, + time_col=time_col, + window_hours=WINDOW_HOURS, + weight_normalized=False, + ) + + @_quiet_asof def gcs_total_readings( events: pl.DataFrame, @@ -780,6 +911,7 @@ def sofa_timeseries( "COMPONENTS", "pf_ratio_readings", "urine_output_24h", + "urine_output_rate", "assessable_keys", "SOFA_SOURCE_CONFIG", "SofaSourceConfig", diff --git a/tests/odyssey/data/test_aki_kdigo.py b/tests/odyssey/data/test_aki_kdigo.py new file mode 100644 index 0000000..4614405 --- /dev/null +++ b/tests/odyssey/data/test_aki_kdigo.py @@ -0,0 +1,562 @@ +"""KDIGO AKI staging beyond creatinine: the RRT and urine-output legs. + +The creatinine legs (baseline-relative rises, the absolute >= 4.0 mg/dL +trigger, and stage ordering) are tested in ``test_concepts.py``; the plain +SOFA renal component and the absolute-volume ``oliguria`` concept in +``test_sepsis3_tasks.py``. This file covers what those two do not: + +- renal-replacement-therapy initiation as an automatic Stage 3 + (:data:`~odyssey.data.concepts.RRT_ITEMIDS`), including mimic-code's two + deliberate exclusions; +- KDIGO's weight-normalized urine-output rate + (:func:`~odyssey.data.sofa.urine_output_rate`) at each of the three + windows the stages use, its daily/admission weight fallback, and the + weight-free anuria branch. +""" + +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Sequence, Tuple + +import polars as pl +import pytest + +from odyssey.data.concepts import ( + RRT_CODE_PATTERN, + RRT_ITEMIDS, + AnyConceptDefinition, + CodeOccurrenceRule, + ConceptDefinition, + DerivedUrineRateRule, + concepts_for_source, + label_concepts, +) +from odyssey.data.sofa import urine_output_24h, urine_output_rate + + +T0 = datetime(2024, 1, 1, 0, 0) +H = timedelta(hours=1) + +# MIMIC-IV code prefixes (code_mapping's mimic_iv table / SOFA_SOURCE_CONFIG) +CREAT = "LAB//RESULT//50912//mg/dL" +URINE = "SUBJECT_FLUID_OUTPUT//226559//mL" # Foley +URINE_VOID = "SUBJECT_FLUID_OUTPUT//226560//mL" # a second collection route +DAILY_WEIGHT = "LAB//224639//kg" +ADMISSION_WEIGHT = "LAB//226512//kg" +# mimic-code rrt.sql: active dialysis, and its two exclusions +DIALYSIS_CATHETER = "PROCEDURE//START//224270" +CRRT_FILTER_CHANGE = "PROCEDURE//START//225436" + +_Row = Tuple[int, str, Optional[float], datetime] + + +def _events(rows: Sequence[_Row]) -> pl.DataFrame: + """Build a synthetic MEDS events frame; every row carries a real time. + + Same shape as ``test_concepts.py``'s ``_events``, with the time + mandatory: every rule exercised here is time-aware (trailing urine + windows, asof-joined weights, first-occurrence onsets). + """ + return pl.DataFrame( + list(rows), + schema={ + "subject_id": pl.Int64, + "code": pl.Utf8, + "numeric_value": pl.Float64, + "time": pl.Datetime("us"), + }, + orient="row", + ) + + +def _aki(*names: str) -> List[AnyConceptDefinition]: + """Return the named AKI concepts, as the mimic_iv expansion builds them.""" + return [c for c in concepts_for_source("mimic_iv") if c.name in names] + + +_ALL_STAGES = ("acute_kidney_injury", "aki_stage_2", "aki_stage_3") + + +def _labels(rows: Sequence[_Row], *names: str) -> Dict[int, Dict[str, object]]: + """Label ``rows`` with the named concepts, keyed by subject id.""" + labeled = label_concepts( + _events(rows), _aki(*(names or _ALL_STAGES)), include_first_time=True + ) + return {row["subject_id"]: row for row in labeled.to_dicts()} + + +def _rates( + rows: Sequence[_Row], *, window_hours: float, weight_normalized: bool = True +) -> Dict[datetime, float]: + """``urine_output_rate`` for the single subject in ``rows``, time -> value.""" + frame = urine_output_rate( + _events(rows), + key="subject_id", + window_hours=window_hours, + weight_normalized=weight_normalized, + ) + return dict(zip(frame["time"].to_list(), frame["value"].to_list())) + + +# --------------------------------------------------------------------------- +# Renal replacement therapy: an automatic Stage 3 +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("itemid", RRT_ITEMIDS) +def test_rrt_fires_on_each_dialysis_itemid_individually(itemid: str) -> None: + """Every one of mimic-code's six dialysis_active item ids stages a 3. + + Parametrized rather than spot-checked: a regex alternation that only + really matched its first branch would pass a single-item test. + """ + rows = [ + (1, CREAT, 1.0, T0), # creatinine charted and flat: no creatinine trigger + (1, f"PROCEDURE//START//{itemid}", None, T0 + 5 * H), + ] + row = _labels(rows)[1] + assert row["aki_stage_3"] == 1 + assert row["aki_stage_3_first_time"] == T0 + 5 * H + # RRT is a Stage 3 criterion only; the lower stages have their own. + assert row["acute_kidney_injury"] == 0 and row["aki_stage_2"] == 0 + + +@pytest.mark.parametrize("code", [DIALYSIS_CATHETER, CRRT_FILTER_CHANGE]) +def test_rrt_excludes_catheter_placement_and_filter_change(code: str) -> None: + """mimic-code's two dialysis_active = 0 rows must not stage anything. + + 224270 is line placement (no therapy yet) and 225436 is maintenance on + an already-running circuit, whose therapy start has its own row. + """ + row = _labels([(1, CREAT, 1.0, T0), (1, code, None, T0 + 5 * H)])[1] + assert row["aki_stage_3"] == 0 + # ... and the subject is still *observed*: procedure data exists, so + # "no dialysis" is a real negative rather than missingness. + assert row["aki_stage_3_observed"] == 1 + + +def test_rrt_pattern_is_anchored_to_procedure_start_and_a_whole_itemid() -> None: + """The alternation must not leak into END rows or longer item ids.""" + rows = [ + (1, CREAT, 1.0, T0), + (1, "PROCEDURE//END//225441", None, T0 + H), # the episode ending + (2, CREAT, 1.0, T0), + (2, "PROCEDURE//START//2254411", None, T0 + H), # a longer id, not ours + (3, CREAT, 1.0, T0), + (3, "PROCEDURE//START//225441//1", None, T0 + H), # a suffixed segment + ] + labeled = _labels(rows) + assert labeled[1]["aki_stage_3"] == 0 + assert labeled[2]["aki_stage_3"] == 0 + assert labeled[3]["aki_stage_3"] == 1 # a real start row with a trailing segment + + +def test_rrt_onset_is_the_first_dialysis_event_not_every_one() -> None: + rows = [ + (1, CREAT, 1.0, T0), + (1, "PROCEDURE//START//225802", None, T0 + 10 * H), # CRRT starts + (1, "PROCEDURE//START//225803", None, T0 + 20 * H), # switched to CVVHD + (1, "PROCEDURE//START//225802", None, T0 + 30 * H), # and back + ] + row = _labels(rows)[1] + assert row["aki_stage_3"] == 1 + assert row["aki_stage_3_first_time"] == T0 + 10 * H + + +def test_rrt_ors_with_the_creatinine_legs_without_double_firing() -> None: + """A patient already Stage 3 by creatinine keeps the earlier onset. + + The concept is one binary label whose rules are OR-ed, so a second + satisfied criterion cannot change the label or move the onset later. + """ + rows = [ + (1, CREAT, 5.0, T0 + 2 * H), # >= 4.0: Stage 3 by creatinine + (1, "PROCEDURE//START//225441", None, T0 + 40 * H), # dialysis later + ] + row = _labels(rows)[1] + assert row["aki_stage_3"] == 1 + assert row["aki_stage_3_first_time"] == T0 + 2 * H + # The mirror case: dialysis first, creatinine crossing later. + rows = [ + (2, "PROCEDURE//START//225441", None, T0 + 2 * H), + (2, CREAT, 5.0, T0 + 40 * H), + ] + row = _labels(rows)[2] + assert row["aki_stage_3"] == 1 and row["aki_stage_3_first_time"] == T0 + 2 * H + + +def test_rrt_rule_is_the_only_occurrence_rule_and_only_on_stage_3() -> None: + """Stage 3 carries the RRT rule; the lower stages must not. + + KDIGO's automatic-Stage-3 fact is about dialysis, and it is expressed + as one rule inside aki_stage_3 rather than as a separate named + concept the bottleneck would need its own head for. + """ + by_name = {c.name: c for c in _aki(*_ALL_STAGES)} + stage_3 = by_name["aki_stage_3"] + assert isinstance(stage_3, ConceptDefinition) + occurrence = [r for r in stage_3.rules if isinstance(r, CodeOccurrenceRule)] + assert [r.code_pattern for r in occurrence] == [RRT_CODE_PATTERN] + assert occurrence[0].observed_families == ("PROCEDURE",) + for name in ("acute_kidney_injury", "aki_stage_2"): + concept = by_name[name] + assert isinstance(concept, ConceptDefinition) + assert not any(isinstance(r, CodeOccurrenceRule) for r in concept.rules) + + +# --------------------------------------------------------------------------- +# Urine-output rate: mL/kg/h arithmetic over KDIGO's windows +# --------------------------------------------------------------------------- + +# 60 kg, a 6 h burst of 60 mL/h then a trickle of 6 mL/h: each window +# length sees a different rate, so a window mix-up cannot pass silently. +_BURST_THEN_TRICKLE: List[_Row] = ( + [(1, DAILY_WEIGHT, 60.0, T0)] + + [(1, URINE, 60.0, T0 + k * H) for k in range(1, 7)] + + [(1, URINE, 6.0, T0 + k * H) for k in range(7, 25)] +) + + +def test_urine_rate_arithmetic_at_each_kdigo_window() -> None: + six = _rates(_BURST_THEN_TRICKLE, window_hours=6.0) + twelve = _rates(_BURST_THEN_TRICKLE, window_hours=12.0) + twentyfour = _rates(_BURST_THEN_TRICKLE, window_hours=24.0) + # 6 h window at 6 h: 6 x 60 mL / 60 kg / 6 h = 1.0 mL/kg/h + assert six[T0 + 6 * H] == pytest.approx(1.0) + # 6 h window at 24 h: 6 x 6 mL / 60 kg / 6 h = 0.1 + assert six[T0 + 24 * H] == pytest.approx(0.1) + # 12 h window at 24 h: 12 x 6 mL / 60 kg / 12 h = 0.1 + assert twelve[T0 + 24 * H] == pytest.approx(0.1) + # 24 h window at 24 h: (360 + 108) mL / 60 kg / 24 h = 0.325 + assert twentyfour[T0 + 24 * H] == pytest.approx(0.325) + + +def test_urine_rate_excludes_windows_without_a_full_window_of_record() -> None: + """Same partial-window exclusion as the 24 h absolute-volume form. + + A window narrower than its length sums less urine only because less + time has passed, which would read as oliguria at every admission. + """ + six = _rates(_BURST_THEN_TRICKLE, window_hours=6.0) + twentyfour = _rates(_BURST_THEN_TRICKLE, window_hours=24.0) + assert min(six) == T0 + 6 * H + assert list(twentyfour) == [T0 + 24 * H] # nothing earlier is assessable + + +def test_urine_rate_sums_every_collection_route() -> None: + """Foley plus void, resolved through the LOINC layer, not one itemid.""" + rows: List[_Row] = [(1, DAILY_WEIGHT, 50.0, T0)] + [ + row + for k in range(1, 7) + for row in ((1, URINE, 20.0, T0 + k * H), (1, URINE_VOID, 5.0, T0 + k * H)) + ] + # 6 x 25 mL / 50 kg / 6 h = 0.5 + assert _rates(rows, window_hours=6.0)[T0 + 6 * H] == pytest.approx(0.5) + + +def test_urine_output_24h_is_the_absolute_volume_special_case() -> None: + """The pre-existing callers (SOFA renal, oliguria) must be unchanged.""" + frame = urine_output_24h(_events(_BURST_THEN_TRICKLE), key="subject_id") + assert frame.columns == ["subject_id", "time", "value"] + assert dict(zip(frame["time"].to_list(), frame["value"].to_list())) == { + T0 + 24 * H: 468.0 + } + assert frame.equals( + urine_output_rate( + _events(_BURST_THEN_TRICKLE), + key="subject_id", + window_hours=24.0, + weight_normalized=False, + ) + ) + + +def test_only_the_weighted_form_needs_a_source_weight_config() -> None: + """The absolute-volume form must keep working where weights are unknown. + + eICU has no :data:`~odyssey.data.sofa.SOFA_SOURCE_CONFIG` entry, and + ``urine_output_24h`` delegates here with ``weight_normalized=False``: + that path must not reach for a config it has no need of. Asking for + the weighted form on such a source is a programming error and says + so, the same way every other SOFA entry point does. + """ + rows: List[_Row] = [(1, "URINE_OUTPUT//mL", 10.0, T0 + k * H) for k in range(0, 30)] + absolute = urine_output_rate( + _events(rows), + source="eicu", + key="subject_id", + window_hours=24.0, + weight_normalized=False, + ) + # Hours 24-29 are the assessable ones; each 24 h window is left-open, + # so it holds 24 of the hourly 10 mL readings, not 25. + assert absolute["value"].to_list() == [240.0] * 6 + with pytest.raises(KeyError): + urine_output_rate(_events(rows), source="eicu", key="subject_id") + + +# --------------------------------------------------------------------------- +# Weight: the fallback order, and abstaining when there is none +# --------------------------------------------------------------------------- + + +def test_daily_weight_is_preferred_over_admission_weight_once_charted() -> None: + """Daily weight is the current weight; admission weight is the fallback. + + Subject 1 has both: windows ending before the first daily weight use + the admission weight (a *later* reading must never be pulled + backwards by the asof join), windows after it use the daily weight. + """ + urine: List[Tuple[int, str, Optional[float], datetime]] = [ + (1, URINE, 30.0, T0 + k * H) for k in range(1, 13) + ] + both = [ + (1, ADMISSION_WEIGHT, 100.0, T0), + (1, DAILY_WEIGHT, 50.0, T0 + 8 * H), + *urine, + ] + rates = _rates(both, window_hours=6.0) + # at 6 h the only weight charted yet is the admission one: 180/100/6 + assert rates[T0 + 6 * H] == pytest.approx(0.3) + # at 12 h the daily weight (charted at 8 h) applies: 180/50/6 + assert rates[T0 + 12 * H] == pytest.approx(0.6) + + admission_only = [(2, ADMISSION_WEIGHT, 100.0, T0)] + [ + (2, code, value, time) for (_, code, value, time) in urine + ] + daily_only = [(3, DAILY_WEIGHT, 50.0, T0)] + [ + (3, code, value, time) for (_, code, value, time) in urine + ] + assert _rates(admission_only, window_hours=6.0)[T0 + 12 * H] == pytest.approx(0.3) + assert _rates(daily_only, window_hours=6.0)[T0 + 12 * H] == pytest.approx(0.6) + + +def test_a_window_with_no_weight_at_all_is_dropped_not_defaulted() -> None: + """No weight anywhere: the rate criterion is unassessable, full stop. + + Weight coverage in the real MIMIC-IV extraction is ~10-17% of + subjects, so this is the common case, and defaulting a weight would + manufacture gold-standard labels out of nothing. + """ + rows: List[_Row] = [(1, URINE, 1.0, T0 + k * H) for k in range(1, 13)] + assert _rates(rows, window_hours=6.0) == {} + # The same rows without normalization are perfectly assessable. The + # window origin is the key's first *event* (here the first urine + # reading, at T0 + 1h), so the earliest full 6 h window ends at + # T0 + 7h -- the same rule urine_output_24h has always applied. + assert _rates(rows, window_hours=6.0, weight_normalized=False)[ + T0 + 7 * H + ] == pytest.approx(6.0) + # a weight charted only AFTER every window is no help either + late = [*rows, (1, DAILY_WEIGHT, 70.0, T0 + 20 * H)] + assert _rates(late, window_hours=6.0) == {} + + +def test_a_non_positive_charted_weight_is_dropped_not_divided_by() -> None: + rows: List[_Row] = [(1, DAILY_WEIGHT, 0.0, T0)] + [ + (1, URINE, 1.0, T0 + k * H) for k in range(1, 13) + ] + assert _rates(rows, window_hours=6.0) == {} + + +def test_urine_rate_is_empty_without_urine_readings() -> None: + """No urine charted at all: no rows, whatever the weight coverage.""" + rows: List[_Row] = [(1, DAILY_WEIGHT, 70.0, T0), (1, CREAT, 1.0, T0 + H)] + assert _rates(rows, window_hours=6.0) == {} + assert _rates(rows, window_hours=12.0, weight_normalized=False) == {} + + +def test_unassessable_urine_rate_is_unobserved_not_a_negative() -> None: + """The observed-mask convention, isolated to the urine-rate rule. + + A urine-rate-only concept must report subjects it cannot score as + unobserved (masked out of supervision), never as label 0. + """ + concept = ConceptDefinition( + "urine_stage_1", + [ + DerivedUrineRateRule( + threshold=0.5, direction="below", window_hours=6.0, source="mimic_iv" + ) + ], + "test", + ) + rows: List[_Row] = [ + # subject 1: urine, no weight -- not assessable + *[(1, URINE, 1.0, T0 + k * H) for k in range(1, 8)], + # subject 2: urine and a weight -- assessable, and oliguric + (2, DAILY_WEIGHT, 60.0, T0), + *[(2, URINE, 1.0, T0 + k * H) for k in range(1, 8)], + # subject 3: urine and a weight, voiding well -- assessable negative + (3, DAILY_WEIGHT, 60.0, T0), + *[(3, URINE, 100.0, T0 + k * H) for k in range(1, 8)], + ] + labeled = label_concepts(_events(rows), [concept]).sort("subject_id") + assert labeled["urine_stage_1"].to_list() == [0, 1, 0] + assert labeled["urine_stage_1_observed"].to_list() == [0, 1, 1] + + +# --------------------------------------------------------------------------- +# Thresholds: strict "<" on the rate, inclusive "<= 0" on anuria +# --------------------------------------------------------------------------- + + +def test_the_rate_threshold_is_strict_exactly_at_kdigos_number() -> None: + """KDIGO says "less than 0.5 mL/kg/h", so exactly 0.5 does not stage. + + The mirror of the ``at_or_above`` convention on Stage 3's creatinine + >= 4.0 trigger: the direction spelled in the rule is the direction + the criterion is written with. + """ + rows: List[_Row] = [ + (1, DAILY_WEIGHT, 50.0, T0), # 25 mL/h / 50 kg = exactly 0.5 + *[(1, URINE, 25.0, T0 + k * H) for k in range(1, 8)], + (2, DAILY_WEIGHT, 50.0, T0), # 24 mL/h / 50 kg = 0.48 + *[(2, URINE, 24.0, T0 + k * H) for k in range(1, 8)], + ] + labeled = _labels(rows, "acute_kidney_injury", "aki_stage_2") + assert labeled[1]["acute_kidney_injury"] == 0 + assert labeled[2]["acute_kidney_injury"] == 1 + assert labeled[2]["acute_kidney_injury_first_time"] == T0 + 6 * H + + +def test_stage_2_needs_twelve_hours_of_the_same_rate_stage_1_needs_six() -> None: + """The stages differ only in window length at 0.5 mL/kg/h.""" + rows: List[_Row] = [ + # subject 1: oliguric for 8 h only -- Stage 1's 6 h window fires, + # Stage 2's 12 h window never has 12 h of oliguria to average over. + (1, DAILY_WEIGHT, 100.0, T0), + *[(1, URINE, 10.0, T0 + k * H) for k in range(1, 9)], # 0.1 mL/kg/h + *[(1, URINE, 300.0, T0 + k * H) for k in range(9, 21)], # 3.0 mL/kg/h + # subject 2: oliguric throughout -- both stages fire. + (2, DAILY_WEIGHT, 100.0, T0), + *[(2, URINE, 10.0, T0 + k * H) for k in range(1, 21)], + ] + labeled = _labels(rows, "acute_kidney_injury", "aki_stage_2") + assert labeled[1]["acute_kidney_injury"] == 1 + assert labeled[1]["acute_kidney_injury_first_time"] == T0 + 6 * H + assert labeled[1]["aki_stage_2"] == 0 + assert labeled[2]["acute_kidney_injury"] == 1 and labeled[2]["aki_stage_2"] == 1 + assert labeled[2]["aki_stage_2_first_time"] == T0 + 12 * H + + +def test_stage_3_rate_leg_is_below_point_three_over_twenty_four_hours() -> None: + rows: List[_Row] = [ + # subject 1: 0.2 mL/kg/h for a full day -- Stage 3 by rate + (1, DAILY_WEIGHT, 100.0, T0), + *[(1, URINE, 20.0, T0 + k * H) for k in range(1, 26)], + # subject 2: 0.4 mL/kg/h -- Stages 1 and 2, but not 3's 0.3 rate + (2, DAILY_WEIGHT, 100.0, T0), + *[(2, URINE, 40.0, T0 + k * H) for k in range(1, 26)], + ] + labeled = _labels(rows) + assert labeled[1]["aki_stage_3"] == 1 + assert labeled[1]["aki_stage_3_first_time"] == T0 + 24 * H + assert labeled[2]["acute_kidney_injury"] == 1 and labeled[2]["aki_stage_2"] == 1 + assert labeled[2]["aki_stage_3"] == 0 + + +def test_anuria_stages_a_3_with_no_weight_reading_anywhere() -> None: + """0 mL over 12 h is 0 mL at any body weight, so it needs no weight. + + This is the leg that keeps Stage 3's urine criterion assessable for + the ~85% of subjects with no charted weight. + """ + rows: List[_Row] = [ + # subject 1: anuric, no weight charted at all + *[(1, URINE, 0.0, T0 + k * H) for k in range(1, 14)], + # subject 2: a trickle, not anuria; no weight, so the rate legs + # cannot rescue it either -- Stage 3 must stay 0, not "unknown = yes" + *[(2, URINE, 1.0, T0 + k * H) for k in range(1, 14)], + ] + labeled = _labels(rows, "aki_stage_3") + assert labeled[1]["aki_stage_3"] == 1 + # First event is the urine reading at T0 + 1h, so the first full 12 h + # window ends at T0 + 13h (the partial-window rule, not an off-by-one). + assert labeled[1]["aki_stage_3_first_time"] == T0 + 13 * H + assert labeled[2]["aki_stage_3"] == 0 + assert labeled[2]["aki_stage_3_observed"] == 1 # urine was assessable + + +def test_anuria_needs_a_full_twelve_hours_of_record() -> None: + """Six hours of no urine is not yet 12 h of anuria.""" + rows: List[_Row] = [(1, URINE, 0.0, T0 + k * H) for k in range(1, 7)] + assert _labels(rows, "aki_stage_3")[1]["aki_stage_3"] == 0 + + +# --------------------------------------------------------------------------- +# Whole-concept integration and per-source expansion +# --------------------------------------------------------------------------- + + +def test_all_three_legs_compose_as_an_or_across_a_mixed_cohort() -> None: + """Creatinine-only, urine-only, RRT-only and combined, in one pass.""" + rows: List[_Row] = [ + # 1: creatinine only, 1.0 -> 3.5 (3.5x): stages 1, 2 and 3 + (1, CREAT, 1.0, T0), + (1, CREAT, 3.5, T0 + 48 * H), + # 2: urine only, 0.2 mL/kg/h for a day on a charted weight: 1, 2, 3 + (2, DAILY_WEIGHT, 100.0, T0), + *[(2, URINE, 20.0, T0 + k * H) for k in range(1, 26)], + # 3: RRT only, creatinine flat and urine fine: 3 alone + (3, CREAT, 1.0, T0), + (3, ADMISSION_WEIGHT, 100.0, T0), + *[(3, URINE, 200.0, T0 + k * H) for k in range(1, 26)], + (3, "PROCEDURE//START//225809", None, T0 + 30 * H), + # 4: creatinine +0.3 in 48 h (stage 1) and anuric 12 h (stage 3), + # no weight: the rate legs abstain, the anuria leg does not + (4, CREAT, 1.0, T0), + (4, CREAT, 1.35, T0 + 24 * H), + *[(4, URINE, 0.0, T0 + k * H) for k in range(1, 14)], + # 5: nothing wrong anywhere -- an observed negative on all three + (5, CREAT, 1.0, T0), + (5, DAILY_WEIGHT, 80.0, T0), + *[(5, URINE, 100.0, T0 + k * H) for k in range(1, 26)], + ] + labeled = _labels(rows) + assert [labeled[i]["acute_kidney_injury"] for i in range(1, 6)] == [1, 1, 0, 1, 0] + assert [labeled[i]["aki_stage_2"] for i in range(1, 6)] == [1, 1, 0, 0, 0] + assert [labeled[i]["aki_stage_3"] for i in range(1, 6)] == [1, 1, 1, 1, 0] + for i in range(1, 6): + for name in _ALL_STAGES: + assert labeled[i][f"{name}_observed"] == 1, (i, name) + # Stage 3 does not require Stage 1 to have fired: the stages stay + # independent binary concepts, as they were before this change. + assert labeled[3]["acute_kidney_injury"] == 0 and labeled[3]["aki_stage_3"] == 1 + + +def test_urine_legs_expand_only_where_the_weight_item_ids_are_known() -> None: + """Like sepsis3: the derived legs need a source config, the rest travels. + + eICU has no :data:`~odyssey.data.sofa.SOFA_SOURCE_CONFIG` entry, so + its AKI concepts keep the creatinine legs (and the harmless + occurrence rule, which is source-agnostic) but not the urine ones, + and the concept is never dropped for want of them. + """ + stages = { + source: { + c.name: c for c in concepts_for_source(source) if c.name in _ALL_STAGES + } + for source in ("mimic_iv", "eicu", "gemini") + } + for source in ("mimic_iv", "eicu", "gemini"): + assert set(stages[source]) == set(_ALL_STAGES) + + def _urine_rules(source: str, name: str) -> List[DerivedUrineRateRule]: + concept = stages[source][name] + assert isinstance(concept, ConceptDefinition) + return [r for r in concept.rules if isinstance(r, DerivedUrineRateRule)] + + assert [ + (r.threshold, r.window_hours, r.weight_normalized) + for r in _urine_rules("mimic_iv", "aki_stage_3") + ] == [(0.3, 24.0, True), (0.0, 12.0, False)] + assert [ + (r.threshold, r.window_hours) for r in _urine_rules("mimic_iv", "aki_stage_2") + ] == [(0.5, 12.0)] + assert [ + (r.threshold, r.window_hours) + for r in _urine_rules("mimic_iv", "acute_kidney_injury") + ] == [(0.5, 6.0)] + for name in _ALL_STAGES: + assert _urine_rules("eicu", name) == [] + assert _urine_rules("gemini", name) == [] diff --git a/tests/odyssey/data/test_value_binning.py b/tests/odyssey/data/test_value_binning.py index 2827838..63f6f1c 100644 --- a/tests/odyssey/data/test_value_binning.py +++ b/tests/odyssey/data/test_value_binning.py @@ -14,6 +14,8 @@ ConceptDefinition, ConceptRule, DerivedGcsTotalRule, + DerivedSofaSignalRule, + DerivedUrineRateRule, SustainedRule, ) from odyssey.data.value_binning import ( @@ -142,6 +144,19 @@ def test_empty_events_frame_is_a_noop() -> None: # --------------------------------------------------------------------------- +#: Rule types with no single fixed threshold on one charted code's value +#: (see :func:`_threshold_rules`); a rule type absent from both this tuple +#: and the handled branches raises, so a newly added one cannot slip +#: through this consistency check unnoticed. +_NO_FIXED_THRESHOLD = ( + BaselineRelativeRule, + DerivedGcsTotalRule, + CodeOccurrenceRule, + DerivedSofaSignalRule, + DerivedUrineRateRule, +) + + def _threshold_rules(concept: ConceptDefinition) -> List[Tuple[str, float]]: """Yield every (code_prefix, threshold) pair reachable from a concept. @@ -153,8 +168,13 @@ def _threshold_rules(concept: ConceptDefinition) -> List[Tuple[str, float]]: a personal baseline, not a fixed value, and :class:`~odyssey.data.concepts.DerivedGcsTotalRule` sums three different codes, not one -- neither maps to a single CLINICAL_RANGES - prefix's bin edge, so both are skipped here. Recurses into - :class:`~odyssey.data.concepts.AnyOf`, which nests further rules. + prefix's bin edge, so both are skipped here, and so are + :class:`~odyssey.data.concepts.DerivedSofaSignalRule` and + :class:`~odyssey.data.concepts.DerivedUrineRateRule`, whose thresholds + are on a *derived* signal (a PaO2/FiO2 ratio, a trailing-window urine + rate in mL/kg/h) that no single charted code's value channel carries. + Recurses into :class:`~odyssey.data.concepts.AnyOf`, which nests + further rules. :class:`~odyssey.data.concepts.CompositeConceptDefinition` (SIRS, qSOFA) is out of scope entirely, not just certain rule types within @@ -171,16 +191,11 @@ def _threshold_rules(concept: ConceptDefinition) -> List[Tuple[str, float]]: for sub_rule in rule.rules: if isinstance(sub_rule, (ConceptRule, SustainedRule)): out.append((sub_rule.code_prefix, sub_rule.threshold)) - elif not isinstance( - sub_rule, - (BaselineRelativeRule, DerivedGcsTotalRule, CodeOccurrenceRule), - ): + elif not isinstance(sub_rule, _NO_FIXED_THRESHOLD): raise TypeError(f"unhandled rule type in AnyOf: {type(sub_rule)!r}") elif isinstance(rule, (ConceptRule, SustainedRule)): out.append((rule.code_prefix, rule.threshold)) - elif not isinstance( - rule, (BaselineRelativeRule, DerivedGcsTotalRule, CodeOccurrenceRule) - ): + elif not isinstance(rule, _NO_FIXED_THRESHOLD): raise TypeError(f"unhandled rule type: {type(rule)!r}") return out