modosaic.cli.pipeline

modosaic.cli.pipeline

Runtime builders used by the Modosaic CLI.

PipelineRun dataclass

PipelineRun(results, experiment_path)

Result of one CLI-triggered pipeline execution.

Attributes:

Name	Type	Description
results	list[Any]	Per-sample pipeline results.
experiment_path	Path	Directory where artifacts were written.

build_dataset

build_dataset(config)

Build an ImageDataset from dataset configuration.

Parameters:

Name	Type	Description	Default
config	DatasetConfig	Dataset configuration.	required

Returns:

Type	Description
ImageDataset	Configured image dataset.

Raises:

Type	Description
ValueError	If the dataset kind is unsupported or required paths are absent.

Source code in modosaic/cli/pipeline.py

def build_dataset(config: DatasetConfig) -> "ImageDataset":
    """Build an `ImageDataset` from dataset configuration.

    Args:
        config: Dataset configuration.

    Returns:
        Configured image dataset.

    Raises:
        ValueError: If the dataset kind is unsupported or required paths are
            absent.
    """
    from modosaic.providers.image_dataset import ImageDataset

    if config.kind == DatasetKind.LOCAL:
        return ImageDataset.from_local_folder(
            root=_required_path(config.root, "root"),
            recursive=config.recursive,
            extensions=config.extensions or None,
        )

    if config.kind == DatasetKind.PARQUET:
        return ImageDataset.from_parquet(
            parquet_path=_required_path(config.parquet_path, "parquet_path"),
            image_column=config.image_column,
            id_column=config.id_column,
            extension_column=config.extension_column,
            metadata_columns=config.metadata_columns or None,
            batch_size=config.batch_size,
        )

    raise ValueError(f"Unsupported dataset kind: {config.kind}")

build_modalities

build_modalities(config)

Build configured modalities for a run.

Parameters:

Name	Type	Description	Default
config	RunConfig	Run configuration.	required

Returns:

Type	Description
list[Modality[Any]]	List of configured modalities in execution order.

Source code in modosaic/cli/pipeline.py

def build_modalities(config: RunConfig) -> list["Modality[Any]"]:
    """Build configured modalities for a run.

    Args:
        config: Run configuration.

    Returns:
        List of configured modalities in execution order.
    """
    builders = {
        ModalityName.IMAGE: _build_image_modality,
        ModalityName.TEXT: _build_text_modality,
        ModalityName.SEGMENTATION: _build_segmentation_modality,
        ModalityName.DEPTH: _build_depth_modality,
        ModalityName.NORMALS: _build_normals_modality,
    }
    selected = set(config.modalities)
    return [builders[name](config, selected) for name in config.modalities]

execute_run

execute_run(config)

Execute a complete run from a normalized config.

Parameters:

Name	Type	Description	Default
config	RunConfig	Normalized run configuration.	required

Returns:

Type	Description
PipelineRun	Pipeline results and experiment path.

Raises:

Type	Description
FileNotFoundError	If the configured dataset path is missing.
ValueError	If the configuration is invalid.

Source code in modosaic/cli/pipeline.py

def execute_run(config: RunConfig) -> PipelineRun:
    """Execute a complete run from a normalized config.

    Args:
        config: Normalized run configuration.

    Returns:
        Pipeline results and experiment path.

    Raises:
        FileNotFoundError: If the configured dataset path is missing.
        ValueError: If the configuration is invalid.
    """
    validate_run_config(config)

    from modosaic.core.pipeline import Pipeline
    from modosaic.services.experiment import ExperimentService
    from modosaic.services.logging import LoggingService
    from modosaic.services.seeding import SeedingService

    LoggingService.setup_logging(config.log_path)
    SeedingService.set_global_seed(config.seed)

    experiment = ExperimentService(
        root=config.experiment_root,
        experiment_name=config.experiment_name,
    )
    pipeline = Pipeline(
        dataset=build_dataset(config.dataset),
        modalities=build_modalities(config),
        experiment=experiment,
    )
    return PipelineRun(
        results=pipeline.run(limit=config.limit),
        experiment_path=experiment.experiment_path,
    )

validate_run_config

validate_run_config(config)

Validate dataset paths and numeric runtime options.

Parameters:

Name	Type	Description	Default
config	RunConfig	Run configuration to validate.	required

Raises:

Type	Description
FileNotFoundError	If a required dataset path is missing.
ValueError	If a required option is absent or invalid.

Source code in modosaic/cli/pipeline.py

def validate_run_config(config: RunConfig) -> None:
    """Validate dataset paths and numeric runtime options.

    Args:
        config: Run configuration to validate.

    Raises:
        FileNotFoundError: If a required dataset path is missing.
        ValueError: If a required option is absent or invalid.
    """
    dataset = config.dataset
    if dataset.kind == DatasetKind.LOCAL:
        if dataset.root is None:
            raise ValueError("--root is required when --dataset local.")
        if not dataset.root.exists():
            raise FileNotFoundError(f"Local dataset folder not found: {dataset.root}")
        if not dataset.root.is_dir():
            raise ValueError(f"Local dataset root must be a directory: {dataset.root}")

    if dataset.kind == DatasetKind.PARQUET:
        if dataset.parquet_path is None:
            raise ValueError("--parquet-path is required when --dataset parquet.")
        if not dataset.parquet_path.exists():
            raise FileNotFoundError(f"Parquet path not found: {dataset.parquet_path}")

    if config.limit is not None and config.limit < 0:
        raise ValueError("--limit must be greater than or equal to 0.")
    if dataset.batch_size <= 0:
        raise ValueError("--batch-size must be greater than 0.")

    validator = config.validators
    non_negative_options = {
        "segmentation_boundary_thickness": validator.segmentation_boundary_thickness,
        "segmentation_tolerance_radius": validator.segmentation_tolerance_radius,
        "depth_boundary_thickness": validator.depth_boundary_thickness,
        "depth_tolerance_radius": validator.depth_tolerance_radius,
    }
    for name, value in non_negative_options.items():
        if value < 0:
            raise ValueError(f"{name} must be greater than or equal to 0.")