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/runner/work/open-iris/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.9.3'¶

class Parameters(*, metadata: PipelineMetadata, pipeline: List[PipelineNode])[source]¶

Bases: Parameters

IRISPipeline parameters, all derived from the input config.

metadata: PipelineMetadata¶

pipeline: List[PipelineNode]¶

estimate(ir_image: IRImage, *args, **kwargs) → Any[source]¶

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

Parameters:

ir_image (IRImage) – Input Infrared image data.
*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(ir_image: IRImage, *args, **kwargs) → Any[source]¶

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

Parameters:

ir_image (IRImage) – Input Infrared image data.
*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_iris_pipeline module¶

class iris.pipelines.multiframe_iris_pipeline.MultiframeIrisPipeline(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_iris_pipeline_output>, error_manager=<function store_error_manager>, disabled_qa=[]))[source]¶

Bases: object

Pipeline that combines IRISPipeline and TemplatesAggregationPipeline. Takes a list of images and eye-side as input, processes each image through IRISPipeline, then aggregates the resulting templates using TemplatesAggregationPipeline.

Uses a unified configuration with two distinct parts: - iris_pipeline: Configuration for individual image processing - templates_aggregation: Configuration for template aggregation

IMPORTANT WARNING: Template Quality and Validation Requirements

To prevent aggregation of templates from different users, different eyes, or of different quality (off-gaze, occlusion, poor focus, etc.), users must pay extra attention to the filtering and validation of provided templates and corresponding thresholds during template creation. It is strongly advised to conduct a dedicated analysis of template quality metrics and establish appropriate validation criteria prior to using this functionality in production environments.

While the pipeline does perform some basic validation of the input templates, users are responsible for ensuring data integrity and appropriate quality thresholds to maintain system accuracy and security.

DEFAULT_PIPELINE_CFG_PATH = '/home/runner/work/open-iris/open-iris/src/iris/pipelines/confs/multiframe_iris_pipeline.yaml'¶

ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_multiframe_iris_pipeline_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])¶

PACKAGE_VERSION = '1.9.3'¶

static derive_iris_template_shape_from_config(iris_pipeline_config: Dict[str, Any]) → Tuple[int, int, int, int][source]¶

Derive the iris template shape from the pipeline configuration.

The shape is determined by scanning the filter_bank node configuration: - n_rows and n_cols from probe_schemas (first two dimensions) - number of filters (third dimension) - number of probe_schemas (fourth dimension)

Returns:: The iris template shape (n_rows, n_cols, n_filters, n_probe_schemas)
Return type:: Tuple[int, int, int, int]

estimate(ir_images: List[IRImage], *args: Any, **kwargs: Any) → Any[source]¶

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

Parameters:

ir_images (List[IRImage]) – List of input images.
*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: Dict[str, Any] | str | None) → Dict[str, Any][source]¶

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

Parameters:: config – Either • a dict already containing your pipeline sections, or • a YAML string (or None) that will be loaded from disk.
Returns:: Dictionary containing the iris_pipeline and templates_aggregation_pipeline configurations.
Return type:: Dict[str, Any]

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

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

Returns:: Initialised pipeline and error (if any).
Return type:: Dict[str, Union[MultiframeIrisPipeline, Optional[Dict[str, Any]]]]

run(ir_images: List[IRImage], *args: Any, **kwargs: Any) → Any[source]¶

Process multiple images through the combined pipeline. :param ir_images: List of input images. :type ir_images: List[IRImage] :param *args: Optional positional arguments for extensibility. :param **kwargs: Optional keyword arguments for extensibility.

Returns:: Output created by builder specified in environment.pipeline_output_builder.
Return type:: Any

iris.pipelines.templates_aggregation_pipeline module¶

class iris.pipelines.templates_aggregation_pipeline.TemplatesAggregationPipeline(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_templates_aggregation_output>, error_manager=<function store_error_manager>, disabled_qa=[]), subconfig_key: str | None = 'templates_aggregation')[source]¶

Bases: BasePipeline

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

DEFAULT_PIPELINE_CFG_PATH = '/home/runner/work/open-iris/open-iris/src/iris/pipelines/confs/templates_aggregation_pipeline.yaml'¶

ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_aggregation_templates_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])¶

PACKAGE_VERSION: str = '1.9.3'¶

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 | None = 'templates_aggregation') → Dict[str, Any][source]¶

