diff --git a/main.nf b/main.nf index 7159939..b50f759 100644 --- a/main.nf +++ b/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,8 +39,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] @@ -102,13 +48,15 @@ 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.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) } } diff --git a/nextflow/configs/profiles/development.config b/nextflow/configs/profiles/development.config index 1dae3aa..ba49c04 100644 --- a/nextflow/configs/profiles/development.config +++ b/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" } } diff --git a/nextflow/configs/profiles/sumner2.config b/nextflow/configs/profiles/sumner2.config index e123ad8..43bceab 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/" @@ -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 diff --git a/nextflow/modules/multi_mouse.nf b/nextflow/modules/multi_mouse.nf index 41413d7..a0c4431 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: diff --git a/nextflow/modules/remote_io.nf b/nextflow/modules/remote_io.nf index 9014f48..a0a4bae 100644 --- a/nextflow/modules/remote_io.nf +++ b/nextflow/modules/remote_io.nf @@ -1,30 +1,117 @@ -process GET_DATA_FROM_T2 { +process CHECK_GLOBUS_AUTH { label "globus" input: - val video_filename + 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. + """ + 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 FILTER_UNPROCESSED_GLOBUS { + label "globus" + + input: + val globus_endpoint + path test_files output: - path video_file + path "unprocessed_files.txt", emit: unprocessed_files script: """ - echo "Not implemented yet!" - exit 1 + 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 PUT_DATA_TO_T2 { +process FILTER_UNPROCESSED_DROPBOX { + label "rclone" + label "dropbox" + + input: + path test_files + val dropbox_prefix + + output: + 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 + if [[ \$? != 0 ]]; then + echo \$test_file >> unprocessed_files.txt + fi + done < ${test_files} + exit 0 + """ +} + +process TRANSFER_GLOBUS { label "globus" input: - path file_to_upload - val folder_name + val globus_src_endpoint + val globus_dst_endpoint + path files_to_transfer + + output: + path "globus_cache_folder.txt", emit: globus_folder script: + // Globus is asynchronous, so we need to capture the task and wait. """ - echo "Not implemented yet!" - exit 1 + while read line; do + line_space_escaped=\$(echo \$line | sed 's: :\\ :g') + 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}) + 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 """ } @@ -33,29 +120,17 @@ process GET_DATA_FROM_DROPBOX { label "dropbox" input: - val video_filename + path files_to_transfer + val dropbox_prefix output: - path ${video_file.baseName}, emit: video_file + path "fetched_files.txt", emit: remote_files script: """ - #!/bin/bash - - rclone ls \$DROPBOX_PREFIX/\$video_filename > \$WORK_DIR/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}' $WORK_DIR/video_file_remote_stats.txt) - available_space=\$(df \$WORK_DIR | 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 + 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 """ } @@ -65,10 +140,11 @@ process PUT_DATA_TO_DROPBOX { input: path file_to_upload - val folder_name + tuple path(result_file), val(publish_filename) + val dropbox_prefix script: """ - rclone copy $file_to_upload $DROPBOX_PREFIX/$folder_name/. + rclone copy --transfers=1 ${result_file} ${dropbox_prefix}/${publish_filename} """ -} \ No newline at end of file +} diff --git a/nextflow/modules/utils.nf b/nextflow/modules/utils.nf index e10942b..927a96d 100644 --- a/nextflow/modules/utils.nf +++ b/nextflow/modules/utils.nf @@ -1,40 +1,81 @@ -process VIDEO_TO_POSE { +/** + * 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" - // Generates a dummy pose file such that the pipeline can start at any step input: - path video_file + val file_name + val file_content output: - tuple path(video_file), path("${video_file.baseName}_pose_est_v0.h5"), emit: files + path file_name, emit: created_file script: """ - touch "${video_file.baseName}_pose_est_v0.h5" + echo "${file_content}" > ${file_name} sleep 10 """ } -process CHECK_FILE { +process FILTER_LOCAL_BATCH { label "r_util" input: - val file_to_check + path input_batch + val ignore_invalid_inputs + val filter_processed + val search_dir output: - val file_to_check, emit: file - // path "file(${file_to_check})", emit: file - // val !file("${file_to_check}").exists(), emit: file_exists + path "files_to_process.txt", emit: process_filelist script: """ - echo "Checking ${file_to_check}" - if [ -f "${file_to_check}" ]; then - echo "File exists" - else - echo "File does not exist" - exit 1 + touch files_to_process.txt + 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 < ${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..." + 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 < files_to_process.txt fi + """ +} + +process VIDEO_TO_POSE { + label "r_util" + + // Generates a dummy pose file such that the pipeline can start at any step + input: + path video_file + + output: + tuple path(video_file), path("${video_file.baseName}_pose_est_v0.h5"), emit: files + + script: + """ + touch "${video_file.baseName}_pose_est_v0.h5" sleep 10 """ } @@ -338,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]}"] @@ -360,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 new file mode 100644 index 0000000..1fb4e1e --- /dev/null +++ b/nextflow/tests/remote/globus_test.nf.test @@ -0,0 +1,76 @@ +nextflow_process { + + name "Test Process TRANSFER_GLOBUS" + tag "remote" + tag "globus" + tag "integration" + 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}:${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 { + + assert process.success + + } + + } + + 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 + + } + } +} diff --git a/nextflow/tests/remote/rclone_test.nf.test b/nextflow/tests/remote/rclone_test.nf.test new file mode 100644 index 0000000..2f2bbdd --- /dev/null +++ b/nextflow/tests/remote/rclone_test.nf.test @@ -0,0 +1,43 @@ +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" + + 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 + input[1] = "labdropbox:\\\"/KumarLab's shared workspace/VideoData/MDS_Tests/\\\"" + """ + } + } + + then { + + assert process.success + // Remote files use glob and it returns with shape [1, n] + assert process.out.remote_files[0].size() == 2 + + } + + } +} diff --git a/nextflow/workflows/io.nf b/nextflow/workflows/io.nf index aebabf3..693fb2b 100644 --- a/nextflow/workflows/io.nf +++ b/nextflow/workflows/io.nf @@ -1,27 +1,47 @@ -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 { FILTER_LOCAL_BATCH; + URLIFY_FILE; + validateInputFile; + validateInputFilelist } from "${projectDir}/nextflow/modules/utils" +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: in_video_file location + skip_urlify main: - { - if (location == "local") - video_file = CHECK_FILE(in_video_file).file - else if (location == "dropbox") - 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 error "${location} is invalid, specify either local or dropbox" - out_file = URLIFY_FILE(video_file, params.path_depth).file + // 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 = 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 + } 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 + if (skip_urlify) { + file_processing_channel = file_batch.readLines().flatMap { line -> file(line) } + } else { + file_processing_channel = URLIFY_FILE(file_batch.readLines().flatMap(), params.path_depth).file } emit: - out_file + file_processing_channel } 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" + +} 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