Implement null detections (calculate what has been processed or not)

ami/jobs/tests/test_tasks.py

@@ -17,7 +17,8 @@
 from ami.jobs.models import Job, JobDispatchMode, JobState, MLJob
 from ami.jobs.tasks import process_nats_pipeline_result
 from ami.main.models import Detection, Project, SourceImage, SourceImageCollection
-from ami.ml.models import Pipeline
+from ami.ml.models import Algorithm, Pipeline
+from ami.ml.models.algorithm import AlgorithmTaskType
 from ami.ml.orchestration.async_job_state import AsyncJobStateManager
 from ami.ml.schemas import PipelineResultsError, PipelineResultsResponse, SourceImageResponse
 from ami.users.models import User
@@ -180,6 +181,15 @@ def test_process_nats_pipeline_result_mixed_results(self, mock_manager_class):
         """
         mock_manager = self._setup_mock_nats(mock_manager_class)
 
+        # Create detection algorithm for the pipeline
+        detection_algorithm = Algorithm.objects.create(
+            name="test-detector",
+            key="test-detector",
+            task_type=AlgorithmTaskType.LOCALIZATION,
+        )
+        # Update pipeline to include detection algorithm
+        self.pipeline.algorithms.add(detection_algorithm)
+
         # For this test, we just want to verify progress tracking works with mixed results
         # We'll skip checking final job completion status since that depends on all stages
 

ami/main/admin.py

@@ -265,6 +265,7 @@ class SourceImageAdmin(AdminBase):
         "checksum",
         "checksum_algorithm",
         "created_at",
+        "get_was_processed",
     )
 
     list_filter = (

ami/main/api/serializers.py

@@ -1246,6 +1246,7 @@ class Meta:
             "source_images",
             "source_images_count",
             "source_images_with_detections_count",
+            "source_images_processed_count",
             "occurrences_count",
             "taxa_count",
             "description",
@@ -1547,6 +1548,7 @@ class EventTimelineIntervalSerializer(serializers.Serializer):
     captures_count = serializers.IntegerField()
     detections_count = serializers.IntegerField()
     detections_avg = serializers.IntegerField()
+    was_processed = serializers.BooleanField()
 
 
 class EventTimelineMetaSerializer(serializers.Serializer):

ami/main/api/views.py

@@ -36,6 +36,7 @@
 from ami.utils.storages import ConnectionTestResult
 
 from ..models import (
+    NULL_DETECTIONS_FILTER,
     Classification,
     Deployment,
     Detection,
@@ -378,7 +379,7 @@ def timeline(self, request, pk=None):
         )
         resolution = datetime.timedelta(minutes=resolution_minutes)
 
-        qs = SourceImage.objects.filter(event=event)
+        qs = SourceImage.objects.filter(event=event).with_was_processed()  # type: ignore
 
         # Bulk update all source images where detections_count is null
         update_detection_counts(qs=qs, null_only=True)
@@ -404,7 +405,7 @@ def timeline(self, request, pk=None):
         source_images = list(
             qs.filter(timestamp__range=(start_time, end_time))
             .order_by("timestamp")
-            .values("id", "timestamp", "detections_count")
+            .values("id", "timestamp", "detections_count", "was_processed")
         )
 
         timeline = []
@@ -421,6 +422,7 @@ def timeline(self, request, pk=None):
                 "captures_count": 0,
                 "detections_count": 0,
                 "detection_counts": [],
+                "was_processed": False,
             }
 
             while image_index < len(source_images) and source_images[image_index]["timestamp"] <= interval_end:
@@ -432,6 +434,9 @@ def timeline(self, request, pk=None):
                 interval_data["detection_counts"] += [image["detections_count"]]
                 if image["detections_count"] >= max(interval_data["detection_counts"]):
                     interval_data["top_capture"] = SourceImage(pk=image["id"])
+                # Track if any image in this interval was processed
+                if image["was_processed"]:
+                    interval_data["was_processed"] = True
                 image_index += 1
 
             # Set a meaningful average detection count to display for the interval
@@ -602,7 +607,7 @@ def prefetch_detections(self, queryset: QuerySet, project: Project | None = None
         score = get_default_classification_threshold(project, self.request)
 
         prefetch_queryset = (
-            Detection.objects.all()
+            Detection.objects.exclude(NULL_DETECTIONS_FILTER)
             .annotate(
                 determination_score=models.Max("occurrence__detections__classifications__score"),
                 # Store whether this occurrence should be included based on default filters
@@ -709,6 +714,7 @@ class SourceImageCollectionViewSet(DefaultViewSet, ProjectMixin):
         SourceImageCollection.objects.all()
         .with_source_images_count()  # type: ignore
         .with_source_images_with_detections_count()
+        .with_source_images_processed_count()
         .prefetch_related("jobs")
     )
     serializer_class = SourceImageCollectionSerializer
@@ -724,6 +730,7 @@ class SourceImageCollectionViewSet(DefaultViewSet, ProjectMixin):
         "method",
         "source_images_count",
         "source_images_with_detections_count",
+        "source_images_processed_count",
         "occurrences_count",
     ]
 
@@ -898,7 +905,7 @@ class DetectionViewSet(DefaultViewSet, ProjectMixin):
     API endpoint that allows detections to be viewed or edited.
     """
 
