Skip to content

Utils

bootstrap

Bootstrap resampling helpers shared by the metric and visualisation modules.

draw_bootstrap_indices 

draw_bootstrap_indices(
    rng: Generator, n_samples: int, n_bootstraps: int
) -> NDArray

Draw all bootstrap resample indices up front, sequentially, so the random draw sequence (and thus reproducibility for a given rng state) is independent of how the resamples are later computed or dispatched.

Parameters:

Name Type Description Default
rng Generator

Random generator the resamples are drawn from.

 required
n_samples int

Number of samples in the data being resampled.

 required
n_bootstraps int

Number of resamples to draw.

 required

Returns:

Type Description
NDArray

Integer array of shape (n_bootstraps, n_samples) whose rows are the resample indices, drawn with replacement.
Source code in medpipe/utils/bootstrap.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def draw_bootstrap_indices(
    rng: np.random.Generator, n_samples: int, n_bootstraps: int
) -> npt.NDArray:
    """Draw all bootstrap resample indices up front, sequentially, so the
    random draw sequence (and thus reproducibility for a given rng state) is
    independent of how the resamples are later computed or dispatched.

    Parameters
    ----------
    rng : np.random.Generator
        Random generator the resamples are drawn from.
    n_samples : int
        Number of samples in the data being resampled.
    n_bootstraps : int
        Number of resamples to draw.

    Returns
    -------
    npt.NDArray
        Integer array of shape (n_bootstraps, n_samples) whose rows are the
        resample indices, drawn with replacement.

    """
    return rng.integers(0, n_samples, size=(n_bootstraps, n_samples))

run_bootstraps 

run_bootstraps(
    worker: Callable[..., Any],
    bootstrap_indices: NDArray,
    *args: Any,
    n_jobs: int | None,
    backend: Literal["threading", "loky"],
) -> list[Any]

Apply worker(idx, *args) to every bootstrap resample, in parallel when n_jobs allows it, and drop resamples the worker rejected.

Parameters:

Name Type Description Default
worker callable

Module-level function taking a resample's indices followed by args and returning its result, or None if the resample is unusable.

 required
bootstrap_indices ndarray

Resample indices, one row per resample, as returned by draw_bootstrap_indices.

 required
*args Any

Extra positional arguments forwarded unchanged to every worker call.

 ()
n_jobs int or None

