Data

---

Available preprocessing operations

The name of a preprocessing operation is the name of a transformer class. PreprocessorRegistry resolves the name by first checking the operations registered with it, then searching these modules in order:

Module	Examples
sklearn.preprocessing	StandardScaler, OrdinalEncoder
sklearn.impute	SimpleImputer

The name must match the class name exactly. Any extra keys of an operation are passed to the class constructor.

[workflow.preprocessing]
preprocess = true
operations = [
  { name = "StandardScaler", columns = ["AGE"], with_mean = true },
]

Adding a preprocessing operation

Decorate a transformer class with PreprocessorRegistry.register. The decorator takes the name to register the class under, which defaults to the class name. The operation can then be used by name in the configuration file.

from sklearn.base import BaseEstimator, OneToOneFeatureMixin, TransformerMixin
from sklearn.utils.validation import validate_data

from medpipe.data import PreprocessorRegistry


@PreprocessorRegistry.register()
class ClipOutliers(OneToOneFeatureMixin, TransformerMixin, BaseEstimator):
    def __init__(self, lower=0.0, upper=1.0):
        self.lower = lower
        self.upper = upper

    def fit(self, X, y=None):
        validate_data(self, X)  # Records the names of the columns
        return self

    def transform(self, X):
        return X.clip(self.lower, self.upper)

The operations are chained as pandas data frames, so a transformer has to know the names of its output columns. OneToOneFeatureMixin provides them for a transformer that keeps one output column per input column, as long as fit records the column names with validate_data. Without them, the run stops with "Unable to configure output ... because set_output is not available".

[workflow.preprocessing]
preprocess = true
operations = [
  { name = "ClipOutliers", columns = ["AGE"], lower = 0, upper = 100 },
]

Register the operation before the pipeline is run. Its name is looked up when the preprocessing is built, not when the configuration is validated.

A complete script that registers a model, a preprocessing operation and a metric is shown in the examples.

---

registry

PreprocessorRegistry

Bases: BaseRegistry[type[Callable]]

Registry for managing and resolving data preprocessing operations.

Source code in medpipe/data/registry.py

class PreprocessorRegistry(BaseRegistry[type[Callable]]):
    """
    Registry for managing and resolving data preprocessing operations.
    """

    _fallback_modules: ClassVar[list[ModuleType]] = [
        sklearn.preprocessing,
        sklearn.impute,
    ]

utils

Utility functions module.

This module provides utility functions for data manipulation.

Functions: - get_split_idx: Returns the indices for the data splits. - split_data: Split data into train and test or train and recalibration sets. - extract_labels: Extracts prediction labels from data.

get_split_idx

get_split_idx(
    idx_list: NDArray,
    column: Series | NDArray,
    values: list[str] | list[int],
) -> tuple[NDArray, NDArray]

Returns the indices for the data splits.

Parameters:

Name	Type	Description	Default
idx_list	NDArray	Indices of the set to split of shape (n_samples,).	required
column	Series | NDArray	Column of shape (n_samples,) used to split the data.	required
values	list[str] | list[int]	Group values that should be in the test or recalibration set.	required

Returns:

Name	Type	Description
train_idx	NDArray	Train indices.
other_idx	NDArray	Other indices for the test or recalibration set.

Raises:

Type	Description
TypeError	If column is not pd.Series or np.ndarray. If values is not a list or a np.ndarray.

Source code in medpipe/data/utils.py

def get_split_idx(
    idx_list: npt.NDArray,
    column: pd.Series | npt.NDArray,
    values: list[str] | list[int],
) -> tuple[npt.NDArray, npt.NDArray]:
    """
    Returns the indices for the data splits.

    Parameters
    ----------
    idx_list : npt.NDArray
        Indices of the set to split of shape (n_samples,).
    column : pd.Series | npt.NDArray
        Column of shape (n_samples,) used to split the data.
    values : list[str] | list[int]
        Group values that should be in the test or recalibration set.

    Returns
    -------
    train_idx : npt.NDArray
        Train indices.
    other_idx : npt.NDArray
        Other indices for the test or recalibration set.

    Raises
    ------
    TypeError
        If column is not pd.Series or np.ndarray.
        If values is not a list or a np.ndarray.

    """
    check_array(idx_list, ensure_2d=False)

    # Standardize groups to numpy
    if isinstance(column, pd.Series):
        column = column.to_numpy()
    elif not isinstance(column, np.ndarray):
        raise TypeError(f"column should be pd.Series or np.array, got {type(column)}")

    check_consistent_length(idx_list, column)  # Ensure dimension match

    # Type checking validation
    if not isinstance(values, (list, np.ndarray)):
        raise TypeError(f"values should be list or np.array, got {type(values)}")

    # Vectorized selection: Find where 'column' matches any value in 'values'
    val_mask = np.isin(column, values)
    other_idx = np.where(val_mask)[0]
    train_idx = np.where(~val_mask)[0]

    if len(other_idx) == 0:
        # Other indices are empty because value was not in column
        raise ValueError(f"{values} not present in column")

    return train_idx, other_idx

