iris.pipelines package

Submodules

iris.pipelines.base_pipeline module

class iris.pipelines.base_pipeline.BasePipeline(config: Dict[str, Any], env: Environment)[source]

Bases: Algorithm, Generic[InputType, OutputType], ABC

Generic base class for IRIS pipelines, abstracting shared logic for pipeline execution, node instantiation, environment setup, and call trace management. Subclasses should implement input/output specifics via _handle_input and _handle_output.

PACKAGE_VERSION: str
estimate(pipeline_input: InputType, *args: Any, **kwargs: Any) → OutputType[source]

Wrap the run method to match the Orb system AI models call interface.

Parameters:

  • pipeline_input (InputType) – Input to the pipeline (type defined by subclass).

  • *args (Any) – Optional positional arguments for extensibility.

  • **kwargs (Any) – Optional keyword arguments for extensibility.

Returns:

Output from the pipeline (type defined by subclass).

Return type:

OutputType

run(pipeline_input: InputType, *args: Any, **kwargs: Any) → OutputType[source]

Main pipeline execution loop. Handles input, node execution, and output. :param pipeline_input: Input to the pipeline (type defined by subclass). :type pipeline_input: InputType :param *args: Optional positional arguments for extensibility. :type *args: Any :param **kwargs: Optional keyword arguments for extensibility. :type **kwargs: Any

Returns:

Output from the pipeline (type defined by subclass).

Return type:

OutputType

iris.pipelines.base_pipeline.load_yaml_config(config: str | None) → Dict[str, Any][source]

Loads YAML config from a YAML string or a path to a YAML file.

Parameters:

config – YAML string or path to YAML file.

Returns:

Deserialized config dict.

Return type:

Dict[str, Any]

Raises:

ValueError – If config is None, not a string, or can’t be parsed.

iris.pipelines.iris_pipeline module

class iris.pipelines.iris_pipeline.IRISPipeline(config: ~typing.Dict[str, ~typing.Any] | str | None = None, env: ~iris.orchestration.environment.Environment = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_simple_iris_pipeline_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[]))[source]

Bases: BasePipeline

Implementation of a fully configurable iris recognition pipeline. Inherits shared logic from BasePipeline and implements input/output specifics.

DEBUGGING_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_simple_iris_pipeline_debugging_output>, error_manager=<function store_error_manager>, disabled_qa=[<class 'iris.nodes.validators.object_validators.Pupil2IrisPropertyValidator'>, <class 'iris.nodes.validators.object_validators.OffgazeValidator'>, <class 'iris.nodes.validators.object_validators.OcclusionValidator'>, <class 'iris.nodes.validators.object_validators.IsPupilInsideIrisValidator'>, <class 'iris.nodes.validators.object_validators.SharpnessValidator'>, <class 'iris.nodes.validators.object_validators.IsMaskTooSmallValidator'>, <class 'iris.nodes.validators.cross_object_validators.EyeCentersInsideImageValidator'>, <class 'iris.nodes.validators.cross_object_validators.ExtrapolatedPolygonsInsideImageValidator'>])
DEFAULT_PIPELINE_CFG_PATH = '/home/rostyslav.shevchenko/Projects/open-iris/src/iris/pipelines/confs/pipeline.yaml'
ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_iris_pipeline_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])
PACKAGE_VERSION: str = '1.8.0'
class Parameters(*, metadata: PipelineMetadata, pipeline: List[PipelineNode])[source]

Bases: Parameters

IRISPipeline parameters, all derived from the input config.

metadata: PipelineMetadata
pipeline: List[PipelineNode]
estimate(img_data: ndarray, eye_side: Literal['left', 'right'], *args, **kwargs) → Any[source]

Wrap the run method to match the Orb system AI models call interface.

Parameters:

  • img_data (np.ndarray) – Input image data.

  • eye_side (Literal["left", "right"]) – Eye side.

  • *args – Optional positional arguments for extensibility.

  • **kwargs – Optional keyword arguments for extensibility.