Number of parallel jobs (the display section's n_jobs).

 required
backend ('threading', 'loky')

Joblib backend. Benchmarks showed numpy-bound resamples (ROC/PR curves) gain from threads, since process start-up and pickling outweigh their millisecond cost, while resamples dominated by Python-level loops (predictive-distribution intervals, spline fitting) only gain from processes.

 "threading"

Returns:

Type Description
list

The non-None results, in resample order.
Source code in medpipe/utils/bootstrap.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def run_bootstraps(
    worker: Callable[..., Any],
    bootstrap_indices: npt.NDArray,
    *args: Any,
    n_jobs: int | None,
    backend: Literal["threading", "loky"],
) -> list[Any]:
    """Apply `worker(idx, *args)` to every bootstrap resample, in parallel
    when `n_jobs` allows it, and drop resamples the worker rejected.

    Parameters
    ----------
    worker : callable
        Module-level function taking a resample's indices followed by `args`
        and returning its result, or `None` if the resample is unusable.
    bootstrap_indices : numpy.ndarray
        Resample indices, one row per resample, as returned by `draw_bootstrap_indices`.
    *args : Any
        Extra positional arguments forwarded unchanged to every `worker`
        call.
    n_jobs : int or None
        Number of parallel jobs (the display section's `n_jobs`).
    backend : {"threading", "loky"}
        Joblib backend. Benchmarks showed numpy-bound resamples (ROC/PR
        curves) gain from threads, since process start-up and pickling
        outweigh their millisecond cost, while resamples dominated by
        Python-level loops (predictive-distribution intervals, spline
        fitting) only gain from processes.

    Returns
    -------
    list
        The non-`None` results, in resample order.

    """
    results = Parallel(n_jobs=n_jobs, backend=backend)(
        delayed(worker)(idx, *args) for idx in bootstrap_indices
    )
    return [r for r in results if r is not None]

io

I/O utilities module.

This module provides helper functions for reading from and writing to files, handling various common I/O tasks.

DataLoaderRegistry

Registry managing file extension mappings to DataFrame reader functions.

Source code in medpipe/utils/io.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class DataLoaderRegistry:
    """Registry managing file extension mappings to DataFrame reader functions."""

    _registry: ClassVar[dict[str, Callable[..., Any]]] = {
        ".csv": pd.read_csv,
        ".tsv": lambda filepath, **kwargs: pd.read_csv(filepath, sep="\t", **kwargs),
        ".txt": pd.read_csv,
        ".parquet": pd.read_parquet,
        ".pq": pd.read_parquet,
        ".feather": pd.read_feather,
        ".xlsx": pd.read_excel,
        ".xls": pd.read_excel,
        ".json": pd.read_json,
        ".jsonl": lambda filepath, **kwargs: pd.read_json(
            filepath, lines=True, **kwargs
        ),
        ".pkl": pd.read_pickle,
        ".pickle": pd.read_pickle,
    }

    @classmethod
    def register(cls, extension: str) -> Callable[[Callable], Callable]:
        """Decorator to register a custom file reader for an extension.

        Parameters
        ----------
        extension : str
            File extension the reader handles, with or without the leading
            dot (e.g., '.rds' or 'rds'). Matching is case-insensitive.

        Returns
        -------
        decorator : Callable[[Callable], Callable]
            Decorator registering a function that accepts a file path and
            returns a DataFrame.

        """

        def decorator(func: Callable[..., pd.DataFrame]) -> Callable[..., pd.DataFrame]:
            ext = cls._normalize_ext(extension)
            cls._registry[ext] = func
            return func

        return decorator

    @classmethod
    def get(cls, extension: str) -> Callable[..., pd.DataFrame]:
        """Retrieve the reader function for a given file extension.

        Parameters
        ----------
        extension : str
            File extension, with or without the leading dot.

        Returns
        -------
        Callable[..., pandas.DataFrame]
            Reader function registered for the extension.

        Raises
        ------
        ValueError
            If no reader is registered for the extension.

        """
        ext = cls._normalize_ext(extension)
        if ext not in cls._registry:
            raise ValueError(
                f"Unsupported file extension '{ext}'. "
                f"Registered extensions: {cls.list_registered()}"
            )
        return cls._registry[ext]

    @classmethod
    def list_registered(cls) -> list[str]:
        """List all supported file extensions.

        Returns
        -------
        list of str
            Registered file extensions, each with a leading dot.

        """
        return list(cls._registry.keys())

    @staticmethod
    def _normalize_ext(extension: str) -> str:
        """Ensure extensions are lowercased and prefixed with a dot."""
        ext = extension.lower().strip()
        return ext if ext.startswith(".") else f".{ext}"

register classmethod 

register(extension: str) -> Callable[[Callable], Callable]

Decorator to register a custom file reader for an extension.

Parameters:

Name Type Description Default
extension str

File extension the reader handles, with or without the leading dot (e.g., '.rds' or 'rds'). Matching is case-insensitive.

 required

Returns:

Name Type Description
decorator Callable[[Callable], Callable]

Decorator registering a function that accepts a file path and returns a DataFrame.
Source code in medpipe/utils/io.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@classmethod
def register(cls, extension: str) -> Callable[[Callable], Callable]:
    """Decorator to register a custom file reader for an extension.

    Parameters
    ----------
    extension : str
        File extension the reader handles, with or without the leading
        dot (e.g., '.rds' or 'rds'). Matching is case-insensitive.

    Returns
    -------
    decorator : Callable[[Callable], Callable]
        Decorator registering a function that accepts a file path and
        returns a DataFrame.

    """

    def decorator(func: Callable[..., pd.DataFrame]) -> Callable[..., pd.DataFrame]:
        ext = cls._normalize_ext(extension)
        cls._registry[ext] = func
        return func

    return decorator

get classmethod 

get(extension: str) -> Callable[..., DataFrame]

Retrieve the reader function for a given file extension.

Parameters:

Name Type Description Default
extension str

File extension, with or without the leading dot.

 required

Returns:

Type Description
Callable[..., DataFrame]

Reader function registered for the extension.

Raises:

Type Description
ValueError

If no reader is registered for the extension.
Source code in medpipe/utils/io.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def get(cls, extension: str) -> Callable[..., pd.DataFrame]:
    """Retrieve the reader function for a given file extension.

    Parameters
    ----------
    extension : str
        File extension, with or without the leading dot.

    Returns
    -------
    Callable[..., pandas.DataFrame]
        Reader function registered for the extension.

    Raises
    ------
    ValueError
        If no reader is registered for the extension.

    """
    ext = cls._normalize_ext(extension)
    if ext not in cls._registry:
        raise ValueError(
            f"Unsupported file extension '{ext}'. "
            f"Registered extensions: {cls.list_registered()}"
        )
    return cls._registry[ext]

list_registered classmethod 

list_registered() -> list[str]

List all supported file extensions.

Returns:

Type Description
list of str

Registered file extensions, each with a leading dot.
Source code in medpipe/utils/io.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def list_registered(cls) -> list[str]:
    """List all supported file extensions.

    Returns
    -------
    list of str
        Registered file extensions, each with a leading dot.

    """
    return list(cls._registry.keys())

register_data_loader 

register_data_loader(
    extension: str, loader_func: Callable[..., Any]
) -> None

Register a custom file reader function for a specific file extension.

Parameters:

Name Type Description Default
extension str

File extension starting with a dot (e.g., '.feather', '.rds').

 required
loader_func Callable[..., Any]

Function accepting a file path and returning a Pandas DataFrame.

 required
Source code in medpipe/utils/io.py 
114
115
116
117
118
119
120
121
122
123
124
125
def register_data_loader(extension: str, loader_func: Callable[..., Any]) -> None:
    """Register a custom file reader function for a specific file extension.

    Parameters
    ----------
    extension : str
        File extension starting with a dot (e.g., '.feather', '.rds').
    loader_func : Callable[..., Any]
        Function accepting a file path and returning a Pandas DataFrame.

    """
    DataLoaderRegistry.register(extension)(loader_func)

load_data 

load_data(
    data_file: str | Path, **kwargs: Any
) -> DataFrame

Reads a supported data file and returns its contents as a DataFrame.

Parameters:

Name Type Description Default
data_file str or Path

Path to the data file to load.

 required
**kwargs Any

Additional keyword arguments passed to the underlying Pandas reader function (e.g., sheet_name for Excel, sep for CSV).

 {}

Returns:

Name Type Description
data DataFrame

Loaded tabular data.

Raises:

Type Description
TypeError

If data_file is not a str or Path.
FileNotFoundError

If data_file does not exist.
IsADirectoryError

If data_file is a directory rather than a file.
ValueError

If data_file has an unsupported file extension.
Source code in medpipe/utils/io.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def load_data(data_file: str | Path, **kwargs: Any) -> pd.DataFrame:
    """Reads a supported data file and returns its contents as a DataFrame.

    Parameters
    ----------
    data_file : str or Path
        Path to the data file to load.
    **kwargs : Any
        Additional keyword arguments passed to the underlying Pandas reader function
        (e.g., `sheet_name` for Excel, `sep` for CSV).

    Returns
    -------
    data : pd.DataFrame
        Loaded tabular data.

    Raises
    ------
    TypeError
        If data_file is not a str or Path.
    FileNotFoundError
        If data_file does not exist.
    IsADirectoryError
        If data_file is a directory rather than a file.
    ValueError
        If data_file has an unsupported file extension.

    """
    file_path = Path(data_file)
    supported_extensions = DataLoaderRegistry.list_registered()

    # Validate the extension case-insensitively, consistent with how
    # DataLoaderRegistry itself resolves loaders (e.g. '.CSV' == '.csv').
    normalized_suffix = DataLoaderRegistry._normalize_ext(file_path.suffix)
    if normalized_suffix not in supported_extensions:
        raise ValueError(
            f"File suffix should be one of {supported_extensions}, "
            f"but got {file_path.suffix}"
        )

    # Extension already validated above, so this only checks existence
    # and that the path is a file rather than a directory.
    file_checks(file_path, file_path.suffix)

    loader = DataLoaderRegistry.get(normalized_suffix)

    return cast(pd.DataFrame, loader(file_path, **kwargs))

read_classifier_toml_configuration 

read_classifier_toml_configuration(
    config_file: str | Path,
) -> MedpipeClassifierConfig

Reads a MedPipe TOML configuration file.

Parameters:

Name Type Description Default
config_file str | Path

Path to the configuration file.

 required

Returns:

Name Type Description
config MedpipeClassifierConfig

Configuration for the pipeline.
Source code in medpipe/utils/io.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def read_classifier_toml_configuration(
    config_file: str | Path,
) -> MedpipeClassifierConfig:
    """
    Reads a MedPipe TOML configuration file.

    Parameters
    ----------
    config_file : str | Path
        Path to the configuration file.

    Returns
    -------
    config : MedpipeClassifierConfig
        Configuration for the pipeline.

    """
    file_checks(config_file, ".toml")

    with open(config_file, "rb") as file:
        raw_config = tomllib.load(file)

    config = MedpipeClassifierConfig.model_validate(raw_config)

    return config

read_regressor_toml_configuration 

read_regressor_toml_configuration(
    config_file: str | Path,
) -> MedpipeRegressorConfig

Reads a MedPipe regression TOML configuration file.

Parameters:

Name Type Description Default
config_file str | Path

Path to the configuration file.

 required

Returns:

Name Type Description
config MedpipeRegressorConfig

Configuration for the regression pipeline.
Source code in medpipe/utils/io.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def read_regressor_toml_configuration(
    config_file: str | Path,
) -> MedpipeRegressorConfig:
    """
    Reads a MedPipe regression TOML configuration file.

    Parameters
    ----------
    config_file : str | Path
        Path to the configuration file.

    Returns
    -------
    config : MedpipeRegressorConfig
        Configuration for the regression pipeline.

    """
    file_checks(config_file, ".toml")

    with open(config_file, "rb") as file:
        raw_config = tomllib.load(file)

    config = MedpipeRegressorConfig.model_validate(raw_config)

    return config

logger

medpipe.utils.logger

Provides centralized logging configuration for the medpipe package. Supports routing high-level logs to the console and detailed debugging logs to specific experiment artifact directories.

CompactProgressFilter

Bases: Filter

Filter that passes only high-level workflow milestone messages during compact mode.

Source code in medpipe/utils/logger.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class CompactProgressFilter(logging.Filter):
    """
    Filter that passes only high-level workflow milestone messages
    during compact mode.
    """

    # Tuple of message prefixes/substrings to allow through in "compact" mode
    PROGRESS_KEYWORDS = (
        "Initialising MedpipeClassifier end-to-end",
        "Initialising MedpipeRegressor end-to-end",
        "Successfully loaded MedpipeClassifier",
        "Successfully loaded MedpipeRegressor",
        "Step 1/",
        "Step 2/",
        "Step 3/",
        "Step 4/",
        "Ingesting data from",
        "Starting model fitting across",
        "--- Starting ",
        "--- Finished ",
        "Full MedpipeClassifier pipeline execution finished",
        "Full MedpipeRegressor pipeline execution finished",
    )

    def filter(self, record: logging.LogRecord) -> bool:
        """Decide whether a log record is emitted in compact mode.

        Parameters
        ----------
        record : logging.LogRecord
            The log record to check.

        Returns
        -------
        bool
            True for records of level WARNING or above and for messages
            containing one of `PROGRESS_KEYWORDS`, False otherwise.

        """
        # Always allow WARNING, ERROR, and CRITICAL messages
        if record.levelno >= logging.WARNING:
            return True

        # Check if the log message contains any progress milestone keyword
        msg = record.getMessage()
        return any(keyword in msg for keyword in self.PROGRESS_KEYWORDS)

filter 

filter(record: LogRecord) -> bool

Decide whether a log record is emitted in compact mode.

Parameters:

Name Type Description Default
record LogRecord

The log record to check.

 required

Returns:

Type Description
bool

True for records of level WARNING or above and for messages containing one of PROGRESS_KEYWORDS, False otherwise.
Source code in medpipe/utils/logger.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def filter(self, record: logging.LogRecord) -> bool:
    """Decide whether a log record is emitted in compact mode.

    Parameters
    ----------
    record : logging.LogRecord
        The log record to check.

    Returns
    -------
    bool
        True for records of level WARNING or above and for messages
        containing one of `PROGRESS_KEYWORDS`, False otherwise.

    """
    # Always allow WARNING, ERROR, and CRITICAL messages
    if record.levelno >= logging.WARNING:
        return True

    # Check if the log message contains any progress milestone keyword
    msg = record.getMessage()
    return any(keyword in msg for keyword in self.PROGRESS_KEYWORDS)

set_verbosity 

set_verbosity(
    verbose: bool | int | str = "compact",
) -> None

Globally configure the console log verbosity level and filters.

Parameters:

Name Type Description Default
verbose str, bool, or int
  • "quiet" / False / logging.WARNING : WARNING level only.
  • "compact" / 1 : Filtered INFO level (progress lines only).
  • "info" / True / logging.INFO : Full unfiltered INFO log output.
  • "debug" / 3 / logging.DEBUG : Complete DEBUG log output.
 "compact"
Source code in medpipe/utils/logger.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def set_verbosity(verbose: bool | int | str = "compact") -> None:
    """
    Globally configure the console log verbosity level and filters.

    Parameters
    ----------
    verbose : str, bool, or int, default="compact"
        - "quiet" / False / logging.WARNING : WARNING level only.
        - "compact" / 1                     : Filtered INFO level (progress lines only).
        - "info" / True / logging.INFO      : Full unfiltered INFO log output.
        - "debug" / 3 / logging.DEBUG       : Complete DEBUG log output.
    """
    root_logger = logging.getLogger("medpipe")

    # Standardize input mode string and handle explicit logging int levels
    if isinstance(verbose, bool):
        mode = "info" if verbose else "quiet"
    elif isinstance(verbose, int):
        mode = {
            logging.WARNING: "quiet",
            logging.INFO: "info",
            logging.DEBUG: "debug",
            0: "quiet",
            1: "compact",
            2: "info",
            3: "debug",
        }.get(verbose, "compact")
    else:
        mode = str(verbose).lower()

    # Resolve level & filter status
    use_compact_filter = False
    if mode in ("quiet", "warning"):
        level = logging.WARNING
    elif mode in ("compact", "progress"):
        level = logging.INFO
        use_compact_filter = True
    elif mode in ("info", "detailed"):
        level = logging.INFO
    elif mode == "debug":
        level = logging.DEBUG
    else:
        level = logging.INFO

    # Update StreamHandler level and filters on root_logger
    for handler in root_logger.handlers:
        if isinstance(handler, logging.StreamHandler) and not isinstance(
            handler, logging.FileHandler
        ):
            handler.setLevel(level)

            # Detach any existing CompactProgressFilter instances
            handler.filters = [
                f for f in handler.filters if not isinstance(f, CompactProgressFilter)
            ]

            if use_compact_filter:
                handler.addFilter(CompactProgressFilter())

get_console_logger 

get_console_logger(
    name: str = "medpipe",
    verbose: bool | int | str | None = None,
) -> Logger

Initialize and return the base console logger for the package with configurable verbosity.

Parameters:

Name Type Description Default
name str

The name of the sub-logger instance.

 "medpipe"
verbose str, bool, or int

Console verbosity setting ("compact", "info", "debug", "quiet")

 "compact"

Returns:

Type Description
Logger

Configured standard library Logger instance.
Source code in medpipe/utils/logger.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_console_logger(
    name: str = "medpipe",
    verbose: bool | int | str | None = None,
) -> logging.Logger:
    """
    Initialize and return the base console logger for the package with
    configurable verbosity.

    Parameters
    ----------
    name : str, default="medpipe"
        The name of the sub-logger instance.
    verbose : str, bool, or int, default="compact"
        Console verbosity setting ("compact", "info", "debug", "quiet")

    Returns
    -------
    logging.Logger
        Configured standard library Logger instance.
    """
    root_logger = logging.getLogger("medpipe")
    root_logger.setLevel(
        logging.DEBUG
    )  # Base logger captures all thresholds for file logging

    has_console_handler = any(
        isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)
        for h in root_logger.handlers
    )

    if not has_console_handler:
        console_handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter(
            fmt="%(asctime)s | %(levelname)s | %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )
        console_handler.setFormatter(formatter)
        root_logger.addHandler(console_handler)

    # Apply log level & filter configuration to stream handler
    if verbose is not None:
        set_verbosity(verbose)

    return logging.getLogger(name)

add_file_handler 

add_file_handler(
    logger: Logger,
    log_dir: str | Path,
    filename: str = "execution.log",
    level: int = DEBUG,
) -> None

Attach a FileHandler to an existing logger to archive detailed run logs.

Parameters:

Name Type Description Default
logger Logger

The logger instance to attach the handler to.

 required
log_dir str or Path

The directory where the log file should be saved (e.g., an artifact folder).

 required
filename str

The name of the log file.

 "execution.log"
level int

The logging level threshold for the file output.

 logging.DEBUG
Source code in medpipe/utils/logger.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def add_file_handler(
    logger: logging.Logger,
    log_dir: str | Path,
    filename: str = "execution.log",
    level: int = logging.DEBUG,
) -> None:
    """Attach a FileHandler to an existing logger to archive detailed run logs.

    Parameters
    ----------
    logger : logging.Logger
        The logger instance to attach the handler to.
    log_dir : str or Path
        The directory where the log file should be saved (e.g., an artifact folder).
    filename : str, default="execution.log"
        The name of the log file.
    level : int, default=logging.DEBUG
        The logging level threshold for the file output.

    """
    # Resolve top-level parent logger ("medpipe") regardless of sub-logger name
    top_logger_name = logger.name.split(".")[0]
    target_logger = logging.getLogger(top_logger_name)

    log_path = Path(log_dir)
    log_path.mkdir(parents=True, exist_ok=True)
    file_path = log_path / filename

    # Avoid duplicate file handlers if initialized multiple times
    for handler in target_logger.handlers:
        if (
            isinstance(handler, logging.FileHandler)
            and Path(handler.baseFilename) == file_path.resolve()
        ):
            return

    file_handler = logging.FileHandler(file_path, mode="w", encoding="utf-8")
    file_handler.setLevel(level)

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(name)s | %(levelname)s | "
        "%(module)s:%(lineno)d | %(message)s"
    )
    file_handler.setFormatter(formatter)
    target_logger.addHandler(file_handler)

