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