New home for algorithm results that are not species classifications

ami/base/model_references.py

@@ -0,0 +1,104 @@
+"""Record ids stored in JSON, resolved to ``{type, id, name}`` so the UI can show and link them.
+
+A model opts in by declaring ``reference_type`` (and ``reference_name_field``, the field or method
+shown as its name; None shows ``#<id>``). A pydantic schema marks a field that holds such an id, or a
+list of them, with ``model_reference()``. ``resolve_model_references`` reads the names with one query
+per type, and the UI maps each type to a page in ``ui/src/utils/model-references.ts``.
+"""
+
+import collections
+import dataclasses
+import functools
+import typing
+
+import pydantic
+from django.apps import apps
+
+
+@dataclasses.dataclass(frozen=True)
+class ModelRef:
+    """A record a stored value names. ``name`` is None when the record no longer exists."""
+
+    type: str
+    id: int
+    name: str | None
+
+
+def model_reference(ref_type: str, default: typing.Any = None, **field_options: typing.Any) -> typing.Any:
+    """Declare a pydantic field that holds the id of a record of ``ref_type``, or a list of such ids.
+
+    ``field_options`` go to ``pydantic.Field``.
+    """
+    return pydantic.Field(default, reference=ref_type, **field_options)
+
+
+def field_references(schema: type[pydantic.BaseModel]) -> dict[str, str]:
+    """The schema's fields declared with ``model_reference()``, mapped to the reference type."""
+    return {
+        name: field.field_info.extra["reference"]
+        for name, field in schema.__fields__.items()
+        if "reference" in field.field_info.extra
+    }
+
+
+def field_titles(schema: type[pydantic.BaseModel]) -> dict[str, str | None]:
+    """Every field of the schema, in declaration order, mapped to its ``title`` or None."""
+    return {name: field.field_info.title for name, field in schema.__fields__.items()}
+
+
+def is_record_id(value: typing.Any) -> bool:
+    """Whether a stored value can be a primary key: an int that is not a bool."""
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def record_ids(value: typing.Any) -> list[int]:
+    """The record ids a stored value holds: one id, a list of ids, or none for anything else."""
+    if is_record_id(value):
+        return [value]
+    if isinstance(value, list) and all(is_record_id(item) for item in value):
+        return value
+    return []
+
+
+@functools.cache
+def reference_types() -> dict[str, tuple[type, str | None]]:
+    """Each declared reference type mapped to its model and name field."""
+    types: dict[str, tuple[type, str | None]] = {}
+    for model in apps.get_models():
+        ref_type = getattr(model, "reference_type", None)
+        if ref_type is None:
+            continue
+        if ref_type in types:
+            raise ValueError(f"Reference type {ref_type!r} is declared by {types[ref_type][0]} and {model}.")
+        types[ref_type] = (model, getattr(model, "reference_name_field", None))
+    return types
+
+
+def unmapped_reference_types(schemas: typing.Iterable[type[pydantic.BaseModel]]) -> set[str]:
+    """Reference types the schemas declare that no model declares; resolving them would fail."""
+    declared = {ref_type for schema in schemas for ref_type in field_references(schema).values()}
+    return declared - set(reference_types())
+
+
+def resolve_model_references(wanted: typing.Iterable[tuple[str, int]]) -> dict[tuple[str, int], ModelRef]:
+    """Each ``(type, id)`` as a ``ModelRef``, reading names with one query per type.
+
+    An unknown type raises KeyError: a schema declared a reference type no model declares.
+    """
+    ids_by_type: dict[str, set[int]] = collections.defaultdict(set)
+    for ref_type, ref_id in wanted:
+        ids_by_type[ref_type].add(ref_id)
+    resolved: dict[tuple[str, int], ModelRef] = {}
+    for ref_type, ids in ids_by_type.items():
+        model, name_field = reference_types()[ref_type]
+        rows = model.objects.filter(pk__in=ids)
+        if name_field is None:
+            names = {pk: f"#{pk}" for pk in rows.values_list("pk", flat=True)}
+        elif callable(getattr(model, name_field)):
+            # A name computed in Python, such as a session's date: load the rows and call it.
+            names = {row.pk: str(getattr(row, name_field)()) for row in rows}
+        else:
+            names = dict(rows.values_list("pk", name_field))
+        for ref_id in ids:
+            resolved[(ref_type, ref_id)] = ModelRef(ref_type, ref_id, names.get(ref_id))
+    return resolved

ami/base/serializers.py

@@ -180,3 +180,11 @@ def clean(self) -> dict[str, typing.Any]:
         if self.is_valid(raise_exception=True):
             return self.validated_data
         raise api_exceptions.ValidationError("Invalid filter parameters")
+
+
+class ModelRefSerializer(serializers.Serializer):
+    """A record a stored value names (see ``ami.base.model_references``). ``name`` is null when it no longer exists."""
+
+    type = serializers.CharField(help_text="What kind of record: taxa_list, capture_set, algorithm, ...")
+    id = serializers.IntegerField()
+    name = serializers.CharField(allow_null=True)

