diff --git a/mimic-iv/concepts/README.md b/mimic-iv/concepts/README.md index f2c55041b..f67fbe502 100644 --- a/mimic-iv/concepts/README.md +++ b/mimic-iv/concepts/README.md @@ -20,6 +20,9 @@ Concepts in this folder: │   ├── icustay_hourly.sql │   ├── icustay_times.sql │   └── weight_durations.sql +├── diagnosticdelay +│   ├── suspected_sepsis_cohort.sql +│   └── diagnostic_delay.sql ├── firstday │   ├── first_day_bg.sql │   ├── first_day_bg_art.sql diff --git a/mimic-iv/concepts/diagnosticdelay/README.md b/mimic-iv/concepts/diagnosticdelay/README.md new file mode 100644 index 000000000..9a68ba31c --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/README.md @@ -0,0 +1,109 @@ +# Diagnostic delay + +Reproducible cohort extraction and diagnostic-delay measurement for +MIMIC-IV, plus a data-quality report generator. + +## What this is + +ICD codes in MIMIC carry no timestamp, so "when was this diagnosed?" is +a recurring community question (e.g. +[#1843](https://github.com/MIT-LCP/mimic-code/issues/1843)). This +module answers a tractable version of it: for ICU stays with a +*suspected-infection* episode (an antibiotic paired with a microbiology +culture, per the `suspicion_of_infection` concept), it measures the +delay between hospital/ICU admission and the first clinically +observable recognition event (first antibiotic order, first culture +draw, suspected-infection time). + +Contents: + +| File | Description | +|---|---| +| `suspected_sepsis_cohort.sql` | Cohort concept: one row per ICU stay for adults admitted via the ED with >= 1 suspected-infection episode. | +| `diagnostic_delay.sql` | Delay measures (hours), onset-window classification, and temporal-plausibility flags, one row per cohort stay. | +| `data_quality_report.py` | Generates a Markdown/JSON data-quality report (coverage, missingness, flag prevalence, delay distributions) from the two tables above. | + +The SQL is written in the BigQuery dialect, following the repository's +concept conventions; PostgreSQL and DuckDB versions are generated +automatically on merge (see the top-level README). + +## Cohort definition (`suspected_sepsis_cohort`) + +* Source: `mimiciv_icu.icustays` x `mimiciv_hosp.admissions` x + `mimiciv_hosp.patients`, inner-joined to + `mimiciv_derived.suspicion_of_infection` (suspected_infection = 1), + aggregated to one row per `stay_id`. +* Inclusion: admission age >= 18 (same `anchor_age` formula as + `age.sql`); `admissions.admission_location = 'EMERGENCY ROOM'`. +* Carries: `subject_id, hadm_id, stay_id, gender, admission_age, race, + admittime, dischtime, admission_type, admission_location, + hospital_expire_flag, hospstay_seq, icu_intime, icu_outtime, + icustay_seq, n_suspicion_episodes, first_suspicion_time, + first_antibiotic_time, first_culture_time`. + +## Delay measures (`diagnostic_delay`) + +All delays are in hours (1 decimal): + +* `admission_to_suspicion_hours`, `icu_to_suspicion_hours`, + `admission_to_antibiotic_hours`, `admission_to_culture_hours`, + `culture_to_antibiotic_hours` (positive = antibiotic after culture). +* `onset_window`: `pre_admission` (suspicion before `admittime`), + `present_on_admission` (within 48 h of admission), `hospital_onset` + (after 48 h). +* Plausibility flags (1 = implausible, 0 = plausible, NULL = + unassessable): `flag_suspicion_before_admission`, + `flag_suspicion_after_icu_outtime`, `flag_suspicion_after_discharge`, + `flag_missing_culture_time`, `flag_missing_antibiotic_time`. + +## Data-quality report + +```bash +python data_quality_report.py \ + --duckdb ~/data/mimic-iv.duckdb \ + --schema mimiciv_derived \ + --raw-schema mimiciv_icu \ + --output delay_data_quality.md \ + --json-output delay_data_quality.json +``` + +`--postgres-dsn` (requires `psycopg2`) can be used instead of +`--duckdb`. `--raw-schema` is optional; when given, the report adds +the cohort capture rate against adult ED ICU stays from the raw +tables. + +## Validation + +Real MIMIC-IV data requires credentialed PhysioNet access, so the +logic was validated without it: + +1. `sqlglot` parse check of both queries (BigQuery dialect) -- the + same check CI runs (`.github/scripts/check_sql_syntax.py`). +2. `sqlfluff` lint with the repository's `.sqlfluff` config. +3. Transpiled both queries to DuckDB with the repo's `mimic_utils` + transpiler and executed them against hand-built synthetic fixtures + mirroring the MIMIC-IV schema (admissions, icustays, patients, + derived suspicion_of_infection). Fixtures cover: a standard + present-on-admission case, a hospital-onset case, a + pre-admission-suspicion case (flagged), multiple episodes per stay, + a pediatric exclusion, a non-ED admission exclusion, a stay with no + suspected infection (excluded), and a NULL `icu_outtime` case + (flag unassessable). +4. Ran `data_quality_report.py` against the resulting DuckDB database + and verified the Markdown/JSON outputs against hand-computed + expectations. + +What still needs real data: end-to-end execution against MIMIC-IV +(v3.1), sanity of the delay distributions at scale, and the capture +rate of the cohort definition. + +## Limitations + +* "Diagnosis time" is proxied by recognition events (antibiotic / + culture); true clinical diagnosis time is not recorded in MIMIC. +* ED arrival time is not used: the module only depends on + `mimiciv_hosp` + `mimiciv_icu` + derived concepts so it runs without + the separate MIMIC-IV-ED module. +* Implausible raw values (see + [#2168](https://github.com/MIT-LCP/mimic-code/issues/2168)) are + flagged via ordering checks, not range-checked here. diff --git a/mimic-iv/concepts/diagnosticdelay/data_quality_report.py b/mimic-iv/concepts/diagnosticdelay/data_quality_report.py new file mode 100644 index 000000000..2eb2fbf9b --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/data_quality_report.py @@ -0,0 +1,413 @@ +#!/usr/bin/env python3 +"""Data-quality report for the diagnostic-delay concept tables. + +Reads the ``suspected_sepsis_cohort`` and ``diagnostic_delay`` tables +(mimic-iv/concepts/diagnosticdelay/) from a DuckDB database (or +PostgreSQL, if psycopg2 is installed) and emits a data-quality report +covering: + +* cohort coverage -- how many ICU stays/patients the cohort captures, + optionally against the raw ``icustays`` denominator (``--raw-schema``) +* missingness of every anchor timestamp and delay measure +* temporal-plausibility flag prevalence + (suspicion before admission, after ICU discharge, after hospital + discharge, missing culture/antibiotic times) +* delay distributions (median / p25 / p75) and onset-window breakdown + +Example (DuckDB, after building the derived concepts):: + + python data_quality_report.py \\ + --duckdb ~/data/mimic-iv.duckdb \\ + --schema mimiciv_derived \\ + --raw-schema mimiciv_icu \\ + --raw-hosp-schema mimiciv_hosp \\ + --output delay_data_quality.md + +The aggregation SQL is plain portable SQL (no engine-specific +functions); quantiles are computed in Python so the numbers are +identical on every backend. +""" + +import argparse +import json +import math +import sys +from datetime import datetime, timezone + + +COHORT_TABLE = "suspected_sepsis_cohort" +DELAY_TABLE = "diagnostic_delay" + +TIMESTAMP_COLUMNS = [ + "admittime", + "dischtime", + "icu_intime", + "icu_outtime", + "first_suspicion_time", + "first_antibiotic_time", + "first_culture_time", +] + +DELAY_COLUMNS = [ + "admission_to_suspicion_hours", + "icu_to_suspicion_hours", + "admission_to_antibiotic_hours", + "admission_to_culture_hours", + "culture_to_antibiotic_hours", +] + +FLAG_COLUMNS = [ + "flag_suspicion_before_admission", + "flag_suspicion_after_icu_outtime", + "flag_suspicion_after_discharge", + "flag_missing_culture_time", + "flag_missing_antibiotic_time", +] + + +def connect(args): + """Open a DB-API connection for the requested engine.""" + if args.postgres_dsn: + try: + import psycopg2 + except ImportError: + sys.exit("psycopg2 is required for --postgres-dsn") + return psycopg2.connect(args.postgres_dsn) + try: + import duckdb + except ImportError: + sys.exit("duckdb is required for --duckdb (pip install duckdb)") + con = duckdb.connect(args.duckdb, read_only=True) + return con + + +def _cursor(con): + """Return something executable for this DB-API connection. + + psycopg2 connections expose no ``execute`` method -- queries must go + through an explicit cursor. DuckDB connections execute directly. + """ + return con if hasattr(con, "execute") else con.cursor() + + +def fetchone(con, sql, params=()): + cur = _cursor(con) + try: + cur.execute(sql, params) + return cur.fetchone() + finally: + if cur is not con: + cur.close() + + +def fetchall(con, sql, params=()): + cur = _cursor(con) + try: + cur.execute(sql, params) + cols = [d[0] for d in cur.description] + return cols, cur.fetchall() + finally: + if cur is not con: + cur.close() + + +def table_exists(con, schema, table): + try: + row = fetchone( + con, f"SELECT 1 FROM {schema}.{table} LIMIT 1" + ) + return row is not None + except Exception: + return False + + +def quantiles(values, qs=(0.25, 0.5, 0.75)): + """Quantiles computed in Python (portable across backends).""" + vals = sorted(v for v in values if v is not None) + if not vals: + return {q: None for q in qs} + out = {} + for q in qs: + pos = q * (len(vals) - 1) + lo = math.floor(pos) + hi = math.ceil(pos) + out[q] = vals[lo] + (vals[hi] - vals[lo]) * (pos - lo) + return out + + +def cohort_stats(con, schema): + cols, rows = fetchall( + con, + f""" + SELECT COUNT(*) AS n_stays + , COUNT(DISTINCT subject_id) AS n_patients + , COUNT(DISTINCT hadm_id) AS n_hadm + , MIN(admittime) AS min_admittime + , MAX(admittime) AS max_admittime + , SUM(CASE WHEN gender = 'F' THEN 1 ELSE 0 END) AS n_female + , SUM(CASE WHEN hospital_expire_flag = 1 THEN 1 ELSE 0 END) + AS n_hospital_deaths + FROM {schema}.{COHORT_TABLE} + """, + ) + stats = dict(zip(cols, rows[0])) + ages = [ + r[0] + for r in fetchall( + con, f"SELECT admission_age FROM {schema}.{COHORT_TABLE}" + )[1] + ] + stats["admission_age_p25"], stats["admission_age_p50"], stats[ + "admission_age_p75" + ] = (lambda q: (q[0.25], q[0.5], q[0.75]))(quantiles(ages)) + return stats + + +def coverage_stats(con, icu_schema, hosp_schema=None): + """Cohort capture rate against the raw icustays denominator. + + ``icustays`` lives in the ICU schema (``mimiciv_icu``); ``admissions`` + and ``patients`` live in the hospital schema (``mimiciv_hosp``), so the + two schema names are accepted separately. Denominator mirrors the + cohort filters (adult, ED origin) without requiring a + suspected-infection episode. Returns None when the raw tables are + unavailable. The age filter reproduces the concept's + ``anchor_age + year(admittime) - anchor_year`` formula with + engine-portable date parts. + """ + hosp_schema = hosp_schema or icu_schema + try: + cols, rows = fetchall( + con, + f""" + SELECT COUNT(*) AS n_adult_ed_icu_stays + , COUNT(DISTINCT ie.subject_id) AS n_patients + FROM {icu_schema}.icustays ie + INNER JOIN {hosp_schema}.admissions adm + ON ie.hadm_id = adm.hadm_id + INNER JOIN {hosp_schema}.patients pat + ON ie.subject_id = pat.subject_id + WHERE pat.anchor_age + + DATE_PART('YEAR', adm.admittime) + - pat.anchor_year >= 18 + AND adm.admission_location = 'EMERGENCY ROOM' + """, + ) + except Exception: + return None + return dict(zip(cols, rows[0])) + + +def missingness(con, schema): + """Per-column missingness for timestamps and delay measures.""" + select = ", ".join( + f"SUM(CASE WHEN {c} IS NULL THEN 1 ELSE 0 END) AS {c}_missing" + for c in TIMESTAMP_COLUMNS + DELAY_COLUMNS + ) + cols, rows = fetchall( + con, f"SELECT COUNT(*) AS n, {select} FROM {schema}.{DELAY_TABLE}" + ) + res = dict(zip(cols, rows[0])) + n = res.pop("n") + return { + "n": n, + "columns": { + c: {"missing": res[f"{c}_missing"], "pct": res[f"{c}_missing"] / n * 100} + for c in TIMESTAMP_COLUMNS + DELAY_COLUMNS + }, + } + + +def flag_prevalence(con, schema): + select = ", ".join( + f"SUM(CASE WHEN {c} = 1 THEN 1 ELSE 0 END) AS {c}_n" + f", SUM(CASE WHEN {c} IS NULL THEN 1 ELSE 0 END) AS {c}_null" + for c in FLAG_COLUMNS + ) + cols, rows = fetchall( + con, f"SELECT COUNT(*) AS n, {select} FROM {schema}.{DELAY_TABLE}" + ) + res = dict(zip(cols, rows[0])) + n = res.pop("n") + return { + "n": n, + "flags": { + c: { + "flagged": res[f"{c}_n"], + "pct": res[f"{c}_n"] / n * 100, + "unassessable_null": res[f"{c}_null"], + } + for c in FLAG_COLUMNS + }, + } + + +def delay_distributions(con, schema): + out = {} + for col in DELAY_COLUMNS: + values = [ + r[0] + for r in fetchall( + con, + f"SELECT {col} FROM {schema}.{DELAY_TABLE}" + f" WHERE {col} IS NOT NULL", + )[1] + ] + q = quantiles(values) + out[col] = { + "n": len(values), + "p25": q[0.25], + "median": q[0.5], + "p75": q[0.75], + "min": min(values) if values else None, + "max": max(values) if values else None, + } + return out + + +def onset_window_counts(con, schema): + cols, rows = fetchall( + con, + f""" + SELECT onset_window, COUNT(*) AS n + FROM {schema}.{DELAY_TABLE} + GROUP BY onset_window + ORDER BY n DESC + """, + ) + return [dict(zip(cols, r)) for r in rows] + + +def build_report(con, schema, raw_schema, raw_hosp_schema=None): + if not table_exists(con, schema, COHORT_TABLE): + sys.exit(f"table {schema}.{COHORT_TABLE} not found") + if not table_exists(con, schema, DELAY_TABLE): + sys.exit(f"table {schema}.{DELAY_TABLE} not found") + report = { + "generated_at": datetime.now(timezone.utc).isoformat(), + "schema": schema, + "cohort": cohort_stats(con, schema), + "missingness": missingness(con, schema), + "flags": flag_prevalence(con, schema), + "delays": delay_distributions(con, schema), + "onset_windows": onset_window_counts(con, schema), + } + if raw_schema: + report["coverage"] = coverage_stats(con, raw_schema, raw_hosp_schema) + return report + + +def render_markdown(report): + c = report["cohort"] + lines = [ + "# Diagnostic-delay data-quality report", + "", + f"Generated {report['generated_at']} from schema `{report['schema']}`.", + "", + "## Cohort coverage", + "", + f"* ICU stays in cohort: **{c['n_stays']}**", + f"* Patients: **{c['n_patients']}**; hospitalizations: **{c['n_hadm']}**", + f"* Admission window: {c['min_admittime']} to {c['max_admittime']}", + f"* Female: {c['n_female']} ({c['n_female'] / c['n_stays'] * 100:.1f}%)", + f"* Hospital deaths: {c['n_hospital_deaths']}" + f" ({c['n_hospital_deaths'] / c['n_stays'] * 100:.1f}%)", + f"* Admission age p25/median/p75: " + f"{c['admission_age_p25']:.0f} / {c['admission_age_p50']:.0f} /" + f" {c['admission_age_p75']:.0f}", + ] + if report.get("coverage"): + cov = report["coverage"] + denom = cov["n_adult_ed_icu_stays"] + lines.append( + f"* Capture rate vs adult ED ICU stays: " + f"{c['n_stays']}/{denom} = {c['n_stays'] / denom * 100:.1f}%" + if denom + else "* Capture rate: denominator unavailable" + ) + lines += ["", "## Missingness", ""] + for col, m in report["missingness"]["columns"].items(): + lines.append( + f"* `{col}`: {m['missing']} missing ({m['pct']:.2f}%)" + ) + lines += ["", "## Temporal-plausibility flags", ""] + for flag, f in report["flags"]["flags"].items(): + lines.append( + f"* `{flag}`: {f['flagged']} flagged ({f['pct']:.2f}%)" + + ( + f", {f['unassessable_null']} unassessable (NULL anchor)" + if f["unassessable_null"] + else "" + ) + ) + lines += ["", "## Delay distributions (hours)", ""] + lines.append("| measure | n | p25 | median | p75 | min | max |") + lines.append("|---|---|---|---|---|---|---|") + for col, d in report["delays"].items(): + lines.append( + f"| `{col}` | {d['n']} | {d['p25']:.1f} | {d['median']:.1f}" + f" | {d['p75']:.1f} | {d['min']:.1f} | {d['max']:.1f} |" + ) + lines += ["", "## Onset windows", ""] + for w in report["onset_windows"]: + lines.append(f"* `{w['onset_window']}`: {w['n']} stays") + lines.append("") + return "\n".join(lines) + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Data-quality report for the diagnostic-delay concepts." + ) + parser.add_argument("--duckdb", help="path to a DuckDB database file") + parser.add_argument( + "--postgres-dsn", help="psycopg2 DSN for a PostgreSQL database" + ) + parser.add_argument( + "--schema", default="mimiciv_derived", help="derived schema name" + ) + parser.add_argument( + "--raw-schema", + default=None, + help="raw ICU schema with icustays (enables the coverage " + "denominator)", + ) + parser.add_argument( + "--raw-hosp-schema", + default=None, + help="raw hospital schema with admissions/patients " + "(defaults to --raw-schema)", + ) + parser.add_argument( + "--output", default=None, help="write the Markdown report here" + ) + parser.add_argument( + "--json-output", default=None, help="write the raw report JSON here" + ) + args = parser.parse_args(argv) + if not args.duckdb and not args.postgres_dsn: + parser.error("one of --duckdb or --postgres-dsn is required") + + con = connect(args) + try: + report = build_report( + con, args.schema, args.raw_schema, args.raw_hosp_schema + ) + finally: + con.close() + + markdown = render_markdown(report) + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(markdown) + print(f"wrote Markdown report to {args.output}") + else: + print(markdown) + if args.json_output: + with open(args.json_output, "w", encoding="utf-8") as f: + json.dump(report, f, indent=2, default=str) + print(f"wrote JSON report to {args.json_output}") + + +if __name__ == "__main__": + main() diff --git a/mimic-iv/concepts/diagnosticdelay/diagnostic_delay.sql b/mimic-iv/concepts/diagnosticdelay/diagnostic_delay.sql new file mode 100644 index 000000000..22d1953a8 --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/diagnostic_delay.sql @@ -0,0 +1,110 @@ +-- ------------------------------------------------------------------ +-- Title: Diagnostic-delay measures for the suspected-sepsis cohort. +-- Description: One row per ICU stay in suspected_sepsis_cohort with the +-- delay (in hours) between hospital/ICU admission and the first +-- recognized sign of infection (suspected-infection time, first +-- antibiotic, first culture). An onset window classifies each stay as +-- pre_admission, present_on_admission (within 48 hours of admission), +-- or hospital_onset (after 48 hours). Plausibility flags mark stays +-- whose timestamps are out of order or missing so they can be +-- investigated rather than silently averaged in. +-- Note: ICD codes in MIMIC carry no timestamp, so "diagnosis time" is +-- proxied here by the clinically observable recognition events +-- (antibiotic order, culture draw) used in the Sepsis-3 definition. +-- Reference: https://github.com/MIT-LCP/mimic-code/issues/1843 +-- ------------------------------------------------------------------ + +WITH dd AS ( + SELECT + c.subject_id + , c.hadm_id + , c.stay_id + , c.admittime + , c.dischtime + , c.icu_intime + , c.icu_outtime + , c.n_suspicion_episodes + , c.first_suspicion_time + , c.first_antibiotic_time + , c.first_culture_time + -- delay measures, in hours (1 decimal) + , ROUND( + DATETIME_DIFF(c.first_suspicion_time, c.admittime, MINUTE) / 60.0 + , 1 + ) AS admission_to_suspicion_hours + , ROUND( + DATETIME_DIFF(c.first_suspicion_time, c.icu_intime, MINUTE) / 60.0 + , 1 + ) AS icu_to_suspicion_hours + , ROUND( + DATETIME_DIFF(c.first_antibiotic_time, c.admittime, MINUTE) / 60.0 + , 1 + ) AS admission_to_antibiotic_hours + , ROUND( + DATETIME_DIFF(c.first_culture_time, c.admittime, MINUTE) / 60.0 + , 1 + ) AS admission_to_culture_hours + , ROUND( + DATETIME_DIFF( + c.first_antibiotic_time, c.first_culture_time, MINUTE + ) / 60.0 + , 1 + ) AS culture_to_antibiotic_hours + FROM `physionet-data.mimiciv_derived.suspected_sepsis_cohort` c +) + +SELECT + dd.subject_id + , dd.hadm_id + , dd.stay_id + , dd.admittime + , dd.dischtime + , dd.icu_intime + , dd.icu_outtime + , dd.n_suspicion_episodes + , dd.first_suspicion_time + , dd.first_antibiotic_time + , dd.first_culture_time + , dd.admission_to_suspicion_hours + , dd.icu_to_suspicion_hours + , dd.admission_to_antibiotic_hours + , dd.admission_to_culture_hours + , dd.culture_to_antibiotic_hours + -- onset window relative to hospital admission. Classified from the raw + -- timestamps rather than the rounded delay measures above: e.g. a + -- suspicion 2 minutes before admission rounds to a delay of -0.0 (not + -- < 0) and a suspicion 48 hours 2 minutes after admission rounds to + -- 48.0 (not > 48), both of which the rounded form misclassified. + , CASE + WHEN dd.first_suspicion_time < dd.admittime THEN 'pre_admission' + WHEN dd.first_suspicion_time <= DATETIME_ADD( + dd.admittime, INTERVAL 48 HOUR + ) THEN 'present_on_admission' + ELSE 'hospital_onset' + END AS onset_window + -- temporal-plausibility flags (1 = implausible, 0 = plausible, + -- NULL = cannot be assessed because an anchor is missing) + , CASE + WHEN dd.first_suspicion_time < dd.admittime THEN 1 + ELSE 0 + END AS flag_suspicion_before_admission + , CASE + WHEN dd.icu_outtime IS NULL THEN NULL + WHEN dd.first_suspicion_time > dd.icu_outtime THEN 1 + ELSE 0 + END AS flag_suspicion_after_icu_outtime + , CASE + WHEN dd.dischtime IS NULL THEN NULL + WHEN dd.first_suspicion_time > dd.dischtime THEN 1 + ELSE 0 + END AS flag_suspicion_after_discharge + , CASE + WHEN dd.first_culture_time IS NULL THEN 1 + ELSE 0 + END AS flag_missing_culture_time + , CASE + WHEN dd.first_antibiotic_time IS NULL THEN 1 + ELSE 0 + END AS flag_missing_antibiotic_time +FROM dd +; diff --git a/mimic-iv/concepts/diagnosticdelay/suspected_sepsis_cohort.sql b/mimic-iv/concepts/diagnosticdelay/suspected_sepsis_cohort.sql new file mode 100644 index 000000000..de45cb97b --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/suspected_sepsis_cohort.sql @@ -0,0 +1,75 @@ +-- ------------------------------------------------------------------ +-- Title: Suspected-sepsis ICU cohort for diagnostic-delay measurement. +-- Description: One row per ICU stay for adult patients admitted through +-- the emergency department who have at least one suspected-infection +-- episode (an antibiotic paired with a microbiology culture, per the +-- suspicion_of_infection concept). Anchor timestamps and basic +-- demographics are carried along so that downstream queries can +-- measure the delay between hospital/ICU admission and the moment the +-- infection was first recognized (first antibiotic or culture). +-- Reference: community questions on diagnosis timing in MIMIC, e.g. +-- https://github.com/MIT-LCP/mimic-code/issues/1843 +-- ------------------------------------------------------------------ + +WITH soi AS ( + SELECT + soi.stay_id + , COUNT(*) AS n_suspicion_episodes + , MIN(soi.suspected_infection_time) AS first_suspicion_time + , MIN(soi.antibiotic_time) AS first_antibiotic_time + , MIN(soi.culture_time) AS first_culture_time + FROM `physionet-data.mimiciv_derived.suspicion_of_infection` soi + WHERE soi.suspected_infection = 1 + GROUP BY soi.stay_id +) + +SELECT + ie.subject_id + , ie.hadm_id + , ie.stay_id + + -- patient level factors + , pat.gender + -- calculate the age as anchor_age plus the difference between + -- the admit year and the anchor year (same formula as age.sql) + , pat.anchor_age + DATETIME_DIFF( + adm.admittime, DATETIME(pat.anchor_year, 1, 1, 0, 0, 0), YEAR + ) AS admission_age + , adm.race + + -- hospital level factors + , adm.admittime + , adm.dischtime + , adm.admission_type + , adm.admission_location + , adm.hospital_expire_flag + , DENSE_RANK() OVER ( + PARTITION BY adm.subject_id ORDER BY adm.admittime + ) AS hospstay_seq + + -- icu level factors + , ie.intime AS icu_intime + , ie.outtime AS icu_outtime + , DENSE_RANK() OVER ( + PARTITION BY ie.hadm_id ORDER BY ie.intime + ) AS icustay_seq + + -- suspected-infection summary (one row per antibiotic in soi) + , soi.n_suspicion_episodes + , soi.first_suspicion_time + , soi.first_antibiotic_time + , soi.first_culture_time +FROM `physionet-data.mimiciv_icu.icustays` ie +INNER JOIN `physionet-data.mimiciv_hosp.admissions` adm + ON ie.hadm_id = adm.hadm_id +INNER JOIN `physionet-data.mimiciv_hosp.patients` pat + ON ie.subject_id = pat.subject_id +-- inner join: only stays with at least one suspected-infection episode +INNER JOIN soi + ON ie.stay_id = soi.stay_id +-- adults only; ED origin only +WHERE pat.anchor_age + DATETIME_DIFF( + adm.admittime, DATETIME(pat.anchor_year, 1, 1, 0, 0, 0), YEAR + ) >= 18 + AND adm.admission_location = 'EMERGENCY ROOM' +; diff --git a/mimic-iv/concepts/diagnosticdelay/test_data_quality_report.py b/mimic-iv/concepts/diagnosticdelay/test_data_quality_report.py new file mode 100644 index 000000000..574977a90 --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/test_data_quality_report.py @@ -0,0 +1,263 @@ +#!/usr/bin/env python3 +"""Regression tests for data_quality_report.py. + +Covers the two report-generator issues fixed for PR #2173 review: + +* [P1] PostgreSQL: ``psycopg2`` connections have no ``execute`` method -- + queries must go through an explicit cursor (which is then closed). +* [P2] Split schemas: ``icustays`` lives in ``mimiciv_icu`` while + ``admissions``/``patients`` live in ``mimiciv_hosp`` -- the coverage + denominator must join across both schemas. + +Run with: ``python -m pytest test_data_quality_report.py -v`` +(duckdb must be installed: ``pip install duckdb``) +""" + +import sys +import os +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import data_quality_report as dqr # noqa: E402 + + +ICU_SCHEMA = "mimiciv_icu" +HOSP_SCHEMA = "mimiciv_hosp" +DERIVED_SCHEMA = "mimiciv_derived" + + +def _cohort_ddl(): + return """ + CREATE TABLE {s}.suspected_sepsis_cohort ( + subject_id INTEGER, + hadm_id INTEGER, + admittime TIMESTAMP, + dischtime TIMESTAMP, + gender VARCHAR, + hospital_expire_flag INTEGER, + admission_age DOUBLE + ) + """.format(s=DERIVED_SCHEMA) + + +def _delay_ddl(): + ts_cols = ",\n".join(f" {c} TIMESTAMP" for c in dqr.TIMESTAMP_COLUMNS) + delay_cols = ",\n".join(f" {c} DOUBLE" for c in dqr.DELAY_COLUMNS) + flag_cols = ",\n".join(f" {c} INTEGER" for c in dqr.FLAG_COLUMNS) + return f""" + CREATE TABLE {DERIVED_SCHEMA}.diagnostic_delay ( +{ts_cols}, +{delay_cols}, +{flag_cols}, + onset_window VARCHAR + ) + """ + + +def build_duckdb_split_schema(): + """In-memory DuckDB mirroring the standard MIMIC-IV schema layout.""" + import duckdb + + con = duckdb.connect(":memory:") + con.execute(f"CREATE SCHEMA {ICU_SCHEMA}") + con.execute(f"CREATE SCHEMA {HOSP_SCHEMA}") + con.execute(f"CREATE SCHEMA {DERIVED_SCHEMA}") + + # raw ICU schema: icustays only + con.execute( + f""" + CREATE TABLE {ICU_SCHEMA}.icustays ( + subject_id INTEGER, hadm_id INTEGER, stay_id INTEGER + ) + """ + ) + con.execute( + f"INSERT INTO {ICU_SCHEMA}.icustays VALUES (100, 200, 300)" + ) + + # raw hospital schema: admissions + patients + con.execute( + f""" + CREATE TABLE {HOSP_SCHEMA}.admissions ( + subject_id INTEGER, + hadm_id INTEGER, + admittime TIMESTAMP, + admission_location VARCHAR + ) + """ + ) + con.execute( + f""" + CREATE TABLE {HOSP_SCHEMA}.patients ( + subject_id INTEGER, anchor_age INTEGER, anchor_year INTEGER + ) + """ + ) + con.execute( + f""" + INSERT INTO {HOSP_SCHEMA}.admissions VALUES + (100, 200, '2020-01-01 10:00:00', 'EMERGENCY ROOM') + """ + ) + con.execute( + f"INSERT INTO {HOSP_SCHEMA}.patients VALUES (100, 60, 2020)" + ) + + # derived concept tables + con.execute(_cohort_ddl()) + con.execute( + f""" + INSERT INTO {DERIVED_SCHEMA}.suspected_sepsis_cohort VALUES + (100, 200, '2020-01-01 10:00:00', '2020-01-10 10:00:00', + 'F', 0, 60.0) + """ + ) + con.execute(_delay_ddl()) + ts_vals = ", ".join(["'2020-01-01 12:00:00'"] * len(dqr.TIMESTAMP_COLUMNS)) + delay_vals = ", ".join(["2.5"] * len(dqr.DELAY_COLUMNS)) + flag_vals = ", ".join(["0"] * len(dqr.FLAG_COLUMNS)) + con.execute( + f""" + INSERT INTO {DERIVED_SCHEMA}.diagnostic_delay VALUES + ({ts_vals}, {delay_vals}, {flag_vals}, 'present_on_admission') + """ + ) + return con + + +class FakePsycopg2Cursor: + """Minimal stand-in for a psycopg2 cursor.""" + + def __init__(self): + self.executed = [] + self.closed = False + + def execute(self, sql, params=()): + self.executed.append((sql, params)) + + def fetchone(self): + return (1,) + + def fetchall(self): + return [(1,)] + + @property + def description(self): + return [("n",)] + + def close(self): + self.closed = True + + +class FakePsycopg2Connection: + """Mimics psycopg2: cursor() exists, execute() does NOT.""" + + def __init__(self): + self.cursors = [] + self.closed = False + + def cursor(self): + cur = FakePsycopg2Cursor() + self.cursors.append(cur) + return cur + + def close(self): + self.closed = True + + +class TestPostgresCursorPath(unittest.TestCase): + """[P1] Queries must go through a cursor on psycopg2 connections.""" + + def test_fetchone_uses_cursor(self): + con = FakePsycopg2Connection() + # Would raise AttributeError before the fix (no .execute on con). + row = dqr.fetchone(con, "SELECT 1") + self.assertEqual(row, (1,)) + self.assertEqual(len(con.cursors), 1) + self.assertTrue(con.cursors[0].closed) + + def test_fetchall_uses_cursor(self): + con = FakePsycopg2Connection() + cols, rows = dqr.fetchall(con, "SELECT 1 AS n") + self.assertEqual(cols, ["n"]) + self.assertEqual(rows, [(1,)]) + self.assertEqual(len(con.cursors), 1) + self.assertTrue(con.cursors[0].closed) + + def test_duckdb_connection_still_executes_directly(self): + import duckdb + + con = duckdb.connect(":memory:") + try: + self.assertIs(dqr._cursor(con), con) + self.assertEqual(dqr.fetchone(con, "SELECT 42"), (42,)) + finally: + con.close() + + +class TestSplitSchemas(unittest.TestCase): + """[P2] Coverage denominator joins icustays (ICU) to admissions/patients + (hospital schema).""" + + def test_coverage_with_split_schemas(self): + con = build_duckdb_split_schema() + try: + cov = dqr.coverage_stats(con, ICU_SCHEMA, HOSP_SCHEMA) + self.assertIsNotNone(cov) + self.assertEqual(cov["n_adult_ed_icu_stays"], 1) + self.assertEqual(cov["n_patients"], 1) + finally: + con.close() + + def test_coverage_defaults_hosp_schema_to_icu_schema(self): + # Backwards compatible: single schema containing all three tables. + import duckdb + + con = duckdb.connect(":memory:") + try: + con.execute("CREATE SCHEMA raw") + con.execute( + "CREATE TABLE raw.icustays" + " (subject_id INTEGER, hadm_id INTEGER)" + ) + con.execute( + "CREATE TABLE raw.admissions (subject_id INTEGER," + " hadm_id INTEGER, admittime TIMESTAMP," + " admission_location VARCHAR)" + ) + con.execute( + "CREATE TABLE raw.patients (subject_id INTEGER," + " anchor_age INTEGER, anchor_year INTEGER)" + ) + con.execute("INSERT INTO raw.icustays VALUES (1, 10)") + con.execute( + "INSERT INTO raw.admissions VALUES" + " (1, 10, '2021-06-01 08:00:00', 'EMERGENCY ROOM')" + ) + con.execute("INSERT INTO raw.patients VALUES (1, 55, 2021)") + cov = dqr.coverage_stats(con, "raw") + self.assertIsNotNone(cov) + self.assertEqual(cov["n_adult_ed_icu_stays"], 1) + finally: + con.close() + + def test_full_report_with_split_schemas(self): + con = build_duckdb_split_schema() + try: + report = dqr.build_report( + con, DERIVED_SCHEMA, ICU_SCHEMA, HOSP_SCHEMA + ) + self.assertIn("coverage", report) + self.assertEqual( + report["coverage"]["n_adult_ed_icu_stays"], 1 + ) + self.assertEqual(report["cohort"]["n_stays"], 1) + md = dqr.render_markdown(report) + self.assertIn("Capture rate", md) + finally: + con.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/mimic-iv/concepts/diagnosticdelay/test_diagnostic_delay_sql.py b/mimic-iv/concepts/diagnosticdelay/test_diagnostic_delay_sql.py new file mode 100644 index 000000000..18778e152 --- /dev/null +++ b/mimic-iv/concepts/diagnosticdelay/test_diagnostic_delay_sql.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Regression tests for the diagnostic_delay.sql onset-window classification. + +The onset window must be classified from the raw timestamps, not the +rounded delay measures: a suspicion 2 minutes before admission rounds to a +delay of -0.0 (which is not < 0) and a suspicion 48 hours 2 minutes after +admission rounds to 48.0 (which is not > 48), so the rounded form +misclassified both boundary cases (PR #2173 review). + +The test transpiles the BigQuery concept SQL to DuckDB with the repository +transpiler (``src/mimic_utils/transpile.py``), runs it against a synthetic +``suspected_sepsis_cohort`` table, and checks the onset window on both +sides of the admission and admission+48h boundaries. + +Run with: ``python -m pytest test_diagnostic_delay_sql.py -v`` +(duckdb and sqlglot must be installed: ``pip install duckdb sqlglot``) +""" + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), + "..", "..", "..", "src")) + +from mimic_utils.transpile import transpile_query # noqa: E402 + +THIS_DIR = os.path.dirname(os.path.abspath(__file__)) +DERIVED_SCHEMA = "mimiciv_derived" + +# (case id, first_suspicion_time, expected onset_window) +# admittime is 2020-01-01 10:00:00 for every case. +BOUNDARY_CASES = [ + # both sides of the admission boundary + ("two_min_before", "2020-01-01 09:58:00", "pre_admission"), + ("one_min_before", "2020-01-01 09:59:00", "pre_admission"), + ("exactly_at_admission", "2020-01-01 10:00:00", "present_on_admission"), + ("one_min_after", "2020-01-01 10:01:00", "present_on_admission"), + # both sides of the admission + 48h boundary + ("just_inside_48h", "2020-01-03 09:59:00", "present_on_admission"), + ("exactly_48h", "2020-01-03 10:00:00", "present_on_admission"), + ("one_min_past_48h", "2020-01-03 10:01:00", "hospital_onset"), + ("two_min_past_48h", "2020-01-03 10:02:00", "hospital_onset"), +] + + +def _transpiled_sql(): + with open(os.path.join(THIS_DIR, "diagnostic_delay.sql")) as f: + bq = f.read() + return transpile_query(bq, "bigquery", "duckdb") + + +def build_duckdb_with_cohort(): + import duckdb + + con = duckdb.connect(":memory:") + con.execute(f"CREATE SCHEMA {DERIVED_SCHEMA}") + con.execute( + f""" + CREATE TABLE {DERIVED_SCHEMA}.suspected_sepsis_cohort ( + subject_id INTEGER, + hadm_id INTEGER, + stay_id INTEGER, + admittime TIMESTAMP, + dischtime TIMESTAMP, + icu_intime TIMESTAMP, + icu_outtime TIMESTAMP, + n_suspicion_episodes INTEGER, + first_suspicion_time TIMESTAMP, + first_antibiotic_time TIMESTAMP, + first_culture_time TIMESTAMP + ) + """ + ) + for i, (case_id, suspicion, _) in enumerate(BOUNDARY_CASES): + con.execute( + f""" + INSERT INTO {DERIVED_SCHEMA}.suspected_sepsis_cohort VALUES ( + 100, 200, {300 + i}, + '2020-01-01 10:00:00', '2020-01-10 10:00:00', + '2020-01-01 11:00:00', '2020-01-05 11:00:00', + 1, '{suspicion}', + '2020-01-01 12:00:00', '2020-01-01 11:30:00' + ) + """ + ) + return con + + +class TestOnsetWindowBoundaries(unittest.TestCase): + """Onset window is classified from raw timestamps at both boundaries.""" + + def test_boundary_cases(self): + con = build_duckdb_with_cohort() + try: + rows = con.execute( + _transpiled_sql() + ).fetchall() + got = {r[2]: r[16] for r in rows} # stay_id -> onset_window + self.assertEqual(len(got), len(BOUNDARY_CASES)) + for i, (case_id, _, expected) in enumerate(BOUNDARY_CASES): + self.assertEqual( + got[300 + i], expected, + f"case {case_id}: expected {expected}, got {got[300 + i]}", + ) + finally: + con.close() + + +if __name__ == "__main__": + unittest.main()