clinops.ingest

clinops.ingest.mimic_tables.MimicTableLoader

MimicTableLoader(
    mimic_path,
    version="auto",
    strict_validation=True,
    chunk_size=None,
)

Pre-built loader for the MIMIC-IV tables researchers use most.

Wraps :class:~clinops.ingest.MimicLoader and adds:

• Pre-validated schemas for chartevents, labevents, admissions, diagnoses_icd, and icustays.
• with_ref_range flag on labevents to retain or drop the reference range columns (noisy on many MIMIC exports).
• primary_only flag on diagnoses_icd to keep only seq_num == 1 (the principal diagnosis).
• with_los_band flag on icustays to add a categorical los_band column (<1d, 1-3d, 3-7d, >7d) useful as a stratification variable.
• A summary() method that prints row counts and null rates for all five tables without loading full data.

Parameters:

Name	Type	Description	Default
mimic_path	str | Path	Root directory of the MIMIC-IV dataset.	required
version	str	MIMIC-IV version string or "auto" (default).	'auto'
strict_validation	bool	Raise on missing required columns when True (default).	True
chunk_size	int | None	Pass through to underlying :class:MimicLoader for large tables.	None

Examples:

>>> tbl = MimicTableLoader("/data/mimic-iv-2.2")
>>> charts = tbl.chartevents(subject_ids=[10000032, 10000980])
>>> dx      = tbl.diagnoses_icd(subject_ids=[10000032], primary_only=True)
>>> stays   = tbl.icustays(subject_ids=[10000032], with_los_band=True)

Source code in clinops/ingest/mimic_tables.py

def __init__(
    self,
    mimic_path: str | Path,
    version: str = "auto",
    strict_validation: bool = True,
    chunk_size: int | None = None,
) -> None:
    self._path = Path(mimic_path)
    self._loader = MimicLoader(
        mimic_path=mimic_path,
        version=version,
        strict_validation=strict_validation,
        chunk_size=chunk_size,
    )
    logger.info(f"MimicTableLoader ready — {self._path}")

chartevents

chartevents(
    subject_ids=None,
    hadm_ids=None,
    stay_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
)

Load ICU charted observations with schema validation.

Returns a DataFrame with columns: subject_id, hadm_id, stay_id, itemid, charttime, valuenum, valueuom.

Parameters:

Name	Type	Description	Default
subject_ids	Sequence[int] | None	Restrict to these patients.	None
hadm_ids	Sequence[int] | None	Restrict to these hospital admissions.	None
stay_ids	Sequence[int] | None	Restrict to these ICU stays.	None
item_ids	Sequence[int] | None	Restrict to these MIMIC itemids (see d_items).	None
start_time	str | None	ISO datetime strings for time range filtering.	None
end_time	str | None	ISO datetime strings for time range filtering.	None

Source code in clinops/ingest/mimic_tables.py

def chartevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    stay_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
) -> pd.DataFrame:
    """
    Load ICU charted observations with schema validation.

    Returns a DataFrame with columns:
    ``subject_id``, ``hadm_id``, ``stay_id``, ``itemid``,
    ``charttime``, ``valuenum``, ``valueuom``.

    Parameters
    ----------
    subject_ids:
        Restrict to these patients.
    hadm_ids:
        Restrict to these hospital admissions.
    stay_ids:
        Restrict to these ICU stays.
    item_ids:
        Restrict to these MIMIC itemids (see ``d_items``).
    start_time, end_time:
        ISO datetime strings for time range filtering.
    """
    df = self._loader.chartevents(
        subject_ids=subject_ids,
        hadm_ids=hadm_ids,
        stay_ids=stay_ids,
        item_ids=item_ids,
        start_time=start_time,
        end_time=end_time,
    )
    df = df.copy()
    df["charttime"] = pd.to_datetime(df["charttime"], errors="coerce")
    _SCHEMAS["chartevents"].validate(df)
    logger.info(f"chartevents: {len(df):,} rows")
    return df

labevents

labevents(
    subject_ids=None,
    hadm_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
    with_ref_range=False,
)

Load hospital laboratory results.

Parameters:

Name	Type	Description	Default
with_ref_range	bool	If False (default), drop ref_range_lower / ref_range_upper columns — they are sparsely populated in most MIMIC exports and add noise to downstream pipelines. Set True to retain them.	False

Source code in clinops/ingest/mimic_tables.py

def labevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    with_ref_range: bool = False,
) -> pd.DataFrame:
    """
    Load hospital laboratory results.

    Parameters
    ----------
    with_ref_range:
        If ``False`` (default), drop ``ref_range_lower`` /
        ``ref_range_upper`` columns — they are sparsely populated in
        most MIMIC exports and add noise to downstream pipelines.
        Set ``True`` to retain them.
    """
    df = self._loader.labevents(
        subject_ids=subject_ids,
        hadm_ids=hadm_ids,
        item_ids=item_ids,
        start_time=start_time,
        end_time=end_time,
    )
    _SCHEMAS["labevents"].validate(df)
    if not with_ref_range:
        drop_cols = [c for c in ["ref_range_lower", "ref_range_upper"] if c in df.columns]
        df = df.drop(columns=drop_cols)
    logger.info(f"labevents: {len(df):,} rows (with_ref_range={with_ref_range})")
    return df

admissions

admissions(subject_ids=None, hadm_ids=None)

Load hospital admission records.

Returns a DataFrame with columns: subject_id, hadm_id, admittime, dischtime, deathtime, admission_type, admission_location, discharge_location, insurance, hospital_expire_flag.

Source code in clinops/ingest/mimic_tables.py

