9.2.3. Pipeline¶

Pipeline components.

class junifer.pipeline.AssetDumperDispatcher¶

Class for helping dynamic asset dumper dispatch.

clear()¶: Not implemented.

popitem()¶: Not implemented.

setdefault(key, value=None)¶

Not implemented.

class junifer.pipeline.AssetLoaderDispatcher¶

Class for helping dynamic asset loader dispatch.

clear()¶: Not implemented.

popitem()¶: Not implemented.

setdefault(key, value=None)¶

Not implemented.

class junifer.pipeline.BaseDataDumpAsset(data, path_without_ext)¶

Abstract base class for a data dump asset.

Parameters:

dataAny: Data to save.
path_without_extpathlib.Path: Path to the asset without extension. The subclass should add the extension when saving.

Initialize the class.

abstract dump()¶

Dump asset.

abstract classmethod load(path)¶

Load asset from path.

class junifer.pipeline.DataObjectDumper¶

Class for pipeline data object dumping.

Overridden to make the class singleton.

dump(data, path, step)¶

Dump data object at path.

Parameters:

datadict: The data object state to dump.
pathpathlib.Path: The path to dump the data object.
stepstr: The step name. Also sets the dump directory.

load(path)¶

Load data object from path.

Parameters:

pathpathlib.Path: The path to the dumped data object.

Returns:

dict: The restored data object dump.

enum junifer.pipeline.ExtDep(value)¶

Accepted external dependencies.

Member Type:: str

Valid values are as follows:

AFNI = <ExtDep.AFNI: 'afni'>¶

FSL = <ExtDep.FSL: 'fsl'>¶

ANTs = <ExtDep.ANTs: 'ants'>¶

FreeSurfer = <ExtDep.FreeSurfer: 'freesurfer'>¶

class junifer.pipeline.MarkerCollection(markers, datareader=None, preprocessors=None, storage=None)¶

Class for marker collection.

Parameters:

markerslist of marker-like: The markers to compute.
datareaderDataReader-like object, optional: The DataReader to use (default None).
preprocessorslist of preprocessing-like, optional: The preprocessors to apply (default None).
storagestorage-like, optional: The storage to use (default None).

Raises:

ValueError: If markers have same names.

fit(input)¶

Fit the pipeline.

Parameters:

inputdict: The input data to fit the pipeline on. Should be the output of indexing the Data Grabber with one element.

Returns:

dict or None: The output of the pipeline. Each key represents a marker name and the values are the computer marker values. If the pipeline has a storage configured, then the output will be None.

validate(datagrabber)¶

Validate the pipeline.

Without doing any computation, check if the marker collection can be fitted without problems i.e., the data required for each marker is present and streamed down the steps. Also, if a storage is configured, check that the storage can handle the markers’ output.

Parameters:

datagrabberDataGrabber-like: The DataGrabber to validate.

class junifer.pipeline.PipelineComponentRegistry(*args, **kwargs)¶

Class for pipeline component registry.

This class is a singleton and is used for managing pipeline components. It serves as a centralized registry for built-in and third-party pipeline components like datagrabbers, datareaders, preprocessors, markers and storage.

Attributes:

stepslist of str: Get valid pipeline steps.
componentsdict: Get registered components for valid pipeline steps.

Initialize the class.

build_component_instance(step, name, baseclass, init_params=None)¶

Build an instance of class registered as name.

Parameters:

stepstr: Name of the pipeline step.
namestr: Name of the component.
baseclassclass: Base class to be checked against.
init_paramsdict or None, optional: Parameters to pass to the class constructor (default None).

Returns:

object: An instance of the class registered as name under step.

Raises:

RuntimeError: If there is a problem creating the instance.
ValueError: If the created object with the given name is not a subclass of the base class baseclass.

property components: Mapping[str, Mapping[str, str | type]]¶: Get registered components for valid pipeline steps.

deregister(step, klass)¶

De-register klass under step.

Parameters:

stepstr: Name of the pipeline step. For valid steps, check steps().
klassclass: Class to be de-registered.

