Merge pull request #54 from KumarLabJax/remote-nextflow-hooks

Remote nextflow hooks

Author: SkepticRaven

main.nf

@@ -10,70 +10,16 @@ include { SINGLE_MOUSE_TRACKING; SPLIT_BY_CORNERS } from './nextflow/workflows/s
 include { SINGLE_MOUSE_V2_FEATURES; SINGLE_MOUSE_V6_FEATURES } from './nextflow/workflows/feature_generation'
 include { MULTI_MOUSE_TRACKING } from './nextflow/workflows/multi_mouse_pipeline'
 include { MANUALLY_CORRECT_CORNERS; INTEGRATE_CORNER_ANNOTATIONS } from './nextflow/workflows/sleap_manual_correction'
-include { ADD_DUMMY_VIDEO; validateInputFile } from './nextflow/modules/utils'
-
-/*
- * Convert input_batch into a single list
- */
-all_files = []
-invalid_files = []
-valid_files = []
-
-if (params.input_batch != null) {
-    def batch_lines = file(params.input_batch).text.readLines()
-
-    // Validate each file in the batch
-    batch_lines.each { file_path ->
-        def (is_valid, error_message) = validateInputFile(file_path, params.workflow)
-        
-        if (is_valid) {
-            valid_files.add(file_path)
-        } else {
-            invalid_files.add([file_path, error_message])
-        }
-    }
-    
-    // Report any invalid files
-    if (invalid_files.size() > 0) {
-        println "The following files failed validation:"
-        invalid_files.each { file_path, error_message ->
-            println "  - ${error_message}"
-        }
-        
-        if (!params.ignore_invalid_inputs) {
-            println "Please check the input files and try again."
-            println "If you want to ignore invalid inputs, please set the parameter ignore_invalid_inputs to true."
-            System.exit(1)
-        }
-
-        // If all files are invalid, exit
-        if (valid_files.size() == 0) {
-            println "No valid files to process. Exiting."
-            System.exit(1)
-        }
-        
-        // Otherwise, continue with valid files and warn the user
-        println "Continuing with ${valid_files.size()} valid files out of ${batch_lines.size()} total files."
-    }
-    
-    all_files.addAll(valid_files)
-}
-
-if (all_files.size() == 0){
-    println "Missing any data to process, please assign either input_data or input_batch"
-    System.exit(1)
-}
+include { ADD_DUMMY_VIDEO } from './nextflow/modules/utils'
 
 /*
  * Run the selected workflow
  */
 workflow{
-    // Download the data locally if necessary
-    PREPARE_DATA(Channel.fromList(all_files), params.location)
-
     // Generate pose files
     if (params.workflow == "single-mouse"){
-        SINGLE_MOUSE_TRACKING(PREPARE_DATA.out.out_file)
+        PREPARE_DATA(params.input_batch, params.location, false)
+        SINGLE_MOUSE_TRACKING(PREPARE_DATA.out.file_processing_channel)
         v2_outputs = SINGLE_MOUSE_TRACKING.out[0]
         all_v6_outputs = SINGLE_MOUSE_TRACKING.out[1]
         // Split and publish pose_v6 files depending on if corners were successful
@@ -93,22 +39,24 @@ workflow{
     if (params.workflow == "single-mouse-corrected-corners"){
         // Integrate annotations back into pose files
         // This branch requires files to be local and already url-ified
-        // Use a channel of `all_files` instead of `PREPARE_DATA.out.out_file`
-        INTEGRATE_CORNER_ANNOTATIONS(Channel.fromList(all_files), params.sleap_file)
+        PREPARE_DATA(params.input_batch, params.location, true)
+        INTEGRATE_CORNER_ANNOTATIONS(PREPATE_DATA.out.file_processing_channel, params.sleap_file)
         ADD_DUMMY_VIDEO(INTEGRATE_CORNER_ANNOTATIONS.out, params.clip_duration)
         paired_video_and_pose = ADD_DUMMY_VIDEO.out[0]
 
         // Pose v6 features
         SINGLE_MOUSE_V6_FEATURES(paired_video_and_pose)
     }
     if (params.workflow == "single-mouse-v6-features"){
+        PREPARE_DATA(params.input_batch, params.location, false)
         // Generate features from pose_v6 files
         ADD_DUMMY_VIDEO(PREPARE_DATA.out.out_file, params.clip_duration)
         paired_video_and_pose = ADD_DUMMY_VIDEO.out[0]
         SINGLE_MOUSE_V6_FEATURES(paired_video_and_pose)
     }
     if (params.workflow == "multi-mouse"){
-        MULTI_MOUSE_TRACKING(PREPARE_DATA.out.video_file, params.num_mice)
+        PREPARE_DATA(params.input_batch, params.location, false)
+        MULTI_MOUSE_TRACKING(PREPARE_DATA.out.file_processing_channel, params.num_mice)
     }
 }
 

nextflow/configs/profiles/development.config

@@ -1,23 +1,25 @@
 // development configuration file
 
-workDir = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/work/"
+artifact_dir = "/media/bgeuther/LL3_Internal/nextflow-artifacts/"
+workDir = "${artifact_dir}nextflow-work/cache/"
 
 params {
     /*
      * Additional Parameters for the development profile
      * These parameters are described in the sumner2 profile
      */
-    pubdir = "/media/bgeuther/Storage/TempStorage/onnx/nf-tests"
+    pubdir = "${artifact_dir}nextflow-work/pubdir/"
     tracking_code_dir = "/kumar_lab_models/mouse-tracking-runtime/"
     gait_code_dir = "/gait-analysis/"
     vfi_code_dir = "/vfi/Code/"
     support_code_dir = "/mouse-tracking-runtime/support_code/"
     heuristic_classifier_folder = "/JABS-postprocess/heuristic_classifiers/"
+    filter_processed = false
 
     jabs_version = "0.18.1"
-    classifier_project_folders = "/media/bgeuther/Storage/TempStorage/jabs-classifiers/project_folders/"
-    classifier_training_file_folder = "/media/bgeuther/Storage/TempStorage/jabs-classifiers/training_files/"
-    exported_classifier_folder = "/media/bgeuther/Storage/TempStorage/jabs-classifiers/exported_classifiers/"
+    classifier_project_folders = "${artifact_dir}jabs-classifiers/project_folders/"
+    classifier_training_file_folder = "${artifact_dir}jabs-classifiers/training_files/"
+    exported_classifier_folder = "${artifact_dir}jabs-classifiers/exported_classifiers/"
     classifier_artifact_suffix = "_classifier_v${jabs_version}.pickle"
     classifier_window_sizes = [2, 5, 10, 20, 30, 60]
     // Classifiers are described as behavior_name: project_folder
@@ -93,6 +95,12 @@ params {
     // Some default parameters for branches in the pipeline
     default_feature_input = ["${projectDir}/nextflow/default-data/DEFAULT_VIDEO.mp4", "${projectDir}/nextflow/default-data/DEFAULT_VIDEO_pose_est_v6.h5"]
     default_manual_correction_input = default_feature_input
+
+    // Remote storage parameters
+    globus_compute_endpoint = "INVALID" // Personal endpoint where compute filesystem is visible
+    globus_remote_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 storage endpoint
+    globus_remote_folder = "/tier2/vkumar/"
+    dropbox_prefix = "labdropbox:\"/KumarLab's shared workspace/VideoData/MDS_Tests/\""
 }
 
 singularity {
@@ -117,33 +125,34 @@ process {
      * Runtime options
      */
     withLabel: "tracking" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/vm/deployment-runtime_2025-03-05.sif"
+        container = "${artifact_dir}vms/deployment-runtime_2025-03-05.sif"
     }
     withLabel: "jabs_classify" {
-        container = "/media/bgeuther/Storage/TempStorage/JABS-GUI_2025-02-12_v0.18.1.sif"
+        container = "${artifact_dir}vms/JABS-GUI_2025-02-12_v0.18.1.sif"
         // Classifiers exist in a folder not bound by default
         containerOptions = "-B /media"
     }
     withLabel: "jabs_postprocess" {
-        container = "/media/bgeuther/Storage/TempStorage/JABS-Postprocessing-2025-03-27_864d687.sif"
+        container = "${artifact_dir}vms/JABS-Postprocessing-2025-03-27_864d687.sif"
     }
     withLabel: "jabs_table_convert" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/vm/support-r-code_2025-02-11.sif"
+        container = "${artifact_dir}vms/support-r-code_2025-02-11.sif"
     }
     withLabel: "gait" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/tests/gait/gaitanalysis/vm/gait-pipeline-2025-03-27.sif"
+        container = "${artifact_dir}vms/gait-pipeline-2025-03-27.sif"
     }
     withLabel: "frailty" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/tests/vfi/vFI-features/vm/vfi-2025-03-27.sif"
+        container = "${artifact_dir}vms/vfi-2025-03-27.sif"
     }
     withLabel: "sleap" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/vm/sleap-1.4.1.sif"
+        container = "${artifact_dir}vms/sleap-1.4.1.sif"
     }
     withLabel: "sleap_io" {
-        container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/vm/sleap-io-0.2.0.sif"
+        container = "${artifact_dir}vms/sleap-io-0.2.0.sif"
     }