def admissions(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """
    Load hospital admission records.

    Returns a DataFrame with columns:
    ``subject_id``, ``hadm_id``, ``admittime``, ``dischtime``,
    ``deathtime``, ``admission_type``, ``admission_location``,
    ``discharge_location``, ``insurance``, ``hospital_expire_flag``.
    """
    df = self._loader.admissions(subject_ids=subject_ids, hadm_ids=hadm_ids)
    _SCHEMAS["admissions"].validate(df)
    logger.info(f"admissions: {len(df):,} rows")
    return df

diagnoses_icd

diagnoses_icd(
    subject_ids=None, hadm_ids=None, primary_only=False
)

Load ICD-9/ICD-10 diagnosis codes per hospital admission.

MIMIC-IV mixes ICD-9-CM and ICD-10-CM codes. The icd_version column contains 9 or 10 — use :class:~clinops.preprocess.ICDMapper to harmonize to a single version before modelling.

Parameters:

Name	Type	Description	Default
primary_only	bool	If True, keep only rows where seq_num == 1 (the principal/primary diagnosis per admission). Default False returns all coded diagnoses.	False

Source code in clinops/ingest/mimic_tables.py

def diagnoses_icd(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    primary_only: bool = False,
) -> pd.DataFrame:
    """
    Load ICD-9/ICD-10 diagnosis codes per hospital admission.

    MIMIC-IV mixes ICD-9-CM and ICD-10-CM codes.  The ``icd_version``
    column contains ``9`` or ``10`` — use
    :class:`~clinops.preprocess.ICDMapper` to harmonize to a single
    version before modelling.

    Parameters
    ----------
    primary_only:
        If ``True``, keep only rows where ``seq_num == 1`` (the
        principal/primary diagnosis per admission).  Default ``False``
        returns all coded diagnoses.
    """
    df = self._load_extra_table("diagnoses_icd")
    if subject_ids is not None:
        df = df[df["subject_id"].isin(subject_ids)]
    if hadm_ids is not None:
        df = df[df["hadm_id"].isin(hadm_ids)]
    if primary_only:
        df = df[df["seq_num"] == 1]
    _SCHEMAS["diagnoses_icd"].validate(df)
    logger.info(f"diagnoses_icd: {len(df):,} rows (primary_only={primary_only})")
    return df.reset_index(drop=True)

icustays

icustays(
    subject_ids=None,
    hadm_ids=None,
    stay_ids=None,
    with_los_band=False,
)

Load ICU stay metadata.

Parameters:

Name	Type	Description	Default
with_los_band	bool	If True, add a los_band categorical column bucketing length-of-stay into <1d, 1-3d, 3-7d, >7d. Useful as a stratification variable for cohort splits.	False

Source code in clinops/ingest/mimic_tables.py

def icustays(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    stay_ids: Sequence[int] | None = None,
    with_los_band: bool = False,
) -> pd.DataFrame:
    """
    Load ICU stay metadata.

    Parameters
    ----------
    with_los_band:
        If ``True``, add a ``los_band`` categorical column bucketing
        length-of-stay into ``<1d``, ``1-3d``, ``3-7d``, ``>7d``.
        Useful as a stratification variable for cohort splits.
    """
    df = self._loader.icustays(
        subject_ids=subject_ids,
        hadm_ids=hadm_ids,
        stay_ids=stay_ids,
    )
    _SCHEMAS["icustays"].validate(df)
    if with_los_band and "los" in df.columns:
        df = df.copy()
        df["los_band"] = pd.cut(
            df["los"],
            bins=[0, 1, 3, 7, float("inf")],
            labels=["<1d", "1-3d", "3-7d", ">7d"],
            right=True,
        )
    logger.info(f"icustays: {len(df):,} rows (with_los_band={with_los_band})")
    return df

summary

summary()

Print a quick-look table of row counts and null rates for all five tables without loading the full data into memory.

Uses pd.read_csv with nrows=0 to read headers, then scans only the first 10,000 rows to estimate null rates.

Returns:

Type	Description
DataFrame	Columns: table, rows_sampled, columns, null_rate_pct

Source code in clinops/ingest/mimic_tables.py

def summary(self) -> pd.DataFrame:
    """
    Print a quick-look table of row counts and null rates for all five
    tables without loading the full data into memory.

    Uses ``pd.read_csv`` with ``nrows=0`` to read headers, then scans
    only the first 10,000 rows to estimate null rates.

    Returns
    -------
    pd.DataFrame
        Columns: ``table``, ``rows_sampled``, ``columns``, ``null_rate_pct``
    """
    records = []
    for table_name in ["chartevents", "labevents", "admissions", "diagnoses_icd", "icustays"]:
        try:
            path = self._resolve_extra_path(table_name)
            sample = pd.read_csv(path, nrows=10_000, low_memory=False)
            null_rate = sample.isnull().mean().mean() * 100
            records.append(
                {
                    "table": table_name,
                    "rows_sampled": len(sample),
                    "columns": len(sample.columns),
                    "null_rate_pct": round(null_rate, 2),
                }
            )
        except FileNotFoundError:
            records.append(
                {
                    "table": table_name,
                    "rows_sampled": 0,
                    "columns": 0,
                    "null_rate_pct": None,
                }
            )
    result = pd.DataFrame(records)
    print(result.to_string(index=False))
    return result

clinops.ingest.mimic.MimicLoader

MimicLoader(
    mimic_path,
    version="auto",
    strict_validation=True,
    chunk_size=None,
)

Loader for MIMIC-IV clinical database tables.

Parameters:

Name	Type	Description	Default
mimic_path	str | Path	Root directory of the MIMIC-IV dataset. Should contain hosp/ and icu/ subdirectories.	required
version	str	MIMIC-IV version string (e.g. "2.2"). Use "auto" to detect from the directory structure.	'auto'
strict_validation	bool	If True (default), raise SchemaValidationError when required columns are missing. If False, log a warning and continue.	True
chunk_size	int | None	If set, return a pd.io.parsers.TextFileReader for large tables instead of loading the full table into memory.	None

Examples:

>>> loader = MimicLoader("/data/mimic-iv-2.2")
>>> charts = loader.chartevents(subject_ids=[10000032])
>>> labs   = loader.labevents(hadm_ids=[20000019])

Source code in clinops/ingest/mimic.py

def __init__(
    self,
    mimic_path: str | Path,
    version: str = "auto",
    strict_validation: bool = True,
    chunk_size: int | None = None,
) -> None:
    self._cfg = MimicLoaderConfig(
        mimic_path=Path(mimic_path),
        version=version,
        strict_validation=strict_validation,
        chunk_size=chunk_size,
    )
    self._version = self._detect_version() if version == "auto" else version
    logger.info(
        f"MimicLoader initialised — path={self._cfg.mimic_path}, version={self._version}"
    )

chartevents

chartevents(
    subject_ids=None,
    hadm_ids=None,
    stay_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
)

Load ICU charted observations (vitals, GCS, ventilator settings, etc.).

Parameters:

Name	Type	Description	Default
subject_ids	Sequence[int] | None	Filter to these patients.	None
hadm_ids	Sequence[int] | None	Filter to these hospital admissions.	None
stay_ids	Sequence[int] | None	Filter to these ICU stays.	None
item_ids	Sequence[int] | None	Filter to these MIMIC-IV itemids (see d_items).	None
start_time	str | None	ISO datetime string — exclude rows before this time.	None
end_time	str | None	ISO datetime string — exclude rows after this time.	None

Source code in clinops/ingest/mimic.py

def chartevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    stay_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
) -> pd.DataFrame:
    """
    Load ICU charted observations (vitals, GCS, ventilator settings, etc.).

    Parameters
    ----------
    subject_ids:
        Filter to these patients.
    hadm_ids:
        Filter to these hospital admissions.
    stay_ids:
        Filter to these ICU stays.
    item_ids:
        Filter to these MIMIC-IV itemids (see d_items).
    start_time:
        ISO datetime string — exclude rows before this time.
    end_time:
        ISO datetime string — exclude rows after this time.
    """
    df = self._load_table("chartevents")
    return self._filter(df, subject_ids, hadm_ids, stay_ids, item_ids, start_time, end_time)

labevents

labevents(
    subject_ids=None,
    hadm_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
)

Load hospital laboratory results.

Source code in clinops/ingest/mimic.py

def labevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
) -> pd.DataFrame:
    """Load hospital laboratory results."""
    df = self._load_table("labevents")
    return self._filter(df, subject_ids, hadm_ids, None, item_ids, start_time, end_time)

admissions

admissions(subject_ids=None, hadm_ids=None)

Load hospital admission records.

Source code in clinops/ingest/mimic.py

