9.2.3. Pipeline

Pipeline components.

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)

Register klass under step.

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.

get_output_type(input_type)

Get output type.

Parameters:

input_typestr

The data type input to the marker.

Returns:

str

The storage type output by the marker.

validate(input)

Validate the the pipeline step.

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.