Source code for iris.orchestration.pipeline_dataclasses

from __future__ import annotations

from typing import Dict, List, Optional, Tuple, Union

from iris.io.class_configs import ImmutableModel




class PipelineMetadata(ImmutableModel):
    """Data holder for input config's metadata."""

    pipeline_name: str
    iris_version: str





class PipelineValue(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."""

    name: str
    index: Optional[int]





class PipelineInput(PipelineValue):
    """Data holder for the reference to an input node."""

    source_node: Union[str, List[Union[str, PipelineValue]]]





class PipelineClass(ImmutableModel):
    """Data holder for the reference to any class: Algorithm, Callback, etc."""

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





class PipelineNode(ImmutableModel):
    """Data holder for one node in a declared pipeline."""

    name: str
    algorithm: PipelineClass
    inputs: List[PipelineInput]
    callbacks: Optional[List[PipelineClass]]
    seed: Optional[str] = None