experimenting with activities

atdb/taskdatabase/migrations/0035_task_activity.py

+# Generated by Django 5.0 on 2024-02-06 12:38
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('taskdatabase', '0034_activity_ingestq_status'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='task',
+            name='activity',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='tasks', to='taskdatabase.activity'),
+        ),
+    ]

atdb/taskdatabase/models.py

@@ -158,6 +158,9 @@ class Task(models.Model):
     predecessor = models.ForeignKey('self', related_name='successors', on_delete=models.SET_NULL, null=True, blank=True)
     joined_output_task = models.ForeignKey('self', related_name='joined_input_tasks', on_delete=models.SET_NULL, null=True, blank=True)
 
+    # pipeline or observation
+    activity = models.ForeignKey(Activity, related_name='tasks', on_delete=models.DO_NOTHING, null=True, blank=True)
+
     def __str__(self):
         return str(self.id) + ' - (' + self.task_type + ') - ' + str(self.sas_id)
 

atdb/taskdatabase/serializers.py

@@ -95,6 +95,7 @@ class TaskReadSerializer(serializers.ModelSerializer):
         required=False,
     )
 
+
     class Meta:
         model = Task
         fields = ['id','task_type','creationTime','filter',
@@ -106,7 +107,8 @@ class TaskReadSerializer(serializers.ModelSerializer):
                   'status','new_status','quality','calculated_qualities',
                   'inputs','outputs','metrics','remarks','status_history',
                   'size_to_process', 'size_processed', 'total_processing_time',
-                  'log_entries','meta_scheduling','environment','archive'
+                  'log_entries','meta_scheduling','environment','archive',
+                  'activity'
                   ]
         read_only_fields = fields
 

atdb/taskdatabase/services/activities.py

@@ -41,6 +41,28 @@ def calculate_ingested_fraction(this_task):
     return result
 
 
+def associate_activity(task):
+
+    if not task.activity:
+
+        try:
+            activity = Activity.objects.get(sas_id=task.sas_id)
+        except:
+            # no activity exists yet, create it
+            logger.info(f'create new activity for sas_id {task.sas_id}')
+
+            activity = Activity(sas_id=task.sas_id,
+                                workflow=task.workflow,
+                                project=task.project,
+                                filter=task.filter,
+                                priority=task.priority,
+                                archive=task.archive)
+            activity.save()
+
+        task.activity = activity
+        task.save()
+
+
 def update_activity(task):
     """
     The activity (SAS_ID level) is updated whenever a task change status.
@@ -69,12 +91,20 @@ def update_activity(task):
                             archive=task.archive)
         activity.save()
 
+    # associate an activity to this tasks
+    if not task.activity:
+        activity = Activity.objects.get(sas_id=task.sas_id)
+        task.activity = activity
+
     # depending on the status transition, perform calculations
     if task.status == State.STORED.value:
         logger.info(f'- calculate_qualities')
         # quality is calculated per task and per sas_id, reget the quality per sas_id
+        try:
             activity.calculated_quality = task.calculated_qualities['per_sasid']
             activity.save()
+        except:
+            pass
 
     # calculate the fraction and list of statusses of ingested tasks of this SAS_ID
     if task.status in [State.SCRUBBED.value, State.ARCHIVING.value, State.ARCHIVED.value, State.FINISHING.value]:

atdb/taskdatabase/templates/taskdatabase/ingest/tasks.html

@@ -23,8 +23,7 @@
                     <a href="{% url 'task-change-priority-sasid' task.pk '10' my_tasks.number %}" class="btn btn-warning btn-sm" role="button">+10</a>
                 {% endif %}
                 </td>
-                <td>{{ task.sasid_ingested_fraction.status }}</td>
-
+                <td>{{ task.activity.ingestq_status }}</td>
                 <td>{{ task.sasid_ingested_fraction.completion }}%</td>
                 <td>
                     {% if task.sas_id_has_archived != None %}

atdb/taskdatabase/urls.py

@@ -129,4 +129,6 @@ urlpatterns = [
     path('tasks/<int:pk>/query-hold/<hold_it>/<query_params>', views.HoldQuery, name='query-hold-resume'),
     path('tasks/<int:pk>/hold/<hold_it>/<page>', views.Hold, name='service-hold-resume'),
     path('tasks/<int:pk>/query-purge/<purge_policy>/<query_params>', views.PurgeQuery, name='query-purge'),
+
+    path('tasks/associate-activities/', views.AssociateActivities, name='associate-activities'),
 ]

atdb/taskdatabase/views.py

@@ -38,7 +38,7 @@ from .serializers import \
     PostProcessingRuleSerializer, \
     MonitorSerializer, LatestMonitorSerializer
 
-from .services import algorithms
+from .services import algorithms, activities
 
 logger = logging.getLogger(__name__)
 
@@ -1710,3 +1710,11 @@ class GetUniqueValuesForKey(generics.ListAPIView):
             })
 
 
+def AssociateActivities(request):
+
+    tasks = Task.objects.all()
+    for task in tasks:
+        activities.associate_activity(task)
+        logger.info(f'{task.id} => {task.sas_id}')
+
+