registry

BaseRegistry

Abstract base class for creating component registries.

Every subclass automatically gets its own isolated _registry dict and _fallback_modules list via __init_subclass__, so state never leaks between subclasses even if they don't redeclare either attribute themselves.

The registry is generic over T, the type of item stored in it (e.g., instances or class types).

Source code in medpipe/utils/registry.py 
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class BaseRegistry[T]:
    """
    Abstract base class for creating component registries.

    Every subclass automatically gets its own isolated `_registry` dict
    and `_fallback_modules` list via `__init_subclass__`, so state never
    leaks between subclasses even if they don't redeclare either
    attribute themselves.

    The registry is generic over ``T``, the type of item stored in it
    (e.g., instances or class types).

    """

    _registry: dict[str, T]
    _fallback_modules: ClassVar[list[Any]] = []

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        # Automatically isolate dictionary state for each subclass
        cls._registry = {}

        # Isolate fallback module state too: give each subclass its own
        # copy of whatever list it would otherwise inherit (empty for a
        # direct BaseRegistry subclass, or the parent's contents for a
        # deeper subclass), so mutating one subclass's list can never leak
        # into another. Subclasses that declare their own _fallback_modules
        # in their class body already own an isolated list and are left
        # untouched.
        if "_fallback_modules" not in cls.__dict__:
            cls._fallback_modules = list(cls._fallback_modules)

    @classmethod
    def register(cls, name: str | None = None) -> Callable[[T], T]:
        """Decorator to register a custom item into the specific subclass registry.

        Parameters
        ----------
        name : str or None, default=None
            The lookup key for the item. Defaults to the item's __name__ attribute.

        Returns
        -------
        decorator : Callable[[T], T]
            Decorator function registering the item.

        """

        def decorator(item: T) -> T:
            reg_name = name if name else getattr(item, "__name__", str(item))
            cls._registry[reg_name] = item
            return item

        return decorator

    @classmethod
    def get(cls, name: str) -> T:
        """Retrieve a registered item, checking fallback modules if necessary.

        Parameters
        ----------
        name : str
            The name identifier of the registered item.

        Returns
        -------
        T
            The registered instance or class.

        Raises
        ------
        ValueError
            If the requested item is not found in the registry or fallbacks.

        """
        if name in cls._registry:
            return cls._registry[name]

        for module in cls._fallback_modules:
            if hasattr(module, name):
                return getattr(module, name)

        raise ValueError(
            f"'{name}' was not found in the custom registry or fallback modules. "
            f"Available options: {cls.list_registered()}"
        )

    @classmethod
    def list_registered(cls) -> list[str]:
        """Return a list of string keys registered in this custom registry.

        Returns
        -------
        List[str]
            List of registered item lookup keys.

        """
        return list(cls._registry.keys())