ami/jobs/job_config.py

@@ -0,0 +1,78 @@
+"""A job's config as labelled rows, with the records it names resolved, for showing it to people.
+
+A post-processing job stores its config in ``params["config"]``, validated by its task's
+``config_schema``. The schema gives each field its title, its order and, through ``model_reference()``,
+the type of record an id names.
+"""
+
+import dataclasses
+import typing
+
+from rest_framework import serializers
+
+from ami.base.model_references import ModelRef, field_references, field_titles, record_ids, resolve_model_references
+from ami.base.serializers import ModelRefSerializer
+from ami.jobs.models import Job
+from ami.ml.post_processing.registry import get_postprocessing_task
+
+
+@dataclasses.dataclass
+class JobConfigField:
+    """One field of a job's config: its title from the task's config schema, and the records it names, if any."""
+
+    key: str
+    label: str
+    value: typing.Any
+    refs: list[ModelRef] = dataclasses.field(default_factory=list)
+
+
+class JobConfigFieldSerializer(serializers.Serializer):
+    """One field of a job's config. ``label`` is the title the task's config schema gives it, else the key."""
+
+    key = serializers.CharField()
+    label = serializers.CharField()
+    value = serializers.JSONField(allow_null=True)
+    refs = ModelRefSerializer(many=True, help_text="The records the field names: one per id it holds, in order.")
+
+
+def job_config(job: Job | None) -> dict | None:
+    """The config a post-processing job ran with, or None for any other job or malformed params."""
+    params = job.params if job is not None else None
+    config = params.get("config") if isinstance(params, dict) else None
+    return config if isinstance(config, dict) else None
+
+
+def job_config_fields(jobs: typing.Iterable[Job]) -> dict[int, list[JobConfigField]]:
+    """Each job's config fields by job id, resolving every record id they name with one query per type."""
+    specs = {job.pk: _field_specs(job) for job in jobs}
+    wanted = [
+        (ref_type, ref_id)
+        for fields in specs.values()
+        for _, _, value, ref_type in fields
+        if ref_type
+        for ref_id in record_ids(value)
+    ]
+    resolved = resolve_model_references(wanted) if wanted else {}
+    return {
+        job_id: [
+            JobConfigField(key, label, value, [resolved[(ref_type, i)] for i in record_ids(value)] if ref_type else [])
+            for key, label, value, ref_type in fields
+        ]
+        for job_id, fields in specs.items()
+    }
+
+
+def _field_specs(job: Job) -> list[tuple[str, str, typing.Any, str | None]]:
+    """``(key, label, value, reference type)`` per field, labelled, typed and ordered by the task's config schema.
+
+    The stored config's key order is not kept (Postgres orders a JSON object's keys), so fields follow the
+    schema and any key it does not declare comes after. A job whose task is not registered shows its
+    config by key, naming no records.
+    """
+    config = job_config(job) or {}
+    task_key = job.params.get("task") if config else None
+    task = get_postprocessing_task(task_key) if isinstance(task_key, str) else None
+    titles = field_titles(task.config_schema) if task else {}
+    references = field_references(task.config_schema) if task else {}
+    keys = [key for key in titles if key in config] + [key for key in config if key not in titles]
+    return [(key, titles.get(key) or key, config[key], references.get(key)) for key in keys]

ami/jobs/models.py

@@ -1,4 +1,5 @@
 import datetime
+import json
 import logging
 import random
 import time
@@ -913,6 +914,12 @@ def run(cls, job: "Job"):
             raise ValueError(f"Unknown post-processing task '{task_key}'")
 
         task = task_cls(job=job, **config)
+        # Store the config as the task's schema validated it, defaults included, so the job
+        # records exactly what this run used. The occurrence history shows it.
+        validated_config = json.loads(task.config.json())
+        if validated_config != config:
+            job.params = {**params, "config": validated_config}
+            job.save(update_fields=["params", "updated_at"])
         task.run()
         job.progress.update_stage(cls.key, status=JobState.SUCCESS, progress=1)
         job.finished_at = datetime.datetime.now()

ami/jobs/tests/test_job_config.py

