From 87a8b9cb61e13a26eb074d56eaea519125f89c20 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Tue, 8 Jul 2025 08:58:35 -0400
Subject: [PATCH 01/20] Changing rclone commands to properly use nextflow
 config vars

---
 nextflow/modules/remote_io.nf | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 9014f48..c7389f3 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -42,20 +42,20 @@ process GET_DATA_FROM_DROPBOX {
     """
     #!/bin/bash
 
-    rclone ls \$DROPBOX_PREFIX/\$video_filename > \$WORK_DIR/video_file_remote_stats.txt"
+    rclone ls ${DROPBOX_PREFIX}/\$video_filename > ./video_file_remote_stats.txt"
     h5_filename=${video_file.baseName}_pose_est_v6.h5
-    rclone ls "\$DROPBOX_PREFIX/\${h5_filename}"
+    rclone ls "${DROPBOX_PREFIX}/\${h5_filename}"
     if [[ \$? == 0 ]]; then
         echo "File already processed. Skipping."
         return 1
     fi
-    required_space=\$(awk '{print \$1}' $WORK_DIR/video_file_remote_stats.txt)
-    available_space=\$(df \$WORK_DIR | awk '{ print \$4 }' | tail -n 1)
+    required_space=\$(awk '{print \$1}' ./video_file_remote_stats.txt)
+    available_space=\$(df . | awk '{ print \$4 }' | tail -n 1)
     if [[ $required_space -gt $available_space ]]; then
         echo "Not enough space to download file. Exiting."
         return 1
     fi
-    rclone copy $DROPBOX_PREFIX/$video_filename $WORK_DIR
+    rclone copy ${DROPBOX_PREFIX}/$video_filename .
     """
 }
 
@@ -69,6 +69,6 @@ process PUT_DATA_TO_DROPBOX {
 
     script:
     """
-    rclone copy $file_to_upload $DROPBOX_PREFIX/$folder_name/.
+    rclone copy $file_to_upload ${DROPBOX_PREFIX}/$folder_name/.
     """
 }
\ No newline at end of file

From 3816540c92be72b85dde65f4d29f8687a0d190e5 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 9 Jul 2025 09:24:54 -0400
Subject: [PATCH 02/20] Adding globus-specific modules

