Support re-processing detections and skipping localizer (#815)

* Set up customizable local processing service

* Set up separate docker compose stack, rename ml backend services

* WIP: README.md

* Improve processing flow

* fix: tests and postgres connection

* Update READMEs with minimal/example setups

* fix: transformers fixed version

* Add tests

* Typos, warn --> warnings

* Add support for Darsa flat-bug

* chore: Change the Pipeline class name to FlatBugDetectorPipeline to avoid shadowing the FlatBugDetector model

* Move README

* Address comment tasks

* Update README

* Pass in pipeline request config, properly cache models, simplifications

* fix: update docker compose instructions & build path

* feat: use ["insect"] for the default zero-shot class

* feat: try to use faster version of zero-shot detector

* feat: use gpu if available

* fix: update minimal docker compose build path

* Add back crop_image_url

* Support re-processing detections and skipping localizer

* fix: correctly pass candidate labels for zero shot object detector

* Support re-processing detections and skipping localizer

* fix: allow empty pipeline request config

* fix: allow empty pipeline request config

* clean up

* fix: ignore detection algorithm during reprocessing

* remove flat bug

* feat: only use zero shot and HF classifier algorithms

* clean up

* Expand support for date formats in image filenames (#809)

* feat: support more date formats in filenames

* feat: update and centralize validation language

* fix: change name of the new docker network (#819)

* docs: clarify new Detection schema/class

* Function for creating detection instances from requests

* Add reprocessing to minimal app

* Add re-processing test

* Fix requirements

* Address review comments

* Only open source image once

* feat: cache huggingface & torch models that are auto-downloaded

* fix: leave gpu passthrough as an example, off by default

* feat: feature flag & pipeline config for reprocessing detections

* fix: spelling in previous feature flag

* chore: migration for new & corrected feature flags

* fix: append detections instead of overriding. add feature flag to tests.

---------

Co-authored-by: mohamedelabbas1996 <hack1996man@gmail.com>
Co-authored-by: Michael Bunsen <notbot@gmail.com>

Author: 3 people

ami/main/api/views.py

@@ -788,7 +788,7 @@ def perform_create(self, serializer):
         obj = serializer.save(user=user)
 
         # Get process_now flag from project feature flags
-        process_now = project.feature_flags.auto_processs_manual_uploads
+        process_now = project.feature_flags.auto_process_manual_uploads
 
         # Create source image from the upload
         source_image = create_source_image_from_upload(

ami/main/migrations/0067_alter_project_feature_flags.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.10 on 2025-08-15 21:01
+
+import ami.main.models
+from django.db import migrations
+import django_pydantic_field.fields
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0066_alter_project_feature_flags_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="project",
+            name="feature_flags",
+            field=django_pydantic_field.fields.PydanticSchemaField(
+                blank=True,
+                config=None,
+                default={"auto_process_manual_uploads": False, "reprocess_existing_detections": False, "tags": False},
+                schema=ami.main.models.ProjectFeatureFlags,
+            ),
+        ),
+    ]

ami/main/models.py

@@ -31,6 +31,7 @@
 from ami.base.models import BaseModel
 from ami.main import charts
 from ami.main.models_future.projects import ProjectSettingsMixin
+from ami.ml.schemas import BoundingBox
 from ami.users.models import User
 from ami.utils.media import calculate_file_checksum, extract_timestamp
 from ami.utils.schemas import OrderedEnum
@@ -202,7 +203,8 @@ class ProjectFeatureFlags(pydantic.BaseModel):
     """
 
     tags: bool = False  # Whether the project supports tagging taxa
-    auto_processs_manual_uploads: bool = False  # Whether to automatically process uploaded images
+    auto_process_manual_uploads: bool = False  # Whether to automatically process uploaded images
+    reprocess_existing_detections: bool = False  # Whether to reprocess existing detections
 
 
 default_feature_flags = ProjectFeatureFlags()
@@ -2263,6 +2265,17 @@ class Detection(BaseModel):
     source_image_id: int
     detection_algorithm_id: int
 
+    def get_bbox(self):
+        if self.bbox:
+            return BoundingBox(
+                x1=self.bbox[0],
+                y1=self.bbox[1],
+                x2=self.bbox[2],
+                y2=self.bbox[3],
+            )
+        else:
+            return None
+
     # def bbox(self):
     #     return (
     #         self.bbox_x,

ami/ml/models/pipeline.py

@@ -5,7 +5,6 @@
 if TYPE_CHECKING:
     from ami.ml.models import ProcessingService, ProjectPipelineConfig
     from ami.jobs.models import Job
-    from ami.main.models import Project
 
 import collections
 import dataclasses
@@ -28,6 +27,7 @@
     Deployment,
     Detection,
     Occurrence,
+    Project,
     SourceImage,
     SourceImageCollection,
     TaxaList,
@@ -39,7 +39,9 @@
 from ami.ml.models.algorithm import Algorithm, AlgorithmCategoryMap
 from ami.ml.schemas import (
     AlgorithmConfigResponse,
+    AlgorithmReference,
     ClassificationResponse,
+    DetectionRequest,
     DetectionResponse,
     PipelineRequest,
     PipelineRequestConfigParameters,
@@ -62,6 +64,7 @@ def filter_processed_images(
     Return only images that need to be processed by a given pipeline.
     An image needs processing if:
     1. It has no detections from the pipeline's detection algorithm
+    or
     2. It has detections but they don't have classifications from all the pipeline's classification algorithms
     """
     pipeline_algorithms = pipeline.algorithms.all()
@@ -175,6 +178,15 @@ def process_images(
         job = Job.objects.get(pk=job_id)
         task_logger = job.logger
 
+    if project_id:
+        project = Project.objects.get(pk=project_id)
+    else:
+        task_logger.warning(f"Pipeline {pipeline} is not associated with a project")
+        project = None
+
+    pipeline_config = pipeline.get_config(project_id=project_id)
+    task_logger.info(f"Using pipeline config: {pipeline_config}")
+
     prefiltered_images = list(images)
     images = list(filter_processed_images(images=prefiltered_images, pipeline=pipeline, task_logger=task_logger))
     if len(images) < len(prefiltered_images):
@@ -192,26 +204,39 @@ def process_images(
     task_logger.info(f"Sending {len(images)} images to Pipeline {pipeline}")
     urls = [source_image.public_url() for source_image in images if source_image.public_url()]
 
-    source_images = [
-        SourceImageRequest(
-            id=str(source_image.pk),
-            url=url,
-        )
-        for source_image, url in zip(images, urls)
-        if url
-    ]
+    source_image_requests: list[SourceImageRequest] = []
+    detection_requests: list[DetectionRequest] = []
+
+    reprocess_existing_detections = False
+    # Check if feature flag is enabled to reprocess existing detections
+    if project and project.feature_flags.reprocess_existing_detections:
+        # Check if the user wants to reprocess existing detections or ignore them
+        if pipeline_config.get("reprocess_existing_detections", True):
+            reprocess_existing_detections = True
+
+    for source_image, url in zip(images, urls):
+        if url:
+            source_image_request = SourceImageRequest(
+                id=str(source_image.pk),
+                url=url,
+            )
+            source_image_requests.append(source_image_request)
 
-    if not project_id:
-        task_logger.warning(f"Pipeline {pipeline} is not associated with a project")
+            if reprocess_existing_detections:
+                detection_requests += collect_detections(source_image, source_image_request)
 
-    config = pipeline.get_config(project_id=project_id)
-    task_logger.info(f"Using pipeline config: {config}")
+    if reprocess_existing_detections:
+        task_logger.info(f"Found {len(detection_requests)} existing detections to reprocess.")
+    else:
+        task_logger.info("Reprocessing of existing detections is disabled, sending images without detections.")
 
     request_data = PipelineRequest(
         pipeline=pipeline.slug,
-        source_images=source_images,
-        config=config,
+        source_images=source_image_requests,
+        config=pipeline_config,
+        detections=detection_requests,
     )
+    task_logger.debug(f"Pipeline request data: {request_data}")
 
     session = create_session()
     resp = session.post(endpoint_url, json=request_data.dict())
@@ -230,7 +255,8 @@ def process_images(
             pipeline=pipeline.slug,
             total_time=0,
             source_images=[
-                SourceImageResponse(id=source_image.id, url=source_image.url) for source_image in source_images
+                SourceImageResponse(id=source_image_request.id, url=source_image_request.url)
+                for source_image_request in source_image_requests
             ],
             detections=[],
             errors=msg,
@@ -251,6 +277,33 @@ def process_images(
     return results
 
 
+def collect_detections(
+    source_image: SourceImage,
+    source_image_request: SourceImageRequest,
+) -> list[DetectionRequest]:
+    """
+    Collect existing detections for a source image and send them with pipeline request.
+    """
+    detection_requests: list[DetectionRequest] = []
+    # Re-process all existing detections if they exist
+    for detection in source_image.detections.all():
+        bbox = detection.get_bbox()
+        if bbox and detection.detection_algorithm:
+            detection_requests.append(
+                DetectionRequest(
+                    source_image=source_image_request,
+                    bbox=bbox,
+                    crop_image_url=detection.url(),
+                    algorithm=AlgorithmReference(
+                        name=detection.detection_algorithm.name,
+                        key=detection.detection_algorithm.key,
+                    ),
+                )
+            )
+
+    return detection_requests
+
+
 def get_or_create_algorithm_and_category_map(
     algorithm_config: AlgorithmConfigResponse,
     logger: logging.Logger = logger,
@@ -351,23 +404,12 @@ def get_or_create_detection(
     serialized_bbox = list(detection_resp.bbox.dict().values())
     detection_repr = f"Detection {detection_resp.source_image_id} {serialized_bbox}"
 
-    assert detection_resp.algorithm, f"No detection algorithm was specified for detection {detection_repr}"
-    try:
-        detection_algo = algorithms_used[detection_resp.algorithm.key]
-    except KeyError:
-        raise ValueError(
-            f"Detection algorithm {detection_resp.algorithm.key} is not a known algorithm. "
-            "The processing service must declare it in the /info endpoint. "
-            f"Known algorithms: {list(algorithms_used.keys())}"
-        )
-
     assert str(detection_resp.source_image_id) == str(
         source_image.pk
     ), f"Detection belongs to a different source image: {detection_repr}"
 
     existing_detection = Detection.objects.filter(
         source_image=source_image,
-        detection_algorithm=detection_algo,
         bbox=serialized_bbox,
     ).first()
 
@@ -387,6 +429,16 @@ def get_or_create_detection(
         detection = existing_detection
 
     else:
+        assert detection_resp.algorithm, f"No detection algorithm was specified for detection {detection_repr}"
+        try:
+            detection_algo = algorithms_used[detection_resp.algorithm.key]
+        except KeyError:
+            raise ValueError(
+                f"Detection algorithm {detection_resp.algorithm.key} is not a known algorithm. "
+                "The processing service must declare it in the /info endpoint. "
+                f"Known algorithms: {list(algorithms_used.keys())}"
+            )
+
         new_detection = Detection(
             source_image=source_image,
             bbox=serialized_bbox,
@@ -1007,7 +1059,7 @@ def collect_images(
         )
 
     def choose_processing_service_for_pipeline(
-        self, job_id: int, pipeline_name: str, project_id: int
+        self, job_id: int | None, pipeline_name: str, project_id: int
     ) -> ProcessingService:
         # @TODO use the cached `last_checked_latency` and a max age to avoid checking every time
 

ami/ml/schemas.py

@@ -112,16 +112,6 @@ class ClassificationResponse(pydantic.BaseModel):
     timestamp: datetime.datetime
 
 
-class DetectionResponse(pydantic.BaseModel):
-    source_image_id: str
-    bbox: BoundingBox
-    inference_time: float | None = None
-    algorithm: AlgorithmReference
-    timestamp: datetime.datetime
-    crop_image_url: str | None = None
-    classifications: list[ClassificationResponse] = []
-
-
 class SourceImageRequest(pydantic.BaseModel):
     # @TODO bring over new SourceImage & b64 validation from the lepsAI repo
     id: str
@@ -144,6 +134,23 @@ class Config:
 ]
 
 
+class DetectionRequest(pydantic.BaseModel):
+    source_image: SourceImageRequest  # the 'original' image
+    bbox: BoundingBox
+    crop_image_url: str | None = None
+    algorithm: AlgorithmReference
+
+
+class DetectionResponse(pydantic.BaseModel):
+    source_image_id: str
+    bbox: BoundingBox
+    inference_time: float | None = None
+    algorithm: AlgorithmReference
+    timestamp: datetime.datetime
+    crop_image_url: str | None = None
+    classifications: list[ClassificationResponse] = []
+
+
 class PipelineRequestConfigParameters(dict):
     """Parameters used to configure a pipeline request.
 
@@ -166,6 +173,7 @@ class PipelineRequestConfigParameters(dict):
 class PipelineRequest(pydantic.BaseModel):
     pipeline: str
     source_images: list[SourceImageRequest]
+    detections: list[DetectionRequest] | None = None
     config: PipelineRequestConfigParameters | dict | None = None