Introduce generic post-processing framework

ami/jobs/migrations/0018_alter_job_job_type_key.py

@@ -0,0 +1,29 @@
+# Generated by Django 4.2.10 on 2025-09-30 12:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("jobs", "0017_alter_job_logs_alter_job_progress"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="job",
+            name="job_type_key",
+            field=models.CharField(
+                choices=[
+                    ("ml", "ML pipeline"),
+                    ("populate_captures_collection", "Populate captures collection"),
+                    ("data_storage_sync", "Data storage sync"),
+                    ("unknown", "Unknown"),
+                    ("data_export", "Data Export"),
+                    ("post_processing", "Post Processing"),
+                ],
+                default="unknown",
+                max_length=255,
+                verbose_name="Job Type",
+            ),
+        ),
+    ]

ami/jobs/models.py

@@ -18,6 +18,7 @@
 from ami.jobs.tasks import run_job
 from ami.main.models import Deployment, Project, SourceImage, SourceImageCollection
 from ami.ml.models import Pipeline
+from ami.ml.post_processing.registry import get_postprocessing_task
 from ami.utils.schemas import OrderedEnum
 
 logger = logging.getLogger(__name__)
@@ -645,6 +646,34 @@ def run(cls, job: "Job"):
         job.update_status(JobState.SUCCESS, save=True)
 
 
+class PostProcessingJob(JobType):
+    name = "Post Processing"
+    key = "post_processing"
+
+    @classmethod
+    def run(cls, job: "Job"):
+        job.progress.add_stage(cls.name, key=cls.key)
+        job.update_status(JobState.STARTED)
+        job.started_at = datetime.datetime.now()
+        job.save()
+
+        params = job.params or {}
+        task_key: str = params.get("task", "")
+        config = params.get("config", {})
+        job.logger.info(f"Post-processing task: {task_key} with params: {job.params}")
+
+        task_cls = get_postprocessing_task(key=task_key)
+        if not task_cls:
+            raise ValueError(f"Unknown post-processing task '{task_key}'")
+
+        task = task_cls(job=job, **config)
+        task.run()
+        job.progress.update_stage(cls.key, status=JobState.SUCCESS, progress=1)
+        job.finished_at = datetime.datetime.now()
+        job.update_status(JobState.SUCCESS)
+        job.save()
+
+
 class UnknownJobType(JobType):
     name = "Unknown"
     key = "unknown"
@@ -654,7 +683,14 @@ def run(cls, job: "Job"):
         raise ValueError(f"Unknown job type '{job.job_type()}'")
 
 
-VALID_JOB_TYPES = [MLJob, SourceImageCollectionPopulateJob, DataStorageSyncJob, UnknownJobType, DataExportJob]
+VALID_JOB_TYPES = [
+    MLJob,
+    SourceImageCollectionPopulateJob,
+    DataStorageSyncJob,
+    UnknownJobType,
+    DataExportJob,
+    PostProcessingJob,
+]
 
 
 def get_job_type_by_key(key: str) -> type[JobType] | None:

ami/main/admin.py

@@ -10,6 +10,7 @@
 
 import ami.utils
 from ami import tasks
+from ami.jobs.models import Job
 from ami.ml.models.project_pipeline_config import ProjectPipelineConfig
 from ami.ml.tasks import remove_duplicate_classifications
 
@@ -594,6 +595,7 @@ class SourceImageCollectionAdmin(admin.ModelAdmin[SourceImageCollection]):
     """Admin panel example for ``SourceImageCollection`` model."""
 
     list_display = ("name", "image_count", "method", "kwargs", "created_at", "updated_at")
+    list_filter = ("project",)
 
     def get_queryset(self, request: HttpRequest) -> QuerySet[Any]:
         return super().get_queryset(request).annotate(image_count=models.Count("images"))