def admissions(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """Load hospital admission records."""
    df = self._load_table("admissions")
    return self._filter(df, subject_ids, hadm_ids)

patients

patients(subject_ids=None)

Load patient demographics.

Source code in clinops/ingest/mimic.py

def patients(
    self,
    subject_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """Load patient demographics."""
    df = self._load_table("patients")
    return self._filter(df, subject_ids)

icustays

icustays(subject_ids=None, hadm_ids=None, stay_ids=None)

Load ICU stay metadata including LOS.

Source code in clinops/ingest/mimic.py

def icustays(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    stay_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """Load ICU stay metadata including LOS."""
    df = self._load_table("icustays")
    return self._filter(df, subject_ids, hadm_ids, stay_ids)

prescriptions

prescriptions(subject_ids=None, hadm_ids=None, drugs=None)

Load medication prescriptions.

Source code in clinops/ingest/mimic.py

def prescriptions(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    drugs: Sequence[str] | None = None,
) -> pd.DataFrame:
    """Load medication prescriptions."""
    df = self._load_table("prescriptions")
    df = self._filter(df, subject_ids, hadm_ids)
    if drugs:
        df = df[df["drug"].str.lower().isin([d.lower() for d in drugs])]
    return df

inputevents

inputevents(subject_ids=None, stay_ids=None)

Load ICU fluid input events.

Source code in clinops/ingest/mimic.py

def inputevents(
    self,
    subject_ids: Sequence[int] | None = None,
    stay_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """Load ICU fluid input events."""
    df = self._load_table("inputevents")
    return self._filter(df, subject_ids, None, stay_ids)

d_items

d_items()

Load ICU item dictionary (maps itemid → label).

Source code in clinops/ingest/mimic.py

def d_items(self) -> pd.DataFrame:
    """Load ICU item dictionary (maps itemid → label)."""
    return self._load_table("d_items")

d_labitems

d_labitems()

Load lab item dictionary (maps itemid → label).

Source code in clinops/ingest/mimic.py

def d_labitems(self) -> pd.DataFrame:
    """Load lab item dictionary (maps itemid → label)."""
    return self._load_table("d_labitems")

clinops.ingest.mimic_iii.MimicIIILoader

MimicIIILoader(
    mimic_path, strict_validation=True, chunk_size=None
)

Loader for the MIMIC-III Clinical Database.

Provides the same filtering interface as MimicLoader (MIMIC-IV) so that analysis code can be reused across both datasets with minimal changes.

Key differences from MIMIC-IV: - Flat directory — no hosp/ / icu/ split. - Uppercase column names in source files — normalised to lowercase on load. - ICU stay key is icustay_id (not stay_id). - ICD-9-CM only — diagnoses_icd has icd9_code, not icd_code. - inputevents is split into inputevents_mv (MetaVision) and inputevents_cv (CareVue); use :meth:inputevents to get both merged.

Parameters:

Name	Type	Description	Default
mimic_path	str | Path	Root directory of the MIMIC-III dataset. Should contain files like CHARTEVENTS.csv.gz, ADMISSIONS.csv.gz, etc.	required
strict_validation	bool	If True (default), raise :exc:SchemaValidationError when required columns are missing. If False, log a warning and continue.	True
chunk_size	int | None	If set, large tables (chartevents, labevents) return a chunked reader instead of loading fully into memory.	None

Examples:

>>> loader = MimicIIILoader("/data/mimic-iii-clinical-database-1.4")
>>> charts = loader.chartevents(subject_ids=[40124])
>>> labs   = loader.labevents(hadm_ids=[198765])
>>> dx     = loader.diagnoses_icd(subject_ids=[40124], primary_only=True)

Source code in clinops/ingest/mimic_iii.py

def __init__(
    self,
    mimic_path: str | Path,
    strict_validation: bool = True,
    chunk_size: int | None = None,
) -> None:
    self._cfg = MimicIIIConfig(
        mimic_path=Path(mimic_path),
        strict_validation=strict_validation,
        chunk_size=chunk_size,
    )
    logger.info(f"MimicIIILoader initialised — path={self._cfg.mimic_path}")

chartevents

chartevents(
    subject_ids=None,
    hadm_ids=None,
    icustay_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
)

Load ICU charted observations.

Parameters:

Name	Type	Description	Default
subject_ids	Sequence[int] | None	Restrict to these patient IDs.	None
hadm_ids	Sequence[int] | None	Restrict to these hospital admission IDs.	None
icustay_ids	Sequence[int] | None	Restrict to these ICU stay IDs (icustay_id in MIMIC-III, equivalent to stay_id in MIMIC-IV).	None
item_ids	Sequence[int] | None	Restrict to these itemid values. See :meth:d_items.	None
start_time	str | None	Exclude rows with charttime before this ISO datetime string.	None
end_time	str | None	Exclude rows with charttime after this ISO datetime string.	None

Returns:

Type	Description
DataFrame	Columns (lowercase): subject_id, hadm_id, icustay_id, itemid, charttime (datetime), valuenum, valueuom.

Source code in clinops/ingest/mimic_iii.py

def chartevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    icustay_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
) -> pd.DataFrame:
    """
    Load ICU charted observations.

    Parameters
    ----------
    subject_ids:
        Restrict to these patient IDs.
    hadm_ids:
        Restrict to these hospital admission IDs.
    icustay_ids:
        Restrict to these ICU stay IDs (``icustay_id`` in MIMIC-III,
        equivalent to ``stay_id`` in MIMIC-IV).
    item_ids:
        Restrict to these ``itemid`` values. See :meth:`d_items`.
    start_time:
        Exclude rows with ``charttime`` before this ISO datetime string.
    end_time:
        Exclude rows with ``charttime`` after this ISO datetime string.

    Returns
    -------
    pd.DataFrame
        Columns (lowercase): ``subject_id``, ``hadm_id``,
        ``icustay_id``, ``itemid``, ``charttime`` (datetime),
        ``valuenum``, ``valueuom``.
    """
    df = self._load("chartevents")
    return self._filter(
        df,
        subject_ids,
        hadm_ids,
        icustay_ids=icustay_ids,
        item_ids=item_ids,
        start_time=start_time,
        end_time=end_time,
        time_col="charttime",
    )

labevents

labevents(
    subject_ids=None,
    hadm_ids=None,
    item_ids=None,
    start_time=None,
    end_time=None,
    with_ref_range=False,
)

Load hospital laboratory results.

Parameters:

Name	Type	Description	Default
subject_ids	Sequence[int] | None	Standard filters — see :meth:chartevents.	None
hadm_ids	Sequence[int] | None	Standard filters — see :meth:chartevents.	None
item_ids	Sequence[int] | None	Standard filters — see :meth:chartevents.	None
start_time	Sequence[int] | None	Standard filters — see :meth:chartevents.	None
end_time	Sequence[int] | None	Standard filters — see :meth:chartevents.	None
with_ref_range	bool	If True, retain ref_range_lower and ref_range_upper. Dropped by default to reduce memory footprint.	False

Returns:

Type	Description
DataFrame	Columns: subject_id, hadm_id, itemid, charttime (datetime), valuenum, valueuom.

Source code in clinops/ingest/mimic_iii.py

def labevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    item_ids: Sequence[int] | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    with_ref_range: bool = False,
) -> pd.DataFrame:
    """
    Load hospital laboratory results.

    Parameters
    ----------
    subject_ids, hadm_ids, item_ids, start_time, end_time:
        Standard filters — see :meth:`chartevents`.
    with_ref_range:
        If ``True``, retain ``ref_range_lower`` and ``ref_range_upper``.
        Dropped by default to reduce memory footprint.

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``hadm_id``, ``itemid``,
        ``charttime`` (datetime), ``valuenum``, ``valueuom``.
    """
    df = self._load("labevents")
    df = self._filter(
        df,
        subject_ids,
        hadm_ids,
        item_ids=item_ids,
        start_time=start_time,
        end_time=end_time,
        time_col="charttime",
    )
    if not with_ref_range:
        drop = [c for c in ["ref_range_lower", "ref_range_upper"] if c in df.columns]
        df = df.drop(columns=drop)
    return df

admissions

admissions(subject_ids=None, hadm_ids=None)

Load hospital admission and discharge records.

Returns:

Type	Description
DataFrame	Columns: subject_id, hadm_id, admittime, dischtime, deathtime, admission_type, and others.

Source code in clinops/ingest/mimic_iii.py

def admissions(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """
    Load hospital admission and discharge records.

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``hadm_id``, ``admittime``,
        ``dischtime``, ``deathtime``, ``admission_type``, and others.
    """
    df = self._load("admissions")
    return self._filter(df, subject_ids, hadm_ids)

diagnoses_icd

diagnoses_icd(
    subject_ids=None,
    hadm_ids=None,
    icd9_codes=None,
    primary_only=False,
)

Load ICD-9-CM diagnosis codes.

MIMIC-III uses ICD-9-CM exclusively. The column name is icd9_code (not icd_code as in MIMIC-IV). For cross-dataset compatibility, a synthetic icd_version column (always 9) is added on load.

Parameters:

Name	Type	Description	Default
subject_ids	Sequence[int] | None	Standard filters.	None
hadm_ids	Sequence[int] | None	Standard filters.	None
icd9_codes	Sequence[str] | None	Restrict to these ICD-9-CM codes (exact match, case-insensitive).	None
primary_only	bool	If True, return only the primary diagnosis (seq_num == 1) per admission.	False

Returns:

Type	Description
DataFrame	Columns: subject_id, hadm_id, seq_num, icd9_code, icd_version (always 9).

Source code in clinops/ingest/mimic_iii.py

def diagnoses_icd(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    icd9_codes: Sequence[str] | None = None,
    primary_only: bool = False,
) -> pd.DataFrame:
    """
    Load ICD-9-CM diagnosis codes.

    MIMIC-III uses ICD-9-CM exclusively. The column name is ``icd9_code``
    (not ``icd_code`` as in MIMIC-IV). For cross-dataset compatibility,
    a synthetic ``icd_version`` column (always 9) is added on load.

    Parameters
    ----------
    subject_ids, hadm_ids:
        Standard filters.
    icd9_codes:
        Restrict to these ICD-9-CM codes (exact match,
        case-insensitive).
    primary_only:
        If ``True``, return only the primary diagnosis (``seq_num == 1``)
        per admission.

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``hadm_id``, ``seq_num``,
        ``icd9_code``, ``icd_version`` (always 9).
    """
    df = self._load("diagnoses_icd")
    df = self._filter(df, subject_ids, hadm_ids)

    if icd9_codes is not None:
        if "icd9_code" in df.columns:
            df = df[df["icd9_code"].str.upper().isin([c.upper() for c in icd9_codes])]
        else:
            logger.warning(
                "Column 'icd9_code' missing from diagnoses_icd; skipping icd9_codes filter."
            )

    if primary_only:
        if "seq_num" in df.columns:
            df = df[df["seq_num"] == 1]
        else:
            logger.warning(
                "Column 'seq_num' missing from diagnoses_icd; "
                "cannot restrict to primary diagnoses."
            )

    # Add synthetic icd_version for cross-dataset compatibility
    if "icd_version" not in df.columns:
        df = df.assign(icd_version=9)

    return df.reset_index(drop=True)

icustays

icustays(subject_ids=None, hadm_ids=None, icustay_ids=None)

Load ICU stay metadata including length of stay.

Note: The ICU stay key in MIMIC-III is icustay_id, not stay_id as in MIMIC-IV.

Returns:

Type	Description
DataFrame	Columns: subject_id, hadm_id, icustay_id, first_careunit, last_careunit, intime (datetime), outtime (datetime), los (days, float).

Source code in clinops/ingest/mimic_iii.py

def icustays(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    icustay_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """
    Load ICU stay metadata including length of stay.

    Note: The ICU stay key in MIMIC-III is ``icustay_id``, not
    ``stay_id`` as in MIMIC-IV.

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``hadm_id``, ``icustay_id``,
        ``first_careunit``, ``last_careunit``, ``intime`` (datetime),
        ``outtime`` (datetime), ``los`` (days, float).
    """
    df = self._load("icustays")
    return self._filter(df, subject_ids, hadm_ids, icustay_ids=icustay_ids)

prescriptions

prescriptions(subject_ids=None, hadm_ids=None, drugs=None)

Load medication prescriptions.

Parameters:

Name	Type	Description	Default
drugs	Sequence[str] | None	Restrict to these drug names (case-insensitive substring match).	None

Returns:

Type	Description
DataFrame	Columns: subject_id, hadm_id, startdate, enddate, drug, dose_val_rx, dose_unit_rx.

Source code in clinops/ingest/mimic_iii.py

def prescriptions(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    drugs: Sequence[str] | None = None,
) -> pd.DataFrame:
    """
    Load medication prescriptions.

    Parameters
    ----------
    drugs:
        Restrict to these drug names (case-insensitive substring match).

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``hadm_id``, ``startdate``,
        ``enddate``, ``drug``, ``dose_val_rx``, ``dose_unit_rx``.
    """
    df = self._load("prescriptions")
    df = self._filter(df, subject_ids, hadm_ids)
    if drugs is not None:
        pattern = "|".join(drugs)
        df = df[df["drug"].str.contains(pattern, case=False, na=False)]
    return df

inputevents

inputevents(
    subject_ids=None,
    hadm_ids=None,
    icustay_ids=None,
    source="mv",
)

Load ICU fluid input events.

MIMIC-III stores MetaVision (INPUTEVENTS_MV) and CareVue (INPUTEVENTS_CV) inputs in separate tables.

Parameters:

Name	Type	Description	Default
source	str	"mv" (MetaVision, default), "cv" (CareVue), or "both" to load and concatenate both tables.	'mv'

Returns:

Type	Description
DataFrame	For MetaVision: subject_id, hadm_id, icustay_id, itemid, starttime, amount, amountuom. When source="both", also includes event_time (unified timestamp) and source ("mv" or "cv").

Source code in clinops/ingest/mimic_iii.py

def inputevents(
    self,
    subject_ids: Sequence[int] | None = None,
    hadm_ids: Sequence[int] | None = None,
    icustay_ids: Sequence[int] | None = None,
    source: str = "mv",
) -> pd.DataFrame:
    """
    Load ICU fluid input events.

    MIMIC-III stores MetaVision (``INPUTEVENTS_MV``) and CareVue
    (``INPUTEVENTS_CV``) inputs in separate tables.

    Parameters
    ----------
    source:
        ``"mv"`` (MetaVision, default), ``"cv"`` (CareVue), or
        ``"both"`` to load and concatenate both tables.

    Returns
    -------
    pd.DataFrame
        For MetaVision: ``subject_id``, ``hadm_id``, ``icustay_id``,
        ``itemid``, ``starttime``, ``amount``, ``amountuom``.
        When ``source="both"``, also includes ``event_time`` (unified
        timestamp) and ``source`` (``"mv"`` or ``"cv"``).
    """
    if source not in {"mv", "cv", "both"}:
        raise ValueError(f"source must be 'mv', 'cv', or 'both' — got {source!r}")

    dfs: list[pd.DataFrame] = []
    if source in {"mv", "both"}:
        df_mv = self._load("inputevents_mv")
        dfs.append(
            self._filter(
                df_mv,
                subject_ids,
                hadm_ids,
                icustay_ids=icustay_ids,
                time_col="starttime",
            )
        )
    if source in {"cv", "both"}:
        df_cv = self._load("inputevents_cv")
        dfs.append(
            self._filter(
                df_cv,
                subject_ids,
                hadm_ids,
                icustay_ids=icustay_ids,
                time_col="charttime",
            )
        )

    # When combining MetaVision and CareVue input events, normalise to a
    # common timestamp column and record the source system so callers have
    # a consistent schema regardless of the underlying table.
    if len(dfs) > 1:
        normalised: list[pd.DataFrame] = []
        for df in dfs:
            df_norm = df.copy()
            if "starttime" in df_norm.columns:
                df_norm["event_time"] = df_norm["starttime"]
                df_norm["source"] = "mv"
            elif "charttime" in df_norm.columns:
                df_norm["event_time"] = df_norm["charttime"]
                df_norm["source"] = "cv"
            else:
                # Fallback: no recognised time column; keep schema
                # consistent but leave event_time/source missing.
                df_norm["event_time"] = pd.NaT
                df_norm["source"] = pd.NA
            normalised.append(df_norm)
        result = pd.concat(normalised, ignore_index=True)
    else:
        result = dfs[0]
    return result

patients

patients(subject_ids=None)

Load patient demographics.

Returns:

Type	Description
DataFrame	Columns: subject_id, gender, dob, dod, dod_hosp, dod_ssn, expire_flag.

Source code in clinops/ingest/mimic_iii.py

def patients(
    self,
    subject_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """
    Load patient demographics.

    Returns
    -------
    pd.DataFrame
        Columns: ``subject_id``, ``gender``, ``dob``, ``dod``,
        ``dod_hosp``, ``dod_ssn``, ``expire_flag``.
    """
    df = self._load("patients")
    return self._filter(df, subject_ids)

d_items

d_items()

Load the item dictionary (itemid → label).

Use this to resolve itemid values in :meth:chartevents.

Returns:

Type	Description
DataFrame	Columns: itemid, label, abbreviation, dbsource, linksto, category, unitname.

Source code in clinops/ingest/mimic_iii.py

def d_items(self) -> pd.DataFrame:
    """
    Load the item dictionary (``itemid`` → label).

    Use this to resolve ``itemid`` values in :meth:`chartevents`.

    Returns
    -------
    pd.DataFrame
        Columns: ``itemid``, ``label``, ``abbreviation``,
        ``dbsource``, ``linksto``, ``category``, ``unitname``.
    """
    return self._load("d_items")

d_labitems

d_labitems()

Load the lab item dictionary (itemid → label).

Use this to resolve itemid values in :meth:labevents.

Returns:

Type	Description
DataFrame	Columns: itemid, label, fluid, category, loinc_code.

Source code in clinops/ingest/mimic_iii.py

def d_labitems(self) -> pd.DataFrame:
    """
    Load the lab item dictionary (``itemid`` → label).

    Use this to resolve ``itemid`` values in :meth:`labevents`.

    Returns
    -------
    pd.DataFrame
        Columns: ``itemid``, ``label``, ``fluid``, ``category``,
        ``loinc_code``.
    """
    return self._load("d_labitems")

clinops.ingest.eicu.EicuLoader

EicuLoader(
    data_dir, strict_validation=True, chunk_size=100000
)

Low-level reader for eICU-CRD CSV tables.

One method per table. Each returns a DataFrame with lowercase columns (eICU CSVs are already lowercase, so no normalisation is applied) and a minimal required-column check. Large tables are read in chunks; an optional patient_unit_stay_ids filter is pushed down into each chunk so a cohort subset never materialises the full table.

Parameters:

Name	Type	Description	Default
data_dir	str | Path	Directory of eICU-CRD CSV files.	required
strict_validation	bool	Raise on missing required columns when True (default).	True
chunk_size	int	Rows per chunk for large tables. Default 100_000.	100000

Examples:

>>> loader = EicuLoader("/data/eicu-crd")
>>> pt = loader.load_patient()
>>> vitals = loader.load_vital_periodic(patient_unit_stay_ids=[141168])

Source code in clinops/ingest/eicu.py

def __init__(
    self,
    data_dir: str | Path,
    strict_validation: bool = True,
    chunk_size: int = 100_000,
) -> None:
    self._cfg = EicuConfig(
        data_dir=Path(data_dir),
        strict_validation=strict_validation,
        chunk_size=chunk_size,
    )
    logger.info("EicuLoader initialised — data_dir=%s", self._cfg.data_dir)

load_patient

load_patient()

Load patient (one row per ICU stay). Small table — read whole.

Source code in clinops/ingest/eicu.py

def load_patient(self) -> pd.DataFrame:
    """Load ``patient`` (one row per ICU stay). Small table — read whole."""
    return self._load_small("patient")

load_diagnosis

load_diagnosis(patient_unit_stay_ids=None)

Load diagnosis (free-text + ICD-9). Small table — read whole.

Source code in clinops/ingest/eicu.py

def load_diagnosis(self, patient_unit_stay_ids: Sequence[int] | None = None) -> pd.DataFrame:
    """Load ``diagnosis`` (free-text + ICD-9). Small table — read whole."""
    df = self._load_small("diagnosis")
    return self._filter_stays(df, patient_unit_stay_ids)

load_vital_periodic

load_vital_periodic(
    patient_unit_stay_ids=None, columns=None
)

Load vitalPeriodic (~5-min resolution vitals). Chunked read.

Parameters:

Name	Type	Description	Default
patient_unit_stay_ids	Sequence[int] | None	Restrict to these ICU stays (pushed down per chunk).	None
columns	Sequence[str] | None	Subset of columns to read (usecols) — cuts memory and time on this wide table. patientunitstayid and observationoffset are always included.	None

Source code in clinops/ingest/eicu.py

def load_vital_periodic(
    self,
    patient_unit_stay_ids: Sequence[int] | None = None,
    columns: Sequence[str] | None = None,
) -> pd.DataFrame:
    """
    Load ``vitalPeriodic`` (~5-min resolution vitals). Chunked read.

    Parameters
    ----------
    patient_unit_stay_ids:
        Restrict to these ICU stays (pushed down per chunk).
    columns:
        Subset of columns to read (``usecols``) — cuts memory and time on
        this wide table. ``patientunitstayid`` and ``observationoffset`` are
        always included.
    """
    return self._load_large("vital_periodic", patient_unit_stay_ids, columns=columns)

load_vital_aperiodic

load_vital_aperiodic(
    patient_unit_stay_ids=None, columns=None
)

Load vitalAperiodic (non-invasive BP etc.). Chunked read.

Source code in clinops/ingest/eicu.py

def load_vital_aperiodic(
    self,
    patient_unit_stay_ids: Sequence[int] | None = None,
    columns: Sequence[str] | None = None,
) -> pd.DataFrame:
    """Load ``vitalAperiodic`` (non-invasive BP etc.). Chunked read."""
    return self._load_large("vital_aperiodic", patient_unit_stay_ids, columns=columns)

load_lab

load_lab(patient_unit_stay_ids=None, labnames=None)

Load lab (long-format lab results). Chunked read.

Parameters:

Name	Type	Description	Default
labnames	Sequence[str] | None	If given, keep only these raw labname values (case-insensitive), applied per chunk to avoid materialising unrelated labs.	None

Source code in clinops/ingest/eicu.py

def load_lab(
    self,
    patient_unit_stay_ids: Sequence[int] | None = None,
    labnames: Sequence[str] | None = None,
) -> pd.DataFrame:
    """
    Load ``lab`` (long-format lab results). Chunked read.

    Parameters
    ----------
    labnames:
        If given, keep only these raw ``labname`` values (case-insensitive),
        applied per chunk to avoid materialising unrelated labs.
    """
    lab_lc = {s.strip().lower() for s in labnames} if labnames else None

    def _row_filter(chunk: pd.DataFrame) -> pd.DataFrame:
        if lab_lc is not None and "labname" in chunk.columns:
            chunk = chunk[chunk["labname"].str.strip().str.lower().isin(lab_lc)]
        return chunk

    return self._load_large("lab", patient_unit_stay_ids, extra_filter=_row_filter)

load_medication

load_medication(patient_unit_stay_ids=None)

Load medication (free-text drug orders). Chunked read.

Source code in clinops/ingest/eicu.py

def load_medication(
    self,
    patient_unit_stay_ids: Sequence[int] | None = None,
) -> pd.DataFrame:
    """Load ``medication`` (free-text drug orders). Chunked read."""
    return self._load_large("medication", patient_unit_stay_ids)

load_nurse_charting

load_nurse_charting(
    patient_unit_stay_ids=None, categories=None
)

Load nurseCharting (~11 GB). Chunked read — never read whole.

Parameters:

Name	Type	Description	Default
categories	Sequence[str] | None	If given, keep only rows whose nursingchartcelltypecat is in this set (e.g. ["Scores"] for GCS). Applied per chunk.	None

Source code in clinops/ingest/eicu.py

def load_nurse_charting(
    self,
    patient_unit_stay_ids: Sequence[int] | None = None,
    categories: Sequence[str] | None = None,
) -> pd.DataFrame:
    """
    Load ``nurseCharting`` (~11 GB). Chunked read — never read whole.

    Parameters
    ----------
    categories:
        If given, keep only rows whose ``nursingchartcelltypecat`` is in
        this set (e.g. ``["Scores"]`` for GCS). Applied per chunk.
    """
    cats = set(categories) if categories else None

    def _row_filter(chunk: pd.DataFrame) -> pd.DataFrame:
        if cats is not None and "nursingchartcelltypecat" in chunk.columns:
            chunk = chunk[chunk["nursingchartcelltypecat"].isin(cats)]
        return chunk

    return self._load_large("nurse_charting", patient_unit_stay_ids, extra_filter=_row_filter)

clinops.ingest.eicu.EicuTableLoader

EicuTableLoader(
    data_dir,
    config=None,
    strict_validation=True,
    chunk_size=100000,
)

Build the ICHI-compatible feature matrix, labels, and sequences from eICU.

Wraps :class:EicuLoader and produces, for the selected cohort:

• a 22-feature hourly matrix (7 vitals, 9 labs, 3 derived, 3 treatment),
• five physiology-only organ-deterioration labels, and
• sliding-window sequences (X, y) ready for model training,

with a programmatic :class:~clinops.ingest.schema.LeakageError guard that refuses to emit any label-defining variable as a feature.

Downstream code (clinops.temporal, clinops.split) does not need to know the data came from eICU — column names are the canonical clinops names.

Parameters:

Name	Type	Description	Default
data_dir	str | Path	Directory of eICU-CRD CSV files.	required
config	EicuCohortConfig | None	:class:EicuCohortConfig. If None, ICHI-study defaults are used.	None
strict_validation	bool	Passed through to the underlying :class:EicuLoader.	True
chunk_size	int	Passed through to the underlying :class:EicuLoader for large tables.	100000

Examples:

>>> loader = EicuTableLoader("/data/eicu-crd")
>>> X, y = loader.build_sequences(observation_hours=24, prediction_hours=6)
>>> meta = loader.build_cohort()[
...     ["uniquepid", "patientunitstayid", "unittype", "age", "gender"]
... ]

Source code in clinops/ingest/eicu.py

def __init__(
    self,
    data_dir: str | Path,
    config: EicuCohortConfig | None = None,
    strict_validation: bool = True,
    chunk_size: int = 100_000,
) -> None:
    self._cfg = config or EicuCohortConfig()
    self._loader = EicuLoader(
        data_dir=data_dir,
        strict_validation=strict_validation,
        chunk_size=chunk_size,
    )
    self._cohort: pd.DataFrame | None = None
    logger.info("EicuTableLoader ready — data_dir=%s", data_dir)

build_cohort

build_cohort()

Select the analysis cohort (one row per included ICU stay).

Applies, in order: adult age filter, first-stay-per-patient, minimum ICU LOS, optional hospital restriction, DNR-within-6h exclusion, and an optional development cap. The hourly-missingness exclusion is applied later (in :meth:build_feature_matrix) because it needs the hourly grid.

Returns:

Type	Description
DataFrame	Columns include patientunitstayid, uniquepid, hospitalid, unittype, age (float, > 89 → 90), gender, los_hours, and apache_ii (may be NaN).

Source code in clinops/ingest/eicu.py

def build_cohort(self) -> pd.DataFrame:
    """
    Select the analysis cohort (one row per included ICU stay).

    Applies, in order: adult age filter, first-stay-per-patient, minimum
    ICU LOS, optional hospital restriction, DNR-within-6h exclusion, and an
    optional development cap. The hourly-missingness exclusion is applied
    later (in :meth:`build_feature_matrix`) because it needs the hourly grid.

    Returns
    -------
    pd.DataFrame
        Columns include ``patientunitstayid``, ``uniquepid``, ``hospitalid``,
        ``unittype``, ``age`` (float, ``> 89`` → 90), ``gender``,
        ``los_hours``, and ``apache_ii`` (may be NaN).
    """
    if self._cohort is not None:
        return self._cohort

    pt = self._loader.load_patient().copy()
    n0 = len(pt)

    pt["age"] = pt["age"].map(parse_eicu_age)
    pt = pt[pt["age"] >= self._cfg.min_age]
    logger.info("Cohort: age>=%d → %d/%d stays", self._cfg.min_age, len(pt), n0)

    if self._cfg.hospital_ids is not None and "hospitalid" in pt.columns:
        pt = pt[pt["hospitalid"].isin(self._cfg.hospital_ids)]
        logger.info("Cohort: hospital filter → %d stays", len(pt))

    # Minimum ICU LOS from the unit discharge offset (minutes → hours).
    pt["los_hours"] = pd.to_numeric(pt["unitdischargeoffset"], errors="coerce") / 60.0
    pt = pt[pt["los_hours"] >= self._cfg.min_los_hours]
    logger.info("Cohort: LOS>=%dh → %d stays", self._cfg.min_los_hours, len(pt))

    # First ICU stay per patient.
    if self._cfg.first_stay_only:
        if "unitvisitnumber" in pt.columns:
            order = ["unitvisitnumber", "patientunitstayid"]
        else:
            order = ["patientunitstayid"]
        pt = pt.sort_values(order).groupby("uniquepid", as_index=False, sort=False).first()
        logger.info("Cohort: first stay per patient → %d stays", len(pt))

    # DNR within the configured window (default 6h) of admission → exclude.
    dnr_stays = self._dnr_stays(within_minutes=self._cfg.dnr_window_hours * 60)
    if dnr_stays:
        before = len(pt)
        pt = pt[~pt["patientunitstayid"].isin(dnr_stays)]
        logger.info("Cohort: DNR<=6h excluded %d stays", before - len(pt))

    if self._cfg.limit_stays is not None:
        pt = pt.head(self._cfg.limit_stays)
        logger.info("Cohort: dev limit_stays → %d stays", len(pt))

    # Stay-level APACHE severity (optional join).
    pt = pt.merge(self._apache_scores(), on="patientunitstayid", how="left")

    self._cohort = pt.reset_index(drop=True)
    logger.info("Cohort finalised: %d stays", len(self._cohort))
    return self._cohort

build_feature_matrix

build_feature_matrix()

Return the hourly 22-feature matrix, leakage-guarded.

Raises:

Type	Description
LeakageError	If any column on the programmatic exclusion list (:data:LEAKAGE_FORBIDDEN) is present in the returned feature set.

Returns:

Type	Description
DataFrame	Columns: patientunitstayid, hour, then :data:FEATURE_COLUMNS.

Source code in clinops/ingest/eicu.py

def build_feature_matrix(self) -> pd.DataFrame:
    """
    Return the hourly 22-feature matrix, leakage-guarded.

    Raises
    ------
    LeakageError
        If any column on the programmatic exclusion list
        (:data:`LEAKAGE_FORBIDDEN`) is present in the returned feature set.

    Returns
    -------
    pd.DataFrame
        Columns: ``patientunitstayid``, ``hour``, then :data:`FEATURE_COLUMNS`.
    """
    hourly = self._build_hourly()
    hourly = self._drop_high_missingness(hourly)

    feature_cols = list(FEATURE_COLUMNS)
    leaked = LEAKAGE_FORBIDDEN & set(feature_cols)
    if leaked:
        raise LeakageError(
            f"Label-defining variable(s) {sorted(leaked)} found in the "
            f"feature matrix. These are derived only to compute labels and "
            f"must never be model inputs."
        )
    for col in feature_cols:
        if col not in hourly.columns:
            hourly[col] = np.nan
    return hourly[["patientunitstayid", "hour", *feature_cols]].reset_index(drop=True)

build_labels

build_labels()

Return the per-(stay, hour) organ-deterioration labels.

Returns:

Type	Description
DataFrame	Columns: patientunitstayid, hour, then :data:ORGAN_LABELS.

Source code in clinops/ingest/eicu.py

def build_labels(self) -> pd.DataFrame:
    """
    Return the per-(stay, hour) organ-deterioration labels.

    Returns
    -------
    pd.DataFrame
        Columns: ``patientunitstayid``, ``hour``, then :data:`ORGAN_LABELS`.
    """
    hourly = self._build_hourly()
    hourly = self._drop_high_missingness(hourly)
    return hourly[["patientunitstayid", "hour", *ORGAN_LABELS]].reset_index(drop=True)

build_sequences

build_sequences(
    observation_hours=None,
    prediction_hours=None,
    max_icu_hours=None,
)

Build sliding-window sequences for model training.

For each stay, an hourly window of observation_hours produces one sample; its label is whether each organ deteriorates at any point in the following prediction_hours. Windows slide hourly.

Parameters:

Name	Type	Description	Default
observation_hours	int | None	Override the configured values for this call.	None
prediction_hours	int | None	Override the configured values for this call.	None
max_icu_hours	int | None	Override the configured values for this call.	None

Returns:

Type	Description
(X, y) : tuple of np.ndarray	X has shape (n_windows, observation_hours, 22); y has shape (n_windows, 5) (one binary label per organ).

Source code in clinops/ingest/eicu.py

def build_sequences(
    self,
    observation_hours: int | None = None,
    prediction_hours: int | None = None,
    max_icu_hours: int | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Build sliding-window sequences for model training.

    For each stay, an hourly window of ``observation_hours`` produces one
    sample; its label is whether each organ deteriorates at any point in the
    following ``prediction_hours``. Windows slide hourly.

    Parameters
    ----------
    observation_hours, prediction_hours, max_icu_hours:
        Override the configured values for this call.

    Returns
    -------
    (X, y) : tuple of np.ndarray
        ``X`` has shape ``(n_windows, observation_hours, 22)``;
        ``y`` has shape ``(n_windows, 5)`` (one binary label per organ).
    """
    # Apply per-call overrides for the duration of the build, then restore —
    # so a one-off call never silently mutates the loader's state, and the
    # overrides reach every consumer consistently (the grid via
    # max_icu_hours, the windowing via observation/prediction_hours, and the
    # baselines via baseline_hours, all read from self._cfg). Overrides are
    # routed through a fresh EicuCohortConfig so they are validated the same
    # way as the constructor (positive values; observation + prediction <=
    # max_icu_hours) rather than silently producing empty/invalid windows.
    overrides = {
        "observation_hours": observation_hours,
        "prediction_hours": prediction_hours,
        "max_icu_hours": max_icu_hours,
    }
    overrides = {k: v for k, v in overrides.items() if v is not None}
    if not overrides:
        return self._build_sequences_impl()

    saved_cfg = self._cfg
    self._cfg = EicuCohortConfig(**{**saved_cfg.model_dump(), **overrides})
    try:
        return self._build_sequences_impl()
    finally:
        self._cfg = saved_cfg

clinops.ingest.eicu.EicuCohortConfig

Bases: BaseModel

Cohort-selection and windowing configuration for :class:EicuTableLoader.

Defaults mirror the ICHI 2026 study design.

Parameters:

Name	Type	Description	Default
min_age		Minimum age in years (adults). Default 18.	required
min_los_hours		Minimum ICU length of stay in hours. Default 48.	required
max_icu_hours		Hours of each stay to materialise on the hourly grid. Default 72.	required
observation_hours		Length of the input window. Default 24.	required
prediction_hours		Length of the forward label horizon. Default 6.	required
baseline_hours		Window (from ICU admission) used to compute each patient's physiological baseline (lowest creatinine, best GCS) for the renal and neurological labels. Default 6. Kept separate from prediction_hours — they are unrelated concepts that merely share a default.	required
missingness_threshold		Stays whose critical-variable missingness on the hourly grid exceeds this fraction are excluded. Default 0.30.	required
dnr_window_hours		Stays with a DNR / care-limitation order within this many hours of admission are excluded. Default 6.	required
first_stay_only		Keep only the first ICU stay per uniquepid. Default True.	required
hospital_ids		Optional restriction to specific hospitalid values (multi-site subgroup analysis).	required
limit_stays		Optional cap on the number of cohort stays (development / testing on a subset). None (default) uses the full cohort.	required
random_state		Seed for any sampling. Default 42.	required

clinops.ingest.fhir.FHIRLoader

FHIRLoader(source)

Load FHIR R4 resources from JSON bundles or NDJSON exports.

Parameters:

Name	Type	Description	Default
source	str | Path	Path to a FHIR JSON Bundle file, an NDJSON file, or a directory of JSON resource files.	required

Examples:

>>> loader = FHIRLoader("/data/fhir_export")
>>> observations = loader.observations()
>>> patients     = loader.patients()

Source code in clinops/ingest/fhir.py

def __init__(self, source: str | Path) -> None:
    self._source = Path(source)
    if not self._source.exists():
        raise FileNotFoundError(f"FHIR source not found: {self._source}")
    logger.info(f"FHIRLoader initialised — source={self._source}")

patients

patients()

Load Patient resources → DataFrame with demographics.

Source code in clinops/ingest/fhir.py

def patients(self) -> pd.DataFrame:
    """Load Patient resources → DataFrame with demographics."""
    records = self._load_resources("Patient")
    rows = []
    for r in records:
        rows.append(
            {
                "patient_id": r.get("id"),
                "gender": r.get("gender"),
                "birth_date": r.get("birthDate"),
                "deceased": r.get("deceasedBoolean", False),
            }
        )
    return pd.DataFrame(rows)

observations

observations(category=None, loinc_codes=None)

Load Observation resources → long-format DataFrame.

Parameters:

Name	Type	Description	Default
category	str | None	Filter to a FHIR observation category (e.g. "vital-signs", "laboratory").	None
loinc_codes	list[str] | None	Filter to specific LOINC codes.	None

Source code in clinops/ingest/fhir.py

def observations(
    self,
    category: str | None = None,
    loinc_codes: list[str] | None = None,
) -> pd.DataFrame:
    """
    Load Observation resources → long-format DataFrame.

    Parameters
    ----------
    category:
        Filter to a FHIR observation category (e.g. "vital-signs", "laboratory").
    loinc_codes:
        Filter to specific LOINC codes.
    """
    records = self._load_resources("Observation")
    rows = []
    for r in records:
        code_obj = r.get("code", {})
        codings = code_obj.get("coding", [])
        loinc = next(
            (c["code"] for c in codings if "loinc" in c.get("system", "").lower()), None
        )
        value = r.get("valueQuantity", {})
        rows.append(
            {
                "observation_id": r.get("id"),
                "patient_id": r.get("subject", {}).get("reference", "").split("/")[-1],
                "loinc_code": loinc,
                "display": code_obj.get("text"),
                "value": value.get("value"),
                "unit": value.get("unit"),
                "effective_time": r.get("effectiveDateTime"),
                "status": r.get("status"),
            }
        )
    df = pd.DataFrame(rows)
    if loinc_codes:
        df = df[df["loinc_code"].isin(loinc_codes)]
    return df

conditions

conditions()

Load Condition resources → DataFrame with ICD/SNOMED codes.

Source code in clinops/ingest/fhir.py

def conditions(self) -> pd.DataFrame:
    """Load Condition resources → DataFrame with ICD/SNOMED codes."""
    records = self._load_resources("Condition")
    rows = []
    for r in records:
        codings = r.get("code", {}).get("coding", [])
        rows.append(
            {
                "condition_id": r.get("id"),
                "patient_id": r.get("subject", {}).get("reference", "").split("/")[-1],
                "code": codings[0].get("code") if codings else None,
                "system": codings[0].get("system") if codings else None,
                "display": codings[0].get("display") if codings else None,
                "clinical_status": r.get("clinicalStatus", {})
                .get("coding", [{}])[0]
                .get("code"),
                "onset": r.get("onsetDateTime"),
            }
        )
    return pd.DataFrame(rows)

clinops.ingest.flat.FlatFileLoader

FlatFileLoader(
    path,
    schema=None,
    id_col=None,
    datetime_cols=None,
    strict=True,
)

Load clinical data from CSV or Parquet flat files with validation.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a CSV (.csv, .csv.gz) or Parquet (.parquet, .pq) file.	required
schema	ClinicalSchema | None	Optional ClinicalSchema for validation after loading.	None
id_col	str | None	Name of the patient/subject identifier column. Used for deduplication reporting.	None
datetime_cols	list[str] | None	Column names to parse as datetimes. If None, auto-detection is attempted for columns with "time", "date", or "dt" in the name.	None
strict	bool	If True, raise on schema violations. If False, warn and continue.	True

Examples:

>>> loader = FlatFileLoader("vitals_export.csv", id_col="patient_id")
>>> df = loader.load()
>>> print(loader.summary())

Source code in clinops/ingest/flat.py

def __init__(
    self,
    path: str | Path,
    schema: ClinicalSchema | None = None,
    id_col: str | None = None,
    datetime_cols: list[str] | None = None,
    strict: bool = True,
) -> None:
    self._path = Path(path)
    self._schema = schema
    self._id_col = id_col
    self._datetime_cols = datetime_cols
    self._strict = strict
    self._loaded_df: pd.DataFrame | None = None

    if not self._path.exists():
        raise FileNotFoundError(f"File not found: {self._path}")

load

load()

Load the file, apply cleaning and validation, return DataFrame.

Returns:

Type	Description
DataFrame

Source code in clinops/ingest/flat.py

def load(self) -> pd.DataFrame:
    """
    Load the file, apply cleaning and validation, return DataFrame.

    Returns
    -------
    pd.DataFrame
    """
    logger.info(f"Loading flat file: {self._path}")
    df = self._read_file()
    df = self._clean(df)
    df = self._parse_datetimes(df)

    if self._schema:
        violations = self._schema.validate(df, strict=self._strict)
        if violations and not self._strict:
            for v in violations:
                logger.warning(v)

    self._loaded_df = df
    logger.info(f"Loaded {len(df):,} rows × {len(df.columns)} cols from {self._path.name}")
    return df

summary

summary()

Return a human-readable summary of the loaded DataFrame.

Source code in clinops/ingest/flat.py

def summary(self) -> str:
    """Return a human-readable summary of the loaded DataFrame."""
    if self._loaded_df is None:
        return "No data loaded yet. Call .load() first."
    df = self._loaded_df
    null_counts = df.isna().sum()
    null_summary = null_counts[null_counts > 0].to_dict()
    lines = [
        f"File:        {self._path.name}",
        f"Rows:        {len(df):,}",
        f"Columns:     {len(df.columns)}",
        f"Dtypes:      {df.dtypes.value_counts().to_dict()}",
        f"Null cols:   {null_summary if null_summary else 'None'}",
    ]
    if self._id_col and self._id_col in df.columns:
        lines.append(f"Unique IDs:  {df[self._id_col].nunique():,}")
    return "\n".join(lines)

clinops.ingest.schema.ClinicalSchema dataclass

ClinicalSchema(
    name, columns=list(), allow_extra_columns=True
)

Declarative schema for a clinical data table.

Parameters:

Name	Type	Description	Default
name	str	Human-readable name for this schema (used in error messages).	required
columns	list[ColumnSpec]	List of ColumnSpec objects describing required and optional columns.	list()
allow_extra_columns	bool	If True (default), columns not in the spec are silently allowed.	True

Example

schema = ClinicalSchema( ... name="vitals", ... columns=[ ... ColumnSpec("subject_id", dtype="int64", nullable=False), ... ColumnSpec("heart_rate", dtype="float64", min_value=0, max_value=300), ... ] ... ) schema.validate(df)

validate

validate(df, strict=True)

Validate a DataFrame against this schema.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame to validate.	required
strict	bool	If True, raise SchemaValidationError on the first violation. If False, collect all violations and return them as a list.	True

Returns:

Type	Description
list[str]	Empty list if valid; list of violation messages otherwise.

Source code in clinops/ingest/schema.py

def validate(self, df: pd.DataFrame, strict: bool = True) -> list[str]:
    """
    Validate a DataFrame against this schema.

    Parameters
    ----------
    df:
        DataFrame to validate.
    strict:
        If True, raise SchemaValidationError on the first violation.
        If False, collect all violations and return them as a list.

    Returns
    -------
    list[str]
        Empty list if valid; list of violation messages otherwise.
    """
    violations: list[str] = []

    for spec in self.columns:
        if spec.name not in df.columns:
            msg = f"[{self.name}] Missing required column: '{spec.name}'"
            violations.append(msg)
            if strict:
                raise SchemaValidationError(msg)
            continue

        col = df[spec.name]

        # Nullability check
        if not spec.nullable and col.isna().any():
            null_count = col.isna().sum()
            msg = (
                f"[{self.name}] Column '{spec.name}' has "
                f"{null_count} null values (nullable=False)"
            )
            violations.append(msg)
            if strict:
                raise SchemaValidationError(msg)

        # Range checks (numeric only)
        if spec.min_value is not None and pd.api.types.is_numeric_dtype(col):
            out_of_range = (col < spec.min_value).sum()
            if out_of_range:
                msg = (
                    f"[{self.name}] Column '{spec.name}' has {out_of_range} values "
                    f"below min_value={spec.min_value}"
                )
                violations.append(msg)
                if strict:
                    raise SchemaValidationError(msg)

        if spec.max_value is not None and pd.api.types.is_numeric_dtype(col):
            out_of_range = (col > spec.max_value).sum()
            if out_of_range:
                msg = (
                    f"[{self.name}] Column '{spec.name}' has {out_of_range} values "
                    f"above max_value={spec.max_value}"
                )
                violations.append(msg)
                if strict:
                    raise SchemaValidationError(msg)

        # Allowed values check
        if spec.allowed_values:
            invalid = ~col.isin(spec.allowed_values)
            invalid_count = invalid.sum()
            if invalid_count:
                msg = (
                    f"[{self.name}] Column '{spec.name}' has {invalid_count} values "
                    f"not in allowed_values={spec.allowed_values}"
                )
                violations.append(msg)
                if strict:
                    raise SchemaValidationError(msg)

    return violations

clinops.ingest.schema.ColumnSpec dataclass

ColumnSpec(
    name,
    dtype=None,
    nullable=True,
    min_value=None,
    max_value=None,
    allowed_values=list(),
)

Specification for a single column in a clinical table.

clinops.ingest.schema.SchemaValidationError

Bases: ValueError

Raised when a loaded table does not match the expected schema.