register classmethod 

register(name: str | None = None) -> Callable[[T], T]

Decorator to register a custom item into the specific subclass registry.

Parameters:

Name Type Description Default
name str or None

The lookup key for the item. Defaults to the item's name attribute.

 None

Returns:

Name Type Description
decorator Callable[[T], T]

Decorator function registering the item.
Source code in medpipe/utils/registry.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
def register(cls, name: str | None = None) -> Callable[[T], T]:
    """Decorator to register a custom item into the specific subclass registry.

    Parameters
    ----------
    name : str or None, default=None
        The lookup key for the item. Defaults to the item's __name__ attribute.

    Returns
    -------
    decorator : Callable[[T], T]
        Decorator function registering the item.

    """

    def decorator(item: T) -> T:
        reg_name = name if name else getattr(item, "__name__", str(item))
        cls._registry[reg_name] = item
        return item

    return decorator

get classmethod 

get(name: str) -> T

Retrieve a registered item, checking fallback modules if necessary.

Parameters:

Name Type Description Default
name str

The name identifier of the registered item.

 required

Returns:

Type Description
T

The registered instance or class.

Raises:

Type Description
ValueError

If the requested item is not found in the registry or fallbacks.
Source code in medpipe/utils/registry.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@classmethod
def get(cls, name: str) -> T:
    """Retrieve a registered item, checking fallback modules if necessary.

    Parameters
    ----------
    name : str
        The name identifier of the registered item.

    Returns
    -------
    T
        The registered instance or class.

    Raises
    ------
    ValueError
        If the requested item is not found in the registry or fallbacks.

    """
    if name in cls._registry:
        return cls._registry[name]

    for module in cls._fallback_modules:
        if hasattr(module, name):
            return getattr(module, name)

    raise ValueError(
        f"'{name}' was not found in the custom registry or fallback modules. "
        f"Available options: {cls.list_registered()}"
    )