-    withLabel: "dropbox" {
-        DROPBOX_PREFIX = "labdropbox:KumarLab's shared workspace/VideoData/MDS_Tests"
+    withLabel: "globus" {
+        container = "${artifact_dir}vms/globus-cli_2025-07-22.sif"
+        containerOptions = "-B /media"
     }
 }
 

nextflow/configs/profiles/sumner2.config

@@ -11,6 +11,7 @@ params {
      * - vfi_code_dir: prefix for the vfi code directory
      * - support_code_dir: prefix for the support code directory
      * - heuristic_classifier_folder: prefix for the heuristic classifier folder
+     * - filter_processed: whether to filter processed files in the batch
 
      * - jabs_version: version of JABS used for the classifiers
      * - classifier_project_folders: directory containing the classifier project folders
@@ -34,6 +35,7 @@ params {
     vfi_code_dir = "/vfi/Code/"
     support_code_dir = "/mouse-tracking-runtime/support_code/"
     heuristic_classifier_folder = "/JABS-postprocess/heuristic_classifiers/"
+    filter_processed = false
 
     jabs_version = "0.18.1"
     classifier_project_folders = "/projects/kumar-lab/multimouse-pipeline/nextflow-artifacts/project_folders/"
@@ -113,6 +115,12 @@ params {
     // Some default parameters for branches in the pipeline
     default_feature_input = ["/projects/kumar-lab/meta/default-data/DEFAULT_VIDEO.mp4", "/projects/kumar-lab/meta/default-data/DEFAULT_VIDEO_pose_est_v6.h5"]
     default_manual_correction_input = default_feature_input
+
+    // Remote storage parameters
+    globus_compute_endpoint = "b8377de1-47c2-11e7-bd5c-22000b9a448b" // JAX endpoint where compute filesystem is visible
+    globus_remote_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 storage endpoint
+    globus_remote_folder = "/tier2/vkumar/"
+    dropbox_prefix = "labdropbox:\"/KumarLab's shared workspace/VideoData/MDS_Tests/\""
 }
 
 apptainer {
@@ -166,6 +174,10 @@ process {
         queue = "compute"
         resourceLimits = [ cpus: 72, memory: 772.GB, time: 72.h ]
     }
+    withLabel: "xfer" {
+        queue = "xfer"
+        resourceLimits = [ cpus: 1, memory: 4.GB, time: 48.h]
+    }
 
     /*
      * Runtime options
@@ -198,9 +210,6 @@ process {
         // executor.queueSize = 1
         container = "/projects/kumar-lab/multimouse-pipeline/rclone.sif"
     }
-    withLabel: "dropbox" {
-        DROPBOX_PREFIX = "labdropbox:KumarLab's shared workspace/VideoData/MDS_Tests"
-    }
 
     /* 
      * Resource scaling labels

nextflow/modules/multi_mouse.nf

@@ -58,7 +58,7 @@ process GENERATE_MULTI_MOUSE_TRACKLETS {
     val num_animals
 
     output:
-    tuple path(video_file), path("${video_file.baseName}_pose_est_v4.h5"), emit: pose_file
+    tuple path(video_file), path("${video_file.baseName}_pose_est_v4.h5"), emit: files
 
     // Number of tracklets is not yet a parameter accepted by code, so num_animals is currently ignored
     script: