diff --git a/.github/workflows/branch.yml b/.github/workflows/branch.yml index 204d5bb..e4b87e9 100644 --- a/.github/workflows/branch.yml +++ b/.github/workflows/branch.yml @@ -21,7 +21,7 @@ jobs: # NOTE - this doesn't currently work if the PR is coming from a fork, due to limitations in GitHub actions secrets - name: Post PR comment if: failure() - uses: mshick/add-pr-comment@b8f338c590a895d50bcbfa6c5859251edc8952fc # v2 + uses: mshick/add-pr-comment@8e4927817251f1ff60c001f04568532b38e0b4a0 # v3 with: message: | ## This PR is against the `${{github.event.pull_request.base.ref}}` branch :x: diff --git a/.github/workflows/linting.yml b/.github/workflows/linting.yml index 8b0f88c..8738ffc 100644 --- a/.github/workflows/linting.yml +++ b/.github/workflows/linting.yml @@ -11,33 +11,31 @@ jobs: pre-commit: runs-on: ubuntu-latest steps: - - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 - - name: Set up Python 3.13 - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 - with: - python-version: "3.13" - - - name: Install pre-commit - run: pip install pre-commit + - name: Install Nextflow + uses: nf-core/setup-nextflow@b4ec1bc7c16a94435159de94a05253542fddf6ef # v3 - - name: Run pre-commit - run: pre-commit run --all-files + - name: Run prek + uses: j178/prek-action@6ad80277337ad479fe43bd70701c3f7f8aa74db3 # v2 nf-core: runs-on: ubuntu-latest steps: - name: Check out pipeline code - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 - name: Install Nextflow - uses: nf-core/setup-nextflow@v2 + uses: nf-core/setup-nextflow@b4ec1bc7c16a94435159de94a05253542fddf6ef # v3 - - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6 with: - python-version: "3.13" + python-version: "3.14" architecture: "x64" + - name: Setup uv + uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0 + - name: read .nf-core.yml uses: pietrobolcato/action-read-yaml@9f13718d61111b69f30ab4ac683e67a56d254e1d # 1.1.0 id: read_yml @@ -45,12 +43,10 @@ jobs: config: ${{ github.workspace }}/.nf-core.yml - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install nf-core==${{ steps.read_yml.outputs['nf_core_version'] }} + run: uv tool install nf-core==${{ steps.read_yml.outputs['nf_core_version'] }} - name: Run nf-core pipelines lint - if: ${{ github.base_ref != 'master' }} + if: ${{ github.base_ref != 'master' || github.base_ref != 'main' }} env: GITHUB_COMMENTS_URL: ${{ github.event.pull_request.comments_url }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} @@ -58,7 +54,7 @@ jobs: run: nf-core -l lint_log.txt pipelines lint --dir ${GITHUB_WORKSPACE} --markdown lint_results.md - name: Run nf-core pipelines lint --release - if: ${{ github.base_ref == 'master' }} + if: ${{ github.base_ref == 'master' || github.base_ref == 'main' }} env: GITHUB_COMMENTS_URL: ${{ github.event.pull_request.comments_url }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} @@ -71,7 +67,7 @@ jobs: - name: Upload linting log file artifact if: ${{ always() }} - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: name: linting-logs path: | diff --git a/.github/workflows/linting_comment.yml b/.github/workflows/linting_comment.yml index d43797d..5b0c24f 100644 --- a/.github/workflows/linting_comment.yml +++ b/.github/workflows/linting_comment.yml @@ -11,7 +11,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Download lint results - uses: dawidd6/action-download-artifact@ac66b43f0e6a346234dd65d4d0c8fbb31cb316e5 # v11 + uses: dawidd6/action-download-artifact@b6e2e70617bc3265edd6dab6c906732b2f1ae151 # v21 with: workflow: linting.yml workflow_conclusion: completed @@ -21,7 +21,7 @@ jobs: run: echo "pr_number=$(cat linting-logs/PR_number.txt)" >> $GITHUB_OUTPUT - name: Post PR comment - uses: marocchino/sticky-pull-request-comment@52423e01640425a022ef5fd42c6fb5f633a02728 # v2 + uses: marocchino/sticky-pull-request-comment@70d2764d1a7d5d9560b100cbea0077fc8f633987 # v3 with: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} number: ${{ steps.pr_number.outputs.pr_number }} diff --git a/.github/workflows/nf-test.yml b/.github/workflows/nf-test.yml index 593c936..2062933 100644 --- a/.github/workflows/nf-test.yml +++ b/.github/workflows/nf-test.yml @@ -76,7 +76,7 @@ jobs: - isMain: false profile: "singularity" NXF_VER: - - "24.10.5" + - "25.10.0" - "latest-everything" env: NXF_ANSI_LOG: false diff --git a/.gitignore b/.gitignore index 8a4540e..a94068b 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,6 @@ testing/ testing* *.pyc null/ +.lineage/ +.claude mylocal \ No newline at end of file diff --git a/.nf-core.yml b/.nf-core.yml index 4ace6f6..a7fb8fd 100644 --- a/.nf-core.yml +++ b/.nf-core.yml @@ -1,6 +1,6 @@ repository_type: pipeline -nf_core_version: 3.3.2 +nf_core_version: 4.0.2 lint: files_unchanged: @@ -20,8 +20,16 @@ lint: - validation.help.afterText - validation.summary.beforeText - validation.summary.afterText - multiqc_config: - - report_comment + - config_defaults: + # value contains single quotes that Nextflow escapes and the linter cannot round-trip + - params.zarr_compression_opts + # resolves to a machine-specific absolute path via ${projectDir} + - params.dask_config + # this pipeline does not produce a MultiQC report + multiqc_config: false + # only subworkflows (no modules) are installed from nf-core/modules, which trips + # the linter's requirement that every repo entry have a "modules" key + modules_json: false files_exist: - CODE_OF_CONDUCT.md - assets/nf-core-nf-cellpose_logo_light.png @@ -33,7 +41,7 @@ lint: template: org: JaneliaSciComp - name: nf-cellpose + name: nf_cellpose description: Cellpose segmentation pipeline author: Cristian Goina version: 1.0.0dev diff --git a/.prettierignore b/.prettierignore index edd29f0..63cde50 100644 --- a/.prettierignore +++ b/.prettierignore @@ -1,6 +1,4 @@ email_template.html -adaptivecard.json -slackreport.json .nextflow* work/ data/ @@ -10,4 +8,7 @@ testing/ testing* *.pyc bin/ +.nf-test/ ro-crate-metadata.json +modules/nf-core/ +subworkflows/nf-core/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f976ba..6f6118c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,14 +3,34 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## v1.0.0dev - [date] +## v1.1.0 - 2026-07-01 + +### `Added` + +- Nextflow V2 parser compliance +- Zarr v3 support +- Conda configuration - allows one to run the pipeline using a conda profile + +### `Changed` +- The pipeline requires Nextflow 25.10.0 because of updated nf-schema plugin + +### `Dependencies` +- Nextflow 25.10.0 + +## v1.0.0 - 2025-09-08 Initial release of JaneliaSciComp/nf-cellpose, created with the [nf-core](https://nf-co.re/) template. ### `Added` -### `Fixed` +- Dask support +- CellposeSAM support ### `Dependencies` - -### `Deprecated` +- Nextflow 25.04.04 +- Modules: + - JaneliaSciComp/nextflow-modules:segtools/distributed/cellpose + - JaneliaSciComp/nextflow-modules:segtools/distributed/cellpose + - JaneliaSciComp/nextflow-modules:omezarrtools/multiscale +- Sumworkflows: + - JaneliaSciComp/nextflow-modules:dask_[start,stop] diff --git a/README.md b/README.md index 64d13a6..5ee0e86 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![GitHub Actions Linting Status](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml/badge.svg)](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml)[![Cite with Zenodo](http://img.shields.io/badge/DOI-10.5281/zenodo.XXXXXXX-1073c8?labelColor=000000)](https://doi.org/10.5281/zenodo.XXXXXXX) [![nf-test](https://img.shields.io/badge/unit_tests-nf--test-337ab7.svg)](https://www.nf-test.com) -[![Nextflow](https://img.shields.io/badge/version-%E2%89%A524.10.5-green?style=flat&logo=nextflow&logoColor=white&color=%230DC09D&link=https%3A%2F%2Fnextflow.io)](https://www.nextflow.io/) +[![Nextflow](https://img.shields.io/badge/version-%E2%89%A525.10.0-green?style=flat&logo=nextflow&logoColor=white&color=%230DC09D&link=https%3A%2F%2Fnextflow.io)](https://www.nextflow.io/) [![nf-core template version](https://img.shields.io/badge/nf--core_template-3.3.2-green?style=flat&logo=nfcore&logoColor=white&color=%2324B064&link=https%3A%2F%2Fnf-co.re)](https://github.com/nf-core/tools/releases/tag/3.3.2) [![run with conda](http://img.shields.io/badge/run%20with-conda-3EB049?labelColor=000000&logo=anaconda)](https://docs.conda.io/en/latest/) [![run with docker](https://img.shields.io/badge/run%20with-docker-0db7ed?labelColor=000000&logo=docker)](https://www.docker.com/) @@ -33,6 +33,14 @@ nextflow run JaneliaSciComp/nf-cellpose \ --outdir ``` +To run with conda: +```bash +nextflow run JaneliaSciComp/nf-cellpose \ + -profile conda -with-conda conf/conda-env.yml \ + --input input.ome.zarr \ + --outdir +``` + > [!WARNING] > Please provide pipeline parameters via the CLI or Nextflow `-params-file` option; an example of such file is available [here](examples/segtest_sample.json). Custom config files including those provided by the `-c` Nextflow option can be used to provide any configuration _**except for parameters**_; see [docs](https://nf-co.re/docs/usage/getting_started/configuration#custom-configuration-files). diff --git a/conf/conda-env.yml b/conf/conda-env.yml index 95af792..9534994 100644 --- a/conf/conda-env.yml +++ b/conf/conda-env.yml @@ -1,8 +1,9 @@ channels: - conda-forge + - bioconda dependencies: - python=3.12 - pip - psutil>=7.1.3 - pip: - - git+https://github.com/JaneliaSciComp/cellpose-tools@695281d + - git+https://github.com/JaneliaSciComp/cellpose-tools@969f729 diff --git a/conf/conda.config b/conf/conda.config new file mode 100644 index 0000000..b58f61c --- /dev/null +++ b/conf/conda.config @@ -0,0 +1,9 @@ +process { + withName: "(.*)SEGMENTATION:(.*)" { + container = null + } + + withName: "(.*):MULTISCALE:(.*)" { + container = null + } +} diff --git a/conf/local-conda.config b/conf/local-conda.config deleted file mode 100644 index e0c69ab..0000000 --- a/conf/local-conda.config +++ /dev/null @@ -1,9 +0,0 @@ -process { - withName: "(.*):SEGMENTATION:CELLPOSE" { - container = null - } - - withName: "(.*):MULTISCALE" { - container = null - } -} diff --git a/conf/segmentation.config b/conf/segmentation.config index 4089523..18143f1 100644 --- a/conf/segmentation.config +++ b/conf/segmentation.config @@ -1,11 +1,17 @@ params { + zarr_format = 3 + zarr_chunksize = '64,64,64' + zarr_sharding_factor = '4,4,4' + zarr_compression = 'zstd' + zarr_compression_opts = '\'{"level": 5}\'' + cellpose_model = 'cpsam' - cellpose_models_dir = "${projectDir}/work/cellpose-models" + cellpose_models_dir = "${System.getProperty('user.home')}/.cellpose/models" cellpose_log_config = "${projectDir}/conf/log_config.ini" cellpose_cpus = 1 cellpose_mem_gb = 0 - headjob_runtime_opts = '' - headjob_cluster_opts = '' + cellpose_runtime_opts = '' // container runtime options for the Cellpose main driver job + cellpose_cluster_opts = '' // cluster options for the Cellpose main driver job // data attributes important for segmentation timeindex = 0 @@ -13,29 +19,41 @@ params { z_axis = '' // for OME-ZARR it's not required since it can be read from the OME metdata, for TIFF 0 (first axis) channel_axis = '' // for OME-ZARR it's not required since it can be read from the OME metdata, for TIFF -1 (last axis) - anisotropy = 1.0 + anisotropy = 0.0 expansion_factor = 1.0 voxel_spacing = '' local_workers = 1 // number of workers in a local cluster when with_dask is false + roi = '' process_blocksize = '256,256,256' // process block partition size blocks_overlap = '' // blocks overlap - default to 10% - gpu_device = '' // which gpu device to use + gpu_device = '0' // which gpu device to use do_3D = true use_gpu = true skip_normalize = false norm_percentile = '1 99' + niter = 0 stitch_threshold = 0.0 cellprob_threshold = 1.0 flow3D_smooth = 1 - cell_diameter = 0 + cell_diameter = '' min_size = 15 merge_labels_distance_th = 1 preprocessing_steps = '' preprocessing_config = '' verbose = false - skip_multiscale = false + // run control + run_cellposemasks = false // skip cellpose model eval; reuse a prior run's labels output + run_mergelabels = false // skip the distributed merge-labels step (runs by default) + no_cellposemaskmerge = false // if true skip the mask merge step in cellpose + + // merge-labels (distributed step) + mergelabels_cpus = 1 + mergelabels_mem_gb = 0 + mergelabels_distance_th = 1 // --label-distance-threshold for the MERGELABELS step + + run_multiscale = false multiscale_cpus = 1 multiscale_mem_gb = 0 @@ -49,20 +67,20 @@ params { dask_scheduler_port = 0 dask_dashboard_port = 0 dask_start_timeout_secs = 300 - dask_worker_runtime_opts = '' - dask_worker_cluster_opts = '' + dask_worker_runtime_opts = '' // container runtime options for the dask workers + dask_worker_cluster_opts = '' // cluster options for the dask workers } process { - withName: "(.*):SEGMENTATION:(.*)" { - container = 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' + withName: "SEGMENTATION:(.*)" { + container = 'ghcr.io/janeliascicomp/cellpose:4.2.1-dask2026.6.0-py12' containerOptions = params.container_runtime_opts + clusterOptions = params.lsf_opts errorStrategy = 'terminate' } withName: "(.*):DASK_STARTMANAGER" { - container = 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' ext.args = [ "--port ${params.dask_scheduler_port}", "--dashboard-address ${params.dask_dashboard_port}", @@ -70,78 +88,110 @@ process { } withName: "(.*):DASK_WAITFORMANAGER" { - container = 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' ext.args = [ - optional_arg_value('--scheduler-start-timeout', params.dask_start_timeout_secs), + params.dask_start_timeout_secs ? "--scheduler-start-timeout ${params.dask_start_timeout_secs}" : '', ].join(' ') } withName: "(.*):DASK_STARTWORKER" { - container = 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' containerOptions = "${params.container_runtime_opts} ${params.dask_worker_runtime_opts}" clusterOptions = "${params.lsf_opts} ${params.dask_worker_cluster_opts}" + ext.args = [ "--nthreads ${params.dask_worker_cpus}", ].join(' ') } withName: "(.*):DASK_WAITFORWORKER" { - container = 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' ext.args = [ - optional_arg_value('--worker-start-timeout', params.dask_start_timeout_secs), + params.dask_start_timeout_secs ? "--worker-start-timeout ${params.dask_start_timeout_secs}" : '', ].join(' ') } - withName: "(.*):CELLPOSE" { + withName: "(.*):SEGTOOLS_DISTRIBUTED_CELLPOSE" { + cpus = { (params.cellpose_cpus as int) ?: 1 } + memory = { ((params.cellpose_mem_gb as int) ?: (params.default_mem_gb_per_cpu as int)*((params.cellpose_cpus as int) ?: 1) ).GB } time = 24.h - containerOptions = "${params.container_runtime_opts} ${params.headjob_runtime_opts}" - clusterOptions = "${params.lsf_opts} ${params.headjob_cluster_opts}" + containerOptions = "${params.container_runtime_opts} ${params.cellpose_runtime_opts}" + clusterOptions = "${params.lsf_opts} ${params.cellpose_cluster_opts}" + ext.args = [ - optional_arg_value('--process-blocksize', params.process_blocksize), - optional_arg_value('--blocks-overlaps', params.blocks_overlap), - optional_arg_value('--device', params.gpu_device), - - bool_arg('--do_3D', params.do_3D), - bool_arg('--no_norm', params.skip_normalize), - - optional_arg_value('--z_axis', params.z_axis), - optional_arg_value('--channel_axis', params.channel_axis), - - optional_arg_value('--input-channels', params.input_channels), - - optional_arg_value('--anisotropy', params.anisotropy), - optional_arg_value('--expansion-factor', params.expansion_factor), - optional_arg_value('--voxel-spacing', params.voxel_spacing), - optional_arg_value('--timeindex', params.timeindex), - optional_arg_value('--norm_percentile', params.norm_percentile), - optional_arg_value('--stitch_threshold', params.stitch_threshold), - optional_arg_value('--cellprob_threshold', params.cellprob_threshold), - optional_arg_value('--flow3D_smooth', params.flow3D_smooth), - optional_arg_value('--diameter', params.cell_diameter), - optional_arg_value('--min_size', params.min_size), - optional_arg_value('--label-distance-threshold', params.merge_labels_distance_th), - - optional_arg_value('--preprocessing-steps', params.preprocessing_steps), - - bool_arg('--use_gpu', params.use_gpu), - - optional_arg_value('--local-dask-workers', params.local_workers), - optional_arg_value('--worker-cpus', params.dask_worker_cpus), - - bool_arg('--verbose', params.verbose), + params.preprocessing_steps ? "--preprocessing-steps ${params.preprocessing_steps}" : '', + + (params.run_mergelabels || + params.no_cellposemaskmerge) ? '--skip-merge-labels' : '', + + params.zarr_format ? "--zarr-format ${params.zarr_format}" : '', + params.zarr_chunksize ? "--output-blocksize ${params.zarr_chunksize}" : '', + params.zarr_sharding_factor ? "--sharding-factor ${params.zarr_sharding_factor}" : '', + params.zarr_compression ? "--compressor ${params.zarr_compression}" : '', + params.zarr_compression_opts ? "--compressor-opts ${params.zarr_compression_opts}" : '', + + params.process_blocksize ? "--process-blocksize ${params.process_blocksize}" : '', + params.blocks_overlap ? "--blocks-overlaps ${params.blocks_overlap}" : '', + + params.roi ? "--roi ${params.roi}" : '', + params.do_3D ? '--do_3D' : '', + params.skip_normalize ? '--no_norm' : '', + + params.timeindex ? "--timeindex ${params.timeindex}" : '', + params.input_channels ? "--input-channels ${params.input_channels}" : '', + params.z_axis ? "--z_axis ${params.z_axis}" : '', + params.channel_axis ? "--channel_axis ${params.channel_axis}" : '', + + params.anisotropy ? "--anisotropy ${params.anisotropy}" : '', + params.expansion_factor ? "--expansion-factor ${params.expansion_factor}" : '', + params.voxel_spacing ? "--voxel-spacing ${params.voxel_spacing}" : '', + params.norm_percentile ? "--norm_percentile ${params.norm_percentile}" : '', + params.stitch_threshold ? "--stitch_threshold ${params.stitch_threshold}" : '', + params.cellprob_threshold ? "--cellprob_threshold ${params.cellprob_threshold}" : '', + params.niter ? "--niter ${params.niter}" : '', + params.flow3D_smooth ? "--flow3D_smooth ${params.flow3D_smooth}" : '', + params.cell_diameter ? "--diameter ${params.cell_diameter}" : '', + params.min_size ? "--min_size ${params.min_size}" : '', + params.merge_labels_distance_th ? "--label-distance-threshold ${params.merge_labels_distance_th}" : '', + + + params.use_gpu ? '--use_gpu' : '', + params.gpu_device ? "--device ${params.gpu_device}" : '', + + params.local_workers ? "--local-dask-workers ${params.local_workers}" : '', + params.dask_worker_cpus ? "--worker-cpus ${params.dask_worker_cpus}" : '', + + params.verbose ? '--verbose' : '', ].join(' ') } -} + withName: "(.*):SEGTOOLS_DISTRIBUTED_MERGELABELS" { + cpus = { (params.mergelabels_cpus as int) ?: 1 } + memory = { ((params.mergelabels_mem_gb as int) ?: (params.default_mem_gb_per_cpu as int)*((params.mergelabels_cpus as int) ?: 1) ).GB } + time = 24.h -def arg_value(arg_flag, arg_value) { - "${arg_flag} ${arg_value}" -} + ext.args = [ + params.roi ? "--roi ${params.roi}" : '', + params.zarr_format ? "--zarr-format ${params.zarr_format}" : '', + params.zarr_sharding_factor ? "--sharding-factor ${params.zarr_sharding_factor}" : '', + params.zarr_compression ? "--compressor ${params.zarr_compression}" : '', + params.zarr_compression_opts ? "--compressor-opts ${params.zarr_compression_opts}" : '', -def bool_arg(arg_flag, arg_value) { - arg_value ? "${arg_flag}" : '' -} + params.process_blocksize ? "--process-blocksize ${params.process_blocksize}" : '', + params.mergelabels_distance_th ? "--label-distance-threshold ${params.mergelabels_distance_th}" : '', + + params.local_workers ? "--local-dask-workers ${params.local_workers}" : '', + params.dask_worker_cpus ? "--worker-cpus ${params.dask_worker_cpus}" : '', + + params.verbose ? '--verbose' : '', + ].join(' ') + } + + withName:"(.*)OMEZARRTOOLS_MULTISCALE" { + cpus = { (params.multiscale_cpus as int) ?: 1 } + memory = { (params.multiscale_mem_gb as int).GB } + time = 8.h + + ext.args = [ + '--data-type segmentation', + ].join(' ') + } -def optional_arg_value(arg_flag, arg_value) { - arg_value ? "${arg_flag} ${arg_value}" : '' } diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md new file mode 100644 index 0000000..6c42f0c --- /dev/null +++ b/docs/CONTRIBUTING.md @@ -0,0 +1,162 @@ +--- +title: Contributing +markdownPlugin: checklist +--- + +# `JaneliaSciComp/nf-cellpose`: Contributing guidelines + +Hi there! +Thanks for taking an interest in improving JaneliaSciComp/nf-cellpose. + +This page describes the recommended nf-core way to contribute to both JaneliaSciComp/nf-cellpose and nf-core pipelines in general, including: + +- [General contribution guidelines](#general-contribution-guidelines): common procedures or guides across all nf-core pipelines. +- [Pipeline-specific contribution guidelines](#pipeline-specific-contribution-guidelines): procedures or guides specific to the development conventions of JaneliaSciComp/nf-cellpose. + +## General contribution guidelines + +### Contribution quick start + +To contribute code to any nf-core pipeline: + +- [ ] Ensure you have Nextflow, nf-core tools, and nf-test installed. See the [nf-core/tools repository](https://github.com/nf-core/tools) for instructions. +- [ ] Check whether a GitHub [issue](https://github.com/JaneliaSciComp/nf-cellpose/issues) about your idea already exists. If an issue does not exist, create one so that others are aware you are working on it. +- [ ] [Fork](https://help.github.com/en/github/getting-started-with-github/fork-a-repo) the [JaneliaSciComp/nf-cellpose repository](https://github.com/JaneliaSciComp/nf-cellpose) to your GitHub account. +- [ ] Create a branch on your forked repository and make your changes following [pipeline conventions](#pipeline-contribution-conventions) (if applicable). +- [ ] To fix major bugs, name your branch `patch` and follow the [patch release](#patch-release) process. +- [ ] Update relevant documentation within the `docs/` folder, use nf-core/tools to update `nextflow_schema.json`, and update `CITATIONS.md`. +- [ ] Run and/or update tests. See [Testing](#testing) for more information. +- [ ] [Lint](#lint-tests) your code with nf-core/tools. +- [ ] Submit a pull request (PR) against the `dev` branch and request a review. + +If you are not used to this workflow with Git, see the [GitHub documentation](https://help.github.com/en/github/collaborating-with-issues-and-pull-requests) or [Git resources](https://try.github.io/) for more information. + +### GitHub Codespaces + +You can contribute to JaneliaSciComp/nf-cellpose without installing a local development environment on your machine by using [GitHub Codespaces](https://github.com/codespaces). + +[GitHub Codespaces](https://github.com/codespaces) is an online developer environment that runs in your browser, complete with VS Code and a terminal. +Most nf-core repositories include a devcontainer configuration, which creates a GitHub Codespaces environment specifically for Nextflow development. +The environment includes pre-installed nf-core tools, Nextflow, and a few other helpful utilities via a Docker container. + +To get started, open the repository in [Codespaces](https://github.com/JaneliaSciComp/nf-cellpose/codespaces). + +### Testing + +Once you have made your changes, run the pipeline with nf-test to test them locally. +For additional information, use the `--verbose` flag to view the Nextflow console log output. + +```bash +nf-test test --tag test --profile +docker --verbose +``` + +If you have added new functionality, ensure you update the test assertions in the `.nf.test` files in the `tests/` directory. +Update the snapshots with the following command: + +```bash +nf-test test --tag test --profile +docker --verbose --update-snapshots +``` + +When you create a pull request with changes, GitHub Actions will run automatic tests. +Pull requests are typically reviewed when these tests are passing. + +Two types of tests are typically run: + +#### Lint tests + +nf-core has a [set of guidelines](https://nf-co.re/docs/specifications/overview) which all pipelines must follow. +To enforce these, run linting with nf-core/tools: + +```bash +nf-core pipelines lint +``` + +If you encounter failures or warnings, follow the linked documentation printed to screen. +For more information about linting tests, see [nf-core/tools API documentation](https://nf-co.re/docs/nf-core-tools/api_reference/latest/pipeline_lint_tests/actions_awsfulltest). + +#### Pipeline tests + +Each nf-core pipeline should be set up with a minimal set of test data. +GitHub Actions runs the pipeline on this data to ensure it runs through and exits successfully. +If there are any failures then the automated tests fail. +These tests are run with the latest available version of Nextflow and the minimum required version specified in the pipeline code. + +### Patch release + +> [!WARNING] +> Only in the unlikely event of a release that contains a critical bug. + +- [ ] Create a new branch `patch` on your fork based on `upstream/main` or `upstream/master`. +- [ ] Fix the bug and use nf-core/tools to bump the version to the next semantic version, for example, `1.2.3` → `1.2.4`. +- [ ] Open a Pull Request from `patch` directly to `main`/`master` with the changes. + +### Pipeline contribution conventions + +nf-core semi-standardises how you write code and other contributions to make the JaneliaSciComp/nf-cellpose code and processing logic more understandable for new contributors and to ensure quality. + +#### Add a new pipeline step + +To contribute a new step to the pipeline, follow the general nf-core coding procedure. +Please also refer to the [pipeline-specific contribution guidelines](#pipeline-specific-contribution-guidelines): + +- [ ] Define the corresponding [input channel](#channel-naming-schemes) into your new process from the expected previous process channel. +- [ ] Install a module with nf-core/tools, or write a local module (see [default processes resource requirements](#default-processes-resource-requirements)), and add it to the target `.nf`. +- [ ] Define the output channel if needed. Mix the version output channel into `ch_versions` and relevant files into `ch_multiqc`. +- [ ] Add new or updated parameters to `nextflow.config` with a [default value](#default-parameter-values). +- [ ] Add new or updated parameters and relevant help text to `nextflow_schema.json` with [nf-core/tools](#default-parameter-values). +- [ ] Add validation for relevant parameters to the pipeline utilisation section of `utils_nfcore_cellpose_pipeline/main.nf` subworkflow. +- [ ] Perform local tests to validate that the new code works as expected. + - [ ] If applicable, add a new test in the `tests` directory. +- [ ] Update `usage.md`, `output.md`, and `citation.md` as appropriate. +- [ ] [Lint](lint) the code with nf-core/tools. +- [ ] Update any diagrams or pipeline images as necessary. + +To update the minimum required Nextflow version, see the [Nextflow version bumping](#nextflow-version-bumping) section below. For more information about pipeline contributions, see [pipeline-specific contribution guidelines](#pipeline-specific-contribution-guidelines). + +#### Channel naming schemes + +Use the following naming schemes for channels to make the channel flow easier to understand: + +- Initial process channel: `ch_output_from_` +- Intermediate and terminal channels: `ch__for_` + +#### Default parameter values + +Parameters should be initialised and defined with default values within the `params` scope in `nextflow.config`. +They should also be documented in the pipeline JSON schema. + +To update `nextflow_schema.json`, run: + +```bash +nf-core pipelines schema build +``` + +The schema builder interface that loads in your browser should automatically update the defaults in the parameter documentation. + +#### Default processes resource requirements + +If you write a local module, specify a default set of resource requirements for the process. + +Sensible defaults for process resource requirements (CPUs, memory, time) should be defined in `conf/base.config`. +Specify these with generic `withLabel:` selectors, so they can be shared across multiple processes and steps of the pipeline. + +nf-core provides a set of standard labels that you should follow where possible, as seen in the [nf-core pipeline template](https://github.com/nf-core/tools/blob/main/nf_core/pipeline-template/conf/base.config). +These labels define resource defaults for single-core processes, modules that require a GPU, and different levels of multi-core configurations with increasing memory requirements. + +Values assigned within these labels can be dynamically passed to a tool using the the `${task.cpus}` and `${task.memory}` Nextflow variables in the `script:` block of a module (see an example in the [modules repository](https://github.com/nf-core/modules/blob/bd1b6a40f55933d94b8c9ca94ec8c1ea0eaf4b82/modules/nf-core/samtools/bam2fq/main.nf#L30)). + +#### Nextflow version bumping + +If you use a new feature from core Nextflow, bump the minimum required Nextflow version in the pipeline with: + +```bash +nf-core pipelines bump-version --nextflow . +``` + +#### Images and figures guidelines + +If you update images or graphics, follow the nf-core [style guidelines](https://nf-co.re/docs/community/brand/workflow-schematics). + +## Pipeline specific contribution guidelines + + diff --git a/examples/sample.json b/examples/sample.json index a2fe4a1..9c9e649 100644 --- a/examples/sample.json +++ b/examples/sample.json @@ -9,8 +9,8 @@ "cellpose_model": "${HOME}/.cellpose/models/cpsam", "cellpose_cpus": 1, - "headjob_runtime_opts": "", - "headjob_cluster_opts": "", + "cellpose_runtime_opts": "", + "cellpose_cluster_opts": "", "input_channels": "", "z_axis": "", diff --git a/examples/segtest.json b/examples/segtest.json index e21484c..ba0ddd1 100644 --- a/examples/segtest.json +++ b/examples/segtest.json @@ -7,10 +7,8 @@ "with_dask": true, "process_blocksize": "128,128,128", - "cellpose_model": "${HOME}/.cellpose/models/cpsam", + "cellpose_model": "cpsam", "cellpose_cpus": 1, - "cellpose_runtime_opts": "", - "cellpose_cluster_opts": "", "input_channels": "1", "z_axis": "1", diff --git a/main.nf b/main.nf index acd4a6a..2448bd0 100644 --- a/main.nf +++ b/main.nf @@ -53,7 +53,6 @@ workflow { params.plaintext_email, params.outdir, params.monochrome_logs, - params.hook_url, ) } diff --git a/modules.json b/modules.json index 72d0dc1..0878748 100644 --- a/modules.json +++ b/modules.json @@ -2,17 +2,22 @@ "name": "JaneliaSciComp/nf-cellpose", "homePage": "https://github.com/JaneliaSciComp/nf-cellpose", "repos": { - "git@github.com:JaneliaSciComp/nextflow-modules.git": { + "https://github.com/JaneliaSciComp/nextflow-modules.git": { "modules": { "janelia": { - "cellpose": { + "omezarrtools/multiscale": { "branch": "main", - "git_sha": "d42e3529e9886c8afbbaa6fc81afdb8fdc49eddf", + "git_sha": "d1bc1904f405348fda80e5a75cafe815a7b0ab28", + "installed_by": ["multiscale"] + }, + "segtools/distributed/cellpose": { + "branch": "main", + "git_sha": "6e2c7cfb8ac03d57757baa9432f387e7460788d7", "installed_by": ["modules"] }, - "cellpose/tests": { + "segtools/distributed/mergelabels": { "branch": "main", - "git_sha": "2a343e2f3a45213be05ca36e5de7d33f54c05ca0", + "git_sha": "6e2c7cfb8ac03d57757baa9432f387e7460788d7", "installed_by": ["modules"] } } @@ -21,12 +26,17 @@ "janelia": { "dask_start": { "branch": "main", - "git_sha": "c2fb6e31c459e53023fcf365276fd83a9449abf4", + "git_sha": "2fbde8548e2677a85bc0c19161ce03b81d5b6048", "installed_by": ["subworkflows"] }, "dask_stop": { "branch": "main", - "git_sha": "d7d82538cad8fc7edc2ae3bc6449c33a614d7b86", + "git_sha": "4261356a66340734ba86409fb1c7b4fe50dde206", + "installed_by": ["subworkflows"] + }, + "multiscale": { + "branch": "main", + "git_sha": "85c14f04783f9e832ed322572adf7901402e02a8", "installed_by": ["subworkflows"] } } @@ -37,17 +47,17 @@ "nf-core": { "utils_nextflow_pipeline": { "branch": "master", - "git_sha": "05954dab2ff481bcb999f24455da29a5828af08d", + "git_sha": "1a545fcbd762911c21a64ced3dbef99b2b51ac75", "installed_by": ["subworkflows"] }, "utils_nfcore_pipeline": { "branch": "master", - "git_sha": "65f5e638d901a51534c68fd5c1c19e8112fb4df1", + "git_sha": "a3fb7351b1fdb2b1de282b765816bbea190e86a8", "installed_by": ["subworkflows"] }, "utils_nfschema_plugin": { "branch": "master", - "git_sha": "e753770db613ce014b3c4bc94f6cba443427b726", + "git_sha": "a7b27fd25bfa8dcc07d299e88bd790585901a436", "installed_by": ["subworkflows"] } } diff --git a/modules/local/multiscale/pyramid/main.nf b/modules/janelia/omezarrtools/multiscale/main.nf similarity index 70% rename from modules/local/multiscale/pyramid/main.nf rename to modules/janelia/omezarrtools/multiscale/main.nf index 9580b78..e81620a 100644 --- a/modules/local/multiscale/pyramid/main.nf +++ b/modules/janelia/omezarrtools/multiscale/main.nf @@ -1,18 +1,15 @@ -process MULTISCALE_PYRAMID { - tag { meta.id } - container { task.ext.container ?: 'ghcr.io/janeliascicomp/zarr-tools:dask2025.11.0-py12-ol9' } - cpus { multiscale_cpus } - memory { "${multiscale_mem_gb} GB" } +process OMEZARRTOOLS_MULTISCALE { + tag "${meta.id}" + container 'ghcr.io/janeliascicomp/zarr-tools:0.3.1-dask2025.11.0-py12-ol9' + label 'process_single' input: tuple val(meta), path(data_container), val(dataset_subpath) tuple val(dask_scheduler), path(dask_config) // this is optional - if undefined pass in as empty list ([]) - val(multiscale_cpus) - val(multiscale_mem_gb) output: - tuple val(meta), env(full_data_container_path), val(dataset_subpath), emit: data - path "versions.yml" , emit: versions + tuple val(meta), env('full_data_container_path'), val(dataset_subpath), emit: data + path "versions.yml" , emit: versions when: task.ext.when == null || task.ext.when @@ -23,8 +20,12 @@ process MULTISCALE_PYRAMID { def dask_scheduler_arg = dask_scheduler ? "--dask-scheduler ${dask_scheduler}" : '' def dask_config_arg = dask_config ? "--dask-config ${dask_config}" : '' """ + case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; + esac # Create command line parameters - full_data_container_path=\$(readlink ${data_container}) + full_data_container_path=\$(\${READLINK_TOOL} -e ${data_container}) echo "Generate pyramid for \${full_data_container_path}:${dataset_subpath}" CMD=( python -m zarr_tools.cli.main_multiscale diff --git a/modules/janelia/omezarrtools/multiscale/meta.yml b/modules/janelia/omezarrtools/multiscale/meta.yml new file mode 100644 index 0000000..05bff3d --- /dev/null +++ b/modules/janelia/omezarrtools/multiscale/meta.yml @@ -0,0 +1,57 @@ +# yaml-language-server: $schema=https://raw.githubusercontent.com/nf-core/modules/master/subworkflows/yaml-schema.json +name: omezarrtools_multiscale +description: Generates a multiscale pyramid for an OME-Zarr container +keywords: + - dask + - image formats + - OME-Zarr + - multiscale + - pyramid +components: [] +tools: + - "zarr-tools": + description: "Tools for working with Zarr and OME-Zarr arrays." + homepage: "https://github.com/JaneliaSciComp/zarr-tools" + tool_dev_url: "https://github.com/JaneliaSciComp/zarr-tools" +input: + - meta: + type: map + description: | + Groovy map containing sample metadata. Must include an 'id' field. + - data_container: + type: directory + description: Path to the OME-Zarr container for which to generate the multiscale pyramid + - dataset_subpath: + type: string + description: | + Relative path to the dataset within the container (for N5 or Zarr containers). + Pass as empty string if the dataset is at the root of the container. + - dask_scheduler: + type: string + description: | + Dask scheduler IP address. If not using a distributed cluster, pass as empty string. + - dask_config: + type: file + description: | + Optional path to a dask configuration file. If not specified, pass as empty list ([]). +output: + - meta: + type: map + description: | + Groovy map containing sample metadata. Must include an 'id' field. + - full_data_container_path: + type: string + description: | + Resolved full path to the OME-Zarr container + - dataset_subpath: + type: string + description: | + Relative path to the dataset within the container + - versions: + type: file + description: File containing software versions + pattern: "versions.yml" +authors: + - "@cgoina" +maintainers: + - "@cgoina" diff --git a/modules/janelia/cellpose/conda-env.yml b/modules/janelia/segtools/distributed/cellpose/conda-env.yml similarity index 57% rename from modules/janelia/cellpose/conda-env.yml rename to modules/janelia/segtools/distributed/cellpose/conda-env.yml index 95af792..1126e13 100644 --- a/modules/janelia/cellpose/conda-env.yml +++ b/modules/janelia/segtools/distributed/cellpose/conda-env.yml @@ -5,4 +5,4 @@ dependencies: - pip - psutil>=7.1.3 - pip: - - git+https://github.com/JaneliaSciComp/cellpose-tools@695281d + - git+https://github.com/JaneliaSciComp/cellpose-tools@969f729 diff --git a/modules/janelia/cellpose/main.nf b/modules/janelia/segtools/distributed/cellpose/main.nf similarity index 75% rename from modules/janelia/cellpose/main.nf rename to modules/janelia/segtools/distributed/cellpose/main.nf index cc7f193..b06ee11 100644 --- a/modules/janelia/cellpose/main.nf +++ b/modules/janelia/segtools/distributed/cellpose/main.nf @@ -1,13 +1,14 @@ -process CELLPOSE { - container { task && task.ext.container ? task.ext.container : 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' } - cpus { cellpose_cpus } - memory "${cellpose_mem_in_gb} GB" - conda 'modules/janelia/cellpose/conda-env.yml' +process SEGTOOLS_DISTRIBUTED_CELLPOSE { + tag "${meta.id}" + container 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' + conda "${moduleDir}/conda-env.yml" input: tuple val(meta), path(image, stageAs: 'cellpose-input/*'), val(image_subpath), + path(mask, stageAs: 'cellpose-mask/*'), // this is optional + val(mask_subpath), path(models_path, stageAs: 'cellpose-models/*'), // this is optional - if undefined pass in as empty list ([]) val(model_name), // model name path(output_dir), @@ -18,16 +19,14 @@ process CELLPOSE { path(dask_config) // this is optional - if undefined pass in as empty list ([]) path(preprocessing_config) // preprocessing config file path(logging_config) // this is optional - if undefined pass in as empty list ([]) - val(cellpose_cpus) - val(cellpose_mem_in_gb) output: tuple val(meta), - env(input_image_fullpath), + env('input_image_fullpath'), val(image_subpath), - eval('(IFS=$"\n"; echo "${output_label_images[@]}")'), - val(output_labels_subpath) , emit: results - path('versions.yml') , emit: versions + eval('printf "%s\n" "\${output_label_images[@]}"'), + val(output_labels_subpath) , emit: results + path('versions.yml') , emit: versions when: task.ext.when == null || task.ext.when @@ -43,7 +42,8 @@ process CELLPOSE { ? "--output-subpath ${labels_subpath}" : '' def set_models_path = models_path - ? "models_fullpath=\$(readlink ${models_path}) && \ + ? "models_fullpath=\$(\${READLINK_TOOL} ${models_path}) && \ + echo \"Set models path: \${models_fullpath}\" && \ mkdir -p \${models_fullpath} && \ export CELLPOSE_LOCAL_MODELS_PATH=\${models_fullpath}" : '' @@ -52,11 +52,13 @@ process CELLPOSE { def models_path_arg = models_path ? "--models-dir \${models_fullpath}" : '' def model_name_arg = model_name ? "--model ${model_name}": '' def subpath_name = image_subpath ? "/${image_subpath.split('/')[-1]}" : '' + def mask_arg = mask ? "--mask \$(\${READLINK_TOOL} ${mask})" : '' + def mask_subpath_arg = mask_subpath ? "--mask-subpath ${mask_subpath}" : '' def working_dirname = working_dir ? working_dir : output_dir def labels_image = labels ?: '' def dask_scheduler_arg = dask_scheduler ? "--dask-scheduler ${dask_scheduler}" : '' def dask_config_arg = dask_config ? "--dask-config ${dask_config}" : '' - (labels_noext, labels_ext) = labels_image.lastIndexOf('.').with { pos -> + def (labels_noext, labels_ext) = labels_image.lastIndexOf('.').with { pos -> pos == -1 ? [labels_image, ''] : [labels_image[0.. log.info "Cellpose path inputs: $it" } - def cellpose_results = CELLPOSE( + def cellpose_results = SEGTOOLS_DISTRIBUTED_CELLPOSE( cellpose_test_data, channel.of(['', []]), params.preprocessing_config ? file(params.preprocessing_config) : [], diff --git a/modules/janelia/cellpose/tests/main.nf.test b/modules/janelia/segtools/distributed/cellpose/tests/main.nf.test similarity index 100% rename from modules/janelia/cellpose/tests/main.nf.test rename to modules/janelia/segtools/distributed/cellpose/tests/main.nf.test diff --git a/modules/janelia/cellpose/tests/nextflow.config b/modules/janelia/segtools/distributed/cellpose/tests/nextflow.config similarity index 100% rename from modules/janelia/cellpose/tests/nextflow.config rename to modules/janelia/segtools/distributed/cellpose/tests/nextflow.config diff --git a/modules/janelia/segtools/distributed/mergelabels/conda-env.yml b/modules/janelia/segtools/distributed/mergelabels/conda-env.yml new file mode 100644 index 0000000..1126e13 --- /dev/null +++ b/modules/janelia/segtools/distributed/mergelabels/conda-env.yml @@ -0,0 +1,8 @@ +channels: + - conda-forge +dependencies: + - python=3.12 + - pip + - psutil>=7.1.3 + - pip: + - git+https://github.com/JaneliaSciComp/cellpose-tools@969f729 diff --git a/modules/janelia/segtools/distributed/mergelabels/main.nf b/modules/janelia/segtools/distributed/mergelabels/main.nf new file mode 100644 index 0000000..c316c22 --- /dev/null +++ b/modules/janelia/segtools/distributed/mergelabels/main.nf @@ -0,0 +1,94 @@ +process SEGTOOLS_DISTRIBUTED_MERGELABELS { + tag "${meta.id}" + container 'ghcr.io/janeliascicomp/cellpose:4.0.8-dask2025.11.0-py12' + conda "${moduleDir}/conda-env.yml" + + input: + tuple val(meta), + path(input_labels, stageAs: 'input-labels/*'), + val(input_subpath), + path(mask, stageAs: 'input-mask/*'), // this is optional + val(mask_subpath), + path(output_labels, stageAs: 'output-labels/*'), + val(output_subpath), + path(working_dir, stageAs: 'cellpose-work/*') // this is optional + tuple val(dask_scheduler), + path(dask_config) // this is optional - if undefined pass in as empty list ([]) + path(logging_config) // this is optional - if undefined pass in as empty list ([]) + + output: + tuple val(meta), + env('input_labels_fullpath'), + val(input_subpath), + env('output_labels_fullpath'), + val("${output_subpath ?: input_subpath}"), emit: results + path('versions.yml') , emit: versions + + when: + task.ext.when == null || task.ext.when + + script: + def args = task.ext.args ?: '' + def input_subpath_arg = input_subpath + ? "--input-subpath ${input_subpath}" + : '' + def mask_arg = mask ? "--mask \$(\${READLINK_TOOL} ${mask})" : '' + def mask_subpath_arg = mask_subpath ? "--mask-subpath ${mask_subpath}" : '' + def set_output_labels = output_labels ? "output_labels_fullpath=\$(\${READLINK_TOOL} -m ${output_labels})" : 'output_labels_fullpath=' + def output_labels_arg = output_labels ? "-o \${output_labels_fullpath}" : '' + def output_subpath_arg = output_subpath + ? "--output-subpath ${output_subpath}" + : '' + def logging_config_arg = logging_config ? "--logging-config ${logging_config}" : '' + def working_dir_arg = working_dir ? "--working-dir \$(\${READLINK_TOOL} -m ${working_dir})" : '' + def dask_scheduler_arg = dask_scheduler ? "--dask-scheduler ${dask_scheduler}" : '' + def dask_config_arg = dask_config ? "--dask-config ${dask_config}" : '' + """ + case \$(uname) in + Darwin) + detected_os=OSX + READLINK_TOOL="greadlink" + ;; + *) + detected_os=Linux + READLINK_TOOL="readlink" + ;; + esac + ${set_output_labels} + input_labels_fullpath=\$(\${READLINK_TOOL} -m ${input_labels}) + echo "Input labels: \${input_labels_fullpath}" + + if [[ -z "\${output_labels_fullpath}" ]]; then + echo "set output_labels_fullpath=\${input_labels_fullpath}" + output_labels_fullpath="\${input_labels_fullpath}" + else + mkdir -p \$(dirname \${output_labels_fullpath}) + fi + + echo "Output labels: \${output_labels_fullpath}" + + CMD=( + python -m tools.main_merge_labels + -i \${input_labels_fullpath} ${input_subpath_arg} + ${output_labels_arg} ${output_subpath_arg} + ${mask_arg} ${mask_subpath_arg} + ${working_dir_arg} + ${dask_scheduler_arg} + ${dask_config_arg} + ${logging_config_arg} + ${args} + ) + + echo "CMD: \${CMD[@]}" + (exec "\${CMD[@]}") + + cellpose_version=\$(python -m tools.main_distributed_cellpose \ + --version | \ + grep "cellpose version" | \ + sed "s/cellpose version:\\s*//") + echo "Cellpose version: \${cellpose_version}" + cat <<-END_VERSIONS > versions.yml + cellpose: \${cellpose_version} + END_VERSIONS + """ +} diff --git a/modules/janelia/segtools/distributed/mergelabels/meta.yml b/modules/janelia/segtools/distributed/mergelabels/meta.yml new file mode 100644 index 0000000..f6fc8bd --- /dev/null +++ b/modules/janelia/segtools/distributed/mergelabels/meta.yml @@ -0,0 +1,87 @@ +# yaml-language-server: $schema=https://raw.githubusercontent.com/nf-core/modules/master/subworkflows/yaml-schema.json +name: segtools_distributed_mergelabels +description: Merge labels across block boundaries after a 3-D distributed segmentation +keywords: + - dask + - segmentation + - cellpose +tools: + - "cellpose": + description: "A generalist algorithm for cell and nucleus segmentation." + homepage: "https://github.com/MouseLand/cellpose" + documentaton: "https://cellpose.readthedocs.io/en/latest/" + tool_dev_url: "https://github.com/MouseLand/cellpose" + license: "['BSD-3-Clause']" +input: + - meta: + type: map + description: | + Groovy map containing: id, input_image_dataset, output_image_name + - input_labels: + type: file + description: Path to the input labels container + - input_subpath: + type: string + description: Relative path to the image dataset if needed (for N5 or zarr containers) + - mask: + type: string + description: | + Optional path to a mask that can be used to select a subvolume to be segmented with cellpose. If this is not specified pass in as empty list. + - mask_subpath: + type: string + description: The mask subpath. + - output_labels: + type: file + description: Path to the output labels container + - output_subpath: + type: string + description: Relative path to the image dataset if needed (for N5 or zarr containers) + - working_dir: + type: directory + description: | + Location of the temporary files created during the segmentation. This must have enough space to hold the temporary labels + before they are reassigned during the merge process + - dask_scheduler: + type: string + description: Dask scheduler IP + - dask_config: + type: file + description: | + Optional path to the dask config. If this is not specified pass in as empty list. + - logging_config: + type: file + description: Optional log config. + - cpus: + type: integer + description: Number of cpus needed to run the main cellpose program + - mem_in_gb: + type: integer + description: Memory needed for the main cellpose program +output: + - meta: + type: map + description: | + Groovy map containing: id, input_image_dataset, output_image_name + - input_labels_fullpath: + type: string + description: | + Full path to the input labels container + - input_subpath: + type: string + description: | + Subpath to the input labels dataset + - output_labels_fullpath: + type: string + description: | + Fullpath to the merged labels container + - output_subpath: + type: string + description: merged labels subpath + - versions: + type: file + description: File containing software versions + pattern: "versions.yml" +authors: + - "@cgoina" +maintainers: + - "@cgoina" \ No newline at end of file diff --git a/nextflow.config b/nextflow.config index 7c8f1a1..004c0f6 100644 --- a/nextflow.config +++ b/nextflow.config @@ -12,6 +12,8 @@ params { input = null input_pattern = '' input_subpath = '1' + input_mask = '' + input_mask_subpath = '' outdir = null labels_container_suffix = 'labels' labels_container_ext = '' // extension of the labels container in case the input has no file extension @@ -22,7 +24,6 @@ params { email_on_fail = null plaintext_email = false monochrome_logs = false - hook_url = null help = false help_full = false show_hidden = false @@ -43,9 +44,9 @@ params { // Schema validation default options validate_params = true - user_id = get_uid('u') - group_id = get_uid('g') - singularity_cache_dir = get_env_value("HOME", "/tmp") + "/.singularity_cache" + user_id = { flag -> "id -${flag}".execute().text.trim() }('u') + group_id = { flag -> "id -${flag}".execute().text.trim() }('g') + singularity_cache_dir = "${System.getProperty('user.home')}/.singularity_cache" // Default memory per CPU core default_mem_gb_per_cpu = 15 @@ -59,7 +60,6 @@ includeConfig 'conf/base.config' profiles { debug { - dumpHashes = true process.beforeScript = 'echo $HOSTNAME' cleanup = false nextflow.enable.configProcessNamesValidation = true @@ -271,14 +271,14 @@ manifest { description = """Cellpose segmentation pipeline""" mainScript = 'main.nf' defaultBranch = 'main' - nextflowVersion = '!>=24.10.5' - version = '1.0.0dev' + nextflowVersion = '!>=25.10.0' + version = '1.1.0' doi = '' } // Nextflow plugins plugins { - id 'nf-schema@2.4.2' // Validation of pipeline parameters and creation of an input channel from a sample sheet + id 'nf-schema@2.7.2' // Validation of pipeline parameters and creation of an input channel from a sample sheet } validation { @@ -290,26 +290,8 @@ validation { fullParameter = "help_full" showHiddenParameter = "show_hidden" } + lenientMode = true } // Load modules.config for DSL2 module specific options includeConfig 'conf/modules.config' - -// Borrowed from nf-virontus/nextflow.config -// Try to get a system environment variable value with some key. If the key -// doesn't exist, then return a default value -def get_env_value(key, default_value = null) { - def env = new HashMap(System.getenv()) - if (env.containsKey(key)) { - env[key] - } else { - default_value - } -} - -def get_uid(flag) { - def p = Runtime.getRuntime().exec("id -${flag}") - try (InputStreamReader pout = new InputStreamReader(p.inputStream)) { - pout.readLine() as Integer - } -} diff --git a/nextflow_schema.json b/nextflow_schema.json index 783bcd2..2d6363b 100644 --- a/nextflow_schema.json +++ b/nextflow_schema.json @@ -26,6 +26,14 @@ "type": "string", "description": "Specifies the input dataset if the input is a ZARR/N5/HDF5 container" }, + "input_mask": { + "type": "string", + "description": "Optional mask container restricting segmentation to masked regions of the input. When empty no mask is used." + }, + "input_mask_subpath": { + "type": "string", + "description": "Specifies the mask dataset within input_mask if it is a ZARR/N5/HDF5 container." + }, "outdir": { "type": "string", "format": "directory-path", @@ -136,7 +144,7 @@ "anisotropy": { "type": "number", "description": "Data anisotropy. If this is not provided - it will be computed from voxel spacing.", - "default": 1 + "default": 0 }, "voxel_spacing": { "type": "string", @@ -154,24 +162,27 @@ "hidden": true }, "cellpose_cpus": { - "type": "number", + "type": "integer", "description": "CPU(s) allocated for the cellpose driver", "default": 1 }, "cellpose_mem_gb": { - "type": "number", + "type": "integer", "description": "Memory in GB allocated for the cellpose driver. If the the memory is computed using default_mem_gb_per_cpu*cellpose_cpus", "default": 0 }, "input_channels": { + "type": ["string", "integer"], + "description": "Comma delimited list of 0-based channels used for segmentation." + }, + "roi": { "type": "string", - "description": "Comma delimited list of 0-based channels used for segmentation.", - "default": "1" + "description": "Restrict segmentation to a region of interest within the input (maps to --roi). When empty the whole input is processed." }, "process_blocksize": { "type": "string", "description": "Define spatial block partitioning for distributing the work to a Dask cluster. Even for 4 or 5-dimensional arrays this only needs to specify the block partitioning in the 3D spatial dimensions only", - "default": "128,128,128" + "default": "256,256,256" }, "local_workers": { "type": "integer", @@ -183,7 +194,7 @@ "description": "Blocks overlap used for distributed processing" }, "gpu_device": { - "type": "string", + "type": ["integer", "string"], "description": "which gpu device to use" }, "do_3D": { @@ -210,12 +221,12 @@ "default": "1 99" }, "stitch_threshold": { - "type": "integer", + "type": "number", "description": "cellpose stitch_threshold parameter", "default": 0 }, "cellprob_threshold": { - "type": "integer", + "type": "number", "description": "cellpose cellprob_threshold parameter", "default": 1 }, @@ -224,11 +235,15 @@ "description": "cellpose flow3D_smooth parameter", "default": 1 }, - "cell_diameter": { - "type": "integer", - "description": "cell diameter", + "niter": { + "type": ["integer", "string"], + "description": "cellpose niter parameter (number of iterations for dynamics; 0 = auto)", "default": 0 }, + "cell_diameter": { + "type": ["integer", "string"], + "description": "cell diameter" + }, "min_size": { "type": "integer", "description": "cell minimum size", @@ -250,29 +265,98 @@ "description": "Preprocessing YAML config file. This contains preprocessing steps configuration.", "hidden": true }, + "run_cellposemasks": { + "type": "boolean", + "description": "If true run the cellpose model evaluation otherwise it will reuse the labels output from a prior run (only valid if the eval has been run before)." + }, + "no_cellposemaskmerge": { + "type": "boolean", + "description": "If true pass --skip-merge-labels to distributed cellpose, skipping internal label merging", + "fa_icon": "fas fa-toggle-on" + }, "verbose": { "type": "boolean", "description": "Log verbosity" }, - "headjob_runtime_opts": { + "cellpose_runtime_opts": { "type": "string", - "description": "Cellpose head job container runtime options. When cellpose is not distributed on a Dask cluster - it requires GPU." + "description": "Cellpose head job container runtime options. When cellpose is not distributed on a Dask cluster - it requires GPU so it may need '--nv' flag is singularity is used." }, - "headjob_cluster_opts": { + "cellpose_cluster_opts": { "type": "string", "description": "Cellpose head job cluster options. When cellpose is not distributed on a Dask cluster - the job needs to be dispatched on a GPU node." } } }, + "zarr_options": { + "title": "Zarr output options", + "type": "object", + "description": "Options controlling how Zarr output containers are written (format, chunking, sharding, compression).", + "default": "", + "properties": { + "zarr_format": { + "type": ["integer", "string"], + "default": 3, + "description": "Zarr format version used for output containers (e.g. 2 or 3)." + }, + "zarr_chunksize": { + "type": "string", + "default": "64,64,64", + "description": "Output Zarr chunk size as a comma-separated list (maps to --output-blocksize)." + }, + "zarr_sharding_factor": { + "type": "string", + "default": "4,4,4", + "description": "Output Zarr sharding factor as a comma-separated list (maps to --sharding-factor). Only used with Zarr format 3." + }, + "zarr_compression": { + "type": "string", + "default": "zstd", + "description": "Compressor used for output Zarr arrays (maps to --compressor), e.g. zstd, blosc, gzip." + }, + "zarr_compression_opts": { + "type": "string", + "default": "'{\"level\": 5}'", + "description": "Compressor options as a JSON string (maps to --compressor-opts), e.g. '{\"level\": 5}'." + } + } + }, + "merge_labels_options": { + "title": "Merge labels options", + "type": "object", + "description": "Options for the distributed merge-labels step that runs after cellpose segmentation.", + "default": "", + "properties": { + "run_mergelabels": { + "type": "boolean", + "description": "If true run the distributed merge-labels step. By default merge-labels in the same process as cellpose segmentation right after mask generation." + }, + "mergelabels_distance_th": { + "type": "integer", + "default": 1, + "description": "Distance threshold (--label-distance-threshold) for the standalone distributed merge-labels step." + }, + "mergelabels_cpus": { + "type": "integer", + "default": 1, + "description": "Number of cpus for the merge-labels driver process." + }, + "mergelabels_mem_gb": { + "type": "integer", + "default": 0, + "description": "Memory resource in GB for the merge-labels driver process." + } + } + }, "multiscale_options": { "title": "Multiscale options", "type": "object", "description": "", "default": "", "properties": { - "skip_multiscale": { + "run_multiscale": { "type": "boolean", - "description": "If true skip generating multiscale pyramid for segmented images" + "description": "If true run generating multiscale pyramid for segmented images" }, "multiscale_cpus": { "type": "integer", @@ -322,12 +406,12 @@ "default": 0 }, "dask_worker_cpus": { - "type": "number", + "type": "integer", "description": "CPU(s) allocated for the cellpose dask worker", "default": 1 }, "dask_worker_mem_gb": { - "type": "number", + "type": "integer", "description": "Memory in GB allocated for the cellpose dask worker. If the the memory is computed using default_mem_gb_per_cpu*dask_worker_cpus", "default": 0 }, @@ -354,6 +438,24 @@ "description": "Less common options for the pipeline, typically set in a config file.", "help_text": "These options are common to all nf-core pipelines and allow you to customise some of the core preferences for how the pipeline runs.\n\nTypically these options would be set in a Nextflow config file loaded for all pipeline runs, such as `~/.nextflow/config`.", "properties": { + "help": { + "type": "boolean", + "description": "Display help text.", + "fa_icon": "fas fa-question-circle", + "hidden": true + }, + "help_full": { + "type": "boolean", + "description": "Display full help text.", + "fa_icon": "fas fa-question-circle", + "hidden": true + }, + "show_hidden": { + "type": "boolean", + "description": "Display all hidden parameters.", + "fa_icon": "fas fa-question-circle", + "hidden": true + }, "version": { "type": "boolean", "description": "Display version and exit.", @@ -389,13 +491,6 @@ "fa_icon": "fas fa-palette", "hidden": true }, - "hook_url": { - "type": "string", - "description": "Incoming hook URL for messaging service", - "fa_icon": "fas fa-people-group", - "help_text": "Incoming hook URL for messaging service. Currently, MS Teams and Slack are supported.", - "hidden": true - }, "validate_params": { "type": "boolean", "description": "Boolean whether to validate parameters against the schema at runtime", @@ -426,10 +521,25 @@ "hidden": true }, "default_mem_gb_per_cpu": { - "type": "number", + "type": "integer", "description": "Default memory for CPU core. This could be used to computed the required memory if only the CPUs are specified.", "default": 15, "hidden": true + }, + "user_id": { + "type": "string", + "description": "User running docker/podman container. It defaults to the current user's id", + "hidden": true + }, + "group_id": { + "type": "string", + "description": "Group running docker/podman container. It defaults to the current user's id", + "hidden": true + }, + "singularity_cache_dir": { + "type": "string", + "description": "Singularity container cache directory", + "hidden": true } } } @@ -444,6 +554,12 @@ { "$ref": "#/$defs/segmentation_options" }, + { + "$ref": "#/$defs/zarr_options" + }, + { + "$ref": "#/$defs/merge_labels_options" + }, { "$ref": "#/$defs/multiscale_options" }, diff --git a/ro-crate-metadata.json b/ro-crate-metadata.json index 18815f7..165c87a 100644 --- a/ro-crate-metadata.json +++ b/ro-crate-metadata.json @@ -23,7 +23,7 @@ "@type": "Dataset", "creativeWorkStatus": "InProgress", "datePublished": "2025-08-12T17:34:08+00:00", - "description": "# JaneliaSciComp/nf-cellpose\n\n[![GitHub Actions CI Status](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/nf-test.yml/badge.svg)](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/nf-test.yml)\n[![GitHub Actions Linting Status](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml/badge.svg)](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml)[![Cite with Zenodo](http://img.shields.io/badge/DOI-10.5281/zenodo.XXXXXXX-1073c8?labelColor=000000)](https://doi.org/10.5281/zenodo.XXXXXXX)\n[![nf-test](https://img.shields.io/badge/unit_tests-nf--test-337ab7.svg)](https://www.nf-test.com)\n\n[![Nextflow](https://img.shields.io/badge/version-%E2%89%A524.10.5-green?style=flat&logo=nextflow&logoColor=white&color=%230DC09D&link=https%3A%2F%2Fnextflow.io)](https://www.nextflow.io/)\n[![nf-core template version](https://img.shields.io/badge/nf--core_template-3.3.2-green?style=flat&logo=nfcore&logoColor=white&color=%2324B064&link=https%3A%2F%2Fnf-co.re)](https://github.com/nf-core/tools/releases/tag/3.3.2)\n[![run with conda](http://img.shields.io/badge/run%20with-conda-3EB049?labelColor=000000&logo=anaconda)](https://docs.conda.io/en/latest/)\n[![run with docker](https://img.shields.io/badge/run%20with-docker-0db7ed?labelColor=000000&logo=docker)](https://www.docker.com/)\n[![run with singularity](https://img.shields.io/badge/run%20with-singularity-1d355c.svg?labelColor=000000)](https://sylabs.io/docs/)\n[![Launch on Seqera Platform](https://img.shields.io/badge/Launch%20%F0%9F%9A%80-Seqera%20Platform-%234256e7)](https://cloud.seqera.io/launch?pipeline=https://github.com/JaneliaSciComp/nf-cellpose)\n\n## Introduction\n\n**JaneliaSciComp/nf-cellpose** is a bioinformatics pipeline that ...\n\n\n\n\n1. Read QC ([`FastQC`](https://www.bioinformatics.babraham.ac.uk/projects/fastqc/))2. Present QC for raw reads ([`MultiQC`](http://multiqc.info/))\n\n## Usage\n\n> [!NOTE]\n> If you are new to Nextflow and nf-core, please refer to [this page](https://nf-co.re/docs/usage/installation) on how to set-up Nextflow. Make sure to [test your setup](https://nf-co.re/docs/usage/introduction#how-to-run-a-pipeline) with `-profile test` before running the workflow on actual data.\n\n\n\nNow, you can run the pipeline using:\n\n\n\n```bash\nnextflow run JaneliaSciComp/nf-cellpose \\\n -profile \\\n --input samplesheet.csv \\\n --outdir \n```\n\n> [!WARNING]\n> Please provide pipeline parameters via the CLI or Nextflow `-params-file` option. Custom config files including those provided by the `-c` Nextflow option can be used to provide any configuration _**except for parameters**_; see [docs](https://nf-co.re/docs/usage/getting_started/configuration#custom-configuration-files).\n\n## Credits\n\nJaneliaSciComp/nf-cellpose was originally written by Cristian Goina.\n\nWe thank the following people for their extensive assistance in the development of this pipeline:\n\n\n\n## Contributions and Support\n\nIf you would like to contribute to this pipeline, please see the [contributing guidelines](.github/CONTRIBUTING.md).\n\n## Citations\n\n\n\n\n\n\nAn extensive list of references for the tools used by the pipeline can be found in the [`CITATIONS.md`](CITATIONS.md) file.\n\nThis pipeline uses code and infrastructure developed and maintained by the [nf-core](https://nf-co.re) community, reused here under the [MIT license](https://github.com/nf-core/tools/blob/main/LICENSE).\n\n> **The nf-core framework for community-curated bioinformatics pipelines.**\n>\n> Philip Ewels, Alexander Peltzer, Sven Fillinger, Harshil Patel, Johannes Alneberg, Andreas Wilm, Maxime Ulysse Garcia, Paolo Di Tommaso & Sven Nahnsen.\n>\n> _Nat Biotechnol._ 2020 Feb 13. doi: [10.1038/s41587-020-0439-x](https://dx.doi.org/10.1038/s41587-020-0439-x).\n", + "description": "# JaneliaSciComp/nf-cellpose\n\n[![GitHub Actions CI Status](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/nf-test.yml/badge.svg)](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/nf-test.yml)\n[![GitHub Actions Linting Status](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml/badge.svg)](https://github.com/JaneliaSciComp/nf-cellpose/actions/workflows/linting.yml)[![Cite with Zenodo](http://img.shields.io/badge/DOI-10.5281/zenodo.XXXXXXX-1073c8?labelColor=000000)](https://doi.org/10.5281/zenodo.XXXXXXX)\n[![nf-test](https://img.shields.io/badge/unit_tests-nf--test-337ab7.svg)](https://www.nf-test.com)\n\n[![Nextflow](https://img.shields.io/badge/version-%E2%89%A525.10.0-green?style=flat&logo=nextflow&logoColor=white&color=%230DC09D&link=https%3A%2F%2Fnextflow.io)](https://www.nextflow.io/)\n[![nf-core template version](https://img.shields.io/badge/nf--core_template-3.3.2-green?style=flat&logo=nfcore&logoColor=white&color=%2324B064&link=https%3A%2F%2Fnf-co.re)](https://github.com/nf-core/tools/releases/tag/3.3.2)\n[![run with conda](http://img.shields.io/badge/run%20with-conda-3EB049?labelColor=000000&logo=anaconda)](https://docs.conda.io/en/latest/)\n[![run with docker](https://img.shields.io/badge/run%20with-docker-0db7ed?labelColor=000000&logo=docker)](https://www.docker.com/)\n[![run with singularity](https://img.shields.io/badge/run%20with-singularity-1d355c.svg?labelColor=000000)](https://sylabs.io/docs/)\n[![Launch on Seqera Platform](https://img.shields.io/badge/Launch%20%F0%9F%9A%80-Seqera%20Platform-%234256e7)](https://cloud.seqera.io/launch?pipeline=https://github.com/JaneliaSciComp/nf-cellpose)\n\n## Introduction\n\n**JaneliaSciComp/nf-cellpose** is a pipeline that runs Cellpose segmentation on a Dask cluster.\n\nThe pipeline generates the image segmentation using [Cellpose](https://github.com/MouseLand/cellpose). Large images can be partitioned into smaller blocks and processed on a distributed Dask cluster. The input can be a single (Tiff) image file or single (OME-ZARR) image container, or a directory containing multiple input files or containers that need to be segmented. In the latter case when multiple inputs need to be segmented, the user must also provide an input pattern, used for selecting the images. The pipeline output is the Cellpose result saved as a Tiff or an OME-ZARR (depending on the input) at the location specified by the `outdir` parameter. For OME-ZARR, there's also an option to generate the multiscale pyramid for the labels image.\n\n\n\n## Usage\n\n> [!NOTE]\n> If you are new to Nextflow and nf-core, please refer to [this page](https://nf-co.re/docs/usage/installation) on how to set-up Nextflow. Make sure to [test your setup](https://nf-co.re/docs/usage/introduction#how-to-run-a-pipeline) with `-profile test` before running the workflow on actual data.\n\nNow, you can run the pipeline using:\n\n```bash\nnextflow run JaneliaSciComp/nf-cellpose \\\n -profile \\\n --input input.ome.zarr \\\n --outdir \n```\n\nTo run with conda:\n```bash\nnextflow run JaneliaSciComp/nf-cellpose \\\n -profile conda -with-conda conf/conda-env.yml \\\n --input input.ome.zarr \\\n --outdir \n```\n\n> [!WARNING]\n> Please provide pipeline parameters via the CLI or Nextflow `-params-file` option; an example of such file is available [here](examples/segtest_sample.json). Custom config files including those provided by the `-c` Nextflow option can be used to provide any configuration _**except for parameters**_; see [docs](https://nf-co.re/docs/usage/getting_started/configuration#custom-configuration-files).\n\n## Credits\n\n[Cellpose-SAM](https://github.com/MouseLand/cellpose) used in this pipeline was developed by Marius Pachitariu, Michael Rariden and Carsen Stringer\n\nThe distributed Cellpose was the contributed to [cellpose](https://github.com/MouseLand/cellpose/pull/408) by Greg Fleishman\n\nJaneliaSciComp/nf-cellpose was originally written by Cristian Goina.\n\nWe thank the following people for their extensive assistance in the development of this pipeline:\n\n\n## Contributions and Support\n\nIf you would like to contribute to this pipeline, please see the [contributing guidelines](.github/CONTRIBUTING.md).\n\n## Citations\n\nAn extensive list of references for the tools used by the pipeline can be found in the [`CITATIONS.md`](CITATIONS.md) file.\n\nThis pipeline uses code and infrastructure developed and maintained by the [nf-core](https://nf-co.re) community, reused here under the [MIT license](https://github.com/nf-core/tools/blob/main/LICENSE).\n\n> **The nf-core framework for community-curated bioinformatics pipelines.**\n>\n> Philip Ewels, Alexander Peltzer, Sven Fillinger, Harshil Patel, Johannes Alneberg, Andreas Wilm, Maxime Ulysse Garcia, Paolo Di Tommaso & Sven Nahnsen.\n>\n> _Nat Biotechnol._ 2020 Feb 13. doi: [10.1038/s41587-020-0439-x](https://dx.doi.org/10.1038/s41587-020-0439-x).\n", "hasPart": [ { "@id": "main.nf" diff --git a/subworkflows/janelia/dask_start/main.nf b/subworkflows/janelia/dask_start/main.nf index 52470c4..91df848 100644 --- a/subworkflows/janelia/dask_start/main.nf +++ b/subworkflows/janelia/dask_start/main.nf @@ -1,3 +1,4 @@ +nextflow.enable.moduleBinaries = true workflow DASK_START { take: @@ -7,11 +8,16 @@ workflow DASK_START { dask_work_path // dask work directory total_workers // int: number of total workers in the cluster required_workers // int: number of required workers in the cluster - dask_worker_cpus // int: number of cores per worker + dask_worker_cpus // int: number of cpus per worker dask_worker_mem_gb // int: worker memory in GB main: - if (distributed) { + if (Boolean.valueOf(distributed)) { + // Helper scripts staged into each task work dir so they are available both + // for local/conda execution and inside containers (docker/podman/singularity). + // Staging guarantees the pipeline's own copy is used even if the image's + // bundled scripts have diverged. + // prepare dask cluster work dir meta -> [ meta, cluster_work_dir ] def dask_work_dir if (dask_work_path) { @@ -27,8 +33,8 @@ workflow DASK_START { meta_and_files, dask_work_dir, ) - | join(meta_and_files, by: 0) - | map { + .join(meta_and_files, by: 0) + .map { it -> def (meta, dask_cluster_work_dir, data_paths) = it def dask_config_path = dask_config ? file(dask_config) : [] def r = [meta, dask_config_path, dask_cluster_work_dir, data_paths] @@ -37,22 +43,24 @@ workflow DASK_START { } // start scheduler - DASK_STARTMANAGER(dask_prepare_result) + DASK_STARTMANAGER( + dask_prepare_result, + ) // wait for manager to start - DASK_WAITFORMANAGER( - dask_prepare_result.map { - def (meta, dask_config_path, dask_cluster_work_dir, data_paths) = it + def wait_manager_results = DASK_WAITFORMANAGER( + dask_prepare_result.map { it -> + def (meta, _dask_config_path, dask_cluster_work_dir, _data_paths) = it [meta, dask_cluster_work_dir] - } + }, ) - def nworkers = total_workers ?: 1 + def nworkers = (total_workers as int) ?: 1 // prepare inputs for dask workers - def dask_workers_input = DASK_WAITFORMANAGER.out.cluster_info - | join(meta_and_files, by: 0) - | flatMap { + def dask_workers_input = wait_manager_results.cluster_info + .join(meta_and_files, by: 0) + .flatMap { it -> def (meta, cluster_work_dir, scheduler_address, dashboard_port, data_paths) = it def dashboard_address_parts = scheduler_address.split(':') dashboard_address_parts[0] = 'http' @@ -71,21 +79,21 @@ workflow DASK_START { // start dask workers DASK_STARTWORKER( dask_workers_input, - dask_worker_cpus, - dask_worker_mem_gb, + (dask_worker_cpus as int), + (dask_worker_mem_gb as int), ) // check dask workers def cluster = DASK_WAITFORWORKERS( - DASK_WAITFORMANAGER.out.cluster_info, + wait_manager_results.cluster_info, nworkers, - required_workers ?: 1, + (required_workers as int) ?: 1, ) dask_context = cluster.cluster_info - | map { + .map { it -> def (meta, cluster_work_dir, scheduler_address, available_workers) = it - dask_info = [ + def dask_info = [ scheduler_address: scheduler_address, cluster_work_dir: cluster_work_dir, available_workers: available_workers, @@ -98,8 +106,8 @@ workflow DASK_START { // do not start a distributed cluster log.debug("No distributed dask cluster") dask_context = meta_and_files - | map { - def (meta, data_paths) = it + .map { it -> + def (meta, _data_paths) = it [meta, [:]] } } @@ -109,82 +117,47 @@ workflow DASK_START { } process DASK_PREPARE { + tag "${meta.id}" label 'process_single' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' } + container 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' input: tuple val(meta), path(data, stageAs: '?/*') path dask_work_dir, stageAs: 'dask_work/*' output: - tuple val(meta), env(cluster_work_fullpath) + tuple val(meta), env('cluster_work_fullpath') when: task.ext.when == null || task.ext.when script: - """ - if [[ "${dask_work_dir}" == "" ]]; then - dwork="dask-\$(date -I)" - mkdir -p \${dwork} - cluster_work_dir=\$(readlink -m \${dwork}) - else - cluster_work_dir=\$(readlink -m ${dask_work_dir}) - fi - cluster_work_fullpath="\${cluster_work_dir}/${meta.id}" - /opt/scripts/daskscripts/prepare.sh "\${cluster_work_fullpath}" - echo "Cluster work dir: \${cluster_work_fullpath}" - """ + template 'prepare.sh' } process DASK_STARTMANAGER { + tag "${meta.id}" label 'process_long' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' } + container 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' input: tuple val(meta), path(dask_config), path(cluster_work_dir, stageAs: 'dask_work/*'), path(data, stageAs: '?/*') output: - tuple val(meta), env(cluster_work_fullpath), emit: cluster_info + tuple val(meta), val(cluster_work_dir), emit: cluster_info path "versions.yml", emit: versions when: task.ext.when == null || task.ext.when script: - def args = task.ext.args ?: '' - def container_engine = workflow.containerEngine - - def set_dask_config_env = dask_config ? "export DASK_CONFIG=\$(readlink ${dask_config})" : '' - def dask_scheduler_pid_file = "${cluster_work_dir}/dask-scheduler.pid" - def dask_scheduler_info_file = "${cluster_work_dir}/dask-scheduler-info.json" - def terminate_file_name = "${cluster_work_dir}/terminate-dask" - - """ - cluster_work_fullpath=\$(readlink ${cluster_work_dir}) - ${set_dask_config_env} - - echo "Scheduler's environment" - env - - /opt/scripts/daskscripts/startmanager.sh \ - --container-engine ${container_engine} \ - --pid-file ${dask_scheduler_pid_file} \ - --scheduler-work-dir ${cluster_work_dir} \ - --scheduler-file ${dask_scheduler_info_file} \ - --terminate-file ${terminate_file_name} \ - ${args} - - dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//" ) - cat <<-END_VERSIONS > versions.yml - "dask": \${dask_version} - END_VERSIONS - """ + template 'startmanager.sh' } process DASK_STARTWORKER { + tag "${meta.id}:${worker_id}" label 'process_long' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' } + container 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' cpus { worker_cpus } memory "${worker_mem_in_gb} GB" @@ -194,100 +167,42 @@ process DASK_STARTWORKER { val worker_mem_in_gb output: - tuple val(meta), env(cluster_work_fullpath), val(scheduler_address), emit: cluster_info + tuple val(meta), val(cluster_work_dir), val(scheduler_address), emit: cluster_info path "versions.yml", emit: versions when: task.ext.when == null || task.ext.when script: - def args = task.ext.args ?: '' - def container_engine = workflow.containerEngine - - def set_dask_config_env = dask_config ? "export DASK_CONFIG=\$(readlink ${dask_config})" : '' - def dask_worker_name = "worker-${worker_id}" - def dask_scheduler_info_file = "${cluster_work_dir}/dask-scheduler-info.json" - def terminate_file_name = "${cluster_work_dir}/terminate-dask" - - def dask_worker_work_dir = "${cluster_work_dir}/${dask_worker_name}" - def dask_worker_pid_file = "${dask_worker_work_dir}/${dask_worker_name}.pid" - - worker_name = dask_worker_name - worker_dir = dask_worker_work_dir - - """ - cluster_work_fullpath=\$(readlink ${cluster_work_dir}) - - ${set_dask_config_env} - echo "Worker's environment" - env - - /opt/scripts/daskscripts/startworker.sh \ - --container-engine ${container_engine} \ - --name ${dask_worker_name} \ - --worker-dir ${dask_worker_work_dir} \ - --scheduler-address ${scheduler_address} \ - --pid-file ${dask_worker_pid_file} \ - --memory-limit "${worker_mem_in_gb}G" \ - --terminate-file ${terminate_file_name} \ - ${args} - - dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//" ) - cat <<-END_VERSIONS > versions.yml - "dask": \${dask_version} - END_VERSIONS - """ + template 'startworker.sh' } process DASK_WAITFORMANAGER { + tag "${meta.id}" label 'process_single' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' } + container 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' input: tuple val(meta), path(cluster_work_dir, stageAs: 'dask_work/*') output: tuple val(meta), - env(cluster_work_fullpath), - env(scheduler_address), - env(dashboard_port), emit: cluster_info + env('cluster_work_fullpath'), + env('scheduler_address'), + env('dashboard_port'), emit: cluster_info path "versions.yml", emit: versions when: task.ext.when == null || task.ext.when script: - def args = task.ext.args ?: '' - def dask_scheduler_info_file = "${cluster_work_dir}/dask-scheduler-info.json" - def terminate_file_name = "${cluster_work_dir}/terminate-dask" - - """ - cluster_work_fullpath=\$(readlink ${cluster_work_dir}) - - /opt/scripts/daskscripts/waitformanager.sh \ - --flist "${dask_scheduler_info_file},${terminate_file_name}" \ - ${args} - - if [[ -e "${dask_scheduler_info_file}" ]] ; then - echo "\$(date): Get cluster info from ${dask_scheduler_info_file}" - scheduler_address=\$(jq ".address" ${dask_scheduler_info_file}) - dashboard_port=\$(jq ".services.dashboard" ${dask_scheduler_info_file}) - else - echo "\$(date): Cluster info file ${dask_scheduler_info_file} not found" - scheduler_address= - dashboard_port= - fi - - dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//" ) - cat <<-END_VERSIONS > versions.yml - "dask": \${dask_version} - END_VERSIONS - """ + template 'waitformanager.sh' } process DASK_WAITFORWORKERS { + tag "${meta.id}" label 'process_single' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' } + container 'ghcr.io/janeliascicomp/dask:2025.5.1-py12-ol9' input: tuple val(meta), path(cluster_work_dir, stageAs: 'dask_work/*'), val(scheduler_address), val(dashboard_port) @@ -295,35 +210,13 @@ process DASK_WAITFORWORKERS { val required_workers output: - tuple val(meta), env(cluster_work_fullpath), val(scheduler_address), env(available_workers), emit: cluster_info + tuple val(meta), env('cluster_work_fullpath'), val(scheduler_address), env('available_workers'), emit: cluster_info path "versions.yml", emit: versions when: task.ext.when == null || task.ext.when script: - def args = task.ext.args ?: '' - def container_engine = workflow.containerEngine - - def terminate_file_name = "${cluster_work_dir}/terminate-dask" - - cluster_work_fullpath = cluster_work_dir.resolveSymLink().toString() - - """ - cluster_work_fullpath=\$(readlink ${cluster_work_dir}) - - # waitforworkers.sh sets available_workers variable - . /opt/scripts/daskscripts/waitforworkers.sh \ - --cluster-work-dir ${cluster_work_dir} \ - --scheduler-address ${scheduler_address} \ - --total-workers ${total_workers} \ - --required-workers ${required_workers} \ - --terminate-file ${terminate_file_name} \ - ${args} - - dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//" ) - cat <<-END_VERSIONS > versions.yml - "dask": \${dask_version} - END_VERSIONS - """ + template 'waitforworkers.sh' } + diff --git a/subworkflows/janelia/dask_start/meta.yml b/subworkflows/janelia/dask_start/meta.yml index 9161b9d..e494212 100644 --- a/subworkflows/janelia/dask_start/meta.yml +++ b/subworkflows/janelia/dask_start/meta.yml @@ -5,5 +5,44 @@ keywords: - dask - infrastructure - distributed -components: - - dask_cluster +input: + - meta_and_files: + type: tuple + description: | + Channel containing metadata and files that need to be accessible by the cluster. + Structure: [ val(meta), path(files)... ] + - distributed: + type: boolean + description: If true, create a distributed Dask cluster; if false, return empty context + - dask_config: + type: file + description: Path to Dask configuration file (optional) + - dask_work_path: + type: directory + description: Path to Dask work directory where cluster files will be stored (optional) + - total_workers: + type: integer + description: Number of total workers to start in the cluster + - required_workers: + type: integer + description: Minimum number of workers required before the cluster is considered ready + - dask_worker_cpus: + type: integer + description: Number of CPU cores allocated per worker + - dask_worker_mem_gb: + type: integer + description: Memory in GB allocated per worker +output: + - dask_context: + type: tuple + description: | + Dask cluster context information. If distributed=true, contains cluster details; + if distributed=false, contains empty map. + Structure: [ val(meta), val(dask_info) ] + Where dask_info is a map containing: + - scheduler_address: Address of the Dask scheduler + - cluster_work_dir: Path to the cluster work directory + - available_workers: Number of workers that joined the cluster +authors: + - "@krokicki" + - "@cgoina" diff --git a/subworkflows/janelia/dask_start/resources/usr/bin/determine_ip.sh b/subworkflows/janelia/dask_start/resources/usr/bin/determine_ip.sh new file mode 100755 index 0000000..c3f0b1a --- /dev/null +++ b/subworkflows/janelia/dask_start/resources/usr/bin/determine_ip.sh @@ -0,0 +1,37 @@ +#!/bin/bash -ue +# Determine the IP address of the current host + +container_engine=$1 + +local_ip= +if command -v ipconfig >/dev/null 2>&1; then + echo "Use ipconfig to get the IP" + # macOS: resolve the interface of the default route, then ask for its IPv4 + iface=$(route -n get default 2>/dev/null | awk '/interface:/{print $2}') + local_ip=$(ipconfig getifaddr "$iface") +elif command -v ip >/dev/null 2>&1; then + echo "Use ip to get the IP" + # Linux fallback if hostname -I unavailable + local_ip=$(ip -4 -o addr show scope global | awk '{print $4}' | cut -d/ -f1 | tail -1) +else + # Last resort + echo "Use ifconfig to get the IP" + local_ip=$(ifconfig 2>/dev/null | awk '/inet /&&$2!="127.0.0.1"{print $2; exit}') +fi +if [[ -z "${local_ip}" ]] ; then + if [ "$container_engine" = "docker" ]; then + for interface in /sys/class/net/{eth*,en*,em*}; do + [ -e $interface ] && \ + [ `cat $interface/operstate` == "up" ] && \ + local_ip=$(ifconfig `basename $interface` | grep "inet " | awk '$1=="inet" {print $2; exit}' | sed s/addr://g) + if [[ "$local_ip" != "" ]]; then + echo "Use IP: $local_ip" + break + fi + done + fi + if [[ "$local_ip" == "" ]]; then + echo "Could not determine local IP: local_ip is empty" + exit 1 + fi +fi diff --git a/subworkflows/janelia/dask_start/resources/usr/bin/waitforanyfile.sh b/subworkflows/janelia/dask_start/resources/usr/bin/waitforanyfile.sh new file mode 100755 index 0000000..0ec8486 --- /dev/null +++ b/subworkflows/janelia/dask_start/resources/usr/bin/waitforanyfile.sh @@ -0,0 +1,43 @@ +#!/bin/bash -ue +# Wait for the specified file to be created +# +declare apid=$1 +declare fs_string=$2 +declare -i wait_timeout=${3:--1} # default to no timeout +declare -i polling_interval=${4:-5} # default to 5s + +IFS=',' read -ra fs_list <<< "${fs_string}" + +if (( ${wait_timeout} > 0 )); then + echo "Check for process=${apid} and files: ${fs_list[@]} with a timeout of ${wait_timeout} seconds (polling interval = ${polling_interval}s)" +else + echo "Check for process=${apid} and files: ${fs_list[@]} with NO timeout (polling interval = ${polling_interval}s)" +fi + +declare -i seconds=0 +while true ; do + + if [[ "${apid}" != "0" ]] ; then + # if apid is set check that the process is alive + if ! kill -0 $apid >/dev/null 2>&1 ; then + echo "Process $apid died" + exit 1 + fi + fi + + # check timeout + if (( ${wait_timeout} > 0 && ${seconds} > ${wait_timeout} )); then + echo "Timed out after ${seconds} seconds while waiting for ${fs_list[@]}" + exit 2 + fi + + for f in "${fs_list[@]}" ; do + if [[ -e "${f}" ]] ; then + echo "Found ${f}" + exit 0 + fi + done + + sleep $polling_interval + seconds=$(( ${seconds} + ${polling_interval} )) +done diff --git a/subworkflows/janelia/dask_start/templates/prepare.sh b/subworkflows/janelia/dask_start/templates/prepare.sh new file mode 100755 index 0000000..dbf157a --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/prepare.sh @@ -0,0 +1,26 @@ +#!/bin/bash -ue +# Prepare the Dask cluster work directory. + +case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; +esac + +if [[ "${dask_work_dir}" == "" ]]; then + dwork="dask-\$(date -I)" + mkdir -p \${dwork} + cluster_work_dir=\$(\${READLINK_TOOL} -m \${dwork}) +else + cluster_work_dir=\$(\${READLINK_TOOL} -m ${dask_work_dir}) +fi +cluster_work_fullpath="\${cluster_work_dir}/${meta.id}" + +if [[ ! -d "\${cluster_work_fullpath}" ]]; then + echo "Creating work directory: \${cluster_work_fullpath}" + mkdir -p "\${cluster_work_fullpath}" +else + echo "Cleaning existing work directory: \${cluster_work_fullpath}" + rm -rf \${cluster_work_fullpath}/* || true +fi + +echo "Cluster work dir: \${cluster_work_fullpath}" diff --git a/subworkflows/janelia/dask_start/templates/startmanager.sh b/subworkflows/janelia/dask_start/templates/startmanager.sh new file mode 100755 index 0000000..09874d4 --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/startmanager.sh @@ -0,0 +1,85 @@ +#!/bin/bash -ue +# Start the Dask scheduler and wait for terminate signal. + +case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; +esac +cluster_work_fullpath=\$(\${READLINK_TOOL} ${cluster_work_dir}) +${dask_config ? 'export DASK_CONFIG=$(${READLINK_TOOL} ' + dask_config + ')' : ''} +args="${task.ext.args ?: ''}" +scheduler_pid_file="\${cluster_work_fullpath}/dask-scheduler.pid" +scheduler_info_file="\${cluster_work_fullpath}/dask-scheduler-info.json" +terminate_file_name="\${cluster_work_fullpath}/terminate-dask" + +echo "Scheduler's environment" +env + +manager_exit_code=0 + +# cleanup running process +function cleanup() { + if [[ -f "\${scheduler_pid_file}" ]]; then + local dpid + dpid=\$(cat "\${scheduler_pid_file}" || true) + if [[ -n "\${dpid}" ]] && kill -0 "\${dpid}" 2>/dev/null; then + kill -TERM "\${dpid}" 2>/dev/null || true + local i=0 + while (( i < 10 )) && kill -0 "\${dpid}" 2>/dev/null; do + sleep 1 + i=\$(( i + 1 )) + done + kill -9 "\${dpid}" 2>/dev/null || true + fi + fi +} + +# INT/TERM: just log; the waitforanyfile.sh calls use || true so the script +# continues to cleanup and the Nextflow epilogue naturally. +function on_term() { + echo "Received termination signal, stopping scheduler" +} + +# EXIT: runs after the epilogue and versions.yml have been written. +function on_exit() { + cleanup + exit \${manager_exit_code} +} + +trap on_term INT TERM +trap on_exit EXIT + +echo "Determining scheduler IP address..." +. determine_ip.sh ${workflow.containerEngine} + +# start scheduler in background +echo "Run: dask scheduler --host \${local_ip} --pid-file \${scheduler_pid_file} --scheduler-file \${scheduler_info_file} \${args}" +dask scheduler \ + --host \${local_ip} \ + --pid-file \${scheduler_pid_file} \ + --scheduler-file \${scheduler_info_file} \ + \${args} \ + 2> >(tee \${cluster_work_fullpath}/dask-scheduler.log >&2) \ + & + +# wait for PID file (or terminate signal); || true so a signal-killed +# subprocess does not trip set -e before the epilogue. +waitforanyfile.sh 0 "\${terminate_file_name},\${scheduler_pid_file}" || true + +if [[ -e "\${scheduler_pid_file}" ]]; then + scheduler_pid=\$(cat "\${scheduler_pid_file}") + echo "Scheduler started: pid=\$scheduler_pid" + echo "Wait for termination event: \${terminate_file_name}" + waitforanyfile.sh \${scheduler_pid} "\${terminate_file_name}" || true +else + echo "Scheduler pid file not found" + scheduler_pid=0 +fi + +echo "Killing dask scheduler" +cleanup + +dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//") +cat <<-END_VERSIONS > versions.yml +"dask": \${dask_version} +END_VERSIONS diff --git a/subworkflows/janelia/dask_start/templates/startworker.sh b/subworkflows/janelia/dask_start/templates/startworker.sh new file mode 100755 index 0000000..a0adafe --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/startworker.sh @@ -0,0 +1,92 @@ +#!/bin/bash -ue +# Start a Dask worker and wait for terminate signal. + +case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; +esac +cluster_work_fullpath=\$(\${READLINK_TOOL} ${cluster_work_dir}) +${dask_config ? 'export DASK_CONFIG=$(${READLINK_TOOL} ' + dask_config + ')' : ''} +args="${task.ext.args ?: ''}" +worker_name="worker-${worker_id}" +worker_dir="\${cluster_work_fullpath}/\${worker_name}" +worker_pid_file="\${worker_dir}/\${worker_name}.pid" +terminate_file_name="\${cluster_work_fullpath}/terminate-dask" +memory_limit="${worker_mem_in_gb / worker_cpus * (worker_cpus + task.attempt - 1) as int}G" + +echo "Worker's environment" +env + +worker_exit_code=0 + +# cleanup running process +function cleanup() { + if [[ -f "\${worker_pid_file}" ]]; then + local wpid + wpid=\$(cat "\${worker_pid_file}" || true) + if [[ -n "\${wpid}" ]] && kill -0 "\${wpid}" 2>/dev/null; then + kill -TERM "\${wpid}" 2>/dev/null || true + local i=0 + while (( i < 10 )) && kill -0 "\${wpid}" 2>/dev/null; do + sleep 1 + i=\$(( i + 1 )) + done + kill -9 "\${wpid}" 2>/dev/null || true + fi + fi +} + +# INT/TERM: just log; the waitforanyfile.sh calls use || true so the script +# continues to cleanup and the Nextflow epilogue naturally. +function on_term() { + echo "Received termination signal, stopping worker \${worker_name}" +} + +# EXIT: runs after the epilogue and versions.yml have been written. +function on_exit() { + cleanup + exit \${worker_exit_code} +} + +trap on_term INT TERM +trap on_exit EXIT + +echo "Determining worker \${worker_name} IP address..." +. determine_ip.sh ${workflow.containerEngine} + +mkdir -p \${worker_dir} + +# start worker in background +echo "Run: dask worker --host \${local_ip} --local-directory \${worker_dir} --name \${worker_name} --pid-file \${worker_pid_file} --memory-limit \${memory_limit} \${args} ${scheduler_address}" +dask worker \ + --host \${local_ip} \ + --local-directory \${worker_dir} \ + --name \${worker_name} \ + --pid-file \${worker_pid_file} \ + --memory-limit \${memory_limit} \ + \${args} \ + ${scheduler_address} \ + 2> >(tee \${worker_dir}/\${worker_name}.log >&2) \ + & + +# wait for PID file (or terminate signal); || true so a signal-killed +# subprocess does not trip set -e before the epilogue. +waitforanyfile.sh 0 "\${terminate_file_name},\${worker_pid_file}" || true + +if [[ -e "\${worker_pid_file}" ]]; then + worker_pid=\$(cat "\${worker_pid_file}") + echo "Worker \${worker_name} started: pid=\$worker_pid" + echo "Worker \${worker_name} - wait for termination event: \${terminate_file_name}" + waitforanyfile.sh \${worker_pid} "\${terminate_file_name}" || true +else + echo "Worker \${worker_name} pid file not found" + worker_pid=0 +fi + +echo "Killing background processes for \${worker_name}" +cleanup + +dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//") +cat <<-END_VERSIONS > versions.yml +"dask": \${dask_version} +END_VERSIONS diff --git a/subworkflows/janelia/dask_start/templates/waitformanager.sh b/subworkflows/janelia/dask_start/templates/waitformanager.sh new file mode 100755 index 0000000..7957d68 --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/waitformanager.sh @@ -0,0 +1,27 @@ +#!/bin/bash -ue +# Wait for the Dask scheduler info file to appear, then extract the address/port. + +case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; +esac +cluster_work_fullpath=\$(\${READLINK_TOOL} ${cluster_work_dir}) +scheduler_info_file="\${cluster_work_fullpath}/dask-scheduler-info.json" +terminate_file_name="\${cluster_work_fullpath}/terminate-dask" + +waitforanyfile.sh 0 "\${scheduler_info_file},\${terminate_file_name}" ${task.ext.scheduler_start_timeout ?: '60'} ${task.ext.scheduler_poll_interval ?: '2'} + +if [[ -e "\${scheduler_info_file}" ]]; then + echo "\$(date): Get cluster info from \${scheduler_info_file}" + scheduler_address=\$(jq ".address" \${scheduler_info_file}) + dashboard_port=\$(jq ".services.dashboard" \${scheduler_info_file}) +else + echo "\$(date): Cluster info file \${scheduler_info_file} not found" + scheduler_address= + dashboard_port= +fi + +dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//") +cat <<-END_VERSIONS > versions.yml +"dask": \${dask_version} +END_VERSIONS diff --git a/subworkflows/janelia/dask_start/templates/waitforworker.sh b/subworkflows/janelia/dask_start/templates/waitforworker.sh new file mode 100755 index 0000000..8211559 --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/waitforworker.sh @@ -0,0 +1,72 @@ +#!/bin/bash -ue +# Start the Dask scheduler process and wait for terminate signal +DIR=$(cd "$(dirname "$0")"; pwd) + +scheduler_address= +cluster_work_dir= +worker_id= +terminate_file= +declare -i worker_start_timeout=-1 +declare -i worker_poll_interval=1 + +while [[ $# > 0 ]]; do + key="$1" + shift # past the key + case $key in + --cluster-work-dir) + cluster_work_dir=$1 + shift + ;; + --worker-start-timeout) + worker_start_timeout=$(($1)) + shift + ;; + --worker-poll-interval) + scheduler_poll_interval=$(($1)) + shift + ;; + --worker-id) + worker_id=$1 + shift + ;; + --terminate-file) + terminate_file=$1 + shift + ;; + --scheduler-address) + scheduler_address=$1 + shift + ;; + *) + ;; + esac +done + +seconds=0 +worker_name="worker-${worker_id}" +worker_log="${cluster_work_dir}/${worker_name}/${worker_name}.log" + +echo "Worker ${worker_name} log: ${worker_log}" +while true; do + if [[ -e ${terminate_file} ]] ; then + # this can happen if the cluster is created on LSF and the workers cannot get nodes + # before the cluster is ended + echo "Termination signal: ${terminate_file} found while waiting for ${worker_name}" + exit 1 + fi + # if worker's log exists check if the worker has connected to the scheduler + if [[ -e "${worker_log}" ]]; then + found=`grep -o "Registered to:.*${scheduler_address}" ${worker_log} || true` + if [[ ! -z ${found} ]]; then + echo "${found}" + echo "Finished waiting for ${worker_name} to connect" + exit 0; + fi + fi + if (( ${worker_start_timeout} > 0 && ${seconds} > ${worker_start_timeout} )); then + echo "Timed out after ${seconds} seconds while waiting for ${worker_name} to connect to scheduler" + exit 2 + fi + sleep ${worker_poll_interval} + seconds=$(( ${seconds} + ${worker_poll_interval} )) +done diff --git a/subworkflows/janelia/dask_start/templates/waitforworkers.sh b/subworkflows/janelia/dask_start/templates/waitforworkers.sh new file mode 100755 index 0000000..5a9d22b --- /dev/null +++ b/subworkflows/janelia/dask_start/templates/waitforworkers.sh @@ -0,0 +1,60 @@ +#!/bin/bash -ue +# Wait for the required number of Dask workers to register with the scheduler. + +case \$(uname) in + Darwin) READLINK_TOOL="greadlink" ;; + *) READLINK_TOOL="readlink" ;; +esac +cluster_work_fullpath=\$(\${READLINK_TOOL} ${cluster_work_dir}) +terminate_file_name="\${cluster_work_fullpath}/terminate-dask" + +declare -i worker_start_timeout=${task.ext.worker_start_timeout ?: '-1'} +declare -i worker_poll_interval=${task.ext.worker_poll_interval ?: '1'} +declare -i total_workers=${total_workers} +declare -i required_workers=${required_workers} + +if (( \${required_workers} > 0 )); then + seconds=0 + while true; do + if [[ -e \${terminate_file_name} ]]; then + # this can happen if the cluster is created on LSF and the workers cannot get nodes + # before the cluster is ended + available_workers=-1 + exit 1 + fi + available_workers=0 + for (( worker_id=1; worker_id<=\${total_workers}; worker_id++ )); do + worker_name="worker-\${worker_id}" + worker_log="\${cluster_work_fullpath}/\${worker_name}/\${worker_name}.log" + # if worker's log exists check if the worker has connected to the scheduler + echo "Check \${worker_log}" + if [[ -e "\${worker_log}" ]]; then + found=\$(grep -o "Registered to:.*${scheduler_address}" \${worker_log} || true) + if [[ ! -z \${found} ]]; then + echo "\${found}" + available_workers=\$(( \${available_workers} + 1 )) + fi + fi + done + echo "Found \${available_workers} after \${seconds}" + # in case somebody forgets to adjust the required workers check also if it is equal to total_workers + if (( \${available_workers} >= \${total_workers} || \${available_workers} >= \${required_workers} )); then + echo "Found \${available_workers} connected workers" + break + fi + if (( \${worker_start_timeout} > 0 && \${seconds} > \${worker_start_timeout} )); then + echo "Timed out after \${seconds} seconds while waiting for at least \${required_workers} workers to connect to scheduler" + available_workers=-1 + exit 2 + fi + sleep \${worker_poll_interval} + seconds=\$(( \${seconds} + \${worker_poll_interval} )) + done +else + available_workers=-1 +fi + +dask_version=\$(dask --version | grep version | sed "s/.*version\\s*//") +cat <<-END_VERSIONS > versions.yml +"dask": \${dask_version} +END_VERSIONS diff --git a/subworkflows/janelia/dask_stop/main.nf b/subworkflows/janelia/dask_stop/main.nf index 664cb3d..cb95ffd 100644 --- a/subworkflows/janelia/dask_stop/main.nf +++ b/subworkflows/janelia/dask_stop/main.nf @@ -1,12 +1,12 @@ process DASK_TERMINATE { label 'process_single' - container { task.ext.container ?: 'ghcr.io/janeliascicomp/dask:2024.12.1-py11-ol9' } + container 'ghcr.io/janeliascicomp/dask:2024.12.1-py11-ol9' input: tuple val(meta), path(cluster_work_dir, stageAs: 'dask_work/*') output: - tuple val(meta), env(cluster_work_fullpath) + tuple val(meta), env('cluster_work_fullpath') when: task.ext.when == null || task.ext.when @@ -15,10 +15,20 @@ process DASK_TERMINATE { def cluster_work_path = cluster_work_dir def terminate_file_name = "${cluster_work_path}/terminate-dask" """ - cluster_work_fullpath=\$(readlink ${cluster_work_dir}) + case \$(uname) in + Darwin) + detected_os=OSX + READLINK_TOOL="greadlink" + ;; + *) + detected_os=Linux + READLINK_TOOL="readlink" + ;; + esac + cluster_work_fullpath=\$(\${READLINK_TOOL} ${cluster_work_dir}) echo "\$(date): Terminate DASK Scheduler: ${cluster_work_path}" - echo $PWD + echo \$PWD cat > ${terminate_file_name} < + def cluster_info = meta_and_context + | filter { _meta, dask_context -> // only terminate the clusters that have a work dir dask_context.cluster_work_dir } diff --git a/subworkflows/janelia/multiscale/main.nf b/subworkflows/janelia/multiscale/main.nf new file mode 100644 index 0000000..9ebaa1f --- /dev/null +++ b/subworkflows/janelia/multiscale/main.nf @@ -0,0 +1,29 @@ +include { OMEZARRTOOLS_MULTISCALE } from '../../../modules/janelia/omezarrtools/multiscale' + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + RUN MULTISCALE +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +workflow MULTISCALE { + take: + ch_meta // ch: [ meta, img_container, img_dataset ] + ch_dask_info // ch: [ dask_scheduler, dask_config ] + generate_pyramid // boolean + + main: + if (generate_pyramid) { + multiscale_results = OMEZARRTOOLS_MULTISCALE( + ch_meta, + ch_dask_info, + ).data + multiscale_results.view { it -> log.debug "Multiscale pyramid: $it" } + } else { + multiscale_results = ch_meta + multiscale_results.view { it -> log.debug "Skip multiscale pyramid: $it" } + } + + emit: + multiscale_results +} diff --git a/subworkflows/local/multiscale/main.nf b/subworkflows/local/multiscale/main.nf deleted file mode 100644 index 389b8a6..0000000 --- a/subworkflows/local/multiscale/main.nf +++ /dev/null @@ -1,39 +0,0 @@ -include { MULTISCALE_PYRAMID } from '../../../modules/local/multiscale/pyramid/main' - -/* -~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - RUN MULTISCALE -~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -*/ - -workflow MULTISCALE { - take: - ch_meta // ch: [ meta, input_img_container, input_img_dataset ] - ch_dask_info // ch: [ dask_scheduler, dask_config ] - skip_flag // boolean - cpus // number - mem_gb // number - - main: - def multiscale_results - def multiscale_version - if (skip_flag) { - multiscale_results = ch_meta - multiscale_results.subscribe { log.debug "Skip multiscale pyramid: $it" } - multiscale_version = channel.empty() - } else { - MULTISCALE_PYRAMID( - ch_meta, - ch_dask_info, - cpus, - mem_gb, - ) - multiscale_results = MULTISCALE_PYRAMID.out.data - multiscale_results.subscribe { log.debug "Multiscale pyramid: $it" } - multiscale_version = MULTISCALE_PYRAMID.out.versions - } - - emit: - results = multiscale_results // [ meta, input_img_container, input_img_dataset ] - versions = multiscale_version -} diff --git a/subworkflows/local/utils_nfcore_cellpose_pipeline/main.nf b/subworkflows/local/utils_nfcore_cellpose_pipeline/main.nf index a7a99b0..3cc9a50 100644 --- a/subworkflows/local/utils_nfcore_cellpose_pipeline/main.nf +++ b/subworkflows/local/utils_nfcore_cellpose_pipeline/main.nf @@ -11,7 +11,6 @@ include { paramsSummaryMap } from 'plugin/nf-schema' include { completionEmail } from '../../nf-core/utils_nfcore_pipeline' include { completionSummary } from '../../nf-core/utils_nfcore_pipeline' -include { imNotification } from '../../nf-core/utils_nfcore_pipeline' include { UTILS_NEXTFLOW_PIPELINE } from '../../nf-core/utils_nextflow_pipeline' include { UTILS_NFCORE_PIPELINE } from '../../nf-core/utils_nfcore_pipeline' @@ -62,6 +61,7 @@ workflow PIPELINE_INITIALISATION { '', '', command, + true, ) // @@ -94,7 +94,6 @@ workflow PIPELINE_COMPLETION { plaintext_email // boolean: Send plain-text email instead of HTML outdir // path: Path to output directory where results will be published monochrome_logs // boolean: Disable ANSI colour codes in log output - hook_url // string: hook URL for notifications main: summary_params = paramsSummaryMap(workflow, parameters_schema: "nextflow_schema.json") @@ -116,9 +115,6 @@ workflow PIPELINE_COMPLETION { } completionSummary(monochrome_logs) - if (hook_url) { - imNotification(summary_params, hook_url) - } } workflow.onError { diff --git a/subworkflows/nf-core/utils_nextflow_pipeline/main.nf b/subworkflows/nf-core/utils_nextflow_pipeline/main.nf index d6e593e..37939ac 100644 --- a/subworkflows/nf-core/utils_nextflow_pipeline/main.nf +++ b/subworkflows/nf-core/utils_nextflow_pipeline/main.nf @@ -73,11 +73,23 @@ def getWorkflowVersion() { def dumpParametersToJSON(outdir) { def timestamp = new java.util.Date().format('yyyy-MM-dd_HH-mm-ss') def filename = "params_${timestamp}.json" - def temp_pf = new File(workflow.launchDir.toString(), ".${filename}") - def jsonStr = groovy.json.JsonOutput.toJson(params) + def temp_pf = workflow.launchDir.resolve(".${filename}") + def jsonGenerator = new groovy.json.JsonGenerator.Options() + .excludeNulls() + .addConverter(Path) { Path path -> path.toUriString() } + .addConverter(Duration) { Duration duration -> duration.toMillis() } + .addConverter(MemoryUnit) { MemoryUnit memory -> memory.toBytes() } + .addConverter(nextflow.script.types.VersionNumber) { nextflow.script.types.VersionNumber version -> version.toString() } + .build() + def jsonStr = jsonGenerator.toJson(params) temp_pf.text = groovy.json.JsonOutput.prettyPrint(jsonStr) - - nextflow.extension.FilesEx.copyTo(temp_pf.toPath(), "${outdir}/pipeline_info/params_${timestamp}.json") + if (outdir instanceof Path) { + temp_pf.copyTo(outdir.resolve("pipeline_info/${filename}")) + } else if (outdir instanceof String) { + temp_pf.copyTo("${outdir}/pipeline_info/params_${timestamp}.json") + } else { + log.warn("Could not determine type of outdir, parameters JSON file will not be copied to output directory!") + } temp_pf.delete() } diff --git a/subworkflows/nf-core/utils_nfcore_pipeline/main.nf b/subworkflows/nf-core/utils_nfcore_pipeline/main.nf index 2f30e9a..afca543 100644 --- a/subworkflows/nf-core/utils_nfcore_pipeline/main.nf +++ b/subworkflows/nf-core/utils_nfcore_pipeline/main.nf @@ -17,7 +17,7 @@ workflow UTILS_NFCORE_PIPELINE { checkProfileProvided(nextflow_cli_args) emit: - valid_config + valid_config = valid_config } /* @@ -353,67 +353,3 @@ def completionSummary(monochrome_logs=true) { log.info("-${colors.purple}[${workflow.manifest.name}]${colors.red} Pipeline completed with errors${colors.reset}-") } } - -// -// Construct and send a notification to a web server as JSON e.g. Microsoft Teams and Slack -// -def imNotification(summary_params, hook_url) { - def summary = [:] - summary_params - .keySet() - .sort() - .each { group -> - summary << summary_params[group] - } - - def misc_fields = [:] - misc_fields['start'] = workflow.start - misc_fields['complete'] = workflow.complete - misc_fields['scriptfile'] = workflow.scriptFile - misc_fields['scriptid'] = workflow.scriptId - if (workflow.repository) { - misc_fields['repository'] = workflow.repository - } - if (workflow.commitId) { - misc_fields['commitid'] = workflow.commitId - } - if (workflow.revision) { - misc_fields['revision'] = workflow.revision - } - misc_fields['nxf_version'] = workflow.nextflow.version - misc_fields['nxf_build'] = workflow.nextflow.build - misc_fields['nxf_timestamp'] = workflow.nextflow.timestamp - - def msg_fields = [:] - msg_fields['version'] = getWorkflowVersion() - msg_fields['runName'] = workflow.runName - msg_fields['success'] = workflow.success - msg_fields['dateComplete'] = workflow.complete - msg_fields['duration'] = workflow.duration - msg_fields['exitStatus'] = workflow.exitStatus - msg_fields['errorMessage'] = (workflow.errorMessage ?: 'None') - msg_fields['errorReport'] = (workflow.errorReport ?: 'None') - msg_fields['commandLine'] = workflow.commandLine.replaceFirst(/ +--hook_url +[^ ]+/, "") - msg_fields['projectDir'] = workflow.projectDir - msg_fields['summary'] = summary << misc_fields - - // Render the JSON template - def engine = new groovy.text.GStringTemplateEngine() - // Different JSON depending on the service provider - // Defaults to "Adaptive Cards" (https://adaptivecards.io), except Slack which has its own format - def json_path = hook_url.contains("hooks.slack.com") ? "slackreport.json" : "adaptivecard.json" - def hf = new File("${workflow.projectDir}/assets/${json_path}") - def json_template = engine.createTemplate(hf).make(msg_fields) - def json_message = json_template.toString() - - // POST - def post = new URL(hook_url).openConnection() - post.setRequestMethod("POST") - post.setDoOutput(true) - post.setRequestProperty("Content-Type", "application/json") - post.getOutputStream().write(json_message.getBytes("UTF-8")) - def postRC = post.getResponseCode() - if (!postRC.equals(200)) { - log.warn(post.getErrorStream().getText()) - } -} diff --git a/subworkflows/nf-core/utils_nfcore_pipeline/tests/main.workflow.nf.test b/subworkflows/nf-core/utils_nfcore_pipeline/tests/main.nf.test similarity index 100% rename from subworkflows/nf-core/utils_nfcore_pipeline/tests/main.workflow.nf.test rename to subworkflows/nf-core/utils_nfcore_pipeline/tests/main.nf.test diff --git a/subworkflows/nf-core/utils_nfcore_pipeline/tests/main.workflow.nf.test.snap b/subworkflows/nf-core/utils_nfcore_pipeline/tests/main.nf.test.snap similarity index 100% rename from subworkflows/nf-core/utils_nfcore_pipeline/tests/main.workflow.nf.test.snap rename to subworkflows/nf-core/utils_nfcore_pipeline/tests/main.nf.test.snap diff --git a/subworkflows/nf-core/utils_nfschema_plugin/main.nf b/subworkflows/nf-core/utils_nfschema_plugin/main.nf index acb3972..9ff0681 100644 --- a/subworkflows/nf-core/utils_nfschema_plugin/main.nf +++ b/subworkflows/nf-core/utils_nfschema_plugin/main.nf @@ -22,6 +22,7 @@ workflow UTILS_NFSCHEMA_PLUGIN { before_text // string: text to show before the help message and parameters summary after_text // string: text to show after the help message and parameters summary command // string: an example command of the pipeline + cli_typecast // boolean: whether to perform typecasting of CLI parameters. Set this to `null` to use the default behaviour main: @@ -34,11 +35,11 @@ workflow UTILS_NFSCHEMA_PLUGIN { fullHelp: help_full, ] if(parameters_schema) { - help_options << [parametersSchema: parameters_schema] + help_options << [parameters_schema: parameters_schema] } log.info paramsHelp( help_options, - params.help instanceof String ? params.help : "", + (help instanceof String && help != "true") ? help : "", ) exit 0 } @@ -50,7 +51,7 @@ workflow UTILS_NFSCHEMA_PLUGIN { summary_options = [:] if(parameters_schema) { - summary_options << [parametersSchema: parameters_schema] + summary_options << [parameters_schema: parameters_schema] } log.info before_text log.info paramsSummaryLog(summary_options, input_workflow) @@ -63,7 +64,10 @@ workflow UTILS_NFSCHEMA_PLUGIN { if(validate_params) { validateOptions = [:] if(parameters_schema) { - validateOptions << [parametersSchema: parameters_schema] + validateOptions << [parameters_schema: parameters_schema] + } + if(cli_typecast != null) { + validateOptions << [cast_cli_params: cli_typecast] } validateParameters(validateOptions) } diff --git a/subworkflows/nf-core/utils_nfschema_plugin/meta.yml b/subworkflows/nf-core/utils_nfschema_plugin/meta.yml index f7d9f02..1d8c75a 100644 --- a/subworkflows/nf-core/utils_nfschema_plugin/meta.yml +++ b/subworkflows/nf-core/utils_nfschema_plugin/meta.yml @@ -25,6 +25,30 @@ input: option. When this input is empty it will automatically use the configured schema or "${projectDir}/nextflow_schema.json" as default. The schema should not be given in this way for meta pipelines. + - help: + type: boolean, string + description: | + Show the help message and exit. When a parameter name is given, show the help message for that parameter instead of the general help message. + - help_full: + type: boolean + description: Show the full help message and exit. + - show_hidden: + type: boolean + description: Show hidden parameters in the help message. + - before_text: + type: string + description: Text to show before the parameters summary and help message. + - after_text: + type: string + description: Text to show after the parameters summary and help message. + - command: + type: string + description: An example command to run the pipeline, to show in the help message and the summary. + - cli_typecast: + type: boolean + description: | + Whether to apply typecasting to the parameters given via the CLI before validation. + Set this to `null` to use the default behavior. output: - dummy_emit: type: boolean diff --git a/subworkflows/nf-core/utils_nfschema_plugin/tests/main.nf.test b/subworkflows/nf-core/utils_nfschema_plugin/tests/main.nf.test index c977917..1fd1eac 100644 --- a/subworkflows/nf-core/utils_nfschema_plugin/tests/main.nf.test +++ b/subworkflows/nf-core/utils_nfschema_plugin/tests/main.nf.test @@ -31,6 +31,7 @@ nextflow_workflow { input[6] = "" input[7] = "" input[8] = "" + input[9] = null """ } } @@ -63,6 +64,7 @@ nextflow_workflow { input[6] = "" input[7] = "" input[8] = "" + input[9] = null """ } } @@ -95,6 +97,7 @@ nextflow_workflow { input[6] = "" input[7] = "" input[8] = "" + input[9] = null """ } } @@ -127,6 +130,7 @@ nextflow_workflow { input[6] = "" input[7] = "" input[8] = "" + input[9] = null """ } } @@ -160,6 +164,7 @@ nextflow_workflow { input[6] = "Before" input[7] = "After" input[8] = "nextflow run test/test" + input[9] = null """ } } diff --git a/subworkflows/nf-core/utils_nfschema_plugin/tests/nextflow.config b/subworkflows/nf-core/utils_nfschema_plugin/tests/nextflow.config index 8d8c737..fd71cb8 100644 --- a/subworkflows/nf-core/utils_nfschema_plugin/tests/nextflow.config +++ b/subworkflows/nf-core/utils_nfschema_plugin/tests/nextflow.config @@ -1,5 +1,5 @@ plugins { - id "nf-schema@2.5.1" + id "nf-schema@2.7.2" } validation { diff --git a/workflows/segmentation.nf b/workflows/segmentation.nf index 742d2cc..5372dae 100644 --- a/workflows/segmentation.nf +++ b/workflows/segmentation.nf @@ -3,15 +3,16 @@ IMPORT MODULES / SUBWORKFLOWS / FUNCTIONS ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ */ -include { paramsSummaryMap } from 'plugin/nf-schema' -include { softwareVersionsToYAML } from '../subworkflows/nf-core/utils_nfcore_pipeline' +include { paramsSummaryMap } from 'plugin/nf-schema' +include { softwareVersionsToYAML } from '../subworkflows/nf-core/utils_nfcore_pipeline' -include { COLLECT_INPUTS } from '../modules/local/collect_inputs' -include { CELLPOSE } from '../modules/janelia/cellpose/main' +include { COLLECT_INPUTS } from '../modules/local/collect_inputs' +include { SEGTOOLS_DISTRIBUTED_CELLPOSE } from '../modules/janelia/segtools/distributed/cellpose/main' +include { SEGTOOLS_DISTRIBUTED_MERGELABELS } from '../modules/janelia/segtools/distributed/mergelabels/main' -include { DASK_START } from '../subworkflows/janelia/dask_start/main' -include { DASK_STOP } from '../subworkflows/janelia/dask_stop/main' -include { MULTISCALE } from '../subworkflows/local/multiscale/main' +include { DASK_START } from '../subworkflows/janelia/dask_start' +include { DASK_STOP } from '../subworkflows/janelia/dask_stop' +include { MULTISCALE } from '../subworkflows/janelia/multiscale' workflow SEGMENTATION { @@ -28,17 +29,17 @@ workflow SEGMENTATION { if (params.cellpose_model) { if (params.cellpose_model.startsWith('/')) { // model is set using an absolute path - def full_model_path = file(cellpose_model) + def full_model_path = file(params.cellpose_model) model_dir = full_model_path.parent model_name = full_model_path.name - log.info("Use model ${model_name} located at: ${model_dir}, derived from: ${params.cellpose_model}") + log.info("Use model ${model_name} located at: ${model_dir} (${params.cellpose_model})") } else { model_dir = params.cellpose_models_dir ? file(params.cellpose_models_dir) : file("${params.workdir}/cellpose_models") model_name = params.cellpose_model - log.info("Use model ${model_name} that will be downloaded to ${model_dir}") + log.info("Use model ${model_name} (${params.cellpose_model}) that will be downloaded to ${model_dir}") } } else { - model_dir = params.cellpose_mnodels_dir ? file(params.cellpose_models_dir) : file("${params.workdir}/cellpose_models") + model_dir = params.cellpose_models_dir ? file(params.cellpose_models_dir) : file("${params.workdir}/cellpose_models") model_name = '' log.info("Use default model that will be downloaded to ${model_dir}") } @@ -50,15 +51,17 @@ workflow SEGMENTATION { model_dir, ] - def dask_cluster = DASK_START( + def ch_dask_cluster = DASK_START( channel.of([meta, dask_data]), params.with_dask, params.dask_config, file("${params.workdir}/dask-work/${workflow.sessionId}"), - params.dask_workers, - params.dask_min_workers, - params.dask_worker_cpus, - params.dask_worker_mem_gb > 0 ? params.dask_worker_mem_gb : params.default_mem_gb_per_cpu * params.dask_worker_cpus, + (params.dask_workers as int), + (params.dask_min_workers as int), + (params.dask_worker_cpus as int), + (params.dask_worker_mem_gb as int) > 0 + ? (params.dask_worker_mem_gb as int) + : (params.default_mem_gb_per_cpu as int) * (params.dask_worker_cpus as int), ) def ch_data_inputs = channel.of([file(params.input), params.input_pattern]) @@ -109,21 +112,23 @@ workflow SEGMENTATION { r } - def cellpose_inputs = dask_cluster + def ch_cellpose_inputs = ch_dask_cluster | combine(ch_data_inputs, by:0) - | multiMap { + | multiMap { row -> def (cellpose_meta, dask_info, input_container, input_subpath, cellpose_output_dir, labels_container, - labels_subpath) = it + labels_subpath) = row def cellpose_data = [ cellpose_meta, input_container, input_subpath, + params.input_mask ? file(params.input_mask) : [], + params.input_mask_subpath, model_dir, model_name, cellpose_output_dir, @@ -137,73 +142,142 @@ workflow SEGMENTATION { params.dask_config ? file(params.dask_config) : [], ] - log.info "Cellpose inputs: $it -> (${cellpose_data}, ${cellpose_cluster})" + log.info "Cellpose inputs: $row -> (${cellpose_data}, ${cellpose_cluster})" cellpose_data: cellpose_data cellpose_cluster: cellpose_cluster } - def cellpose_outputs = CELLPOSE( - cellpose_inputs.cellpose_data, - cellpose_inputs.cellpose_cluster, - params.preprocessing_config ? file(params.preprocessing_config) : [], - params.cellpose_log_config ? file(params.cellpose_log_config) : [], - params.cellpose_cpus, - params.cellpose_mem_gb > 0 ? params.cellpose_mem_gb : params.default_mem_gb_per_cpu * params.cellpose_cpus, - ) - - def cellpose_results = cellpose_outputs.results - cellpose_results.subscribe { - log.debug "Cellpose results: $it" + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + // Cellpose model evaluation (or reuse a prior run's output) + // ch_labels shape: [ meta, image, image_subpath, labels_containers(\n-joined), labels_subpath ] + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + def ch_labels + if (params.run_cellposemasks) { + def cellpose_outputs = SEGTOOLS_DISTRIBUTED_CELLPOSE( + ch_cellpose_inputs.cellpose_data, + ch_cellpose_inputs.cellpose_cluster, + params.preprocessing_config ? file(params.preprocessing_config) : [], + params.cellpose_log_config ? file(params.cellpose_log_config) : [], + ) + ch_versions = ch_versions.concat(cellpose_outputs.versions) + ch_labels = cellpose_outputs.results + ch_labels.subscribe { result -> log.debug "Cellpose results: $result" } + } else { + // reuse an existing segmentation output (requires a prior successful eval) + ch_labels = ch_cellpose_inputs.cellpose_data + | map { cp_row -> + def (sk_meta, image, image_subpath, + _mask, _mask_subpath, + _models_dir, _model_name, + output_dir, labels_container, labels_subpath, + _working_dir) = cp_row + def existing_labels = "${output_dir}/${labels_container}" + log.debug "run_cellposemasks=false: reusing existing labels ${existing_labels}" + [ sk_meta, image, image_subpath, existing_labels, labels_subpath ?: image_subpath ] + } } - // append cellpose version - ch_versions = ch_versions.concat (cellpose_outputs.versions) + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + // Optionally merge labels across blocks (distributed). + // ch_segmentation_results shape (per labels container): [ meta, labels_container, labels_subpath ] + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + def ch_segmentation_results + if (params.run_mergelabels) { + def mergelabels_inputs = ch_labels + | combine(ch_dask_cluster, by: 0) + | flatMap { row -> + def (ml_meta, + _image, _image_subpath, + labels_containers, labels_subpath, + dask_info) = row + labels_containers.split('\n') + .findAll { lc -> lc } + .collect { labels_container -> + def ml_data = [ + ml_meta, + file(labels_container), + labels_subpath, + [], // mask + '', // mask_subpath + [], // output_labels => in-place (overwrite input container) + labels_subpath, // output_subpath + file("${params.workdir}/cellpose-work/${workflow.sessionId}"), + ] + def ml_cluster = [ + dask_info ? dask_info.scheduler_address : '', + params.dask_config ? file(params.dask_config) : [], + ] + log.info "Merge labels input: ${ml_data} on ${ml_cluster}" + [ ml_data, ml_cluster ] + } + } - def multiscale_inputs = cellpose_results - | flatMap { - def (cellpose_meta, - input_container, input_subpath, - labels_containers, labels_subpath) = it - labels_containers.split('\n') - .findAll { it } - .collect { labels_container -> - def r = [ - cellpose_meta, labels_container, labels_subpath, - ] - log.info "Multiscale inputs: $r" - r + def ch_mergelabels_outputs = SEGTOOLS_DISTRIBUTED_MERGELABELS( + mergelabels_inputs.map { pair -> pair[0] }, + mergelabels_inputs.map { pair -> pair[1] }, + params.cellpose_log_config ? file(params.cellpose_log_config) : [], + ) + ch_versions = ch_versions.concat(ch_mergelabels_outputs.versions) + + // ml.results: [ meta, input_labels_fullpath, input_subpath, output_labels_fullpath, output_subpath ] + ch_segmentation_results = ch_mergelabels_outputs.results + | map { ml_row -> + def (sr_meta, _in_labels, _in_subpath, out_labels, out_subpath) = ml_row + [ sr_meta, out_labels, out_subpath ] + } + } else { + ch_segmentation_results = ch_labels + | flatMap { row -> + def (sr_meta, + _image, _image_subpath, + labels_containers, labels_subpath) = row + labels_containers.split('\n') + .findAll { lc -> lc } + .collect { labels_container -> + [ sr_meta, labels_container, labels_subpath ] + } } } - // the dask_info used for multiscale will be the same as the one used for cellpose - // this is acceptable now because for now there's only going to a single dask cluster - def multiscale_outputs = MULTISCALE( - multiscale_inputs, - cellpose_inputs.cellpose_cluster, // this can be questionable here but there's a single dask cluster - params.skip_multiscale, - params.multiscale_cpus, - params.multiscale_mem_gb > 0 ? params.multiscale_mem_gb : params.default_mem_gb_per_cpu * params.multiscale_cpus, + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + // Multiscale pyramid for the segmentation results (per labels container), + // attaching the dask cluster info per item. + // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + def ch_multiscale_inputs = ch_segmentation_results + | combine(ch_dask_cluster, by: 0) + | map { row -> + def (ms_meta, labels_container, labels_subpath, dask_info) = row + def ms_data = [ ms_meta, labels_container, labels_subpath ] + def ms_cluster = [ + dask_info ? dask_info.scheduler_address : '', + params.dask_config ? file(params.dask_config) : [], + ] + log.info "Multiscale inputs: ${ms_data} on ${ms_cluster}" + [ ms_data, ms_cluster ] + } + + def ch_multiscale_outputs = MULTISCALE( + ch_multiscale_inputs.map { pair -> pair[0] }, + ch_multiscale_inputs.map { pair -> pair[1] }, + params.run_multiscale, ) // wait until all multiscale finish - this is still needed in case multiple inputs are being segmented - def all_multiscale_results = multiscale_outputs.results + def ch_all_multiscale_results = ch_multiscale_outputs | groupTuple(by:0) - // append multiscale version - ch_versions = ch_versions.concat (multiscale_outputs.versions) - // stop the dask cluster - dask_cluster - | join(all_multiscale_results, by:0) - | map { - def (cluster_meta, cluster_context) = it - log.debug "Prepare to stop the dask cluster -> $it" + ch_dask_cluster + | join(ch_all_multiscale_results, by:0) + | map { row -> + def (cluster_meta, cluster_context) = row + log.debug "Prepare to stop the dask cluster -> $row" [ cluster_meta, cluster_context ] } | DASK_STOP emit: - results = multiscale_outputs.results + results = ch_multiscale_outputs versions = ch_versions // channel: [ path(versions.yml) ] }