@@ -619,7 +621,31 @@ def populate_collection_async(self, request: HttpRequest, queryset: QuerySet[Sou
             f"Populating {len(queued_tasks)} collection(s) background tasks: {queued_tasks}.",
         )
 
-    actions = [populate_collection, populate_collection_async]
+    @admin.action(description="Run Small Size Filter post-processing task (async)")
+    def run_small_size_filter(self, request: HttpRequest, queryset: QuerySet[SourceImageCollection]) -> None:
+        jobs = []
+        for collection in queryset:
+            job = Job.objects.create(
+                name=f"Post-processing: SmallSizeFilter on Collection {collection.pk}",
+                project=collection.project,
+                job_type_key="post_processing",
+                params={
+                    "task": "small_size_filter",
+                    "config": {
+                        "source_image_collection_id": collection.pk,
+                    },
+                },
+            )
+            job.enqueue()
+            jobs.append(job.pk)
+
+        self.message_user(request, f"Queued Small Size Filter for {queryset.count()} collection(s). Jobs: {jobs}")
+
+    actions = [
+        populate_collection,
+        populate_collection_async,
+        run_small_size_filter,
+    ]
 
     # Hide images many-to-many field from form. This would list all source images in the database.
     exclude = ("images",)

ami/main/migrations/0075_alter_project_options.py

@@ -0,0 +1,59 @@
+# Generated by Django 4.2.10 on 2025-10-14 05:01
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0074_taxon_cover_image_credit_taxon_cover_image_url_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="project",
+            options={
+                "ordering": ["-priority", "created_at"],
+                "permissions": [
+                    ("create_identification", "Can create identifications"),
+                    ("update_identification", "Can update identifications"),
+                    ("delete_identification", "Can delete identifications"),
+                    ("create_job", "Can create a job"),
+                    ("update_job", "Can update a job"),
+                    ("run_ml_job", "Can run/retry/cancel ML jobs"),
+                    ("run_populate_captures_collection_job", "Can run/retry/cancel Populate Collection jobs"),
+                    ("run_data_storage_sync_job", "Can run/retry/cancel Data Storage Sync jobs"),
+                    ("run_data_export_job", "Can run/retry/cancel Data Export jobs"),
+                    ("run_single_image_ml_job", "Can process a single capture"),
+                    ("run_post_processing_job", "Can run/retry/cancel Post-Processing jobs"),
+                    ("delete_job", "Can delete a job"),
+                    ("create_deployment", "Can create a deployment"),
+                    ("delete_deployment", "Can delete a deployment"),
+                    ("update_deployment", "Can update a deployment"),
+                    ("sync_deployment", "Can sync images to a deployment"),
+                    ("create_sourceimagecollection", "Can create a collection"),
+                    ("update_sourceimagecollection", "Can update a collection"),
+                    ("delete_sourceimagecollection", "Can delete a collection"),
+                    ("populate_sourceimagecollection", "Can populate a collection"),
+                    ("create_sourceimage", "Can create a source image"),
+                    ("update_sourceimage", "Can update a source image"),
+                    ("delete_sourceimage", "Can delete a source image"),
+                    ("star_sourceimage", "Can star a source image"),
+                    ("create_sourceimageupload", "Can create a source image upload"),
+                    ("update_sourceimageupload", "Can update a source image upload"),
+                    ("delete_sourceimageupload", "Can delete a source image upload"),
+                    ("create_s3storagesource", "Can create storage"),
+                    ("delete_s3storagesource", "Can delete storage"),
+                    ("update_s3storagesource", "Can update storage"),
+                    ("test_s3storagesource", "Can test storage connection"),
+                    ("create_site", "Can create a site"),
+                    ("delete_site", "Can delete a site"),
+                    ("update_site", "Can update a site"),
+                    ("create_device", "Can create a device"),
+                    ("delete_device", "Can delete a device"),
+                    ("update_device", "Can update a device"),
+                    ("view_private_data", "Can view private data"),
+                    ("trigger_exports", "Can trigger data exports"),
+                ],
+            },
+        ),
+    ]

ami/main/migrations/0077_merge_20251014_1426.py

@@ -0,0 +1,12 @@
+# Generated by Django 4.2.10 on 2025-10-14 14:26
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0075_alter_project_options"),
+        ("main", "0076_add_occurrence_composite_indexes"),
+    ]
+
+    operations = []

ami/main/migrations/0078_classification_applied_to.py