split_data

split_data(
    features: DataFrame,
    labels: NDArray,
    strategy: Literal["random", "group"],
    group_column: str | None = None,
    values: list[str] | list[int] | None = None,
    test_size: float | None = None,
    recalibration_size: float | None = None,
    random_state: int | None = None,
) -> tuple[DataFrame, NDArray, DataFrame, NDArray]

Split data into train and test or train and recalibration sets.

If strategy is group then group_column and values must be specified. If strategy is random then test_size or recalibration_size must be specified. If both are specified, the selected value is test_size.

Parameters:

Name	Type	Description	Default
features	DataFrame	Features to split.	required
labels	NDArray	Labels to split.	required
strategy	('random', 'group')	Strategy used to split the data.	"random"
group_column	str | None	Name of the column used to split with if strategy is group.	None
values	list[str] | list[int] | None	Values of the group column that do not belong to the train set.	None
test_size	float | None	Test set size if the strategy is random.	None
recalibration_size	float | None	Recalibration set size if the strategy is random.	None
random_state	int | None	Seed controlling the shuffling applied before the random split, for reproducibility. Ignored when strategy is group.	None

Returns:

Type	Description
X_train, X_test : pd.DataFrame	Train and test / recalibration set.
y_train, y_test : npt.NDArray	Train and test / recalibration labels.

Raises:

Type	Description
TypeError	If features is not a pd.DataFrame. If labels is not a np.ndarray.
ValueError	If group_colum and values not specified with group strategy. If test_size or recalibration_size not specified with random strategy.

Source code in medpipe/data/utils.py

def split_data(
    features: pd.DataFrame,
    labels: npt.NDArray,
    strategy: Literal["random", "group"],
    group_column: str | None = None,
    values: list[str] | list[int] | None = None,
    test_size: float | None = None,
    recalibration_size: float | None = None,
    random_state: int | None = None,
) -> tuple[pd.DataFrame, npt.NDArray, pd.DataFrame, npt.NDArray]:
    """
    Split data into train and test or train and recalibration sets.

    If strategy is group then group_column and values must be specified.
    If strategy is random then test_size or recalibration_size must be
    specified. If both are specified, the selected value is test_size.

    Parameters
    ----------
    features : pd.DataFrame
        Features to split.
    labels : npt.NDArray
        Labels to split.
    strategy : {"random", "group"}
        Strategy used to split the data.
    group_column : str | None, default: None
        Name of the column used to split with if strategy is group.
    values : list[str] | list[int] | None, default: None
        Values of the group column that do not belong to the train set.
    test_size : float | None, default: None
        Test set size if the strategy is random.
    recalibration_size : float | None, default: None
        Recalibration set size if the strategy is random.
    random_state : int | None, default: None
        Seed controlling the shuffling applied before the random split, for
        reproducibility. Ignored when strategy is group.

    Returns
    -------
    X_train, X_test : pd.DataFrame
        Train and test / recalibration set.
    y_train, y_test : npt.NDArray
        Train and test / recalibration labels.

    Raises
    ------
    TypeError
        If features is not a pd.DataFrame.
        If labels is not a np.ndarray.
    ValueError
        If group_colum and values not specified with group strategy.
        If test_size or recalibration_size not specified with random strategy.

    """
    if not isinstance(features, pd.DataFrame):
        raise TypeError(f"features should be a pd.DataFrame, but got {type(features)}")
    if not isinstance(labels, np.ndarray):
        raise TypeError(f"labels should be a np.array, but got {type(labels)}")

    if strategy == "group":
        if not group_column or not values:
            raise ValueError(
                "group_column and values must be specified with group strategy"
            )

        train_idx, test_idx = get_split_idx(
            np.arange(len(features)),
            features[group_column],  # type: ignore
            values,  # type: ignore
        )

        X_train = features.iloc[train_idx]
        y_train = labels[train_idx]
        X_test = features.iloc[test_idx]
        y_test = labels[test_idx]

    elif strategy == "random":
        if test_size:
            size = test_size
        elif recalibration_size:
            size = recalibration_size
        else:
            raise ValueError(
                "test_size or recalibration_size must be specified with random strategy"
            )

        X_train, X_test, y_train, y_test = train_test_split(
            features, labels, test_size=size, random_state=random_state
        )

    else:
        raise ValueError(f"strategy should be random or group, but got {strategy}")

    return (
        cast(pd.DataFrame, X_train),
        cast(npt.NDArray, y_train),
        cast(pd.DataFrame, X_test),
        cast(npt.NDArray, y_test),
    )

