Refactor job creation code

app/grandchallenge/algorithms/forms.py

@@ -55,6 +55,7 @@
 from grandchallenge.components.forms import ContainerImageForm
 from grandchallenge.components.models import (
     ComponentInterface,
+    ComponentInterfaceValue,
     ComponentJob,
     ImportStatusChoices,
     InterfaceKindChoices,
@@ -93,16 +94,27 @@ class JobCreateForm(SaveFormInitMixin, Form):
     algorithm_model = ModelChoiceField(
         queryset=None, disabled=True, required=False, widget=HiddenInput
     )
+    creator = ModelChoiceField(
+        queryset=None, disabled=True, required=False, widget=HiddenInput
+    )
+    time_limit = IntegerField(
+        disabled=True, required=False, widget=HiddenInput
+    )
 
     def __init__(self, *args, algorithm, user, **kwargs):
         super().__init__(*args, **kwargs)
 
         self.helper = FormHelper()
 
         self._user = user
+        self.fields["creator"].queryset = get_user_model().objects.filter(
+            pk=self._user.pk
+        )
+        self.fields["creator"].initial = self._user
 
         self._algorithm = algorithm
         self._algorithm_image = self._algorithm.active_image
+        self.fields["time_limit"].initial = self._algorithm.time_limit
 
         active_model = self._algorithm.active_model
 
@@ -119,12 +131,36 @@ def __init__(self, *args, algorithm, user, **kwargs):
             self.fields["algorithm_model"].initial = active_model
 
         for inp in self._algorithm.inputs.all():
+            if (
+                inp.requires_file
+                and inp.slug in self._algorithm.values_for_interfaces.keys()
+            ):
+                existing_civs = ComponentInterfaceValue.objects.filter(
+                    id__in=self._algorithm.values_for_interfaces[inp.slug]
+                )
+            else:
+                # Images are handled separately in FlexibleImageWidget
+                # Values stored in the DB need to be re-entered,
+                # we re-use existing CIVs for values later as well,
+                # but on the form level it does not make sense to check for those
+                existing_civs = None
+
+            if inp.slug in self.data:
+                if inp.kind == ComponentInterface.Kind.ANY:
+                    initial = self.data.getlist(inp.slug)
+                else:
+                    initial = self.data[inp.slug]
+            else:
+                initial = None
+
             self.fields[inp.slug] = InterfaceFormField(
                 instance=inp,
-                initial=inp.default_value,
+                initial=initial if initial else inp.default_value,
                 user=self._user,
-                required=inp.value_required,
+                required=True,
                 help_text=clean(inp.description) if inp.description else "",
+                existing_civs=existing_civs,
+                form_data=self.data,
             ).field
 
     @cached_property
@@ -143,6 +179,16 @@ def clean(self):
         if self.jobs_limit < 1:
             raise ValidationError("You have run out of algorithm credits")
 
+        if Job.objects.get_jobs_with_same_inputs(
+            data=cleaned_data,
+            algorithm_image=cleaned_data["algorithm_image"],
+            algorithm_model=cleaned_data["algorithm_model"],
+        ):
+            raise ValidationError(
+                "A result for these inputs with the current image "
+                "and model already exists."
+            )
+
         return cleaned_data
 
 

app/grandchallenge/algorithms/models.py

@@ -10,9 +10,8 @@
 from django.core.exceptions import ObjectDoesNotExist
 from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models
-from django.db.models import Count, Min, Q, Sum
+from django.db.models import Count, Min, Q, QuerySet, Sum
 from django.db.models.signals import post_delete
-from django.db.transaction import on_commit
 from django.dispatch import receiver
 from django.template.defaultfilters import truncatechars
 from django.utils import timezone
@@ -26,14 +25,18 @@
 from stdimage import JPEGField
 
 from grandchallenge.anatomy.models import BodyStructure
+from grandchallenge.cases.models import Image, RawImageUploadSession
 from grandchallenge.charts.specs import stacked_bar
-from grandchallenge.components.models import (
+from grandchallenge.components.models import (  # noqa: F401
+    CIVForObjectMixin,
     ComponentImage,
     ComponentInterface,
+    ComponentInterfaceValue,
     ComponentJob,
     ComponentJobManager,
     ImportStatusChoices,
     Tarball,
+    ValuesForInterfacesMixin,
 )
 from grandchallenge.core.guardian import get_objects_for_group
 from grandchallenge.core.models import RequestBase, UUIDModel
@@ -57,14 +60,20 @@
 from grandchallenge.publications.models import Publication
 from grandchallenge.reader_studies.models import DisplaySet
 from grandchallenge.subdomains.utils import reverse
+from grandchallenge.uploads.models import UserUpload
 from grandchallenge.workstations.models import Workstation
 
 logger = logging.getLogger(__name__)
 
 JINJA_ENGINE = sandbox.ImmutableSandboxedEnvironment()
 
 
-class Algorithm(UUIDModel, TitleSlugDescriptionModel, HangingProtocolMixin):
+class Algorithm(
+    UUIDModel,
+    TitleSlugDescriptionModel,
+    ValuesForInterfacesMixin,
+    HangingProtocolMixin,
+):
     editors_group = models.OneToOneField(
         Group,
         on_delete=models.PROTECT,
@@ -420,6 +429,16 @@ def add_user(self, user):
     def remove_user(self, user):
         return user.groups.remove(self.users_group)
 
+    @property
+    def civ_set_lookup(self):
+        return "inputs"
+
+    @property
+    def civ_sets_related_manager(self):
+        return Job.objects.filter(
+            algorithm_image__in=self.algorithm_container_images.all()
+        )
+
     @cached_property
     def user_statistics(self):
         return (
@@ -629,6 +648,118 @@ def create(
 
         return obj
 
+    @property
+    def non_interface_fields(self):
+        return ["algorithm_image", "algorithm_model", "creator", "time_limit"]
+
+    def retrieve_existing_civs(self, *, civ_data):
+        """
+        Checks if there are existing CIVs for the provided data and returns those.
+
+        Parameters
+        ----------
+        civ_data
+            A dictionary with interface slugs as keys and CIV values as values.
+
+        Returns
+        -------
+        A list of ComponentInterfaceValues
+
+        """
+        existing_civs = []
+        for interface, value in civ_data.items():
+            if isinstance(value, ComponentInterfaceValue):
+                existing_civs.append(value)
+            elif isinstance(value, Image):
+                ci = ComponentInterface.objects.get(slug=interface)
+                try:
+                    civ = ComponentInterfaceValue.objects.filter(
+                        interface=ci, image=value
+                    ).get()
+                    existing_civs.append(civ)
+                except ObjectDoesNotExist:
+                    continue
+            elif isinstance(
+                value, (RawImageUploadSession, UserUpload, QuerySet)
+            ):
+                # uploads will create new CIVs, so ignore these
+                continue
+            else:
+                # values can be of different types
+                ci = ComponentInterface.objects.get(slug=interface)
+                try:
+                    civ = ComponentInterfaceValue.objects.filter(
+                        interface=ci, value=value
+                    ).get()
+                    existing_civs.append(civ)
+                except ObjectDoesNotExist:
+                    continue
+
+        return existing_civs
+
+    def get_jobs_with_same_inputs(
+        self, *, data, algorithm_image, algorithm_model
+    ):
+        civ_data = {
+            k: v for k, v in data.items() if k not in self.non_interface_fields
+        }
+        existing_civs = self.retrieve_existing_civs(civ_data=civ_data)
+        unique_kwargs = {
+            "algorithm_image": algorithm_image,
+        }
+        input_interface_count = algorithm_image.algorithm.inputs.count()
+        if algorithm_model:
+            unique_kwargs["algorithm_model"] = algorithm_model
+        else:
+            unique_kwargs["algorithm_model__isnull"] = True
+        existing_jobs = (
+            Job.objects.filter(**unique_kwargs)
+            .annotate(
+                inputs_match_count=Count(
+                    "inputs", filter=Q(inputs__in=existing_civs)
+                )
+            )
+            .filter(inputs_match_count=input_interface_count)
+        )
+        return existing_jobs
+
+    def create_job_and_process_inputs(
+        self,
+        *,
+        data,
+        extra_viewer_groups=None,
+        extra_logs_viewer_groups=None,
+    ):
+        civ_data = {
+            k: v for k, v in data.items() if k not in self.non_interface_fields
+        }
+        non_civ_data = {
+            k: v for k, v in data.items() if k in self.non_interface_fields
+        }
+        job = Job.objects.create(
+            **non_civ_data,
+            extra_viewer_groups=extra_viewer_groups,
+            extra_logs_viewer_groups=extra_logs_viewer_groups,
+        )
+        # local import to avoid circular dependency
+        from grandchallenge.algorithms.tasks import (
+            execute_algorithm_job_for_inputs,
+        )
+
+        linked_task = execute_algorithm_job_for_inputs.signature(
+            kwargs={"job_pk": job.pk}, immutable=True
+        )
+
+        for key, value in civ_data.items():
+            job.create_civ(
+                ci_slug=key,
+                new_value=value,
+                user=data["creator"],
+                linked_task=linked_task,
+            )
+
+        return job
+
     def spent_credits(self, user):
         now = timezone.now()
         period = timedelta(days=30)
@@ -712,7 +843,7 @@ class AlgorithmModelGroupObjectPermission(GroupObjectPermissionBase):
     )
 
 
-class Job(UUIDModel, ComponentJob):
+class Job(UUIDModel, CIVForObjectMixin, ComponentJob):
     objects = JobManager.as_manager()
 
     algorithm_image = models.ForeignKey(
@@ -770,6 +901,20 @@ def container(self):
     def output_interfaces(self):
         return self.algorithm_image.algorithm.outputs
 
+    @cached_property
+    def inputs_complete(self):
+        # check if all inputs are present and if they all have a value
+        input_interfaces = self.algorithm_image.algorithm.inputs.all()
+        existing_inputs = [
+            civ.interface for civ in self.inputs.all() if civ.has_value
+        ]
+        missing_inputs = [
+            interface
+            for interface in input_interfaces
+            if interface not in existing_inputs
+        ]
+        return not missing_inputs
+
     @cached_property
     def rendered_result_text(self) -> str:
         try:
@@ -883,24 +1028,9 @@ def add_viewer(self, user):
     def remove_viewer(self, user):
         return user.groups.remove(self.viewers)
 
-    def sort_inputs_and_execute(
-        self, upload_session_pks=None, user_upload_pks=None
-    ):
-        # Local import to avoid circular dependency
-        from grandchallenge.algorithms.tasks import (
-            run_algorithm_job_for_inputs,
-        )
-
-        on_commit(
-            run_algorithm_job_for_inputs.signature(
-                kwargs={
-                    "job_pk": self.pk,
-                    "upload_session_pks": upload_session_pks,
-                    "user_upload_pks": user_upload_pks,
-                },
-                immutable=True,
-            ).apply_async
-        )
+    @property
+    def base_object(self):
+        return self.algorithm_image.algorithm
 
     @property
     def executor_kwargs(self):