-    queryset = Detection.objects.all().select_related("source_image", "detection_algorithm")
+    queryset = Detection.objects.exclude(NULL_DETECTIONS_FILTER).select_related("source_image", "detection_algorithm")
     serializer_class = DetectionSerializer
     filterset_fields = ["source_image", "detection_algorithm", "source_image__project"]
     ordering_fields = ["created_at", "updated_at", "detection_score", "timestamp"]

ami/main/models.py

@@ -85,6 +85,8 @@ class TaxonRank(OrderedEnum):
     ]
 )
 
+NULL_DETECTIONS_FILTER = Q(bbox__isnull=True) | Q(bbox=[])
+
 
 def get_media_url(path: str) -> str:
     """
@@ -1775,6 +1777,17 @@ def with_taxa_count(self, project: Project | None = None, request=None):
             taxa_count=Coalesce(models.Subquery(taxa_subquery, output_field=models.IntegerField()), 0)
         )
 
+    def with_was_processed(self):
+        """
+        Annotate each SourceImage with a boolean `was_processed` indicating
+        whether any detections exist for that image.
+
+        This mirrors `SourceImage.get_was_processed()` but as a queryset
+        annotation for efficient bulk queries.
+        """
+        processed_exists = models.Exists(Detection.objects.filter(source_image_id=models.OuterRef("pk")))
+        return self.annotate(was_processed=processed_exists)
+
 
 class SourceImageManager(models.Manager.from_queryset(SourceImageQuerySet)):
     pass
@@ -1874,7 +1887,15 @@ def size_display(self) -> str:
             return filesizeformat(self.size)
 
     def get_detections_count(self) -> int:
-        return self.detections.distinct().count()
+        # Detections count excludes detections without bounding boxes
+        # Detections with null bounding boxes are valid and indicates the image was successfully processed
+        return self.detections.exclude(NULL_DETECTIONS_FILTER).count()
+
+    def get_was_processed(self, algorithm_key: str | None = None) -> bool:
+        if algorithm_key:
+            return self.detections.filter(detection_algorithm__key=algorithm_key).exists()
+        else:
+            return self.detections.exists()
 
     def get_base_url(self) -> str | None:
         """