Raises:

ValueError: If the step is invalid.

get_class(step, name)¶

Get the class registered under name for step.

Parameters:

stepstr: Name of the pipeline step.
namestr: Name of the component.

Returns:

class: Registered class.

Raises:

ValueError: If the step or name is invalid.

register(step, klass)¶

Parameters:

stepstr: Name of the pipeline step. For valid steps, check steps().
klassclass: Class to be registered.

Raises:

ValueError: If the step is invalid.

step_components(step)¶

Get registered components for step.

Parameters:

stepstr: Name of the pipeline step.

Returns:

list of str: List of registered component classes.

Raises:

ValueError: If the step is invalid.

property steps: list[str]¶: Get valid pipeline steps.

class junifer.pipeline.PipelineStepMixin¶

Mixin class for a pipeline step.

fit_transform(input, **kwargs)¶

Fit and transform.

Parameters:

inputdict: The Junifer Data object.
**kwargsdict: Extra keyword arguments passed to the concrete class’ _fit_transform().

Returns:

dict: The processed output of the pipeline step.

validate_component(input)¶

Validate the pipeline component.

Parameters:

inputlist of str: The input to the pipeline step.

Returns:

list of str: The output of the pipeline step.

validate_input(input)¶

Validate the input to the pipeline step.

Parameters:

inputlist of str: The input to the pipeline step. The list must contain the available Junifer Data dictionary keys.

Returns:

list of str: The actual elements of the input that will be processed by this pipeline step.

Raises:

ValueError: If the input does not have the required data.

class junifer.pipeline.UpdateMetaMixin¶

Mixin class for updating meta.

update_meta(input, step_name)¶

Update metadata.

Parameters:

inputdict or list of dict: The data object to update.
step_namestr: The name of the pipeline step.

class junifer.pipeline.WorkDirManager(*args, **kwargs)¶

Class for working directory manager.

This class is a singleton and is used for managing temporary and working directories used across the pipeline by datagrabbers, preprocessors, markers and so on. It maintains a single super-directory and provides directories on-demand and cleans after itself thus keeping the user filesystem clean.

Parameters:

workdirstr or pathlib.Path, optional: The path to the super-directory. If None, “TMPDIR/junifer” is used where TMPDIR is the platform-dependent temporary directory.
cleanupbool, optional: If False, the directories are not cleaned up after the object is destroyed. This is useful for debugging purposes (default True).

Attributes:

workdirpathlib.Path: Get working directory.
elementdirpathlib.Path: Get element directory.
root_tempdirpathlib.Path or None: Get root temporary directory.

Initialize the class.

cleanup_elementdir()¶

Clean up element directory.

It should preferably be used after fitting a marker or something similar in the element-specific scope. If called between components, can lead to required intermediate files not being found.

delete_element_tempdir(tempdir)¶

Delete an element-scoped temporary directory.

Parameters:

tempdirpathlib.Path: The temporary directory path to be deleted.

delete_tempdir(tempdir)¶

Delete a component-scoped temporary directory.

Parameters:

tempdirpathlib.Path: The temporary directory path to be deleted.

property elementdir: Path¶: Get element directory.

get_element_tempdir(prefix=None, suffix=None)¶

Get an element-scoped temporary directory.

This directory should be available only for the lifetime of an element.

Parameters:

prefixstr, optional: The temporary directory prefix. If None, a default prefix is used (default None).
suffixstr, optional: The temporary directory suffix. If None, no suffix is added (default None).

Returns:

pathlib.Path: The path to the temporary directory.

get_tempdir(prefix=None, suffix=None)¶

Get a component-scoped temporary directory.

This directory should be available only for the lifetime of a component like a preprocessor or marker.

Parameters:

prefixstr, optional: The temporary directory prefix. If None, a default prefix is used (default None).
suffixstr, optional: The temporary directory suffix. If None, no suffix is added (default None).

Returns:

pathlib.Path: The path to the temporary directory.

property root_tempdir: Path | None¶: Get root temporary directory.

property workdir: Path¶: Get working directory.