@@ -0,0 +1,25 @@
+# Generated by Django 4.2.10 on 2025-10-14 21:32
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0077_merge_20251014_1426"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="classification",
+            name="applied_to",
+            field=models.ForeignKey(
+                blank=True,
+                help_text="If this classification was produced by a post-processing algorithm, this field references the original classification it was applied to.",
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="derived_classifications",
+                to="main.classification",
+            ),
+        ),
+    ]

ami/main/models.py

@@ -333,6 +333,7 @@ class Permissions:
         RUN_POPULATE_CAPTURES_COLLECTION_JOB = "run_populate_captures_collection_job"
         RUN_DATA_STORAGE_SYNC_JOB = "run_data_storage_sync_job"
         RUN_DATA_EXPORT_JOB = "run_data_export_job"
+        RUN_POST_PROCESSING_JOB = "run_post_processing_job"
         DELETE_JOB = "delete_job"
 
         # Deployment permissions
@@ -395,6 +396,7 @@ class Meta:
             ("run_data_storage_sync_job", "Can run/retry/cancel Data Storage Sync jobs"),
             ("run_data_export_job", "Can run/retry/cancel Data Export jobs"),
             ("run_single_image_ml_job", "Can process a single capture"),
+            ("run_post_processing_job", "Can run/retry/cancel Post-Processing jobs"),
             ("delete_job", "Can delete a job"),
             # Deployment permissions
             ("create_deployment", "Can create a deployment"),
@@ -2204,7 +2206,7 @@ class Classification(BaseModel):
 
     taxon = models.ForeignKey("Taxon", on_delete=models.SET_NULL, null=True, related_name="classifications")
     score = models.FloatField(null=True)
-    timestamp = models.DateTimeField()
+    timestamp = models.DateTimeField()  # Is this to represent when classification was made? why not use created_at?
     terminal = models.BooleanField(
         default=True, help_text="Is this the final classification from a series of classifiers in a pipeline?"
     )
@@ -2225,7 +2227,17 @@ class Classification(BaseModel):
         related_name="classifications",
     )
     # job = models.CharField(max_length=255, null=True)
-
+    applied_to = models.ForeignKey(
+        "self",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="derived_classifications",
+        help_text=(
+            "If this classification was produced by a post-processing algorithm, "
+            "this field references the original classification it was applied to."
+        ),
+    )
     objects = ClassificationManager()
 
     # Type hints for auto-generated fields

ami/ml/migrations/0025_alter_algorithm_task_type.py

@@ -0,0 +1,42 @@
+# Generated by Django 4.2.10 on 2025-10-14 05:01
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("ml", "0024_fix_classifications_missing_category_maps"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="algorithm",
+            name="task_type",
+            field=models.CharField(
+                choices=[
+                    ("detection", "Detection"),
+                    ("localization", "Localization"),
+                    ("segmentation", "Segmentation"),
+                    ("classification", "Classification"),
+                    ("embedding", "Embedding"),
+                    ("tracking", "Tracking"),
+                    ("tagging", "Tagging"),
+                    ("regression", "Regression"),
+                    ("captioning", "Captioning"),
+                    ("generation", "Generation"),
+                    ("translation", "Translation"),
+                    ("summarization", "Summarization"),
+                    ("question_answering", "Question Answering"),
+                    ("depth_estimation", "Depth Estimation"),
+                    ("pose_estimation", "Pose Estimation"),
+                    ("size_estimation", "Size Estimation"),
+                    ("post_processing", "Post Processing"),
+                    ("other", "Other"),
+                    ("unknown", "Unknown"),
+                ],
+                default="unknown",
+                max_length=255,
+                null=True,
+            ),
+        ),
+    ]

ami/ml/models/algorithm.py

@@ -176,6 +176,7 @@ class AlgorithmTaskType(str, enum.Enum):
     DEPTH_ESTIMATION = "depth_estimation"
     POSE_ESTIMATION = "pose_estimation"
     SIZE_ESTIMATION = "size_estimation"
+    POST_PROCESSING = "post_processing"
     OTHER = "other"
     UNKNOWN = "unknown"
 

ami/ml/post_processing/__init__.py

@@ -0,0 +1 @@
+from . import small_size_filter  # noqa: F401