Returns:

Output created by builder specified in environment.pipeline_output_builder.

Return type:

Any

classmethod load_config(config: str | None) → Dict[str, Any][source]

Load and deserialize the pipeline configuration. :param config: YAML string or None for default config. :type config: Optional[str]

Returns:

Deserialized config dict.

Return type:

Dict[str, Any]

classmethod load_from_config(config: str) → Dict[str, IRISPipeline | Dict[str, Any] | None][source]

Given an iris config string in base64, initialise an IRISPipeline with this config. :param config: Base64-encoded config string. :type config: str

Returns:

Initialised pipeline and error (if any).

Return type:

Dict[str, Union[IRISPipeline, Optional[Dict[str, Any]]]]

run(img_data: ndarray, eye_side: Literal['left', 'right'], *args, **kwargs) → Any[source]

Wrap the run method to match the Orb system AI models call interface.

Parameters:

  • img_data (np.ndarray) – Input Infrared image data.

  • eye_side (Literal["left", "right"]) – Eye side.

  • *args – Optional positional arguments for extensibility.

  • **kwargs – Optional keyword arguments for extensibility.

Returns:

Output created by builder specified in environment.pipeline_output_builder.

Return type:

Any

update_config(config: str) → None[source]

Update the pipeline configuration based on the provided base64-encoded string.

Parameters:

config (str) – Base64-encoded string of the new configuration.

iris.pipelines.multiframe_aggregation_pipeline module

class iris.pipelines.multiframe_aggregation_pipeline.MultiframeAggregationPipeline(config: ~typing.Dict[str, ~typing.Any] | str | None = None, env: ~iris.orchestration.environment.Environment = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_simple_multiframe_aggregation_output>, error_manager=<function store_error_manager>, disabled_qa=[]), subconfig_key: str = 'templates_aggregation')[source]

Bases: BasePipeline

Pipeline for multiframe iris template aggregation. Inherits shared logic from BasePipeline and implements input/output specifics.

DEFAULT_PIPELINE_CFG_PATH = '/home/rostyslav.shevchenko/Projects/open-iris/src/iris/pipelines/confs/multiframe_aggregation_pipeline.yaml'
ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_aggregation_multiframe_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])
PACKAGE_VERSION: str = '1.8.0'
class Parameters(*, metadata: PipelineMetadata, pipeline: List[PipelineNode])[source]

Bases: Parameters

metadata: PipelineMetadata
pipeline: List[PipelineNode]
classmethod load_config(config: Dict[str, Any] | str | None, keyword: str = 'templates_aggregation') → Dict[str, Any][source]

Load and deserialize the pipeline configuration (for multiframe aggregation).

Parameters:

  • config – Either • a dict already containing your pipeline sections, or • a YAML string (or None) that will be loaded from disk.

  • keyword – If non‐empty, the top‐level key to extract from the final dict.

Returns:

The sub-dict at keyword (or the entire dict if keyword==””).

Raises:

ValueError – if keyword is non-empty and not found in the config.

classmethod load_from_config(config: str) → Dict[str, MultiframeAggregationPipeline | Dict[str, Any] | None][source]

Given an iris config string in base64, initialise a MultiframeAggregationPipeline with this config. :param config: Base64-encoded config string. :type config: str

Returns:

Initialised pipeline and error (if any).

Return type:

Dict[str, Union[MultiframeAggregationPipeline, Optional[Dict[str, Any]]]]

run(templates: List[IrisTemplate], *args: Any, **kwargs: Any) → Any[source]

Main pipeline execution loop. Handles input, node execution, and output. :param pipeline_input: Input to the pipeline (type defined by subclass). :type pipeline_input: InputType :param *args: Optional positional arguments for extensibility. :type *args: Any :param **kwargs: Optional keyword arguments for extensibility. :type **kwargs: Any

Returns:

Output from the pipeline (type defined by subclass).

Return type:

OutputType

Module contents