Utils

source

store_variables


def store_variables(
    pkl_fn:str | pathlib.Path, size:list, apply_reorder:bool, target_spacing:int | list
):

Save variable values in a pickle file.

source

load_variables


def load_variables(
    pkl_fn:(<class 'str'>, <class 'pathlib.Path'>)
):

Loads stored variable values from a pickle file.

Args: pkl_fn: File path of the pickle file to be loaded.

Returns: The deserialized value of the pickled data.

Patch-based inference settings

source

store_patch_variables


def store_patch_variables(
    pkl_fn:str | pathlib.Path, patch_size:list, patch_overlap:int | float | list, aggregation_mode:str,
    apply_reorder:bool=False, target_spacing:list=None, sampler_type:str='uniform', label_probabilities:dict=None,
    samples_per_volume:int=8, queue_length:int=300, queue_num_workers:int=4, keep_largest_component:bool=False
):

Save patch-based training and inference configuration to a pickle file.

Args: pkl_fn: Path to save the pickle file. patch_size: Size of patches [x, y, z]. patch_overlap: Overlap for inference (int, float 0-1, or list). aggregation_mode: GridAggregator mode (‘crop’, ‘average’, ‘hann’). apply_reorder: Whether to reorder to canonical (RAS+) orientation. target_spacing: Target voxel spacing [x, y, z]. sampler_type: Type of sampler used during training. label_probabilities: Label probabilities for LabelSampler. samples_per_volume: Number of patches extracted per volume during training. queue_length: Maximum number of patches in queue buffer. queue_num_workers: Number of workers for parallel patch extraction. keep_largest_component: If True, keep only the largest connected component in binary segmentation predictions during inference.

Example: >>> store_patch_variables( … ‘patch_settings.pkl’, … patch_size=[96, 96, 96], … patch_overlap=0.5, … aggregation_mode=‘hann’, … apply_reorder=True, … target_spacing=[1.0, 1.0, 1.0], … samples_per_volume=16, … keep_largest_component=True … )

source

load_patch_variables


def load_patch_variables(
    pkl_fn:str | pathlib.Path
)->dict:

Load patch-based training and inference configuration from a pickle file.

Args: pkl_fn: Path to the pickle file.

Returns: Dictionary with patch configuration including: - patch_size, patch_overlap, aggregation_mode - apply_reorder, target_spacing, sampler_type, label_probabilities - samples_per_volume, queue_length, queue_num_workers

Example: >>> config = load_patch_variables(‘patch_settings.pkl’) >>> from fastMONAI.vision_patch import PatchConfig >>> patch_config = PatchConfig(**config)

source

ModelTrackingCallback


def ModelTrackingCallback(
    model_name:str, loss_function:str, item_tfms:list, size:list, target_spacing:list, apply_reorder:bool,
    experiment_name:str=None, run_name:str=None, auto_start:bool=False, patch_config:dict=None,
    extra_params:dict=None, extra_tags:dict=None, dataset_version:str=None, log_split:bool=True
):

A FastAI callback for comprehensive MLflow experiment tracking.

This callback automatically logs hyperparameters, metrics, model artifacts, and configuration to MLflow during training. If a checkpoint callback (SaveModelCallback, EMACheckpoint, or any TrackerCallback with fname) is present, the best model checkpoint will also be logged as an artifact.

Supports auto-managed runs when created via create_mlflow_callback().

source

create_mlflow_callback


def create_mlflow_callback(
    learn, experiment_name:str=None, run_name:str=None, auto_start:bool=True, model_name:str=None,
    extra_params:dict=None, extra_tags:dict=None, dataset_version:str=None, log_split:bool=True
)->ModelTrackingCallback:

Create MLflow tracking callback with auto-extracted configuration.

This factory function automatically extracts configuration from the Learner, eliminating the need to manually specify parameters like size, transforms, loss function, etc.

Auto-extracts from Learner: - Preprocessing: apply_reorder, target_spacing, size/patch_size - Transforms: item_tfms or pre_patch_tfms - Training: loss_func, model architecture

Args: learn: fastai Learner instance experiment_name: MLflow experiment name. If None, uses model name. run_name: MLflow run name. If None, auto-generates with timestamp. auto_start: If True, auto-starts/stops MLflow run in before_fit/after_fit. model_name: Override auto-extracted model name for registration. extra_params: Additional parameters to log (e.g., {‘dropout’: 0.5}). extra_tags: MLflow tags to set on the run. dataset_version: Dataset version hash string for tracking. log_split: If True, auto-logs train/val split CSV when dls has split_df.

Returns: ModelTrackingCallback ready to use with learn.fit()

Example: >>> # Instead of this (6 manual params): >>> # mlflow_callback = ModelTrackingCallback( >>> # model_name=f”{task}_{model._get_name()}“, >>> # loss_function=loss_func.loss_func._get_name(), >>> # item_tfms=item_tfms, >>> # size=size, >>> # target_spacing=target_spacing, >>> # apply_reorder=True, >>> # ) >>> # with mlflow.start_run(run_name=”training”): >>> # learn.fit_one_cycle(30, lr, cbs=[mlflow_callback]) >>> >>> # Do this (zero manual params): >>> callback = create_mlflow_callback(learn, experiment_name=“Task02_Heart”) >>> learn.fit_one_cycle(30, lr, cbs=[callback, save_best])

# Test auto-extraction helper functions
from fastcore.test import test_eq, test_fail
from dataclasses import dataclass

# Test _detect_patch_workflow
class MockStandardDls:
    bs = 4
    after_item = None
mock_std = MockStandardDls()
test_eq(_detect_patch_workflow(mock_std), False)

@dataclass
class MockPatchConfig:
    patch_size: list = None
    patch_overlap: float = 0.5
    samples_per_volume: int = 8
    sampler_type: str = 'uniform'
    label_probabilities: dict = None
    queue_length: int = 300
    aggregation_mode: str = 'hann'
    padding_mode: int = 0
    keep_largest_component: bool = False
    apply_reorder: bool = True
    target_spacing: list = None
    
    def __post_init__(self):
        if self.patch_size is None:
            self.patch_size = [96, 96, 96]

class MockPatchDls:
    bs = 4
    patch_config = MockPatchConfig()
mock_patch = MockPatchDls()
test_eq(_detect_patch_workflow(mock_patch), True)

# Test _extract_size_from_transforms with mock transform
class MockPadOrCrop:
    def __init__(self, target_shape):
        self.target_shape = target_shape

class MockTransform:
    def __init__(self, target_shape):
        self.pad_or_crop = MockPadOrCrop(target_shape)

tfms = [MockTransform([128, 128, 64])]
test_eq(_extract_size_from_transforms(tfms), [128, 128, 64])
test_eq(_extract_size_from_transforms(None), None)
test_eq(_extract_size_from_transforms([]), None)

print("All auto-extraction helper tests passed!")
All auto-extraction helper tests passed!

source

MLflowUIManager


def MLflowUIManager(
    
):

Initialize self. See help(type(self)) for accurate signature.