Load and deserialize the pipeline configuration (for templates 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 None or empty string, the entire dict is returned. Otherwise, extracts the sub-dict at this key.

Returns:

The sub-dict at keyword (or the entire dict if keyword is None or empty).

Raises:

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

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

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

Returns:: Initialised pipeline and error (if any).
Return type:: Dict[str, Union[TemplatesAggregationPipeline, Optional[Dict[str, Any]]]]

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

Returns:: Output from the pipeline (type defined by subclass).
Return type:: OutputType

Module contents¶

class iris.pipelines.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/runner/work/open-iris/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.9.3'¶

class Parameters(*, metadata: PipelineMetadata, pipeline: List[PipelineNode])[source]¶

Bases: Parameters

IRISPipeline parameters, all derived from the input config.

metadata: PipelineMetadata¶

pipeline: List[PipelineNode]¶

estimate(ir_image: IRImage, *args, **kwargs) → Any[source]¶

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

Parameters:

ir_image (IRImage) – Input Infrared image data.
*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(ir_image: IRImage, *args, **kwargs) → Any[source]¶

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

Parameters:

ir_image (IRImage) – Input Infrared image data.
*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.

class iris.pipelines.MultiframeIrisPipeline(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_iris_pipeline_output>, error_manager=<function store_error_manager>, disabled_qa=[]))[source]¶

Bases: object

Uses a unified configuration with two distinct parts: - iris_pipeline: Configuration for individual image processing - templates_aggregation: Configuration for template aggregation

IMPORTANT WARNING: Template Quality and Validation Requirements

DEFAULT_PIPELINE_CFG_PATH = '/home/runner/work/open-iris/open-iris/src/iris/pipelines/confs/multiframe_iris_pipeline.yaml'¶

ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_multiframe_iris_pipeline_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])¶

PACKAGE_VERSION = '1.9.3'¶

static derive_iris_template_shape_from_config(iris_pipeline_config: Dict[str, Any]) → Tuple[int, int, int, int][source]¶

Derive the iris template shape from the pipeline configuration.

Returns:: The iris template shape (n_rows, n_cols, n_filters, n_probe_schemas)
Return type:: Tuple[int, int, int, int]

estimate(ir_images: List[IRImage], *args: Any, **kwargs: Any) → Any[source]¶

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

Parameters:

ir_images (List[IRImage]) – List of input images.
*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: Dict[str, Any] | str | None) → Dict[str, Any][source]¶

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

Parameters:: config – Either • a dict already containing your pipeline sections, or • a YAML string (or None) that will be loaded from disk.
Returns:: Dictionary containing the iris_pipeline and templates_aggregation_pipeline configurations.
Return type:: Dict[str, Any]

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

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

Returns:: Initialised pipeline and error (if any).
Return type:: Dict[str, Union[MultiframeIrisPipeline, Optional[Dict[str, Any]]]]

run(ir_images: List[IRImage], *args: Any, **kwargs: Any) → Any[source]¶

Returns:: Output created by builder specified in environment.pipeline_output_builder.
Return type:: Any

class iris.pipelines.TemplatesAggregationPipeline(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_templates_aggregation_output>, error_manager=<function store_error_manager>, disabled_qa=[]), subconfig_key: str | None = 'templates_aggregation')[source]¶

Bases: BasePipeline

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

DEFAULT_PIPELINE_CFG_PATH = '/home/runner/work/open-iris/open-iris/src/iris/pipelines/confs/templates_aggregation_pipeline.yaml'¶

ORB_ENVIRONMENT = Environment(call_trace_initialiser=<function PipelineCallTraceStorage.initialise>, pipeline_output_builder=<function build_aggregation_templates_orb_output>, error_manager=<function store_error_manager>, disabled_qa=[])¶

PACKAGE_VERSION: str = '1.9.3'¶

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 | None = 'templates_aggregation') → Dict[str, Any][source]¶

Load and deserialize the pipeline configuration (for templates 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 None or empty string, the entire dict is returned. Otherwise, extracts the sub-dict at this key.

Returns:

The sub-dict at keyword (or the entire dict if keyword is None or empty).

Raises:

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

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

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

Returns:: Initialised pipeline and error (if any).
Return type:: Dict[str, Union[TemplatesAggregationPipeline, Optional[Dict[str, Any]]]]

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

Returns:: Output from the pipeline (type defined by subclass).
Return type:: OutputType