Post Process Batch Job (#312)

* initial post process batch job

* add collection support and linting

* linting

* addressing comments

* fix for inactive jobs

Author: BryonLewis

client/README.md

@@ -102,7 +102,7 @@ The client is broken into 4 main folders which separate different parts of the s
 
 ## Unified API Specification
 
-To resuse as much code as possible between the Desktop and Web versions there is a unified API which provides the capability to export/import/save/delete annotations and metadata as well as run training and configuration pipelines.  This API allows both versions to share the `/dive-common` and `/src` folders while handling these calls differently.
+To reuse as much code as possible between the Desktop and Web versions there is a unified API which provides the capability to export/import/save/delete annotations and metadata as well as run training and configuration pipelines.  This API allows both versions to share the `/dive-common` and `/src` folders while handling these calls differently.
 
 ## Annotation Viewer Organization
 

client/platform/web-girder/views/Admin/AdminJobs.vue

@@ -55,7 +55,7 @@ export default defineComponent({
     const initTypes = async () => {
       const typesAndStatus = (await getJobTypesStatus());
       jobTypes.value = typesAndStatus.data.types;
-      filterTypes.value = ['convert', 'Dive Metadata Slicer CLI Batch', 'slicer_cli_web_job', 'Slicer CLI Metadata Run'];
+      filterTypes.value = ['convert', 'Dive Metadata Slicer CLI Batch', 'slicer_cli_web_job', 'Slicer CLI Metadata Run', 'DIVE Batch Postprocess'];
     };
     const getData = async () => {
       const statusNums = filterStatus.value.map(

server/dive_server/crud_rpc.py

@@ -202,7 +202,7 @@ def postprocess(
     additive=False,
     additivePrepend='',
     logic='replace',
-) -> types.GirderModel:
+) -> dict:
     """
     Post-processing to be run after media/annotation import
 
@@ -214,12 +214,18 @@ def postprocess(
 
     In either case, the following may run synchronously:
         Conversion of CSV annotations into track JSON
+
+    Returns:
+        dict: Contains 'folder' (the processed folder) and 'job_ids' (list of created job IDs)
     """
     job_is_private = user.get(constants.UserPrivateQueueEnabledMarker, False)
     isClone = dsFolder.get(constants.ForeignMediaIdMarker, None) is not None
     # add default confidence filter threshold to folder metadata
     dsFolder['meta'][constants.ConfidenceFiltersMarker] = {'default': 0.1}
 
+    # Track job IDs for batch processing
+    created_job_ids = []
+
     # Validate user-supplied metadata fields are present
     if fromMeta(dsFolder, constants.FPSMarker) is None:
         raise RestException(f'{constants.FPSMarker} missing from metadata')
@@ -255,7 +261,8 @@ def postprocess(
             newjob.job[constants.JOBCONST_DATASET_ID] = str(item["folderId"])
             newjob.job[constants.JOBCONST_CREATOR] = str(user['_id'])
             Job().save(newjob.job)
-            return dsFolder
+            created_job_ids.append(newjob.job['_id'])
+            return {'folder': dsFolder, 'job_ids': created_job_ids}
 
         if not isClone:
             # transcode VIDEO if necessary
@@ -281,6 +288,7 @@ def postprocess(
                 newjob.job[constants.JOBCONST_PRIVATE_QUEUE] = job_is_private
                 newjob.job[constants.JOBCONST_DATASET_ID] = dsFolder["_id"]
                 Job().save(newjob.job)
+                created_job_ids.append(newjob.job['_id'])
 
             # transcode IMAGERY if necessary
             imageItems = Folder().childItems(
@@ -305,11 +313,12 @@ def postprocess(
                 newjob.job[constants.JOBCONST_PRIVATE_QUEUE] = job_is_private
                 newjob.job[constants.JOBCONST_DATASET_ID] = dsFolder["_id"]
                 Job().save(newjob.job)
+                created_job_ids.append(newjob.job['_id'])
 
             elif imageItems.count() > 0:
                 dsFolder["meta"][constants.DatasetMarker] = True
 
             Folder().save(dsFolder)
     print(f'Processing Items: {user}')
     process_items(dsFolder, user, additive, additivePrepend)
-    return dsFolder
+    return {'folder': dsFolder, 'job_ids': created_job_ids}

server/dive_server/event.py

@@ -2,14 +2,21 @@
 import os
 
 from bson.objectid import ObjectId
+import cherrypy
 from girder import logger
+from girder.api.rest import getApiUrl
+from girder.models.collection import Collection
 from girder.models.folder import Folder
 from girder.models.item import Item
 from girder.models.setting import Setting
+from girder.models.token import Token
 from girder.models.user import User
 from girder.settings import SettingKey
 from girder.utility.mail_utils import renderTemplate, sendMail
+from girder_jobs.models.job import Job
+from girder_worker.girder_plugin.utils import getWorkerApiUrl
 
+from dive_tasks.dive_batch_postprocess import DIVEBatchPostprocessTaskParams
 from dive_utils import asbool, fromMeta
 from dive_utils.constants import (
     AssetstoreSourceMarker,
@@ -22,8 +29,6 @@
     videoRegex,
 )
 
-from . import crud_rpc
-
 
 def send_new_user_email(event):
     try:
@@ -66,11 +71,11 @@ def process_assetstore_import(event, meta: dict):
         userId = parentFolder['creatorId'] or parentFolder['baseParentId']
         user = User().findOne({'_id': ObjectId(userId)})
         foldername = f'Video {item["name"]}'
-        # resuse existing folder if it already exists with same name
+        # reuse existing folder if it already exists with same name
         dest = Folder().createFolder(parentFolder, foldername, creator=user, reuseExisting=True)
         now = datetime.now()
         if now - dest['created'] > timedelta(hours=1):
-            # Remove the old  referenced item, replace it with the new one.
+            # Remove the old referenced item, replace it with the new one.
             oldItem = Item().findOne({'folderId': dest['_id'], 'name': item['name']})
             if oldItem is not None:
                 if oldItem['meta'].get('codec', False):
@@ -108,12 +113,38 @@ def process_assetstore_import(event, meta: dict):
 
 
 def convert_video_recrusive(folder, user):
-    subFolders = list(Folder().childFolders(folder, 'folder', user))
-    for child in subFolders:
-        if child.get('meta', {}).get(MarkForPostProcess, False):
-            Folder().save(child)
-            crud_rpc.postprocess(user, child, False, True)
-        convert_video_recrusive(child, user)
+    """
+    Start a batch postprocess job for all folders with MarkForPostProcess flag.
+    This replaces the manual recursive postprocess calls with a single batch job.
+    """
+    # Create a token for the batch job
+
+    token = Token().createToken(user=user, days=2)
+
+    dive_batch_postprocess_task_params: DIVEBatchPostprocessTaskParams = {
+        "source_folder_id": str(folder['_id']),
+        "skipJobs": False,  # Allow jobs to run (transcoding, etc.)
+        "skipTranscoding": True,  # Skip transcoding if not needed
+        "additive": False,
+        "additivePrepend": '',
+        "userId": str(user['_id']),
+        "girderToken": str(token['_id']),
+        "girderApiUrl": getWorkerApiUrl(),
+    }
+    if not Setting().get('worker.api_url'):
+        Setting().set('worker.api_url', getApiUrl())
+    job = Job().createLocalJob(
+        module='dive_tasks.dive_batch_postprocess',
+        function='batchPostProcessingTaskLauncher',
+        kwargs={'params': dive_batch_postprocess_task_params, 'url': cherrypy.url()},
+        title='Batch process Dive Batch Postprocess',
+        type='DIVE Batch Postprocess',
+        user=user,
+        public=True,
+        asynchronous=True,
+    )
+    job = Job().save(job)
+    Job().scheduleJob(job)
 
 
 class DIVES3Imports:
@@ -132,6 +163,14 @@ def process_s3_import_after(self, event):
             userId = destinationFolder['creatorId'] or destinationFolder['baseParentId']
             user = User().findOne({'_id': ObjectId(userId)})
             convert_video_recrusive(destinationFolder, user)
+        if self.destinationType == 'collection' and self.destinationId is not None:
+            destinationCollection = Collection().findOne({"_id": ObjectId(self.destinationId)})
+            userId = destinationCollection['creatorId'] or destinationCollection['baseParentId']
+            user = User().findOne({'_id': ObjectId(userId)})
+            child_folders = Folder().find({'parentId': ObjectId(self.destinationId)})
+            for child in child_folders:
+                convert_video_recrusive(child, user)
+
         self.destinationId = None
         self.destinationType = None
 

server/dive_server/views_metadata.py

@@ -14,9 +14,9 @@
 from girder.models.file import File
 from girder.models.folder import Folder
 from girder.models.item import Item
-from girder.models.upload import Upload
 from girder.models.setting import Setting
 from girder.models.token import Token
+from girder.models.upload import Upload
 from girder.utility import path as path_util
 from girder_jobs.models.job import Job
 from girder_worker.girder_plugin.utils import getWorkerApiUrl
@@ -28,11 +28,11 @@
     DIVEMetadataClonedFilter,
     DIVEMetadataClonedFilterBase,
     DIVEMetadataFilter,
-    DIVEMetadataMarker,
     DIVEMetadataHistoryMarker,
+    DIVEMetadataMarker,
+    csvRegex,
     jsonRegex,
     ndjsonRegex,
-    csvRegex
 )
 from dive_utils.metadata.models import DIVE_Metadata, DIVE_MetadataKeys
 from dive_utils.types import DiveDatasetList, DIVEMetadataSlicerCLITaskParams
@@ -192,9 +192,7 @@ def bulk_metadata_update_process(user, rootFolder, updates, replace=False):
             raise RestException('Metadata Updates need either DIVEDataset or Filename', code=400)
         if dive_metadata:
             # Find the DIVE_Metadata entry for this dataset and root
-            dataset = Folder().load(
-                dive_metadata['DIVEDataset'], level=AccessType.READ, user=user
-            )
+            dataset = Folder().load(dive_metadata['DIVEDataset'], level=AccessType.READ, user=user)
             updated_keys = []
             errors = []
             # initial pass for all metadata keys:
@@ -1367,16 +1365,21 @@ def bulk_metadata_process_file(self, user, rootFolder, updates, replace=False):
         if len(errors) > 0:
             return results
         # Check and find DIVEMetadataMarker folder or create it if it doesn't exist
-        metadata_folder = Folder().findOne({
-            "name": DIVEMetadataHistoryMarker,
-            "parent": str(rootFolder["_id"]),
-            f'meta.{DIVEMetadataHistoryMarker}': {'$in': TRUTHY_META_VALUES},
-
-        }, user=user, level=AccessType.WRITE)
+        metadata_folder = Folder().findOne(
+            {
+                "name": DIVEMetadataHistoryMarker,
+                "parent": str(rootFolder["_id"]),
+                f'meta.{DIVEMetadataHistoryMarker}': {'$in': TRUTHY_META_VALUES},
+            },
+            user=user,
+            level=AccessType.WRITE,
+        )
         if not metadata_folder:
             metadata_folder = Folder().createFolder(
-                rootFolder, DIVEMetadataHistoryMarker,
-                reuseExisting=True, creator=user,
+                rootFolder,
+                DIVEMetadataHistoryMarker,
+                reuseExisting=True,
+                creator=user,
             )
             Folder().setMetadata(metadata_folder, {DIVEMetadataHistoryMarker: True})
         # now save the previous_data with a timestamp for the name in the folder
@@ -1473,10 +1476,7 @@ def bulk_update_metadata_file(self, rootFolder, replace):
             required=True,
             paramType="body",
             requireArray=True,
-            default=[
-                {
-                }
-            ],
+            default=[{}],
         )
         .param(
             "replace",
@@ -1485,7 +1485,6 @@ def bulk_update_metadata_file(self, rootFolder, replace):
             required=False,
             default=False,
         )
-
     )
     def bulk_update_metadata(self, rootFolder, updates, replace=False):
         user = self.getCurrentUser()

server/dive_server/views_rpc.py

@@ -96,7 +96,7 @@ def __init__(self, resourceName):
         )
     )
     def postprocess(self, folder, skipJobs, skipTranscoding, additive, additivePrepend, logic):
-        return crud_rpc.postprocess(
+        result = crud_rpc.postprocess(
             self.getCurrentUser(),
             folder,
             skipJobs,
@@ -105,6 +105,8 @@ def postprocess(self, folder, skipJobs, skipTranscoding, additive, additivePrepe
             additivePrepend,
             logic,
         )
+        # Return the folder for backward compatibility, but also include job_ids
+        return result
 
     @access.user
     @autoDescribeRoute(

server/dive_server/web_client/JobStatus.js

@@ -5,7 +5,7 @@ events.on('g:appload.before', () => {
     const JobStatus = girder.plugins.jobs.JobStatus;
     const jobPluginIsCancelable = JobStatus.isCancelable;
     JobStatus.isCancelable = function (job) {
-        if (job.get('type').startsWith('Dive Metadata Slicer CLI Batch')) {
+        if (job.get('type').startsWith('Dive Metadata Slicer CLI Batch') || job.get('type').startsWith('DIVE Batch Postprocess')) {
             return ![JobStatus.CANCELED, JobStatus.WORKER_CANCELING || 824,
                 JobStatus.SUCCESS, JobStatus.ERROR].includes(job.get('status'));
         }

server/dive_tasks/__init__.py

@@ -24,4 +24,9 @@ def __init__(self, app, *args, **kwargs):
     def task_imports(self):
         # Return a list of python importable paths to the
         # plugin's path directory
-        return ["dive_tasks.tasks", "dive_tasks.dive_metadata_slicer_cli", "dive_tasks.sam_tasks"]
+        return [
+            "dive_tasks.tasks",
+            "dive_tasks.dive_metadata_slicer_cli",
+            "dive_tasks.sam_tasks",
+            "dive_tasks.dive_batch_postprocess",
+        ]