list_registered classmethod 

list_registered() -> list[str]

Return a list of string keys registered in this custom registry.

Returns:

Type Description
List[str]

List of registered item lookup keys.
Source code in medpipe/utils/registry.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def list_registered(cls) -> list[str]:
    """Return a list of string keys registered in this custom registry.

    Returns
    -------
    List[str]
        List of registered item lookup keys.

    """
    return list(cls._registry.keys())

reproducibility

medpipe.utils.reproducibility

Provides utilities for capturing runtime environment metadata, hashing configurations and datasets, and managing artifact serialization for experiment reproducibility.

ArtifactManager

Manages the creation of experiment artifact directories and metadata persistence.

Parameters:

Name Type Description Default
base_artifact_dir str or Path

Root directory where experiment runs will be saved.

 "artifacts"

Attributes:

Name Type Description
base_dir Path

Resolved base directory path.

Methods:

Name Description
create_run_directory

Create a new versioned directory for storing experiment artifacts.
save_json

Saves object as a JSON file.
saved_resolved_config

Persist the resolved configuration dictionary as a JSON file.
save_toml_config

Copy the original TOML configuration file into the artifact directory.
save_env_state

Capture and persist the environment state to env_state.json.
Source code in medpipe/utils/reproducibility.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
class ArtifactManager:
    """Manages the creation of experiment artifact directories and
    metadata persistence.

    Parameters
    ----------
    base_artifact_dir : str or Path, default="artifacts"
        Root directory where experiment runs will be saved.

    Attributes
    ----------
    base_dir : Path
        Resolved base directory path.

    Methods
    -------
    create_run_directory()
        Create a new versioned directory for storing experiment
        artifacts.
    save_json(obj, destination_dir)
        Saves object as a JSON file.
    saved_resolved_config(config, destination_dir)
        Persist the resolved configuration dictionary as a JSON file.
    save_toml_config(source_path, destination_dir)
        Copy the original TOML configuration file into the artifact
        directory.
    save_env_state(destination_dir, dataset_path)
        Capture and persist the environment state to `env_state.json`.

    """

    def __init__(self, base_artifact_dir: str | Path = "artifacts") -> None:
        self.base_dir = Path(base_artifact_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def _get_next_version_number(self) -> int:
        """Scan the base directory and return the next auto-incremented run version.

        Returns
        -------
        int
            Next available experiment version number (e.g., 1 for 'v1').

        """
        existing_versions = []
        for path in self.base_dir.iterdir():
            if path.is_dir() and path.name.startswith("v"):
                version_str = path.name[1:].split("_")[
                    0
                ]  # Extracts number from v1_hash
                if version_str.isdigit():
                    existing_versions.append(int(version_str))

        return max(existing_versions, default=0) + 1

    def create_run_directory(self) -> Path:
        """Create a new versioned directory for storing experiment artifacts.

        Returns
        -------
        Path
            Path to the newly created run directory (e.g., `artifacts/v1`).

        """
        version_num = self._get_next_version_number()
        dir_name = f"v{version_num}"

        run_dir = self.base_dir / dir_name
        run_dir.mkdir(parents=True, exist_ok=True)
        return run_dir

    def save_json(self, obj: Any, destination_dir: str | Path, filename: str) -> Path:
        """Saves object as a JSON file.

        The destination directory is created if it does not exist.

        Parameters
        ----------
        obj : Any
            Object to save.
        destination_dir : str or Path
            Directory where the file will be written.
        filename : str
            Name of the file to save.

        Returns
        -------
        Path
            Path to the saved JSON file.

        """
        dest_dir = Path(destination_dir)
        dest_dir.mkdir(exist_ok=True, parents=True)

        dest = dest_dir / filename
        with open(dest, "w", encoding="utf-8") as f:
            json.dump(obj, f, indent=4, default=str)
        return dest

    def save_resolved_config(
        self, config: dict[str, Any], destination_dir: str | Path
    ) -> Path:
        """Persist the resolved configuration dictionary as a JSON file.

        Parameters
        ----------
        config : dict
            Fully resolved configuration to save.
        destination_dir : str or Path
            Directory where `resolved_config.json` will be written.

        Returns
        -------
        Path
            Path to the saved JSON file.

        """
        return self.save_json(config, destination_dir, "resolved_config.json")

    def save_toml_config(
        self, source_path: str | Path, destination_dir: str | Path
    ) -> Path:
        """Copy the original TOML configuration file into the artifact
        directory.

        The destination directory is created if it does not exist.

        Parameters
        ----------
        source_path : str or Path
            Path to the original TOML configuration file.
        destination_dir : str or Path
            Directory where `config.toml` will be written.

        Returns
        -------
        Path
            Path to the copied TOML file.

        """
        dest_dir = Path(destination_dir)
        dest_dir.mkdir(exist_ok=True, parents=True)

        dest = dest_dir / "config.toml"
        shutil.copy2(source_path, dest)
        return dest

    def save_env_state(
        self,
        destination_dir: str | Path,
        config: dict[str, Any],
        dataset_path: str | Path | None = None,
    ) -> Path:
        """Capture and persist the environment state to `env_state.json`.

        Parameters
        ----------
        destination_dir : str or Path
            Directory where `env_state.json` will be written.
        config : dict
            Fully resolved configuration dictionary.
        dataset_path : str or Path, optional
            Path to the dataset file to hash and log.

        Returns
        -------
        Path
            Path to the saved JSON file.

        """
        env_state = capture_environment_state(config, dataset_path=dataset_path)
        return self.save_json(env_state, destination_dir, "env_state.json")

create_run_directory 

create_run_directory() -> Path

Create a new versioned directory for storing experiment artifacts.

Returns:

Type Description
Path

Path to the newly created run directory (e.g., artifacts/v1).
Source code in medpipe/utils/reproducibility.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def create_run_directory(self) -> Path:
    """Create a new versioned directory for storing experiment artifacts.

    Returns
    -------
    Path
        Path to the newly created run directory (e.g., `artifacts/v1`).

    """
    version_num = self._get_next_version_number()
    dir_name = f"v{version_num}"

    run_dir = self.base_dir / dir_name
    run_dir.mkdir(parents=True, exist_ok=True)
    return run_dir

save_json 

save_json(
    obj: Any, destination_dir: str | Path, filename: str
) -> Path

Saves object as a JSON file.

The destination directory is created if it does not exist.

Parameters:

Name Type Description Default
obj Any

Object to save.

 required
destination_dir str or Path

Directory where the file will be written.

 required
filename str

Name of the file to save.

 required

Returns:

Type Description
Path

Path to the saved JSON file.
Source code in medpipe/utils/reproducibility.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def save_json(self, obj: Any, destination_dir: str | Path, filename: str) -> Path:
    """Saves object as a JSON file.

    The destination directory is created if it does not exist.

    Parameters
    ----------
    obj : Any
        Object to save.
    destination_dir : str or Path
        Directory where the file will be written.
    filename : str
        Name of the file to save.

    Returns
    -------
    Path
        Path to the saved JSON file.

    """
    dest_dir = Path(destination_dir)
    dest_dir.mkdir(exist_ok=True, parents=True)

    dest = dest_dir / filename
    with open(dest, "w", encoding="utf-8") as f:
        json.dump(obj, f, indent=4, default=str)
    return dest

save_resolved_config 

save_resolved_config(
    config: dict[str, Any], destination_dir: str | Path
) -> Path

Persist the resolved configuration dictionary as a JSON file.

Parameters:

Name Type Description Default
config dict

Fully resolved configuration to save.

 required
destination_dir str or Path

Directory where resolved_config.json will be written.

 required

Returns:

Type Description
Path

Path to the saved JSON file.
Source code in medpipe/utils/reproducibility.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def save_resolved_config(
    self, config: dict[str, Any], destination_dir: str | Path
) -> Path:
    """Persist the resolved configuration dictionary as a JSON file.

    Parameters
    ----------
    config : dict
        Fully resolved configuration to save.
    destination_dir : str or Path
        Directory where `resolved_config.json` will be written.

    Returns
    -------
    Path
        Path to the saved JSON file.

    """
    return self.save_json(config, destination_dir, "resolved_config.json")

save_toml_config 

save_toml_config(
    source_path: str | Path, destination_dir: str | Path
) -> Path

Copy the original TOML configuration file into the artifact directory.

The destination directory is created if it does not exist.

Parameters:

Name Type Description Default
source_path str or Path

Path to the original TOML configuration file.

 required
destination_dir str or Path

Directory where config.toml will be written.

 required

Returns:

Type Description
Path

Path to the copied TOML file.
Source code in medpipe/utils/reproducibility.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def save_toml_config(
    self, source_path: str | Path, destination_dir: str | Path
) -> Path:
    """Copy the original TOML configuration file into the artifact
    directory.

    The destination directory is created if it does not exist.

    Parameters
    ----------
    source_path : str or Path
        Path to the original TOML configuration file.
    destination_dir : str or Path
        Directory where `config.toml` will be written.

    Returns
    -------
    Path
        Path to the copied TOML file.

    """
    dest_dir = Path(destination_dir)
    dest_dir.mkdir(exist_ok=True, parents=True)

    dest = dest_dir / "config.toml"
    shutil.copy2(source_path, dest)
    return dest

save_env_state 

save_env_state(
    destination_dir: str | Path,
    config: dict[str, Any],
    dataset_path: str | Path | None = None,
) -> Path

Capture and persist the environment state to env_state.json.

Parameters:

Name Type Description Default
destination_dir str or Path

Directory where env_state.json will be written.

 required
config dict

Fully resolved configuration dictionary.

 required
dataset_path str or Path

Path to the dataset file to hash and log.

 None

Returns:

Type Description
Path

Path to the saved JSON file.
Source code in medpipe/utils/reproducibility.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def save_env_state(
    self,
    destination_dir: str | Path,
    config: dict[str, Any],
    dataset_path: str | Path | None = None,
) -> Path:
    """Capture and persist the environment state to `env_state.json`.

    Parameters
    ----------
    destination_dir : str or Path
        Directory where `env_state.json` will be written.
    config : dict
        Fully resolved configuration dictionary.
    dataset_path : str or Path, optional
        Path to the dataset file to hash and log.

    Returns
    -------
    Path
        Path to the saved JSON file.

    """
    env_state = capture_environment_state(config, dataset_path=dataset_path)
    return self.save_json(env_state, destination_dir, "env_state.json")

compute_file_hash 

compute_file_hash(
    file_path: str | Path, algorithm: str = "sha256"
) -> str

Compute the cryptographic hash of a file on disk.

Parameters:

Name Type Description Default
file_path str or Path

Path to the file to be hashed.

 required
algorithm str

Hashing algorithm to use (e.g., 'md5', 'sha256').

 "sha256"

Returns:

Type Description
str

Hexadecimal hash string representing the file contents.

Raises:

Type Description
FileNotFoundError

If the specified file_path does not exist.
Source code in medpipe/utils/reproducibility.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def compute_file_hash(file_path: str | Path, algorithm: str = "sha256") -> str:
    """Compute the cryptographic hash of a file on disk.

    Parameters
    ----------
    file_path : str or Path
        Path to the file to be hashed.
    algorithm : str, default="sha256"
        Hashing algorithm to use (e.g., 'md5', 'sha256').

    Returns
    -------
    str
        Hexadecimal hash string representing the file contents.

    Raises
    ------
    FileNotFoundError
        If the specified `file_path` does not exist.

    """
    path = Path(file_path)
    if not path.is_file():
        raise FileNotFoundError(f"File not found for hashing: {path}")

    hasher = hashlib.new(algorithm)
    # Read in chunks to prevent memory overload on large datasets
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            hasher.update(chunk)

    return hasher.hexdigest()

compute_config_hash 

compute_config_hash(config: dict[str, Any]) -> str

Generate a deterministic SHA-256 hash from a configuration dictionary.

Parameters:

Name Type Description Default
config dict

Fully resolved configuration dictionary.

 required

Returns:

Type Description
str

The SHA-256 hash representing the config state.
Source code in medpipe/utils/reproducibility.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def compute_config_hash(config: dict[str, Any]) -> str:
    """Generate a deterministic SHA-256 hash from a configuration dictionary.

    Parameters
    ----------
    config : dict
        Fully resolved configuration dictionary.

    Returns
    -------
    str
        The SHA-256 hash representing the config state.

    """
    # Canonicalize dictionary to ensure key ordering does not alter hash
    encoded_config = json.dumps(config, sort_keys=True, default=str).encode("utf-8")
    return hashlib.sha256(encoded_config).hexdigest()

get_git_commit_hash 

get_git_commit_hash() -> str | None

Retrieve the current Git commit hash of the repository.

Returns:

Type Description
str or None

The 40-character SHA-1 commit hash, or None if Git is not installed or the code is not running inside a Git repository.
Source code in medpipe/utils/reproducibility.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_git_commit_hash() -> str | None:
    """Retrieve the current Git commit hash of the repository.

    Returns
    -------
    str or None
        The 40-character SHA-1 commit hash, or None if Git is not installed
        or the code is not running inside a Git repository.

    """
    try:
        git_hash = subprocess.check_output(
            ["git", "rev-parse", "HEAD"], stderr=subprocess.DEVNULL
        )
        return git_hash.decode("ascii").strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return None

capture_environment_state 

capture_environment_state(
    config: dict[str, Any],
    dataset_path: str | Path | None = None,
) -> dict[str, Any]

Collect system, Python, package, and data metadata for reproducibility.

Parameters:

Name Type Description Default
config dict

Fully resolved configuration to save.

 required
dataset_path str or Path

Path to the dataset file to hash. If None, dataset_hash is omitted.

 None

Returns:

Type Description
Dict[str, Any]

Dictionary containing environment and runtime metadata.
Source code in medpipe/utils/reproducibility.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def capture_environment_state(
    config: dict[str, Any],
    dataset_path: str | Path | None = None,
) -> dict[str, Any]:
    """Collect system, Python, package, and data metadata for reproducibility.

    Parameters
    ----------
    config : dict
        Fully resolved configuration to save.
    dataset_path : str or Path, optional
        Path to the dataset file to hash. If None, dataset_hash is omitted.

    Returns
    -------
    Dict[str, Any]
        Dictionary containing environment and runtime metadata.

    """
    env_state: dict[str, Any] = {
        "timestamp_utc": datetime.now(UTC).isoformat(),
        "config_hash": compute_config_hash(config),
        "python_version": sys.version.split()[0],
        "platform": platform.platform(),
        "git_commit_hash": get_git_commit_hash(),
        "dataset_hash": None,
    }

    if dataset_path is not None:
        try:
            env_state["dataset_hash"] = compute_file_hash(dataset_path)
        except FileNotFoundError:
            env_state["dataset_hash"] = "FILE_NOT_FOUND"

    return env_state

validation

Validation functions module.

This module provides functions for validating files and paths, raising the appropriate exception when a check fails.

Functions: - file_checks: Checks if the file is correct. - path_checks: Checks if the path is correct.

file_checks 

file_checks(
    file: str | Path,
    extension: str | list[str],
    exists: bool = True,
) -> None

Performs checks to ensure that a file and extension are correct.

Parameters:

Name Type Description Default
file str | Path

File to check.

 required
extension str | list[str]

Extension or list of extensions of the file to check.

 required
exists bool

Flag to indicate if the file should exists.

 True

Returns:

Type Description
None

Nothing is returned.

Raises:

Type Description
TypeError

If file is not a str or a Path. If extension is not a str or list[str].
FileNotFoundError

If file does not exist.
IsADirectoryError

If file is a directory.
ValueError

If file extension is not correct.
Source code in medpipe/utils/validation.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def file_checks(
    file: str | Path, extension: str | list[str], exists: bool = True
) -> None:
    """
    Performs checks to ensure that a file and extension are correct.

    Parameters
    ----------
    file : str | Path
        File to check.
    extension : str | list[str]
        Extension or list of extensions of the file to check.
    exists : bool, default: True
        Flag to indicate if the file should exists.

    Returns
    -------
    None
        Nothing is returned.

    Raises
    ------
    TypeError
        If file is not a str or a Path.
        If extension is not a str or list[str].
    FileNotFoundError
        If file does not exist.
    IsADirectoryError
        If file is a directory.
    ValueError
        If file extension is not correct.

    """
    if not isinstance(file, (str, Path)):
        raise TypeError("File should be a string or Path")

    if not isinstance(extension, (str, list)):
        raise TypeError("Extension should be a string or list of strings")

    path_object = Path(file).expanduser().resolve()  # Create a Path object

    path_checks(str(path_object.parent))

    if not path_object.exists() and exists:
        raise FileNotFoundError(f"{file} does not exist")

    if not path_object.is_file() and exists:
        raise IsADirectoryError(f"{file} should be a file")

    suffix = path_object.suffix
    if isinstance(extension, str):
        if suffix != extension:
            raise ValueError(f"File suffix should be {extension}, but got {suffix}")
    else:
        if suffix not in extension:
            raise ValueError(
                f"File suffix should be one of {extension}, but got {suffix}"
            )

path_checks 

path_checks(path: str | Path) -> None

Performs checks to ensure that a path is correct and creates it if it does not exist.

Parameters:

Name Type Description Default
path str | Path

Path to check.

 required

Returns:

Type Description
None

Nothing is returned.

Raises:

Type Description
TypeError

If path is not a str or Path.
FileNotFoundError

If path does not exist.
NotADirectoryError

If path is not a directory.
Source code in medpipe/utils/validation.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def path_checks(path: str | Path) -> None:
    """
    Performs checks to ensure that a path is correct and creates it
    if it does not exist.

    Parameters
    ----------
    path : str | Path
        Path to check.

    Returns
    -------
    None
        Nothing is returned.

    Raises
    ------
    TypeError
        If path is not a str or Path.
    FileNotFoundError
        If path does not exist.
    NotADirectoryError
        If path is not a directory.

    """
    if not isinstance(path, (str, Path)):
        raise TypeError("Path should be a string or a Path")

    path_object = Path(path).expanduser().resolve()  # Create a Path object

    if not path_object.exists():
        path_object.mkdir(parents=True)

    if not path_object.is_dir():
        raise NotADirectoryError(f"{path} should be a directory")