create aggregation task

atdb/taskdatabase/models.py

@@ -8,7 +8,7 @@ import json
 import logging
 
 from .services import calculated_qualities as qualities
-from .services.common import State, AggregationStrategy
+from .services.common import State, AggregationStrategy, ACTIVITY_RESET_STATUSSEN
 
 logger = logging.getLogger(__name__)
 
@@ -224,8 +224,8 @@ class Task(models.Model):
                     # so that it won't be picked up again.
 
                     # TODO: only activate when the aggregator service actually picks this up
-                    #self.new_status = State.AGGREGATE.value
-                    pass
+                    self.new_status = State.AGGREGATE.value
+                    #pass
 
 
         except Exception as error:
@@ -260,6 +260,9 @@ class Task(models.Model):
         if (self.status != State.PROCESSED.value) & (self.new_status == State.PROCESSED.value):
             self.handle_aggregation()
 
+        if self.status in ACTIVITY_RESET_STATUSSEN:
+            self.is_aggregated = False
+
         # check in the outputs if this task should be considered to be summary task
         self.is_summary = check_if_summary(self)
 

atdb/taskdatabase/serializers.py

@@ -45,7 +45,7 @@ class TaskWriteSerializer(serializers.ModelSerializer):
         model = Task
         fields = ('id','task_type','filter','predecessor','successors',
                   'joined_output_task',
-                  'project','sas_id','priority','purge_policy','cleanup_policy','resume',
+                  'project','sas_id','priority','purge_policy','cleanup_policy','resume','is_aggregated',
                   'new_workflow_id','new_workflow_uri','workflow',
                   'stage_request_id',
                   'status','new_status','quality','calculated_qualities',

atdb/taskdatabase/services/activities_handler.py

@@ -178,7 +178,7 @@ def update_processed_and_aggregate(task):
     current_is_processed = activity.is_processed
     activity.is_processed = True
     non_discarded_found = False
-    for t in Task.objects.filter(sas_id=task.sas_id):
+    for t in Task.objects.filter(sas_id=task.sas_id,task_type='regular'):
         if t.status not in PROCESSED_STATUSSES:
             activity.is_processed = False
             break
@@ -202,7 +202,9 @@ def update_processed_and_aggregate(task):
     if (task.workflow.aggregation_strategy == AggregationStrategy.COLLECT_H5.value):
 
         # check if the activity is ready to collect H5 data
-        if (not activity.is_aggregated and activity.status != State.COLLECTING_DATA.value):
+        if (not activity.is_aggregated and
+                activity.status != State.COLLECTING_DATA.value and
+                activity.status != State.AGGREGATE.value):
 
             # check if there is already a storage_location, if not, add it.
             if not activity.storage_location:

atdb/taskdatabase/templates/taskdatabase/index.html

@@ -31,7 +31,7 @@
             {% include 'taskdatabase/pagination.html' %}
         </div>
     </div>
-    <p class="footer"> Version 29 Jul 2024
+    <p class="footer"> Version 30 Jul 2024
 </div>
 
 {% include 'taskdatabase/refresh.html' %}