extract_labels

extract_labels(
    data: DataFrame, labels: list[str]
) -> tuple[DataFrame, NDArray]

Extracts the prediction labels from the training data.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame to manipulate.	required
labels	list[str]	List of labels to extract from the data.	required

Returns:

Name	Type	Description
X	DataFrame	DataFrame containing the data.
y	NDArray	Array containing the prediction labels.

Raises:

Type	Description
TypeError	If data is not a pd.DataFrame.
TypeError	If labels is not list[str].
ValueError	If a prediction label is not a valid key.

Source code in medpipe/data/utils.py

def extract_labels(
    data: pd.DataFrame, labels: list[str]
) -> tuple[pd.DataFrame, npt.NDArray]:
    """
    Extracts the prediction labels from the training data.

    Parameters
    ----------
    data : pd.DataFrame
        DataFrame to manipulate.
    labels : list[str]
        List of labels to extract from the data.

    Returns
    -------
    X : pd.DataFrame
        DataFrame containing the data.
    y : npt.NDArray
        Array containing the prediction labels.

    Raises
    ------
    TypeError
        If data is not a pd.DataFrame.
    TypeError
        If labels is not list[str].
    ValueError
        If a prediction label is not a valid key.

    """
    if not isinstance(data, pd.DataFrame):
        raise TypeError(f"data should be a pd.DataFrame, but got {type(data)}")
    if not isinstance(labels, list):
        raise TypeError(f"labels should be a list, but got {type(labels)}")
    if labels and not isinstance(labels[0], str):
        raise TypeError(f"labels should contain strings, but got {type(labels[0])}")

    # .drop() and column selection are already highly optimized in Pandas
    for label in labels:
        if label not in data.columns:
            raise ValueError(f"{label} was not found in data")

    X = data.drop(columns=labels)
    y = data[labels].to_numpy()

    return X, y

resolve_subgroup_mask

resolve_subgroup_mask(
    df: DataFrame, column: str, group: Any
) -> Series

Generate a boolean mask for a dataset matching a stratum group definition.

Ranges are inclusive so [18, 50] will include 18 and 50 in the range.

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataset containing the stratum column.	required
column	str	Name of the column to stratify on.	required
group	Any	Group definition. Can be a discrete scalar ('M'), a range tuple (18, 65), a pd.Interval.	required

Returns:

Type	Description
Series	Boolean mask of shape (n_samples,) matching the group criteria.

Raises:

Type	Description
KeyError	If column is not present in df.
ValueError	If group is a tuple or list that does not have exactly 2 elements.

Source code in medpipe/data/utils.py

def resolve_subgroup_mask(
    df: pd.DataFrame,
    column: str,
    group: Any,
) -> pd.Series:
    """
    Generate a boolean mask for a dataset matching a stratum group definition.

    Ranges are inclusive so [18, 50] will include 18 and 50 in the range.

    Parameters
    ----------
    df : pd.DataFrame
        Dataset containing the stratum column.
    column : str
        Name of the column to stratify on.
    group : Any
        Group definition. Can be a discrete scalar ('M'), a range tuple (18, 65),
        a pd.Interval.

    Returns
    -------
    pd.Series
        Boolean mask of shape (n_samples,) matching the group criteria.

    Raises
    ------
    KeyError
        If `column` is not present in `df`.
    ValueError
        If `group` is a tuple or list that does not have exactly 2 elements.

    """
    if column not in df.columns:
        raise KeyError(f"Stratum column '{column}' not found in DataFrame.")

    col_data = df[column]

    # Column is already a Pandas Categorical with pd.Interval values
    if isinstance(group, pd.Interval):
        return col_data == group

    # Group passed as a tuple/list of bounds: (min, max)
    elif isinstance(group, (tuple, list)) and len(group) == 2:
        lower, upper = group[0], group[1]
        return (col_data >= lower) & (col_data <= upper)

    # A tuple/list of any other length is not a valid range definition, and
    # would otherwise silently fall through to a positional/length-mismatched
    # equality comparison below.
    elif isinstance(group, (tuple, list)):
        raise ValueError(
            "Range group definitions must have exactly 2 elements (min, max), "
            f"but got {len(group)}: {group!r}"
        )

    # Standard discrete scalar equality (e.g. string, int, float)
    return col_data == group