iris.orchestration package

Submodules

iris.orchestration.environment module

class iris.orchestration.environment.Environment(*, call_trace_initialiser: Callable[[Dict[str, Algorithm], List[PipelineNode]], PipelineCallTraceStorage], pipeline_output_builder: Callable[[PipelineCallTraceStorage], Any], error_manager: Callable[[PipelineCallTraceStorage, Exception], None], disabled_qa: List[type] = [])

Bases: ImmutableModel

Data holder for the pipeline environment properties.

call_trace_initialiser is responsible for initialising the PipelineCallTraceStorage instance in the pipeline.

pipeline_output_builder is responsible for building the pipeline output from the call_trace, which kept all intermediary results so far.

error_manager is responsible for the pipeline’s behaviour in case of an exception

disabled_qa stores a list of Algorithm and/or Callbacks types to be disabled.

call_trace_initialiser: Callable[[Dict[str, Algorithm], List[PipelineNode]], PipelineCallTraceStorage]

disabled_qa: List[type]

error_manager: Callable[[PipelineCallTraceStorage, Exception], None]

pipeline_output_builder: Callable[[PipelineCallTraceStorage], Any]

iris.orchestration.error_managers module

iris.orchestration.error_managers.raise_error_manager(call_trace: PipelineCallTraceStorage, exception: Exception) → None

Error manager for the Orb.

Parameters:

• call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

• exception (Exception) – Exception raised during the pipeline call.

Raises:

Exception – Reraise the exception parameter.

iris.orchestration.error_managers.store_error_manager(call_trace: PipelineCallTraceStorage, exception: Exception) → None

Error manager for debugging.

Parameters:

• call_trace (PipelineCallTraceStorage) – Pipeline call results storage.

• exception (Exception) – Exception raised during the pipeline call.

iris.orchestration.output_builders module

iris.orchestration.output_builders.build_aggregation_multiframe_orb_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct multiframe aggregation ORB output with safe serialization.

iris.orchestration.output_builders.build_iris_pipeline_debugging_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct full debugging output: wrap simple_debugging and ensure the iris_template is also safely serialized.

iris.orchestration.output_builders.build_iris_pipeline_orb_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct ORB output with serialized iris_template.

iris.orchestration.output_builders.build_simple_iris_pipeline_debugging_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct debugging output with intermediate results (raw values).

iris.orchestration.output_builders.build_simple_iris_pipeline_orb_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct simple ORB output: raw iris_template, error, and metadata.

iris.orchestration.output_builders.build_simple_multiframe_aggregation_output(call_trace: PipelineCallTraceStorage) → Dict[str, Any]

Construct simple multiframe aggregation output (raw values).

iris.orchestration.pipeline_dataclasses module

class iris.orchestration.pipeline_dataclasses.PipelineClass(*, class_name: str, params: Dict[str, int | float | str | PipelineClass | Tuple[int, float, str, PipelineClass] | List[int | float | str | PipelineClass]])

Bases: ImmutableModel

Data holder for the reference to any class: Algorithm, Callback, etc.

class_name: str

params: Dict[str, int | float | str | PipelineClass | Tuple[int, float, str, PipelineClass] | List[int | float | str | PipelineClass]]

class iris.orchestration.pipeline_dataclasses.PipelineInput(*, name: str, index: int | None = None, source_node: str | List[str | PipelineValue])

Bases: PipelineValue

Data holder for the reference to an input node.

source_node: str | List[str | PipelineValue]

class iris.orchestration.pipeline_dataclasses.PipelineMetadata(*, pipeline_name: str, iris_version: str)

Bases: ImmutableModel

Data holder for input config’s metadata.

iris_version: str

pipeline_name: str

class iris.orchestration.pipeline_dataclasses.PipelineNode(*, name: str, algorithm: PipelineClass, inputs: List[PipelineInput], callbacks: List[PipelineClass] | None = None, seed: str | None = None)

Bases: ImmutableModel

Data holder for one node in a declared pipeline.

algorithm: PipelineClass

callbacks: List[PipelineClass] | None

inputs: List[PipelineInput]

name: str

seed: str | None

class iris.orchestration.pipeline_dataclasses.PipelineValue(*, name: str, index: int | None = None)

Bases: ImmutableModel

Data holder for pipeline value that flows through the system with optional index to specify value this holder refers to if value is of Iterable type.

index: int | None

name: str

iris.orchestration.validators module

iris.orchestration.validators.pipeline_config_duplicate_node_name_check(cls: type, v: List[Any], field: ModelField) → List[Any]

Check if all pipeline nodes have distinct names.

Parameters:

• cls (type) – Class type.

• v (List[Any]) – Value to check.

• field (fields.ModelField) – Field descriptor.

Raises:

IRISPipelineError – Raised if pipeline nodes aren’t unique.

Returns:

v sent for further processing.

Return type:

List[Any]

iris.orchestration.validators.pipeline_metadata_version_check(cls: type, v: str, field: ModelField, expected_version: str) → str

Check if the version provided in the input config matches the expected version (e.g. iris.__version__).

Module contents