@@ -0,0 +1,83 @@
+from unittest import mock
+
+import pydantic
+from django.test import TestCase
+
+from ami.base.model_references import ModelRef, model_reference, unmapped_reference_types
+from ami.jobs.job_config import JobConfigField, job_config_fields
+from ami.jobs.models import Job
+from ami.main.models import TaxaList
+from ami.ml.models import Algorithm
+from ami.ml.post_processing.registry import POSTPROCESSING_TASKS
+from ami.tests.fixtures.main import setup_test_project
+
+
+class JobConfigFieldsTestCase(TestCase):
+    """A job's config as labelled rows, in its task's schema order, with the records it names resolved."""
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.project, _ = setup_test_project(reuse=False)
+
+    def _fields(self, params) -> list[JobConfigField]:
+        job = Job.objects.create(project=self.project, name="Run", job_type_key="post_processing", params=params)
+        return job_config_fields([job])[job.pk]
+
+    def test_labels_order_and_references_come_from_the_task_config_schema(self):
+        taxa_list = TaxaList.objects.create(name="Kept species")
+        classifier = Algorithm.objects.create(name="Classifier", key="config-test-classifier")
+        config = {
+            "reweight": True,
+            "algorithm_id": classifier.pk,
+            "taxa_list_id": taxa_list.pk,
+            "occurrence_id": True,
+            "source_image_collection_id": None,
+        }
+        self.assertEqual(
+            self._fields({"task": "class_masking", "config": config}),
+            [
+                JobConfigField("source_image_collection_id", "Capture set", None),
+                # A value that is not an id names no record, whatever the schema declares.
+                JobConfigField("occurrence_id", "Occurrence", True),
+                JobConfigField(
+                    "taxa_list_id", "Species list", taxa_list.pk, [ModelRef("taxa_list", taxa_list.pk, "Kept species")]
+                ),
+                JobConfigField(
+                    "algorithm_id", "Classifier", classifier.pk, [ModelRef("algorithm", classifier.pk, "Classifier")]
+                ),
+                JobConfigField("reweight", "Re-weighted scores", True),
+            ],
+        )
+
+    def test_a_list_of_ids_names_one_record_per_id_in_order(self):
+        kept, other = TaxaList.objects.create(name="Kept"), TaxaList.objects.create(name="Other")
+
+        class ListConfig(pydantic.BaseModel):
+            taxa_list_ids: list[int] = model_reference("taxa_list", [], title="Species lists")
+
+        task = mock.Mock(config_schema=ListConfig)
+        with mock.patch("ami.jobs.job_config.get_postprocessing_task", return_value=task):
+            fields = self._fields({"task": "lists", "config": {"taxa_list_ids": [other.pk, kept.pk, 999999]}})
+        self.assertEqual(
+            fields[0].refs,
+            [
+                ModelRef("taxa_list", other.pk, "Other"),
+                ModelRef("taxa_list", kept.pk, "Kept"),
+                ModelRef("taxa_list", 999999, None),
+            ],
+        )
+
+    def test_a_job_without_a_registered_task_shows_its_config_by_key_with_no_references(self):
+        self.assertEqual(
+            self._fields({"config": {"taxa_list_id": 3, "size_threshold": 0.01}}),
+            [
+                JobConfigField("taxa_list_id", "taxa_list_id", 3),
+                JobConfigField("size_threshold", "size_threshold", 0.01),
+            ],
+        )
+
+    def test_a_job_whose_params_are_not_an_object_has_no_config(self):
+        self.assertEqual(self._fields(["not", "an", "object"]), [])
+
+    def test_every_reference_a_task_declares_can_be_resolved(self):
+        self.assertEqual(unmapped_reference_types(task.config_schema for task in POSTPROCESSING_TASKS.values()), set())

ami/jobs/tests/test_jobs.py

@@ -1799,10 +1799,11 @@ def test_query_count_does_not_grow_with_the_number_of_jobs(self):
         """No per-job work: listing eight jobs runs the same queries as listing three, for a member
         whose permissions would otherwise be resolved on every row. Cachalot is off so every query
         counts."""
-        from cachalot.api import cachalot_disabled
         from django.db import connection
         from django.test.utils import CaptureQueriesContext
 
+        from ami.tests.fixtures.queries import no_query_cache
+
         self.client.force_authenticate(self.user)  # The owner is a member of their project.
 
         def list_choices() -> tuple[int, int]:
@@ -1811,17 +1812,12 @@ def list_choices() -> tuple[int, int]:
             self.assertEqual(response.status_code, status.HTTP_200_OK)
             return len(response.json()["results"]), len(queries.captured_queries)
 
-        disabled = cachalot_disabled()
-        disabled.__enter__()
-        try:
+        with no_query_cache():
             three_rows, three_rows_queries = list_choices()
             Job.objects.bulk_create(
                 Job(project=self.project, name=f"More {index}", job_type_key=MLJob.key) for index in range(5)
             )
             eight_rows, eight_rows_queries = list_choices()
-        finally:
-            # cachalot_disabled() does not restore itself when the block raises.
-            disabled.__exit__(None, None, None)
         self.assertEqual((three_rows, eight_rows), (3, 8))
         self.assertEqual(eight_rows_queries, three_rows_queries)
 

ami/main/admin.py

@@ -605,6 +605,8 @@ class ClassificationAdmin(IdSearchAdminMixin, admin.ModelAdmin[Classification]):
     # every taxon / detection / classification — the latter makes the change page
     # unusable on a large database.
     autocomplete_fields = ("detection", "taxon", "algorithm", "category_map", "applied_to")
+    # Algorithm results have no admin to search, so the link is an id input rather than a select.
+    raw_id_fields = ("algorithm_result",)
     # A digit term jumps to that classification by id (IdSearchAdminMixin); text searches taxon name.
     search_fields = ("taxon__name",)
     # Order by -id (indexed PK) rather than the model's -created_at (no index).