---
 nextflow/modules/remote_io.nf | 36 +++++++++++++++++++++++------------
 1 file changed, 24 insertions(+), 12 deletions(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index c7389f3..e2b2166 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -1,30 +1,42 @@
-process GET_DATA_FROM_T2 {
+process CHECK_GLOBUS_AUTH {
     label "globus"
     
     input:
-    val video_filename
-
-    output:
-    path video_file
+    val globus_endpoint
 
     script:
+    // TODO:
+    // If the command fails, globus will print a message to re-authenticate
+    // This message should be sent to the user via email.
     """
-    echo "Not implemented yet!"
-    exit 1
+    globus ls ${globus_endpoint}:/
+    if [[ \$? != 0 ]]; then
+        echo "Globus authentication failed. Please re-authenticate."
+        exit 1
+    fi
     """
+
+    // TODO: This check could be improved.
+    // "globus session show -F json" can return a json containing auth_time
+    // But this needs to be parsed and compared with the endpoint expiration
 }
 
-process PUT_DATA_TO_T2 {
+process TRANSFER_GLOBUS {
     label "globus"
     
     input:
-    path file_to_upload
-    val folder_name
+    val globus_src_endpoint
+    val globus_dst_endpoint
+    val video_filename
+
+    output:
+    path video_file
 
     script:
+    // Globus is asynchronous, so we need to capture the task and wait.
     """
-    echo "Not implemented yet!"
-    exit 1
+    id=$(globus transfer --jq "task_id" --format=UNIX ${globus_src_endpoint}:/${video_filename} ${globus_dst_endpoint}:/${video_filename})
+    globus task wait --polling-interval=10 \$id
     """
 }
 

From 24a04c21c904a09fb05129aae3693abbcfb3acd0 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 9 Jul 2025 11:09:43 -0400
Subject: [PATCH 03/20] Adding filters for files Some syntax correction

---
 nextflow/modules/remote_io.nf | 71 +++++++++++++++++++++++++----------
 1 file changed, 51 insertions(+), 20 deletions(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index e2b2166..7fc2064 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -21,6 +21,52 @@ process CHECK_GLOBUS_AUTH {
     // But this needs to be parsed and compared with the endpoint expiration
 }
 
+process FILTER_UNPROCESSED_GLOBUS {
+    label "globus"
+
+    input:
+    val globus_endpoint
+    path test_files
+
+    output:
+    path "unprocessed_files.txt", emit unprocessed_files
+
+    script:
+    """
+    touch unprocessed_files.txt
+    while read test_file; do
+        test_pose=\${test_file/.*}_pose_est_v6.h5
+        globus ls ${globus_endpoint}:/\${test_pose} > /dev/null 2>&1
+        if [[ \$? != 0 ]]; then
+            echo \$test_file >> unprocessed_files.txt
+        fi
+    done < ${test_files}
+    """
+}
+
+process FILTER_UNPROCESSED_DROPBOX {
+    label "rclone"
+    label "dropbox"
+
+    input:
+    path test_files
+
+    output:
+    path "unprocessed_files.txt", emit unprocessed_files
+
+    script:
+    """
+    touch unprocessed_files.txt
+    while read test_file; do
+        test_pose=\${test_file/.*}_pose_est_v6.h5
+        rclone ls ${DROPBOX_PREFIX}/\${test_pose} > /dev/null 2>&1
+        if [[ \$? != 0 ]]; then
+            echo \$test_file >> unprocessed_files.txt
+        fi
+    done < ${test_files}
+    """
+}
+
 process TRANSFER_GLOBUS {
     label "globus"
     
@@ -48,26 +94,11 @@ process GET_DATA_FROM_DROPBOX {
     val video_filename
 
     output:
-    path ${video_file.baseName}, emit: video_file
+    path "${video_filename}", emit: video_file
 
     script:
     """
-    #!/bin/bash
-
-    rclone ls ${DROPBOX_PREFIX}/\$video_filename > ./video_file_remote_stats.txt"
-    h5_filename=${video_file.baseName}_pose_est_v6.h5
-    rclone ls "${DROPBOX_PREFIX}/\${h5_filename}"
-    if [[ \$? == 0 ]]; then
-        echo "File already processed. Skipping."
-        return 1
-    fi
-    required_space=\$(awk '{print \$1}' ./video_file_remote_stats.txt)
-    available_space=\$(df . | awk '{ print \$4 }' | tail -n 1)
-    if [[ $required_space -gt $available_space ]]; then
-        echo "Not enough space to download file. Exiting."
-        return 1
-    fi
-    rclone copy ${DROPBOX_PREFIX}/$video_filename .
+    rclone copy ${DROPBOX_PREFIX}/${video_filename} ./${video_filename}
     """
 }
 
@@ -77,10 +108,10 @@ process PUT_DATA_TO_DROPBOX {
     
     input:
     path file_to_upload
-    val folder_name
+    tuple path(result_file), val(publish_filename)
 
     script:
     """
-    rclone copy $file_to_upload ${DROPBOX_PREFIX}/$folder_name/.
+    rclone copy ${result_file} ${DROPBOX_PREFIX}/${publish_filename}
     """
-}
\ No newline at end of file
+}

From 6a853659e3846197e35d4974ad001ac465421bcd Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 9 Jul 2025 11:26:35 -0400
Subject: [PATCH 04/20] Starting the remote workflow logic changes.

---
 nextflow/configs/profiles/development.config |  5 +++++
 nextflow/configs/profiles/sumner2.config     |  5 +++++
 nextflow/workflows/io.nf                     | 19 ++++++++++++-------
 3 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 1dae3aa..e8e3d4f 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -93,6 +93,11 @@ 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_t1_endpoint = "INVALID" // Personal endpoint where compute filesystem is visible
+    globus_t2_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 endpoint
+    globus_t2_folder = "/tier2/vkumar/"
 }
 
 singularity {
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index 348af65..5e56ba2 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/nextflow/configs/profiles/sumner2.config
@@ -113,6 +113,11 @@ 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_t1_endpoint = "b8377de1-47c2-11e7-bd5c-22000b9a448b" // JAX endpoint where compute filesystem is visible
+    globus_t2_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 endpoint
+    globus_t2_folder = "/tier2/vkumar/"
 }
 
 apptainer {
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index aebabf3..af15f70 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -1,6 +1,10 @@
 include { CHECK_FILE; URLIFY_FILE } from "${projectDir}/nextflow/modules/utils"
-include { GET_DATA_FROM_T2; PUT_DATA_TO_T2 } from "${projectDir}/nextflow/modules/remote_io"
-include { GET_DATA_FROM_DROPBOX; PUT_DATA_TO_DROPBOX } from "${projectDir}/nextflow/modules/remote_io"
+include { CHECK_GLOBUS_AUTH;
+          FILTER_UNPROCESSED_GLOBUS;
+          FILTER_UNPROCESSED_DROPBOX;
+          TRANSFER_GLOBUS;
+          GET_DATA_FROM_DROPBOX;
+        } from "${projectDir}/nextflow/modules/remote_io"
 
 workflow PREPARE_DATA {
     take:
@@ -11,13 +15,14 @@ workflow PREPARE_DATA {
     {
         if (location == "local")
             video_file = CHECK_FILE(in_video_file).file
+        // TODO: Change remote retrieval to be serialized to not DDOS the network
         else if (location == "dropbox")
+            CHECK_GLOBUS_AUTH()
+            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
             video_file = GET_DATA_FROM_DROPBOX(in_video_file).out.video_file
-        // T2 retrieval not implemented yet, due to globus permission issue.
-        // else if (location == "t2")
-        // """
-        //    GET_DATA_FROM_T2(${in_video_file})
-        // """
+        else if (location == "t2")
+            in_video_list = FILTER_UNPROCESSED_GLOBUS(in_video_file).unprocessed_files
+            video_file = TRANSFER_GLOBUS(params.globus_t2_endpoint, params.globus_t1_endpoint, in_video_file).out.video_file
         else error "${location} is invalid, specify either local or dropbox"
         out_file = URLIFY_FILE(video_file, params.path_depth).file
     }

From ee5896073595200a9cc2661f1703ee45b27ab1e0 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 9 Jul 2025 11:35:14 -0400
Subject: [PATCH 05/20] Adjusting var names to read more clearly

---
 nextflow/configs/profiles/development.config |  6 +++---
 nextflow/configs/profiles/sumner2.config     |  6 +++---
 nextflow/workflows/io.nf                     | 10 +++++-----
 3 files changed, 11 insertions(+), 11 deletions(-)

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index e8e3d4f..8657da8 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -95,9 +95,9 @@ params {
     default_manual_correction_input = default_feature_input
 
     // Remote storage parameters
-    globus_t1_endpoint = "INVALID" // Personal endpoint where compute filesystem is visible
-    globus_t2_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 endpoint
-    globus_t2_folder = "/tier2/vkumar/"
+    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/"
 }
 
 singularity {
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index 5e56ba2..8435cf6 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/nextflow/configs/profiles/sumner2.config
@@ -115,9 +115,9 @@ params {
     default_manual_correction_input = default_feature_input
 
     // Remote storage parameters
-    globus_t1_endpoint = "b8377de1-47c2-11e7-bd5c-22000b9a448b" // JAX endpoint where compute filesystem is visible
-    globus_t2_endpoint = "70850914-722d-11e7-aa01-22000bf2d287" // JAX T2 endpoint
-    globus_t2_folder = "/tier2/vkumar/"
+    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/"
 }
 
 apptainer {
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index af15f70..940be88 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -17,13 +17,13 @@ workflow PREPARE_DATA {
             video_file = CHECK_FILE(in_video_file).file
         // TODO: Change remote retrieval to be serialized to not DDOS the network
         else if (location == "dropbox")
-            CHECK_GLOBUS_AUTH()
             in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
             video_file = GET_DATA_FROM_DROPBOX(in_video_file).out.video_file
-        else if (location == "t2")
-            in_video_list = FILTER_UNPROCESSED_GLOBUS(in_video_file).unprocessed_files
-            video_file = TRANSFER_GLOBUS(params.globus_t2_endpoint, params.globus_t1_endpoint, in_video_file).out.video_file
-        else error "${location} is invalid, specify either local or dropbox"
+        else if (location == "globus")
+            CHECK_GLOBUS_AUTH()
+            in_video_list = FILTER_UNPROCESSED_GLOBUS(params.globus_remote_endpoint, in_video_file).unprocessed_files
+            video_file = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_file).out.video_file
+        else error "${location} is invalid, specify either local, dropbox, or globus"
         out_file = URLIFY_FILE(video_file, params.path_depth).file
     }
 

From a5da72d9b19a4de5a86a2f4b52b151d64e515db8 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Thu, 10 Jul 2025 15:02:25 -0400
Subject: [PATCH 06/20] Multimouse workflow used wrong property

---
 main.nf | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/main.nf b/main.nf
index cdcdd1f..0a16b86 100644
--- a/main.nf
+++ b/main.nf
@@ -108,7 +108,7 @@ workflow{
         SINGLE_MOUSE_V6_FEATURES(paired_video_and_pose)
     }
     if (params.workflow == "multi-mouse"){
-        MULTI_MOUSE_TRACKING(PREPARE_DATA.out.video_file, params.num_mice)
+        MULTI_MOUSE_TRACKING(PREPARE_DATA.out.out_file, params.num_mice)
     }
 }
 

From 5fc673d5bbd65969770254b20f9d5d809ac48afb Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Mon, 14 Jul 2025 13:16:14 -0400
Subject: [PATCH 07/20] Reworking filtering structure and prepare data
 workflows to be more uniform. Also attempting to restructure all to operate
 on file lists for retrieval.

---
 main.nf                       | 53 +----------------------------------
 nextflow/modules/remote_io.nf | 11 ++++----
 nextflow/modules/utils.nf     | 52 ++++++++++++++++++++++------------
 nextflow/workflows/io.nf      | 24 ++++++++++------
 4 files changed, 57 insertions(+), 83 deletions(-)

diff --git a/main.nf b/main.nf
index 0a16b86..58d42fa 100644
--- a/main.nf
+++ b/main.nf
@@ -12,64 +12,13 @@ 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.pipeline)
-        
-        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)
-}
 
 /*
  * Run the selected workflow
  */
 workflow{
     // Download the data locally if necessary
-    PREPARE_DATA(Channel.fromList(all_files), params.location)
+    PREPARE_DATA(params.input_batch, params.location)
 
     // Generate pose files
     if (params.workflow == "single-mouse"){
diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 7fc2064..be7f321 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -73,16 +73,17 @@ process TRANSFER_GLOBUS {
     input:
     val globus_src_endpoint
     val globus_dst_endpoint
-    val video_filename
+    path files_to_transfer
 
     output:
-    path video_file
+    path "globus_cache_folder.txt", emit: globus_folder
 
     script:
     // Globus is asynchronous, so we need to capture the task and wait.
     """
     id=$(globus transfer --jq "task_id" --format=UNIX ${globus_src_endpoint}:/${video_filename} ${globus_dst_endpoint}:/${video_filename})
     globus task wait --polling-interval=10 \$id
+    echo \${pwd} > globus_cache_folder.txt
     """
 }
 
@@ -91,14 +92,14 @@ process GET_DATA_FROM_DROPBOX {
     label "dropbox"
     
     input:
-    val video_filename
+    path files_to_transfer
 
     output:
-    path "${video_filename}", emit: video_file
+    path "retrieved_files/*", emit: remote_files
 
     script:
     """
-    rclone copy ${DROPBOX_PREFIX}/${video_filename} ./${video_filename}
+    rclone copy --include-from ${files_to_transfer} ${DROPBOX_PREFIX}/ retrieved_files/.
     """
 }
 
diff --git a/nextflow/modules/utils.nf b/nextflow/modules/utils.nf
index e10942b..4503620 100644
--- a/nextflow/modules/utils.nf
+++ b/nextflow/modules/utils.nf
@@ -1,40 +1,56 @@
-process VIDEO_TO_POSE {
+process FILTER_LOCAL_BATCH {
     label "r_util"
 
-    // Generates a dummy pose file such that the pipeline can start at any step
     input:
-    path video_file
+    path input_batch
+    val ignore_invalid_inputs
+    val filter_processed
+    val search_dir
 
     output:
-    tuple path(video_file), path("${video_file.baseName}_pose_est_v0.h5"), emit: files
+    path "files_to_process.txt", emit: process_filelist
 
     script:
     """
-    touch "${video_file.baseName}_pose_est_v0.h5"
-    sleep 10
+    touch files_to_process.txt
+    for file in ${input_batch}; do
+        if [[ ! -f "\${file}" && ${ignore_invalid_inputs} != "true" ]]; then
+            echo "File does not exist: \${file}"
+            exit 1
+        else
+            echo "\${file}" >> files_to_process.txt
+        fi
+    done
+    
+    if [[ ${filter_processed} == "true" ]]; then
+        mv files_to_process.txt all_files.txt
+        touch files_to_process.txt
+        echo "Filtering out already processed files..."
+        for file in \$(cat files_to_process.txt); do
+            pose_file="${search_dir}/\${file/.*}_pose_est_v6.h5"
+            if [[ -f "\${pose_file}" ]]; then
+                echo "File \${file} already processed, skipping."
+            else
+                echo "\${file}" >> files_to_process.txt
+            fi
+        done
+    fi
     """
 }
 
-process CHECK_FILE {
+process VIDEO_TO_POSE {
     label "r_util"
 
+    // Generates a dummy pose file such that the pipeline can start at any step
     input:
-    val file_to_check
+    path video_file
 
     output:
-    val file_to_check, emit: file
-    // path "file(${file_to_check})", emit: file
-    // val !file("${file_to_check}").exists(), emit: file_exists
+    tuple path(video_file), path("${video_file.baseName}_pose_est_v0.h5"), emit: files
 
     script:
     """
-    echo "Checking ${file_to_check}"
-    if [ -f "${file_to_check}" ]; then
-        echo "File exists"
-    else
-        echo "File does not exist"
-        exit 1
-    fi
+    touch "${video_file.baseName}_pose_est_v0.h5"
     sleep 10
     """
 }
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index 940be88..702aa89 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -1,4 +1,5 @@
-include { CHECK_FILE; URLIFY_FILE } from "${projectDir}/nextflow/modules/utils"
+include { FILTER_LOCAL_BATCH;
+          URLIFY_FILE } from "${projectDir}/nextflow/modules/utils"
 include { CHECK_GLOBUS_AUTH;
           FILTER_UNPROCESSED_GLOBUS;
           FILTER_UNPROCESSED_DROPBOX;
@@ -13,20 +14,27 @@ workflow PREPARE_DATA {
 
     main:
     {
+
         if (location == "local")
-            video_file = CHECK_FILE(in_video_file).file
-        // TODO: Change remote retrieval to be serialized to not DDOS the network
+            input_batch = FILTER_LOCAL_BATCH(in_video_file, params.ignore_invalid_inputs, params.filter_processed, params.pubdir).process_filelist
+            video_file_batch = Channel.fromPath(video_file_batch)
         else if (location == "dropbox")
             in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
-            video_file = GET_DATA_FROM_DROPBOX(in_video_file).out.video_file
+            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_file).remote_files
         else if (location == "globus")
             CHECK_GLOBUS_AUTH()
             in_video_list = FILTER_UNPROCESSED_GLOBUS(params.globus_remote_endpoint, in_video_file).unprocessed_files
-            video_file = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_file).out.video_file
-        else error "${location} is invalid, specify either local, dropbox, or globus"
-        out_file = URLIFY_FILE(video_file, params.path_depth).file
+            globus_out_folder = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_list).globus_folder
+            video_file_batch = Channel.fromPath(file(globus_out_folder).text)
+        else if (location == "dropbox")
+            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
+            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_list).remote_files
+        else error "${location} is invalid, specify local, dropbox, or globus"
+
+        // Files should be appropriately URLified to avoid collisions within the pipeline
+        input_video_channel = URLIFY_FILE(video_file_batch, params.path_depth).file
     }
 
     emit:
-    out_file
+    input_video_channel
 }

From 42e522acc9c168f34524538e50e03e54dd3567a1 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 16 Jul 2025 15:57:48 -0400
Subject: [PATCH 08/20] First version of failing test, but can get nf-tests
 running

---
 nextflow/modules/remote_io.nf       |  4 ++-
 nextflow/modules/utils.nf           | 24 ++++++++++++++++++
 nextflow/tests/remote_tests.nf.test | 38 +++++++++++++++++++++++++++++
 3 files changed, 65 insertions(+), 1 deletion(-)
 create mode 100644 nextflow/tests/remote_tests.nf.test

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index be7f321..841f0e3 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -80,8 +80,10 @@ process TRANSFER_GLOBUS {
 
     script:
     // Globus is asynchronous, so we need to capture the task and wait.
+    // TODO: Input file is assumed to have the to/from prefixes.
+    // This should be checked or properly documented.
     """
-    id=$(globus transfer --jq "task_id" --format=UNIX ${globus_src_endpoint}:/${video_filename} ${globus_dst_endpoint}:/${video_filename})
+    id=\$(globus transfer --jq "task_id" --format=UNIX --batch ${files_to_transfer} ${globus_src_endpoint} ${globus_dst_endpoint})
     globus task wait --polling-interval=10 \$id
     echo \${pwd} > globus_cache_folder.txt
     """
diff --git a/nextflow/modules/utils.nf b/nextflow/modules/utils.nf
index 4503620..f26c247 100644
--- a/nextflow/modules/utils.nf
+++ b/nextflow/modules/utils.nf
@@ -1,3 +1,27 @@
+/**
+ * Lazy nextflow module for creating files, useful for testing.
+ *
+ * @param file_name The name of the file to be created
+ * @param file_content The content to be written to the file
+ * @return A path to the created file
+ */
+process CREATE_FILE {
+    label "r_util"
+
+    input:
+    val file_name
+    val file_content
+
+    output:
+    path file_name, emit: created_file
+
+    script:
+    """
+    echo "${file_content}" > ${file_name}
+    sleep 10
+    """
+}
+
 process FILTER_LOCAL_BATCH {
     label "r_util"
 
diff --git a/nextflow/tests/remote_tests.nf.test b/nextflow/tests/remote_tests.nf.test
new file mode 100644
index 0000000..5c83dc6
--- /dev/null
+++ b/nextflow/tests/remote_tests.nf.test
@@ -0,0 +1,38 @@
+nextflow_process {
+
+    name "Test Process GET_DATA_FROM_DROPBOX"
+    script "../modules/remote_io.nf"
+    process "GET_DATA_FROM_DROPBOX"
+
+    test("Retrieve Example Data from Dropbox") {
+
+        setup {
+            run ("CREATE_FILE") {
+                script "../modules/utils.nf"
+                process {
+                    """
+                    input[0] = "test_files.txt"
+                    input[1] = "Maze_FED3_B6J/NV7-CBAX2/2023-08-28/Maze0116_0_2023-08-28_10-59-58.avi\\nB6J_3M_ethanol_stranger_26day/NV5-CBAX2/2020-05-22/MDX0174 p2_2020-05-22_06-59-58.avi"
+                    """
+                }
+            }
+        }
+
+        when {
+            process {
+                """
+                test_batch = CREATE_FILE.out.created_file
+                input[0] = test_batch
+                """
+            }
+        }
+
+        then {
+
+            assert process.success
+            assert process.trace.tasks().size() == 1
+
+        }
+
+    }
+}
\ No newline at end of file

From 3e3f8d1efba8e60202f6b1d69bceb0b692082b26 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Fri, 18 Jul 2025 08:59:08 -0400
Subject: [PATCH 09/20] Getting dropbox retrieval operating.

---
 .gitignore                                   |  1 -
 nextflow/configs/profiles/development.config |  4 +---
 nextflow/configs/profiles/sumner2.config     |  4 +---
 nextflow/modules/remote_io.nf                | 12 ++++++++----
 nextflow/tests/remote_tests.nf.test          |  4 +++-
 5 files changed, 13 insertions(+), 12 deletions(-)

diff --git a/.gitignore b/.gitignore
index a00ac16..59900d1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,5 +11,4 @@
 __pycache__
 models
 work
-tests
 !mouse-tracking-runtime/models
diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 8657da8..da5bc17 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -98,6 +98,7 @@ params {
     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 {
@@ -147,9 +148,6 @@ process {
     withLabel: "sleap_io" {
         container = "/media/bgeuther/Storage/TempStorage/onnx/onnx-pipelines/vm/sleap-io-0.2.0.sif"
     }
-    withLabel: "dropbox" {
-        DROPBOX_PREFIX = "labdropbox:KumarLab's shared workspace/VideoData/MDS_Tests"
-    }
 }
 
 executor {
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index 8435cf6..b37c1c7 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/nextflow/configs/profiles/sumner2.config
@@ -118,6 +118,7 @@ params {
     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 {
@@ -203,9 +204,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
diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 841f0e3..99772f6 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -80,9 +80,11 @@ process TRANSFER_GLOBUS {
 
     script:
     // Globus is asynchronous, so we need to capture the task and wait.
-    // TODO: Input file is assumed to have the to/from prefixes.
-    // This should be checked or properly documented.
     """
+    while read line; do
+        line_space_escaped=\$(echo \$line | sed 's: :\\ :g')
+        echo ${globus_src_endpoint}/\$line_space_escaped ${globus_dst_endpoint}/\$line_space_escaped >> batch_to_from.txt
+    done < ${files_to_transfer}
     id=\$(globus transfer --jq "task_id" --format=UNIX --batch ${files_to_transfer} ${globus_src_endpoint} ${globus_dst_endpoint})
     globus task wait --polling-interval=10 \$id
     echo \${pwd} > globus_cache_folder.txt
@@ -95,13 +97,15 @@ process GET_DATA_FROM_DROPBOX {
     
     input:
     path files_to_transfer
+    val dropbox_prefix
 
     output:
-    path "retrieved_files/*", emit: remote_files
+    path "retrieved_files/**", emit: remote_files
 
     script:
     """
-    rclone copy --include-from ${files_to_transfer} ${DROPBOX_PREFIX}/ retrieved_files/.
+    echo ${dropbox_prefix}
+    rclone copy --include-from ${files_to_transfer} ${dropbox_prefix} retrieved_files/.
     """
 }
 
diff --git a/nextflow/tests/remote_tests.nf.test b/nextflow/tests/remote_tests.nf.test
index 5c83dc6..77f89f1 100644
--- a/nextflow/tests/remote_tests.nf.test
+++ b/nextflow/tests/remote_tests.nf.test
@@ -23,6 +23,7 @@ nextflow_process {
                 """
                 test_batch = CREATE_FILE.out.created_file
                 input[0] = test_batch
+                input[1] = "labdropbox:\\\"/KumarLab's shared workspace/VideoData/MDS_Tests/\\\""
                 """
             }
         }
@@ -30,7 +31,8 @@ nextflow_process {
         then {
 
             assert process.success
-            assert process.trace.tasks().size() == 1
+            // Remote files use glob and it returns with shape [1, n]
+            assert process.out.remote_files[0].size() == 2
 
         }
 

From 389ac855983c3aed3f47506dbdeb36e46029c6c5 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Fri, 18 Jul 2025 09:04:32 -0400
Subject: [PATCH 10/20] Adjusting other rclone-based funcs to use adjusted
 variable

---
 nextflow/modules/remote_io.nf | 6 ++++--
 nextflow/workflows/io.nf      | 7 ++-----
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 99772f6..b4b2366 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -50,6 +50,7 @@ process FILTER_UNPROCESSED_DROPBOX {
 
     input:
     path test_files
+    val dropbox_prefix
 
     output:
     path "unprocessed_files.txt", emit unprocessed_files
@@ -59,7 +60,7 @@ process FILTER_UNPROCESSED_DROPBOX {
     touch unprocessed_files.txt
     while read test_file; do
         test_pose=\${test_file/.*}_pose_est_v6.h5
-        rclone ls ${DROPBOX_PREFIX}/\${test_pose} > /dev/null 2>&1
+        rclone ls ${dropbox_prefix}/\${test_pose} > /dev/null 2>&1
         if [[ \$? != 0 ]]; then
             echo \$test_file >> unprocessed_files.txt
         fi
@@ -116,9 +117,10 @@ process PUT_DATA_TO_DROPBOX {
     input:
     path file_to_upload
     tuple path(result_file), val(publish_filename)
+    val dropbox_prefix
 
     script:
     """
-    rclone copy ${result_file} ${DROPBOX_PREFIX}/${publish_filename}
+    rclone copy ${result_file} ${dropbox_prefix}/${publish_filename}
     """
 }
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index 702aa89..5f2b9b4 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -19,16 +19,13 @@ workflow PREPARE_DATA {
             input_batch = FILTER_LOCAL_BATCH(in_video_file, params.ignore_invalid_inputs, params.filter_processed, params.pubdir).process_filelist
             video_file_batch = Channel.fromPath(video_file_batch)
         else if (location == "dropbox")
-            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
-            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_file).remote_files
+            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file, params.dropbox_prefix).unprocessed_files
+            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_file, params.dropbox_prefix).remote_files
         else if (location == "globus")
             CHECK_GLOBUS_AUTH()
             in_video_list = FILTER_UNPROCESSED_GLOBUS(params.globus_remote_endpoint, in_video_file).unprocessed_files
             globus_out_folder = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_list).globus_folder
             video_file_batch = Channel.fromPath(file(globus_out_folder).text)
-        else if (location == "dropbox")
-            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file).unprocessed_files
-            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_list).remote_files
         else error "${location} is invalid, specify local, dropbox, or globus"
 
         // Files should be appropriately URLified to avoid collisions within the pipeline

From 24a7ed540ab4713d40c599166d78b2528c1e12a1 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Mon, 21 Jul 2025 09:14:42 -0400
Subject: [PATCH 11/20] Restricting rclone to 1 transfer at a time

---
 nextflow/modules/remote_io.nf | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index b4b2366..f1939a6 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -106,7 +106,7 @@ process GET_DATA_FROM_DROPBOX {
     script:
     """
     echo ${dropbox_prefix}
-    rclone copy --include-from ${files_to_transfer} ${dropbox_prefix} retrieved_files/.
+    rclone copy --transfers=1 --include-from ${files_to_transfer} ${dropbox_prefix} retrieved_files/.
     """
 }
 
@@ -121,6 +121,6 @@ process PUT_DATA_TO_DROPBOX {
 
     script:
     """
-    rclone copy ${result_file} ${dropbox_prefix}/${publish_filename}
+    rclone copy --transfers=1 ${result_file} ${dropbox_prefix}/${publish_filename}
     """
 }

From 510e74a81c286445a620a90fcf1a3327d6a03a7c Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Mon, 21 Jul 2025 09:18:05 -0400
Subject: [PATCH 12/20] Adding globus remote test

---
 nextflow/tests/remote/globus_test.nf.test     | 40 +++++++++++++++++++
 .../rclone_test.nf.test}                      |  8 ++--
 nf-test.config                                |  8 ++++
 3 files changed, 53 insertions(+), 3 deletions(-)
 create mode 100644 nextflow/tests/remote/globus_test.nf.test
 rename nextflow/tests/{remote_tests.nf.test => remote/rclone_test.nf.test} (89%)
 create mode 100644 nf-test.config

diff --git a/nextflow/tests/remote/globus_test.nf.test b/nextflow/tests/remote/globus_test.nf.test
new file mode 100644
index 0000000..e0a6d99
--- /dev/null
+++ b/nextflow/tests/remote/globus_test.nf.test
@@ -0,0 +1,40 @@
+nextflow_process {
+    
+    name "Test Process TRANSFER_GLOBUS"
+    tag "remote"
+    tag "globus"
+    script "../../modules/remote_io.nf"
+    process "TRANSFER_GLOBUS"
+
+    test("Transfer Files via Globus from remote to compute") {
+
+        setup {
+            run ("CREATE_FILE") {
+                script "../../modules/utils.nf"
+                process {
+                    """
+                    input[0] = "files_to_transfer.txt"
+                    input[1] = "kumarlab-new/dataset-releases/strain-survey-open-field/data/LL1-B2B/2018-04-17_SPD/WT001G2N21717M-17-PSY.mp4\\nkumarlab-new/dataset-releases/strain-survey-open-field/data/LL1-B2B/2018-04-17_SPD/WT001G2N21717M-17-PSY_pose_est_v6.h5"
+                    """
+                }
+            }
+        }
+
+        when {
+            process {
+                """
+                input[0] = "${params.globus_remote_endpoint}"
+                input[1] = "${params.globus_compute_endpoint}"
+                input[2] = CREATE_FILE.out.created_file
+                """
+            }
+        }
+
+        then {
+
+            assert process.success
+
+        }
+
+    }
+}
diff --git a/nextflow/tests/remote_tests.nf.test b/nextflow/tests/remote/rclone_test.nf.test
similarity index 89%
rename from nextflow/tests/remote_tests.nf.test
rename to nextflow/tests/remote/rclone_test.nf.test
index 77f89f1..ca34d91 100644
--- a/nextflow/tests/remote_tests.nf.test
+++ b/nextflow/tests/remote/rclone_test.nf.test
@@ -1,14 +1,16 @@
 nextflow_process {
 
     name "Test Process GET_DATA_FROM_DROPBOX"
-    script "../modules/remote_io.nf"
+    tag "remote"
+    tag "rclone"
+    script "../../modules/remote_io.nf"
     process "GET_DATA_FROM_DROPBOX"
 
     test("Retrieve Example Data from Dropbox") {
 
         setup {
             run ("CREATE_FILE") {
-                script "../modules/utils.nf"
+                script "../../modules/utils.nf"
                 process {
                     """
                     input[0] = "test_files.txt"
@@ -37,4 +39,4 @@ nextflow_process {
         }
 
     }
-}
\ No newline at end of file
+}
diff --git a/nf-test.config b/nf-test.config
new file mode 100644
index 0000000..e531416
--- /dev/null
+++ b/nf-test.config
@@ -0,0 +1,8 @@
+config {
+
+    testsDir "tests"
+    workDir ".nf-test"
+    configFile "nextflow.config"
+    profile "development"
+
+}

From 2841f69ec0dca62c42c005d3b121b8c857e1b1e0 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Tue, 22 Jul 2025 09:23:48 -0400
Subject: [PATCH 13/20] Updating resource paths due to failing hard drive. Also
 making them simpler to update in the future.

---
 nextflow/configs/profiles/development.config | 27 ++++++++++----------
 1 file changed, 14 insertions(+), 13 deletions(-)

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index da5bc17..8849751 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -1,13 +1,14 @@
 // 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/"
@@ -15,9 +16,9 @@ params {
     heuristic_classifier_folder = "/JABS-postprocess/heuristic_classifiers/"
 
     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
@@ -123,30 +124,30 @@ 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"
     }
 }
 

From a8fcd629dc7130f96ddce0e722dbdd5c0e7057eb Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Tue, 22 Jul 2025 10:03:36 -0400
Subject: [PATCH 14/20] Adding proper config to test globus transfers

---
 nextflow/configs/profiles/development.config | 3 +++
 nextflow/configs/profiles/sumner2.config     | 4 ++++
 nextflow/tests/remote/globus_test.nf.test    | 7 ++++---
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 8849751..289cbff 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -149,6 +149,9 @@ process {
     withLabel: "sleap_io" {
         container = "${artifact_dir}vms/sleap-io-0.2.0.sif"
     }
+    withLabel: "globus" {
+        container = "${artifact_dir}vms/globus-cli_2025-07-22.sif"
+    }
 }
 
 executor {
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index b37c1c7..a4d6fc2 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/nextflow/configs/profiles/sumner2.config
@@ -172,6 +172,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
diff --git a/nextflow/tests/remote/globus_test.nf.test b/nextflow/tests/remote/globus_test.nf.test
index e0a6d99..82d5228 100644
--- a/nextflow/tests/remote/globus_test.nf.test
+++ b/nextflow/tests/remote/globus_test.nf.test
@@ -1,5 +1,5 @@
 nextflow_process {
-    
+
     name "Test Process TRANSFER_GLOBUS"
     tag "remote"
     tag "globus"
@@ -23,8 +23,9 @@ nextflow_process {
         when {
             process {
                 """
-                input[0] = "${params.globus_remote_endpoint}"
-                input[1] = "${params.globus_compute_endpoint}"
+                input[0] = "${params.globus_remote_endpoint}:${params.globus_remote_folder}"
+                // This test placed the files onto T1, not locally. Cache folder can only be checked manually.
+                input[1] = "b8377de1-47c2-11e7-bd5c-22000b9a448b:/projects/kumar-lab/multimouse-pipeline/nextflow-test-results/globus_test/"
                 input[2] = CREATE_FILE.out.created_file
                 """
             }

From 531049ef876a0e71fdfeec15e4019ad547544705 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Tue, 22 Jul 2025 15:41:16 -0400
Subject: [PATCH 15/20] Getting the globus test to actually ping globus.
 Currently goes into an infinite wait loop if files don't exist.

---
 nextflow/configs/profiles/development.config | 1 +
 nextflow/modules/remote_io.nf                | 4 ++--
 nextflow/tests/remote/globus_test.nf.test    | 4 +++-
 support_code/globus-cli.def                  | 6 ++++++
 4 files changed, 12 insertions(+), 3 deletions(-)
 create mode 100644 support_code/globus-cli.def

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 289cbff..003bd0a 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -151,6 +151,7 @@ process {
     }
     withLabel: "globus" {
         container = "${artifact_dir}vms/globus-cli_2025-07-22.sif"
+        containerOptions = "-B /media"
     }
 }
 
diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index f1939a6..ce225a1 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -84,9 +84,9 @@ process TRANSFER_GLOBUS {
     """
     while read line; do
         line_space_escaped=\$(echo \$line | sed 's: :\\ :g')
-        echo ${globus_src_endpoint}/\$line_space_escaped ${globus_dst_endpoint}/\$line_space_escaped >> batch_to_from.txt
+        echo \${line_space_escaped} \${line_space_escaped} >> batch_to_from.txt
     done < ${files_to_transfer}
-    id=\$(globus transfer --jq "task_id" --format=UNIX --batch ${files_to_transfer} ${globus_src_endpoint} ${globus_dst_endpoint})
+    id=\$(globus transfer --jq "task_id" --format=UNIX --batch batch_to_from.txt ${globus_src_endpoint} ${globus_dst_endpoint})
     globus task wait --polling-interval=10 \$id
     echo \${pwd} > globus_cache_folder.txt
     """
diff --git a/nextflow/tests/remote/globus_test.nf.test b/nextflow/tests/remote/globus_test.nf.test
index 82d5228..aaf43c7 100644
--- a/nextflow/tests/remote/globus_test.nf.test
+++ b/nextflow/tests/remote/globus_test.nf.test
@@ -23,8 +23,10 @@ nextflow_process {
         when {
             process {
                 """
-                input[0] = "${params.globus_remote_endpoint}:${params.globus_remote_folder}"
+                // input[0] = "${params.globus_remote_endpoint}:${params.globus_remote_folder}"
+                input[0] = "70850914-722d-11e7-aa01-22000bf2d287:/tier2/vkumar/"
                 // This test placed the files onto T1, not locally. Cache folder can only be checked manually.
+                // input[1] = "${params.globus_compute_endpoint}:[folder_on_compute]"
                 input[1] = "b8377de1-47c2-11e7-bd5c-22000b9a448b:/projects/kumar-lab/multimouse-pipeline/nextflow-test-results/globus_test/"
                 input[2] = CREATE_FILE.out.created_file
                 """
diff --git a/support_code/globus-cli.def b/support_code/globus-cli.def
new file mode 100644
index 0000000..58b1922
--- /dev/null
+++ b/support_code/globus-cli.def
@@ -0,0 +1,6 @@
+Bootstrap: docker
+from: python:3.10-bookworm
+
+%post
+	# Install python dependencies
+	pip install globus-cli

From d5582b81926b273137bf547e6618f05d9a00958a Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 23 Jul 2025 10:02:51 -0400
Subject: [PATCH 16/20] Adding a catch for infinite task wait loop

---
 nextflow/modules/remote_io.nf | 22 +++++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index ce225a1..41389db 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -87,7 +87,27 @@ process TRANSFER_GLOBUS {
         echo \${line_space_escaped} \${line_space_escaped} >> batch_to_from.txt
     done < ${files_to_transfer}
     id=\$(globus transfer --jq "task_id" --format=UNIX --batch batch_to_from.txt ${globus_src_endpoint} ${globus_dst_endpoint})
-    globus task wait --polling-interval=10 \$id
+    while true; do
+        globus task wait --timeout 60 --timeout-exit-code 2 \$id
+        // Task succeeded
+        if [[ \$? == 0 ]]; then
+            break
+        // Task failed
+        elif [[ \$? == 1 ]]; then
+            echo "Globus transfer failed."
+            exit 1
+        // Timeout, still running. Figure out if something is wrong.
+        elif [[ \$? == 2 ]]; then
+            // To get all the task info:
+            // globus task show --format=UNIX \$id > globus_task_info.txt
+            fault_count=\$(globus task show --format=UNIX -jq "faults" \$id)
+            if [[ \$fault_count -gt 0 ]]; then
+                echo "Globus transfer failed with faults."
+                globus task cancel \$id
+                exit 1
+            fi
+        fi
+    done
     echo \${pwd} > globus_cache_folder.txt
     """
 }

From a4e095aefa8fbc8a5f2017060e3b06110c99affd Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Wed, 23 Jul 2025 10:25:12 -0400
Subject: [PATCH 17/20] Adding an invalid file test for globus

---
 nextflow/tests/remote/globus_test.nf.test | 32 +++++++++++++++++++++++
 1 file changed, 32 insertions(+)

diff --git a/nextflow/tests/remote/globus_test.nf.test b/nextflow/tests/remote/globus_test.nf.test
index aaf43c7..6e3d4b6 100644
--- a/nextflow/tests/remote/globus_test.nf.test
+++ b/nextflow/tests/remote/globus_test.nf.test
@@ -40,4 +40,36 @@ nextflow_process {
         }
 
     }
+
+    test("Fail globus transfer for files that don't exist") {
+        setup {
+            run ("CREATE_FILE") {
+                script "../../modules/utils.nf"
+                process {
+                    """
+                    input[0] = "files_to_transfer.txt"
+                    input[1] = "FILE_THAT_DOESNT_EXIST.mp4"
+                    """
+                }
+            }
+        }
+        when {
+            process {
+                """
+                // input[0] = "${params.globus_remote_endpoint}:${params.globus_remote_folder}"
+                input[0] = "70850914-722d-11e7-aa01-22000bf2d287:/tier2/vkumar/"
+                // This test placed the files onto T1, not locally. Cache folder can only be checked manually.
+                // input[1] = "${params.globus_compute_endpoint}:[folder_on_compute]"
+                input[1] = "b8377de1-47c2-11e7-bd5c-22000b9a448b:/projects/kumar-lab/multimouse-pipeline/nextflow-test-results/globus_test/"
+                input[2] = CREATE_FILE.out.created_file
+                """
+            }
+        }
+
+        then {
+            // TODO: This test may need to get a time limit set, because failing this test may be an infinite wait.
+            assert process.exitStatus == 1
+
+        }
+    }
 }

From 7b000d3f9d502d34748854c026f899aa33f25cec Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Fri, 25 Jul 2025 10:23:23 -0400
Subject: [PATCH 18/20] Review comment changes

---
 main.nf                                      | 15 ++--
 nextflow/configs/profiles/development.config |  1 +
 nextflow/configs/profiles/sumner2.config     |  2 +
 nextflow/modules/remote_io.nf                | 10 +--
 nextflow/modules/utils.nf                    | 82 +++++++++++++++-----
 nextflow/tests/remote/globus_test.nf.test    |  1 +
 nextflow/tests/remote/rclone_test.nf.test    |  1 +
 nextflow/workflows/io.nf                     | 44 +++++++----
 8 files changed, 107 insertions(+), 49 deletions(-)

diff --git a/main.nf b/main.nf
index 58d42fa..2abb27a 100644
--- a/main.nf
+++ b/main.nf
@@ -10,19 +10,17 @@ 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'
+include { ADD_DUMMY_VIDEO } from './nextflow/modules/utils'
 
 
 /*
  * Run the selected workflow
  */
 workflow{
-    // Download the data locally if necessary
-    PREPARE_DATA(params.input_batch, 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
@@ -42,8 +40,8 @@ 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]
 
@@ -51,13 +49,14 @@ workflow{
         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.out_file, params.num_mice)
+        MULTI_MOUSE_TRACKING(PREPARE_DATA.out.file_processing_channel, params.num_mice)
     }
 }
 
diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 003bd0a..6a0d02f 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -14,6 +14,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 = "${artifact_dir}jabs-classifiers/project_folders/"
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index a4d6fc2..529d86c 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/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/"
diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 41389db..2bc5ddf 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -89,17 +89,17 @@ process TRANSFER_GLOBUS {
     id=\$(globus transfer --jq "task_id" --format=UNIX --batch batch_to_from.txt ${globus_src_endpoint} ${globus_dst_endpoint})
     while true; do
         globus task wait --timeout 60 --timeout-exit-code 2 \$id
-        // Task succeeded
+        # Task succeeded
         if [[ \$? == 0 ]]; then
             break
-        // Task failed
+        # Task failed
         elif [[ \$? == 1 ]]; then
             echo "Globus transfer failed."
             exit 1
-        // Timeout, still running. Figure out if something is wrong.
+        # Timeout, still running. Figure out if something is wrong.
         elif [[ \$? == 2 ]]; then
-            // To get all the task info:
-            // globus task show --format=UNIX \$id > globus_task_info.txt
+            # To get all the task info:
+            # globus task show --format=UNIX \$id > globus_task_info.txt
             fault_count=\$(globus task show --format=UNIX -jq "faults" \$id)
             if [[ \$fault_count -gt 0 ]]; then
                 echo "Globus transfer failed with faults."
diff --git a/nextflow/modules/utils.nf b/nextflow/modules/utils.nf
index f26c247..927a96d 100644
--- a/nextflow/modules/utils.nf
+++ b/nextflow/modules/utils.nf
@@ -37,27 +37,28 @@ process FILTER_LOCAL_BATCH {
     script:
     """
     touch files_to_process.txt
-    for file in ${input_batch}; do
+    while IFS="" read -r file; do
         if [[ ! -f "\${file}" && ${ignore_invalid_inputs} != "true" ]]; then
             echo "File does not exist: \${file}"
             exit 1
         else
+            echo "\${file} exists, adding to process list."
             echo "\${file}" >> files_to_process.txt
         fi
-    done
+    done < ${input_batch}
     
     if [[ ${filter_processed} == "true" ]]; then
         mv files_to_process.txt all_files.txt
         touch files_to_process.txt
         echo "Filtering out already processed files..."
-        for file in \$(cat files_to_process.txt); do
+        while IFS="" read -f file; do
             pose_file="${search_dir}/\${file/.*}_pose_est_v6.h5"
             if [[ -f "\${pose_file}" ]]; then
                 echo "File \${file} already processed, skipping."
             else
                 echo "\${file}" >> files_to_process.txt
             fi
-        done
+        done < files_to_process.txt
     fi
     """
 }
@@ -378,21 +379,6 @@ def validateInputFile(String file_path, String pipeline_type) {
     
     def extension = file_path.substring(file_path.lastIndexOf('.'))
     
-    // Check if file exists
-    if (!file.exists()) {
-        return [false, "File does not exist: ${file_path}"]
-    }
-    
-    // Check if file is readable
-    if (!file.canRead()) {
-        return [false, "File is not readable: ${file_path}"]
-    }
-    
-    // Check if file is non-empty
-    if (file.size() == 0) {
-        return [false, "File is empty: ${file_path}"]
-    }
-    
     // Check file extension against allowed extensions for pipeline type
     if (!valid_extensions[pipeline_type].contains(extension.toLowerCase())) {
         return [false, "Invalid file extension: ${extension}. For pipeline ${pipeline_type}, expected one of: ${valid_extensions[pipeline_type]}"]
@@ -400,3 +386,61 @@ def validateInputFile(String file_path, String pipeline_type) {
     
     return [true, ""]
 }
+
+/**
+ * Subsets an input file list by the formats allowed for a specific pipeline type.
+ *
+ * @param in_file_list The path to the file that contains the list of intut files
+ * @param pipeline_type The type of pipeline being run. See validateInputFile for valid types.
+ * @return A list of valid file paths that match the allowed formats for the specified pipeline type.
+ */
+def validateInputFilelist(String in_file_list, String pipeline_type) {
+    def all_valid_files = []
+    def invalid_files = []
+    def valid_files = []
+
+    def batch_lines = file(in_file_list).text.readLines()
+
+    // Validate each file in the batch
+    batch_lines.each { file_path ->
+        def (is_valid, error_message) = validateInputFile(file_path, pipeline_type)
+
+        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_valid_files.addAll(valid_files)
+
+    if (all_valid_files.size() == 0){
+        println "Missing any data to process, please assign either input_data or input_batch"
+        System.exit(1)
+    }
+
+    return all_valid_files
+}
diff --git a/nextflow/tests/remote/globus_test.nf.test b/nextflow/tests/remote/globus_test.nf.test
index 6e3d4b6..1fb4e1e 100644
--- a/nextflow/tests/remote/globus_test.nf.test
+++ b/nextflow/tests/remote/globus_test.nf.test
@@ -3,6 +3,7 @@ nextflow_process {
     name "Test Process TRANSFER_GLOBUS"
     tag "remote"
     tag "globus"
+    tag "integration"
     script "../../modules/remote_io.nf"
     process "TRANSFER_GLOBUS"
 
diff --git a/nextflow/tests/remote/rclone_test.nf.test b/nextflow/tests/remote/rclone_test.nf.test
index ca34d91..2f2bbdd 100644
--- a/nextflow/tests/remote/rclone_test.nf.test
+++ b/nextflow/tests/remote/rclone_test.nf.test
@@ -3,6 +3,7 @@ nextflow_process {
     name "Test Process GET_DATA_FROM_DROPBOX"
     tag "remote"
     tag "rclone"
+    tag "integration"
     script "../../modules/remote_io.nf"
     process "GET_DATA_FROM_DROPBOX"
 
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index 5f2b9b4..81477d9 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -1,5 +1,7 @@
 include { FILTER_LOCAL_BATCH;
-          URLIFY_FILE } from "${projectDir}/nextflow/modules/utils"
+          URLIFY_FILE;
+          validateInputFile;
+          validateInputFilelist } from "${projectDir}/nextflow/modules/utils"
 include { CHECK_GLOBUS_AUTH;
           FILTER_UNPROCESSED_GLOBUS;
           FILTER_UNPROCESSED_DROPBOX;
@@ -11,27 +13,35 @@ workflow PREPARE_DATA {
     take:
     in_video_file
     location
+    skip_urlify
 
     main:
-    {
+    // Validate input file extensions
+    // TODO: This needs to be a file, not a list. Having the list here wrapping files as paths for filtering.
+    // all_valid_files = validateInputFilelist(in_video_file, params.workflow)
+    all_valid_files = in_video_file
 
-        if (location == "local")
-            input_batch = FILTER_LOCAL_BATCH(in_video_file, params.ignore_invalid_inputs, params.filter_processed, params.pubdir).process_filelist
-            video_file_batch = Channel.fromPath(video_file_batch)
-        else if (location == "dropbox")
-            in_video_list = FILTER_UNPROCESSED_DROPBOX(in_video_file, params.dropbox_prefix).unprocessed_files
-            video_file_batch = GET_DATA_FROM_DROPBOX(in_video_file, params.dropbox_prefix).remote_files
-        else if (location == "globus")
-            CHECK_GLOBUS_AUTH()
-            in_video_list = FILTER_UNPROCESSED_GLOBUS(params.globus_remote_endpoint, in_video_file).unprocessed_files
-            globus_out_folder = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_list).globus_folder
-            video_file_batch = Channel.fromPath(file(globus_out_folder).text)
-        else error "${location} is invalid, specify local, dropbox, or globus"
+    if (location == "local") {
+        file_batch = FILTER_LOCAL_BATCH(all_valid_files, params.ignore_invalid_inputs, params.filter_processed, params.pubdir).process_filelist
+    } else if (location == "dropbox") {
+        in_video_list = FILTER_UNPROCESSED_DROPBOX(all_valid_files, params.dropbox_prefix).unprocessed_files
+        file_batch = GET_DATA_FROM_DROPBOX(in_video_list, params.dropbox_prefix).remote_files
+    } else if (location == "globus") {
+        CHECK_GLOBUS_AUTH()
+        in_video_list = FILTER_UNPROCESSED_GLOBUS(params.globus_remote_endpoint, all_valid_files).unprocessed_files
+        globus_out_folder = TRANSFER_GLOBUS(params.globus_remote_endpoint, params.globus_compute_endpoint, in_video_list).globus_folder
+        file_batch = Channel.fromPath(file(globus_out_folder).text)
+    } else {
+        error "${location} is invalid, specify local, dropbox, or globus"
+    }
 
-        // Files should be appropriately URLified to avoid collisions within the pipeline
-        input_video_channel = URLIFY_FILE(video_file_batch, params.path_depth).file
+    // Files should be appropriately URLified to avoid collisions within the pipeline
+    if (skip_urlify) {
+        file_processing_channel = file_batch
+    } else {
+        file_processing_channel = URLIFY_FILE(file_batch, params.path_depth).file
     }
 
     emit:
-    input_video_channel
+    file_processing_channel
 }

From 31ca86fce80c0d5cf6a68794ba943a0967721edd Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Fri, 25 Jul 2025 16:03:58 -0400
Subject: [PATCH 19/20] Patching some issues with failing full integration
 tests (manual calling of nextflow main ...)

---
 nextflow/configs/profiles/development.config |  2 +-
 nextflow/configs/profiles/sumner2.config     |  2 +-
 nextflow/modules/remote_io.nf                | 12 ++++++++----
 nextflow/workflows/io.nf                     |  6 +++---
 4 files changed, 13 insertions(+), 9 deletions(-)

diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config
index 6a0d02f..ba49c04 100644
--- a/nextflow/configs/profiles/development.config
+++ b/nextflow/configs/profiles/development.config
@@ -100,7 +100,7 @@ params {
     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/\\\""
+    dropbox_prefix = "labdropbox:\"/KumarLab's shared workspace/VideoData/MDS_Tests/\""
 }
 
 singularity {
diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config
index 529d86c..b9505f2 100644
--- a/nextflow/configs/profiles/sumner2.config
+++ b/nextflow/configs/profiles/sumner2.config
@@ -120,7 +120,7 @@ params {
     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/\\\""
+    dropbox_prefix = "labdropbox:\"/KumarLab's shared workspace/VideoData/MDS_Tests/\""
 }
 
 apptainer {
diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf
index 2bc5ddf..a0a4bae 100644
--- a/nextflow/modules/remote_io.nf
+++ b/nextflow/modules/remote_io.nf
@@ -29,7 +29,7 @@ process FILTER_UNPROCESSED_GLOBUS {
     path test_files
 
     output:
-    path "unprocessed_files.txt", emit unprocessed_files
+    path "unprocessed_files.txt", emit: unprocessed_files
 
     script:
     """
@@ -53,18 +53,21 @@ process FILTER_UNPROCESSED_DROPBOX {
     val dropbox_prefix
 
     output:
-    path "unprocessed_files.txt", emit unprocessed_files
+    path "unprocessed_files.txt", emit: unprocessed_files
 
     script:
     """
+    #!/bin/bash
+
     touch unprocessed_files.txt
     while read test_file; do
         test_pose=\${test_file/.*}_pose_est_v6.h5
-        rclone ls ${dropbox_prefix}/\${test_pose} > /dev/null 2>&1
+        rclone ls ${dropbox_prefix}\${test_pose} > /dev/null 2>&1
         if [[ \$? != 0 ]]; then
             echo \$test_file >> unprocessed_files.txt
         fi
     done < ${test_files}
+    exit 0
     """
 }
 
@@ -121,12 +124,13 @@ process GET_DATA_FROM_DROPBOX {
     val dropbox_prefix
 
     output:
-    path "retrieved_files/**", emit: remote_files
+    path "fetched_files.txt", emit: remote_files
 
     script:
     """
     echo ${dropbox_prefix}
     rclone copy --transfers=1 --include-from ${files_to_transfer} ${dropbox_prefix} retrieved_files/.
+    find \$(pwd)/retrieved_files/ -type f > fetched_files.txt
     """
 }
 
diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf
index 81477d9..693fb2b 100644
--- a/nextflow/workflows/io.nf
+++ b/nextflow/workflows/io.nf
@@ -19,7 +19,7 @@ workflow PREPARE_DATA {
     // Validate input file extensions
     // TODO: This needs to be a file, not a list. Having the list here wrapping files as paths for filtering.
     // all_valid_files = validateInputFilelist(in_video_file, params.workflow)
-    all_valid_files = in_video_file
+    all_valid_files = file(in_video_file)
 
     if (location == "local") {
         file_batch = FILTER_LOCAL_BATCH(all_valid_files, params.ignore_invalid_inputs, params.filter_processed, params.pubdir).process_filelist
@@ -37,9 +37,9 @@ workflow PREPARE_DATA {
 
     // Files should be appropriately URLified to avoid collisions within the pipeline
     if (skip_urlify) {
-        file_processing_channel = file_batch
+        file_processing_channel = file_batch.readLines().flatMap { line -> file(line) }
     } else {
-        file_processing_channel = URLIFY_FILE(file_batch, params.path_depth).file
+        file_processing_channel = URLIFY_FILE(file_batch.readLines().flatMap(), params.path_depth).file
     }
 
     emit:

From 26b75bad9b3cc40ec5980c4cdb6eda5cb7a0f6a1 Mon Sep 17 00:00:00 2001
From: Brian Geuther <Brian.Geuther@jax.org>
Date: Mon, 28 Jul 2025 11:10:28 -0400
Subject: [PATCH 20/20] Minor fixes to get this running on sumner2

---
 main.nf                         | 1 +
 nextflow/modules/multi_mouse.nf | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/main.nf b/main.nf
index 2abb27a..de86d61 100644
--- a/main.nf
+++ b/main.nf
@@ -56,6 +56,7 @@ workflow{
         SINGLE_MOUSE_V6_FEATURES(paired_video_and_pose)
     }
     if (params.workflow == "multi-mouse"){
+        PREPARE_DATA(params.input_batch, params.location, false)
         MULTI_MOUSE_TRACKING(PREPARE_DATA.out.file_processing_channel, params.num_mice)
     }
 }
diff --git a/nextflow/modules/multi_mouse.nf b/nextflow/modules/multi_mouse.nf
index 759e97d..f8c155b 100644
--- a/nextflow/modules/multi_mouse.nf
+++ b/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: