Merge pull request #178 from worldcoin/dev

Fix Error Handling and None Value Processing in Templates Aggregation Pipeline (#177)

Author: ShevchenkoRostyslav

colab/ConfiguringCustomPipeline.ipynb

@@ -126,7 +126,7 @@
     "```yaml\n",
     "metadata:\n",
     "  pipeline_name: iris_pipeline\n",
-    "  iris_version: 1.9.6\n",
+    "  iris_version: 1.9.7\n",
     "```\n",
     "\n",
     "The top YAML file contains `IRISPipeline` metadata, used to both describe `IRISPipeline` and specify package parameters that are later used to verify compatibility between `iris` package version/release and later, specified in the `pipeline` YAML file section, pipeline's graph.\n",
@@ -206,7 +206,7 @@
    "outputs": [],
    "source": [
     "default_pipeline_conf = {\n",
-    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.6\"},\n",
+    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.7\"},\n",
     "    \"pipeline\": [\n",
     "        {\n",
     "            \"name\": \"segmentation\",\n",
@@ -481,7 +481,7 @@
    "outputs": [],
    "source": [
     "new_pipeline_conf = {\n",
-    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.6\"},\n",
+    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.7\"},\n",
     "    \"pipeline\": [\n",
     "        {\n",
     "            \"name\": \"segmentation\",\n",
@@ -760,7 +760,7 @@
    "outputs": [],
    "source": [
     "default_pipeline_conf = {\n",
-    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.6\"},\n",
+    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.7\"},\n",
     "    \"pipeline\": [\n",
     "        {\n",
     "            \"name\": \"segmentation\",\n",
@@ -1035,7 +1035,7 @@
    "outputs": [],
    "source": [
     "new_pipeline_conf = {\n",
-    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.6\"},\n",
+    "    \"metadata\": {\"pipeline_name\": \"iris_pipeline\", \"iris_version\": \"1.9.7\"},\n",
     "    \"pipeline\": [\n",
     "        {\n",
     "            \"name\": \"segmentation\",\n",

docs/source/examples/custom_pipeline.rst

@@ -18,7 +18,7 @@ When the ``IRISPipeline`` pipeline is created with default parameters, it's grap
 
     metadata:
     pipeline_name: iris_pipeline
-    iris_version: 1.9.6
+    iris_version: 1.9.7
 
 The top YAML file contains ``IRISPipeline`` metadata, used to both describe ``IRISPipeline`` and specify package parameters that are later used to verify compatibility between ``iris`` package version/release and later, specified in the ``pipeline`` YAML file section, pipeline's graph.
 
@@ -93,7 +93,7 @@ First let's intantiate ``IRISPipeline`` with default configuration and see ``iri
 .. code-block:: python
 
     default_pipeline_conf = {
-        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.6"},
+        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.7"},
         "pipeline": [
             {
                 "name": "segmentation",
@@ -320,7 +320,7 @@ As expected all threshold values are set to default ``0.5`` value. Now, let's mo
 .. code-block:: python
 
     new_pipeline_conf = {
-        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.6"},
+        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.7"},
         "pipeline": [
             {
                 "name": "segmentation",
@@ -552,7 +552,7 @@ First let's instantiate ``IRISPipeline`` with default configuration and see node
 .. code-block:: python
 
     default_pipeline_conf = {
-        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.6"},
+        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.7"},
         "pipeline": [
             {
                 "name": "segmentation",
@@ -783,7 +783,7 @@ As expected, ``input_polygons`` argument of the ``run`` method is taken from the
 .. code-block:: python
 
     new_pipeline_conf = {
-        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.6"},
+        "metadata": {"pipeline_name": "iris_pipeline", "iris_version": "1.9.7"},
         "pipeline": [
             {
                 "name": "segmentation",

src/iris/_version.py

@@ -1 +1 @@
-__version__ = "1.9.6"
+__version__ = "1.9.7"

src/iris/io/errors.py

@@ -158,3 +158,9 @@ class IdentityValidationError(Exception):
     """Identity validation Error class."""
 
     pass
+
+
+class DifferentImageIdsTemplatesListLenError(Exception):
+    """Error raised when the number of image_ids doesn't match the number of templates."""
+
+    pass

src/iris/orchestration/output_builders.py

@@ -267,7 +267,9 @@ def __get_multiframe_iris_pipeline_metadata(call_trace: PipelineCallTraceStorage
     ),
     OutputFieldSpec(
         key="templates_aggregation_metadata",
-        extractor=lambda ct: {k: v for k, v in ct.get("aggregation_result", {}).items() if k != "iris_template"},
+        extractor=lambda ct: None
+        if (agg := ct.get("aggregation_result")) is None
+        else {k: v for k, v in agg.items() if k != "iris_template"},
         safe_serialize=False,
     ),
 ]
@@ -292,7 +294,9 @@ def __get_multiframe_iris_pipeline_metadata(call_trace: PipelineCallTraceStorage
     ),
     OutputFieldSpec(
         key="templates_aggregation_metadata",
-        extractor=lambda ct: {k: v for k, v in ct.get("aggregation_result", {}).items() if k != "iris_template"},
+        extractor=lambda ct: None
+        if (agg := ct.get("aggregation_result")) is None
+        else {k: v for k, v in agg.items() if k != "iris_template"},
         safe_serialize=False,
     ),
 ]

src/iris/pipelines/confs/multiframe_iris_pipeline.yaml

@@ -1,12 +1,12 @@
 metadata:
   pipeline_name: multiframe_iris_pipeline
-  iris_version: 1.9.6
+  iris_version: 1.9.7
 
 # Configuration for individual image processing (IRISPipeline)
 iris_pipeline:
   metadata:
     pipeline_name: iris_pipeline
-    iris_version: 1.9.6
+    iris_version: 1.9.7
 
   pipeline:
     - name: segmentation
@@ -315,7 +315,7 @@ iris_pipeline:
 # Configuration for template aggregation (TemplatesAggregationPipeline)
 templates_aggregation_pipeline:
   metadata:
-    iris_version: 1.9.6
+    iris_version: 1.9.7
     pipeline_name: templates_aggregation
 
   pipeline:

src/iris/pipelines/confs/pipeline.yaml

@@ -1,6 +1,6 @@
 metadata:
   pipeline_name: iris_pipeline
-  iris_version: 1.9.6
+  iris_version: 1.9.7
 
 pipeline:
   - name: segmentation

src/iris/pipelines/confs/templates_aggregation_pipeline.yaml

@@ -1,6 +1,6 @@
 metadata:
   pipeline_name: iris_pipeline
-  iris_version: 1.9.6
+  iris_version: 1.9.7
 
 pipeline:
   - name: segmentation
@@ -307,7 +307,7 @@ pipeline:
 
 templates_aggregation:
   metadata:
-    iris_version: 1.9.6
+    iris_version: 1.9.7
     pipeline_name: templates_aggregation
 
   pipeline:

src/iris/pipelines/templates_aggregation_pipeline.py

@@ -8,6 +8,7 @@
 from iris.callbacks.pipeline_trace import PipelineCallTraceStorage
 from iris.io.class_configs import Algorithm
 from iris.io.dataclasses import IrisTemplate, IrisTemplateWithId
+from iris.io.errors import DifferentImageIdsTemplatesListLenError
 from iris.orchestration.environment import Environment
 from iris.orchestration.error_managers import store_error_manager
 from iris.orchestration.output_builders import (
@@ -67,17 +68,21 @@ def __init__(
     def run(
         self, templates: List[IrisTemplate], image_ids: Optional[List[str]] = None, *args: Any, **kwargs: Any
     ) -> Any:
-        # Validate input consistency
-        if image_ids is not None and len(image_ids) != len(templates):
-            raise ValueError(
-                f"Number of image_ids ({len(image_ids)}) must match number of templates ({len(templates)})"
-            )
-
-        # Create IrisTemplateWithId
-        templates_with_ids = []
-        for i, template in enumerate(templates):
-            image_id = image_ids[i] if image_ids else f"frame_{i}"
-            templates_with_ids.append(IrisTemplateWithId.from_template(template, image_id))
+        try:
+            # Validate input consistency
+            if image_ids is not None and len(image_ids) != len(templates):
+                raise DifferentImageIdsTemplatesListLenError(
+                    f"Number of image_ids ({len(image_ids)}) must match number of templates ({len(templates)})"
+                )
+
+            # Create IrisTemplateWithId
+            templates_with_ids = []
+            for i, template in enumerate(templates):
+                image_id = image_ids[i] if image_ids else f"frame_{i}"
+                templates_with_ids.append(IrisTemplateWithId.from_template(template, image_id))
+        except Exception as e:
+            self.env.error_manager(self.call_trace, e)
+            return self._handle_output(*args, **kwargs)
 
         pipeline_input = {"templates_with_ids": templates_with_ids}
         return super().run(pipeline_input, *args, **kwargs)

tests/e2e_tests/pipelines/test_e2e_multiframe_aggregation_pipeline.py

@@ -508,3 +508,136 @@ def test_pipeline_reproducibility(self, standalone_aggregation_config):
                 np.testing.assert_array_equal(
                     result1["iris_template"].mask_codes[i], result2["iris_template"].mask_codes[i]
                 )
+
+
+class TestE2ETemplatesAggregationFailures:
+    """End-to-end tests for templates aggregation pipeline failure scenarios."""
+
+    @pytest.fixture
+    def failure_prone_config_no_clusters(self):
+        """Configuration that will cause no identity clusters to be found."""
+        return {
+            "metadata": {"pipeline_name": "templates_aggregation", "iris_version": "1.7.1"},
+            "pipeline": [
+                {
+                    "name": "templates_alignment",
+                    "algorithm": {
+                        "class_name": "iris.nodes.templates_alignment.hamming_distance_based.HammingDistanceBasedAlignment",
+                        "params": {
+                            "rotation_shift": 15,
+                            "use_first_as_reference": False,
+                            "normalise": True,
+                            "reference_selection_method": "linear",
+                        },
+                    },
+                    "inputs": [{"name": "templates_with_ids", "source_node": "input"}],
+                    "callbacks": [],
+                },
+                {
+                    "name": "identity_validation",
+                    "algorithm": {
+                        "class_name": "iris.nodes.templates_filter.single_identity_filter.TemplateIdentityFilter",
+                        "params": {
+                            "identity_distance_threshold": 0.05,  # Very low threshold
+                            "identity_validation_action": "raise_error",  # Will raise error
+                            "min_templates_after_validation": 1,
+                        },
+                    },
+                    "inputs": [{"name": "aligned_templates", "source_node": "templates_alignment"}],
+                    "callbacks": [],
+                },
+                {
+                    "name": "templates_aggregation",
+                    "algorithm": {
+                        "class_name": "iris.nodes.templates_aggregation.majority_vote.MajorityVoteAggregation",
+                        "params": {
+                            "consistency_threshold": 0.75,
+                            "mask_threshold": 0.01,
+                            "use_inconsistent_bits": True,
+                            "inconsistent_bit_threshold": 0.4,
+                        },
+                    },
+                    "inputs": [{"name": "templates", "source_node": "identity_validation"}],
+                    "callbacks": [],
+                },
+            ],
+        }
+
+    @pytest.fixture
+    def very_different_templates(self):
+        """Create templates that are very different from each other (high Hamming distances)."""
+        templates = []
+
+        for i in range(4):
+            iris_codes = []
+            mask_codes = []
+
+            for wavelet in range(2):
+                # Create very different patterns for each template
+                if i == 0:
+                    # Template 0: mostly zeros
+                    iris_code = np.zeros((8, 32, 2), dtype=bool)
+                elif i == 1:
+                    # Template 1: mostly ones
+                    iris_code = np.ones((8, 32, 2), dtype=bool)
+                elif i == 2:
+                    # Template 2: checkerboard pattern
+                    iris_code = np.zeros((8, 32, 2), dtype=bool)
+                    iris_code[::2, ::2] = True
+                    iris_code[1::2, 1::2] = True
+                else:
+                    # Template 3: inverse checkerboard
+                    iris_code = np.ones((8, 32, 2), dtype=bool)
+                    iris_code[::2, ::2] = False
+                    iris_code[1::2, 1::2] = False
+
+                # All templates have good masks
+                mask_code = np.ones((8, 32, 2), dtype=bool)
+
+                iris_codes.append(iris_code)
+                mask_codes.append(mask_code)
+
+            template = IrisTemplate(iris_codes=iris_codes, mask_codes=mask_codes, iris_code_version="v2.1")
+            templates.append(template)
+
+        return templates
+
+    def test_e2e_no_identity_clusters_found(self, failure_prone_config_no_clusters, very_different_templates):
+        """Test end-to-end pipeline when no identity clusters are found."""
+        env = Environment(
+            pipeline_output_builder=build_simple_templates_aggregation_output,
+            error_manager=store_error_manager,
+            call_trace_initialiser=PipelineCallTraceStorage.initialise,
+        )
+
+        pipeline = TemplatesAggregationPipeline(config=failure_prone_config_no_clusters, env=env, subconfig_key="")
+
+        image_ids = [f"image_{i}" for i in range(len(very_different_templates))]
+        result = pipeline.run(very_different_templates, image_ids)
+
+        # Should have error due to no clusters found
+        assert result["error"] is not None
+        assert "no identity clusters" in result["error"]["message"]
+        assert result["iris_template"] is None
+
+        # Metadata should still be generated
+        assert result["metadata"] is not None
+        assert result["metadata"]["input_templates_count"] == len(very_different_templates)
+
+    def test_e2e_empty_templates_list(self):
+        """Test end-to-end pipeline with empty templates list."""
+        env = Environment(
+            pipeline_output_builder=build_simple_templates_aggregation_output,
+            error_manager=store_error_manager,
+            call_trace_initialiser=PipelineCallTraceStorage.initialise,
+        )
+
+        pipeline = TemplatesAggregationPipeline(env=env, subconfig_key="")
+
+        result = pipeline.run(templates=[], image_ids=["image1"])
+
+        assert result["error"] is not None
+        assert "must match number of templates" in result["error"]["message"]
+        assert result["iris_template"] is None
+        assert result["metadata"] is not None
+        assert result["metadata"]["input_templates_count"] is None