@@ -2044,6 +2065,7 @@ def update_detection_counts(qs: models.QuerySet[SourceImage] | None = None, null
 
     subquery = models.Subquery(
         Detection.objects.filter(source_image_id=models.OuterRef("pk"))
+        .exclude(NULL_DETECTIONS_FILTER)
         .values("source_image_id")
         .annotate(count=models.Count("id"))
         .values("count"),
@@ -2514,6 +2536,15 @@ def save(self, *args, **kwargs):
         super().save(*args, **kwargs)
 
 
+class DetectionQuerySet(BaseQuerySet):
+    def null_detections(self):
+        return self.filter(NULL_DETECTIONS_FILTER)
+
+
+class DetectionManager(models.Manager.from_queryset(DetectionQuerySet)):
+    pass
+
+
 @final
 class Detection(BaseModel):
     """An object detected in an image"""
@@ -2582,6 +2613,8 @@ class Detection(BaseModel):
     source_image_id: int
     detection_algorithm_id: int
 
+    objects = DetectionManager()
+
     def get_bbox(self):
         if self.bbox:
             return BoundingBox(
@@ -3752,7 +3785,18 @@ def with_source_images_count(self):
     def with_source_images_with_detections_count(self):
         return self.annotate(
             source_images_with_detections_count=models.Count(
-                "images", filter=models.Q(images__detections__isnull=False), distinct=True
+                "images",
+                filter=(~models.Q(images__detections__bbox__isnull=True) & ~models.Q(images__detections__bbox=[])),
+                distinct=True,
+            )
+        )
+
+    def with_source_images_processed_count(self):
+        return self.annotate(
+            source_images_processed_count=models.Count(
+                "images",
+                filter=models.Q(images__detections__isnull=False),
+                distinct=True,
             )
         )
 
@@ -3863,7 +3907,10 @@ def source_images_count(self) -> int | None:
 
     def source_images_with_detections_count(self) -> int | None:
         # This should always be pre-populated using queryset annotations
-        # return self.images.filter(detections__isnull=False).count()
+        return None
+
+    def source_images_processed_count(self) -> int | None:
+        # This should always be pre-populated using queryset annotations
         return None
 
     def occurrences_count(self) -> int | None:

ami/ml/models/pipeline.py

@@ -84,6 +84,9 @@ def filter_processed_images(
             task_logger.debug(f"Image {image} needs processing: has no existing detections from pipeline's detector")
             # If there are no existing detections from this pipeline, send the image
             yield image
+        elif existing_detections.null_detections().exists():  # type: ignore
+            task_logger.debug(f"Image {image} has a null detection from pipeline {pipeline}, skipping! ")
+            continue
         elif existing_detections.filter(classifications__isnull=True).exists():
             # Check if there are detections with no classifications
             task_logger.debug(
@@ -406,13 +409,17 @@ def get_or_create_detection(
 
     :return: A tuple of the Detection object and a boolean indicating whether it was created
     """
-    serialized_bbox = list(detection_resp.bbox.dict().values())
+    if detection_resp.bbox is not None:
+        serialized_bbox = list(detection_resp.bbox.dict().values())
+    else:
+        serialized_bbox = None
     detection_repr = f"Detection {detection_resp.source_image_id} {serialized_bbox}"
 
     assert str(detection_resp.source_image_id) == str(
         source_image.pk
     ), f"Detection belongs to a different source image: {detection_repr}"
 
+    # When reprocessing, we don't care which detection algorithm created the existing detection
     existing_detection = Detection.objects.filter(
         source_image=source_image,
         bbox=serialized_bbox,
@@ -485,6 +492,7 @@ def create_detections(
 
     existing_detections: list[Detection] = []
     new_detections: list[Detection] = []
+
     for detection_resp in detections:
         source_image = source_image_map.get(detection_resp.source_image_id)
         if not source_image:
@@ -810,6 +818,37 @@ class PipelineSaveResults:
     total_time: float
 
 
+def create_null_detections_for_undetected_images(
+    results: PipelineResultsResponse,
+    detection_algorithm: Algorithm,
+    logger: logging.Logger = logger,
+) -> list[DetectionResponse]:
+    """
+    Create null DetectionResponse objects (empty bbox) for images that have no detections.
+
+    :param results: The PipelineResultsResponse from the processing service
+    :param algorithms_known: Dictionary of algorithms keyed by algorithm key
+
+    :return: List of DetectionResponse objects with null bbox
+    """
+    source_images_with_detections = {detection.source_image_id for detection in results.detections}
+    null_detections_to_add = []
+    detection_algorithm_reference = AlgorithmReference(name=detection_algorithm.name, key=detection_algorithm.key)
+
+    for source_img in results.source_images:
+        if source_img.id not in source_images_with_detections:
+            null_detections_to_add.append(
+                DetectionResponse(
+                    source_image_id=source_img.id,
+                    bbox=None,
+                    algorithm=detection_algorithm_reference,
+                    timestamp=now(),
+                )
+            )
+
+    return null_detections_to_add
+
+
 @celery_app.task(soft_time_limit=60 * 4, time_limit=60 * 5)
 def save_results(
     results: PipelineResultsResponse | None = None,
@@ -857,6 +896,13 @@ def save_results(
         )
 
     algorithms_known: dict[str, Algorithm] = {algo.key: algo for algo in pipeline.algorithms.all()}
+    try:
+        detection_algorithm = pipeline.algorithms.get(task_type__in=Algorithm.detection_task_types)
+    except Algorithm.DoesNotExist:
+        raise ValueError("Pipeline does not have a detection algorithm")
+    except Algorithm.MultipleObjectsReturned:
+        raise NotImplementedError("Multiple detection algorithms per pipeline are not supported")
+
     job_logger.info(f"Algorithms registered for pipeline: \n{', '.join(algorithms_known.keys())}")
 
     if results.algorithms:
@@ -866,6 +912,15 @@ def save_results(
             "Algorithms and category maps must be registered before processing, using /info endpoint."
         )
 
+    # Ensure all images have detections
+    # if not, add a NULL detection (empty bbox) to the results
+    null_detections = create_null_detections_for_undetected_images(
+        results=results,
+        detection_algorithm=detection_algorithm,
+        logger=job_logger,
+    )
+    results.detections = results.detections + null_detections
+
     detections = create_detections(
         detections=results.detections,
         algorithms_known=algorithms_known,

ami/ml/schemas.py

@@ -163,14 +163,14 @@ class Config:
 
 class DetectionRequest(pydantic.BaseModel):
     source_image: SourceImageRequest  # the 'original' image
-    bbox: BoundingBox
+    bbox: BoundingBox | None = None
     crop_image_url: str | None = None
     algorithm: AlgorithmReference
 
 
 class DetectionResponse(pydantic.BaseModel):
     source_image_id: str
-    bbox: BoundingBox
+    bbox: BoundingBox | None = None
     inference_time: float | None = None
     algorithm: AlgorithmReference
     timestamp: datetime.datetime

ami/ml/tests.py

@@ -735,6 +735,30 @@ def test_project_pipeline_config(self):
         final_config = self.pipeline.get_config(self.project.pk)
         self.assertEqual(final_config["test_param"], "project_value")
 
+    def test_image_with_null_detection(self):
+        """
+        Test saving results for a pipeline that returns null detections for some images.
+        """
+        image = self.test_images[0]
+        results = self.fake_pipeline_results([image], self.pipeline)
+
+        # Manually change the results for a single image to a list of empty detections
+        results.detections = []
+
+        save_results(results)
+
+        image.save()
+        self.assertEqual(image.get_detections_count(), 0)  # detections_count should exclude null detections
+        total_num_detections = image.detections.distinct().count()
+        self.assertEqual(total_num_detections, 1)
+
+        was_processed = image.get_was_processed()
+        self.assertEqual(was_processed, True)
+
+        # Also test filtering by algorithm
+        was_processed = image.get_was_processed(algorithm_key="random-detector")
+        self.assertEqual(was_processed, True)
+
 
 class TestAlgorithmCategoryMaps(TestCase):
     def setUp(self):