feat: compute group:value pairs for tag status (#1673)

* feat: compute group:value pairs for tag status

* chore: lint

* tests: add coverage for properties

Author: dbirman

src/aind_data_schema/core/quality_control.py

@@ -146,7 +146,7 @@ class QualityControl(DataCoreModel):
 
     @property
     def tags(self) -> List[str]:
-        """Get all unique tags from all metrics
+        """Get all unique tag values from all metrics
 
         Returns
         -------
@@ -158,6 +158,21 @@ def tags(self) -> List[str]:
             all_tags.extend(metric.tags.values())
         return list(set(all_tags))
 
+    @property
+    def tag_pairs(self) -> List[str]:
+        """Get all unique tag key:value pairs from all metrics
+
+        Returns
+        -------
+        List[str]
+            List of all unique tag key:value pairs across all metrics in 'key:value' format
+        """
+        all_tag_pairs = []
+        for metric in self.metrics:
+            for key, value in metric.tags.items():
+                all_tag_pairs.append(f"{key}:{value}")
+        return list(set(all_tag_pairs))
+
     @property
     def modalities(self) -> List[Modality.ONE_OF]:
         """Get all unique modalities from all metrics
@@ -192,9 +207,9 @@ def compute_status(self):
         if self.metrics:
             computed_status = {}
 
-            # Compute tag statuses
-            for tag in self.tags:
-                computed_status[tag] = self.evaluate_status(tag=tag)
+            # Compute tag statuses (using key:value format)
+            for tag_pair in self.tag_pairs:
+                computed_status[tag_pair] = self.evaluate_status(tag=tag_pair)
 
             # Compute modality statuses
             for modality in self.modalities:
@@ -335,23 +350,36 @@ def _get_filtered_statuses(
     tag_filter: Optional[List[str]] = None,
     allow_tag_failures: List[str] = [],
 ):
-    """Get the status of metrics filtered by modality, stage, tag, and date."""
+    """Get the status of metrics filtered by modality, stage, tag, and date.
+
+    tag_filter can contain either 'key:value' pairs or just tag values for backward compatibility.
+    allow_tag_failures can contain either 'key:value' pairs or just tag values.
+    """
     filtered_statuses = []
     for metric in metrics:
         # Apply filters
         if modality_filter and metric.modality not in modality_filter:
             continue
         if stage_filter and metric.stage not in stage_filter:
             continue
-        if tag_filter and not (metric.tags and any(t in metric.tags.values() for t in tag_filter)):
-            continue
+        if tag_filter:
+            # Check if any of the filter tags match this metric's tags
+            # Support both 'key:value' format and just values for backward compatibility
+            metric_tag_pairs = [f"{k}:{v}" for k, v in metric.tags.items()]
+            metric_tag_values = list(metric.tags.values())
+            if not any(t in metric_tag_pairs or t in metric_tag_values for t in tag_filter):
+                continue
 
         # Get status at the specified date using the helper function
         status = _get_status_by_date(metric, date)
-        # Check if any of our tag values are in the allow_tag_failures list
+        # Check if any of our tag key:value pairs or values are in the allow_tag_failures list
         if status == Status.FAIL and metric.tags:
-            metric_tag_values = set(metric.tags.values())
-            if any(tag_value in allow_tag_failures for tag_value in metric_tag_values):
+            metric_tag_pairs = [f"{k}:{v}" for k, v in metric.tags.items()]
+            metric_tag_values = list(metric.tags.values())
+            if any(
+                tag_pair in allow_tag_failures or tag_value in allow_tag_failures
+                for tag_pair, tag_value in zip(metric_tag_pairs, metric_tag_values)
+            ):
                 status = Status.PASS
         filtered_statuses.append(status)
 

tests/test_quality_control.py

@@ -47,6 +47,28 @@ def test_tags_list_to_dict_conversion(self):
         self.assertIsInstance(metric.tags, dict)
         self.assertEqual(metric.tags, {"tag_1": "tag1", "tag_2": "tag2", "tag_3": "tag3"})
 
+    def test_tags_property(self):
+        """test that QualityControl.tags returns all unique tag values"""
+        tags = quality_control.tags
+        self.assertIsInstance(tags, list)
+        self.assertIn("Probe A", tags)
+        self.assertIn("Probe B", tags)
+        self.assertIn("Probe C", tags)
+        self.assertIn("Video 1", tags)
+        self.assertIn("Video 2", tags)
+        self.assertEqual(len(tags), 5)
+
+    def test_tag_pairs_property(self):
+        """test that QualityControl.tag_pairs returns all unique key:value pairs"""
+        tag_pairs = quality_control.tag_pairs
+        self.assertIsInstance(tag_pairs, list)
+        self.assertIn("probe:Probe A", tag_pairs)
+        self.assertIn("probe:Probe B", tag_pairs)
+        self.assertIn("probe:Probe C", tag_pairs)
+        self.assertIn("video:Video 1", tag_pairs)
+        self.assertIn("video:Video 2", tag_pairs)
+        self.assertEqual(len(tag_pairs), 5)
+
     def test_overall_status(self):
         """test that overall status goes to pass/pending/fail correctly"""
 
@@ -186,7 +208,7 @@ def test_evaluation_status(self):
             )
         )
 
-        self.assertEqual(qc.evaluate_status(tag="Drift map"), Status.FAIL)
+        self.assertEqual(qc.evaluate_status(tag="group:Drift map"), Status.FAIL)
 
     def test_allowed_failed_metrics(self):
         """Test that if you set the flag to allow failures that tags pass"""
@@ -224,17 +246,17 @@ def test_allowed_failed_metrics(self):
             default_grouping=["group"],
         )
 
-        self.assertEqual(qc.evaluate_status(tag="Drift map"), Status.PENDING)
+        self.assertEqual(qc.evaluate_status(tag="group:Drift map"), Status.PENDING)
 
         # Replace the pending evaluation with a fail, evaluation should not evaluate to pass
         qc.metrics[1].status_history[0].status = Status.FAIL
 
-        self.assertEqual(qc.evaluate_status(tag="Drift map"), Status.FAIL)
+        self.assertEqual(qc.evaluate_status(tag="group:Drift map"), Status.FAIL)
 
         # Now add the tag to allow_tag_failures
-        qc.allow_tag_failures = ["Drift map"]
+        qc.allow_tag_failures = ["group:Drift map"]
 
-        self.assertEqual(qc.evaluate_status(tag="Drift map"), Status.PASS)
+        self.assertEqual(qc.evaluate_status(tag="group:Drift map"), Status.PASS)
 
     def test_metric_history_order(self):
         """Test that the order of the metric status history list is preserved when dumping"""
@@ -441,10 +463,10 @@ def test_status_filters(self):
                 "behavior": Status.FAIL,
                 "behavior-videos": Status.PENDING,
                 "ecephys": Status.PASS,
-                # Tags
-                "test_group": Status.PASS,
-                "test_group2": Status.FAIL,
-                "tag1": Status.PENDING,
+                # Tags (now in key:value format)
+                "group:test_group": Status.PASS,
+                "group:test_group2": Status.FAIL,
+                "type:tag1": Status.PENDING,
             },
         )
 
@@ -454,7 +476,7 @@ def test_status_filters(self):
         self.assertEqual(q.evaluate_status(modality=[Modality.ECEPHYS, Modality.BEHAVIOR]), Status.FAIL)
         self.assertEqual(q.evaluate_status(stage=Stage.RAW), Status.FAIL)
         self.assertEqual(q.evaluate_status(stage=Stage.PROCESSING), Status.PASS)
-        self.assertEqual(q.evaluate_status(tag="tag1"), Status.PENDING)
+        self.assertEqual(q.evaluate_status(tag="type:tag1"), Status.PENDING)
 
     def test_status_date(self):
         """QualityControl.status(date=) should return the correct status for the given date"""
@@ -635,7 +657,7 @@ def test_get_filtered_statuses_helper(self):
         shared_tag_statuses = _get_filtered_statuses(
             metrics=all_metrics,
             date=test_date,
-            tag_filter=["shared_tag"],
+            tag_filter=["group:shared_tag"],
         )
         self.assertEqual(len(shared_tag_statuses), 2)  # Our BEHAVIOR and OPHYS test metrics
         self.assertIn(Status.PASS, shared_tag_statuses)
@@ -655,7 +677,7 @@ def test_get_filtered_statuses_helper(self):
         time_test_statuses = _get_filtered_statuses(
             metrics=all_metrics,
             date=earlier_date,
-            tag_filter=["time_test"],
+            tag_filter=["test:time_test"],
         )
         self.assertEqual(len(time_test_statuses), 1)
         self.assertEqual(time_test_statuses[0], Status.FAIL)  # Should get the earlier FAIL status
@@ -664,8 +686,8 @@ def test_get_filtered_statuses_helper(self):
         ophys_fail_statuses = _get_filtered_statuses(
             metrics=all_metrics,
             date=test_date,
-            tag_filter=["ophys_tag"],
-            allow_tag_failures=["ophys_tag"],
+            tag_filter=["type:ophys_tag"],
+            allow_tag_failures=["type:ophys_tag"],
         )
         self.assertEqual(len(ophys_fail_statuses), 1)
         self.assertEqual(ophys_fail_statuses[0], Status.PASS)  # FAIL converted to PASS
@@ -756,16 +778,16 @@ def test_helper_functions_integration(self):
         # Test status at different times
         early_date = datetime.fromisoformat("2020-02-01T00:00:00+00:00")
         # At early date: metric 1 is FAIL, metric 2 is PASS -> overall FAIL
-        early_status = qc.evaluate_status(date=early_date, tag="time_sensitive")
+        early_status = qc.evaluate_status(date=early_date, tag="group:time_sensitive")
         self.assertEqual(early_status, Status.FAIL)
 
         # At test date: metric 1 is PASS, metric 2 is PASS -> overall PASS
-        test_status = qc.evaluate_status(date=test_date, tag="time_sensitive")
+        test_status = qc.evaluate_status(date=test_date, tag="group:time_sensitive")
         self.assertEqual(test_status, Status.PASS)
 
         # At late date: metric 1 is PASS, metric 2 is FAIL -> overall FAIL
         late_date = datetime.fromisoformat("2020-08-01T00:00:00+00:00")
-        late_status = qc.evaluate_status(date=late_date, tag="time_sensitive")
+        late_status = qc.evaluate_status(date=late_date, tag="group:time_sensitive")
         self.assertEqual(late_status, Status.FAIL)
 
     def test_backwards_compatibility_default_grouping(self):