NenuFlow

Nextflow pipeline for NenuFAR Cosmic Dawn calibration. Supports two execution modes (fast-track and self-calibration) and three compute backends (local, HyperQueue, SLURM).

Dependencies

• Nextflow ≥ 23 (needs a Java runtime — see below)
• Singularity — all heavy processes (DP3, WSClean, aoquality) run inside a dawntools container image
• nenucal-cd ≥ 0.13 — modeltool, aostats, nenudata must be available on the host (outside the container)

Installing Java + Nextflow

Nextflow needs a Java 11+ runtime (17+ recommended) on the host — it does not run inside the dawntools container. Check for an existing one first:

java -version

If that's missing or too old, install a JDK (e.g. via your package manager or SDKMAN). Then install Nextflow itself (see docs.seqera.io/nextflow/install for details/alternatives):

curl -s https://get.nextflow.io | bash
chmod +x nextflow
mkdir -p ~/.local/bin && mv nextflow ~/.local/bin/   # make sure this is on $PATH
nextflow -version

Setup

pip install git+https://gitlab.com/nenufar-cosmic-dawn/nenuflow.git

mkdir -p ~/nenufar-ops/my-field && cd ~/nenufar-ops/my-field
nenuflow init

nenuflow init (like git init) writes a default nextflow.config into the current directory. It only refuses to run if one is already there, so it's safe to use inside a directory set up by another tool (e.g. nenuprepipe init). Edit its one CHANGEME placeholder:

Parameter	What to set
container	Absolute path to dawntools.sif

intrinsic_catalog_model/intrinsic_catalog_model_cluster default to null (build from the web / auto-cluster) — override with an absolute path only if you want a fixed sky model instead.

Then generate an mslist with nenudata (an existing nenucal-cd command — not part of this package):

nenudata make_ms_list L2 20231208_NT04:SW03 -c data_handler.toml --target ms_lists -C

mslist naming convention

The mslist filename encodes the observation metadata used downstream:

{obs_id}_{level}_{sw}

Examples: 20231208_NT04_L2_SW03, 20240115_NT04_L2a_SW03

• Tokens starting with SW identify the sub-band
• Tokens starting with L followed by digits identify the processing level
• Everything before the level token is the obs_id

The obs_id is parsed automatically and used to label bad-station JSON output. nenudata make_ms_list already produces one file per (obs_id, sw) following this exact convention.

Running

Local (single node)

nenuflow run ms_lists/20231208_NT04_L2_SW03 -profile local

nenuflow run MSLIST [NEXTFLOW_ARGS...] is a thin wrapper: it resolves the bundled pipeline's main.nf and runs nextflow run <that path> --mslist MSLIST -work-dir work/<name> -resume [NEXTFLOW_ARGS...] (<name> is MSLIST's filename stem, e.g. 20231208_NT04_L2_SW03), so any additional Nextflow/pipeline flag (-profile, --mode, -stub-run, -c, ...) works exactly as it would with a raw nextflow run invocation. It must be run from the directory nenuflow init set up (or one containing your own nextflow.config), since Nextflow auto-discovers ./nextflow.config.

Concurrency is capped by the local executor's own cpu accounting: 'dp3' and WSCleanImage processes both declare cpus = params.cpus, so at most floor(available cpus / cpus) of them run at once.

HyperQueue (multi-node)

Start the HQ server once on the submission node, then one worker per compute node:

hq server start &
hq worker start --cpus=auto --resource "mem=bytes:128000000000" &

Then run the pipeline:

nenuflow run ms_lists/20231208_NT04_L2_SW03 -profile hq

HQ uses mem_per_job for admission control. Set it to roughly (node_RAM − 8 GB) / max_concurrent_jobs. For a 128 GB node with 3 jobs: 40 GB.

You can also manage the HQ cluster with the bundled helper script — this lives in the git repo, not the pip package (nenuflow init/run don't need it), so clone the repo to use it:

python scripts/hq_cluster.py start  -c scripts/hq_cluster.yaml
python scripts/hq_cluster.py stop   -c scripts/hq_cluster.yaml
python scripts/hq_cluster.py status -c scripts/hq_cluster.yaml

Edit scripts/hq_cluster.yaml to declare your node list and per-node resources.

SLURM

nenuflow run ms_lists/20231208_NT04_L2_SW03 -profile slurm

Adjust queue, time, memory, and clusterOptions in nextflow.config for your site.

Execution modes

Set params.mode in nextflow.config.

fast_track

Calibrates against a fixed intrinsic sky model. No WSClean imaging.

build_ms_channel
  └─ CalibrateDI   (attenuate model → DI calibration → A-team subtraction)
       └─ CalibrateDD   (DD calibration → source subtraction)  [if do_subtract]
            └─ PostProcess

Sky model resolution order:

1. intrinsic_catalog_model if set, otherwise ModelToolBuild fetches from the web using modeltool_build_catalog and modeltool_build_radius
2. intrinsic_catalog_model_cluster if set, otherwise MakeClusters clusters the catalog model into number_of_clusters directions

selfcal

Iteratively builds the sky model via WSClean self-calibration (n_selfcal_cycles cycles, max 5).

build_ms_channel
  └─ CalibrateDI
       └─ SelfCalCycle × n_selfcal_cycles   (WSClean → attenuate → DI cal)
            └─ CalibrateDD   (DD cal on final model)  [if do_subtract]
                 └─ PostProcess

In selfcal mode intrinsic_catalog_model_cluster is ignored — the cluster is always built from the final WSClean model.

Resuming after a completed stage

-resume doesn't work reliably here: several parsets mutate the MS in place (msout=.), so the same MS path can have different content across attempts, which defeats Nextflow's file-based cache keys.

Instead, use the skip_* params to pick up after a stage that already completed. Each one trusts that the stage's output (an MS column, or a file under resultspath/) is already on disk from a previous run, and reconstructs the channel from that fixed location instead of re-running the processes:

Flag	Skips	Assumes already on disk
skip_di	(first) DI calibration	CORRECTED_DATA_L2_A column
skip_wsclean	WSClean imaging + deattenuate (selfcal only)	results/models/selfcal_intrinsic.skymodel
skip_di_img	second DI round on produced model (selfcal only)	CORRECTED_DATA_L2_A column (final)
skip_dd	DD calibration + subtraction	SUBTRACTED_DATA_L3 column

All default to false. They are independent flags, not a single checkpoint — set the combination that matches how far the previous run actually got, e.g.:

nenuflow run ms_lists/20231208_NT04_L2_SW03 -profile local --skip_di true --skip_wsclean true

Output

Results are published under params.resultspath (default: one subdirectory per mslist, ./results/<mslist basename>, e.g. ./results/20231208_NT04_L2_SW03 — h5/qs/log filenames are only namespaced by stage label, not by obs_id/SW, so two mslist runs sharing one resultspath would otherwise overwrite each other's output):

results/20231208_NT04_L2_SW03/
  models/          sky models (catalog, clustered, apparent)
  aoquality/       combined .qs files, diagnostic plots, bad_stations_*.json
  solutions/       combined h5parm + solution plots
  wsclean/         FITS images and logs (selfcal mode only)
  report.html, timeline.html, trace.html

Per-MS logs (and, for selfcal, per-MS skymodels/h5 solutions) are published next to each MS itself rather than under resultspath, since multiple mslist runs can share the same underlying MS.

dag.png is off by default (needs Graphviz); re-enable per-run with -with-dag.

Post-run: bad-station import

Each run writes a per-observation bad_stations_{label}.json in the aoquality output directory. After one or more runs, merge them into the master bad_stations.json via nenudata:

nenudata bad-stations import "results/*/aoquality/bad_stations_l2a.json" \
    --note "auto-detected from run XX" \
    -c data_handler.toml

Use nenudata bad-stations show-history to audit past imports.

Testing pipeline changes

Before pushing a change to main.nf, workflows/*.nf, or processes/*.nf, validate it locally with nextflow lint (syntax) and the -stub-run suite under test/stub/ (DAG composition and channel wiring). -stub-run executes every process's stub: block instead of the real DP3/WSClean/modeltool command, so it needs no container and no real data — it still exercises real channel joins/maps, skip_* flag branches, and file staging, just not the actual calibration.

cd test/stub
./setup.sh   # (re)generate a throwaway mslist + fake .MS dirs
./run.sh     # runs fast_track/selfcal, with and without do_subtract/skip_di

Adding a new process requires a matching stub: block (touch/mkdir whatever files its output: declares — content doesn't matter, only presence/shape, since every downstream consumer is stubbed too).

Key parameters

Parameter	Default	Description
mode	selfcal	fast_track or selfcal
do_subtract	true	Run DD calibration and source subtraction
skip_di / skip_wsclean / skip_di_img / skip_dd	false	Skip a completed stage instead of using -resume; see Resuming after a completed stage
cpus	10	WSClean's -j; DP3's numthreads= is set separately in each parsets/*.parset and must be kept in sync by hand
mem_per_job	40.GB	RAM ceiling per job (HQ admission control)
number_of_clusters	10	DD calibration directions
solint	30	DI solution interval (time steps)
solint_target	30	DD solution interval (time steps)
ntimesteps_to_average	3	Time averaging before DD calibration
bad_stations_stat	DStd	Statistic used by aostats find-bad-stations
modeltool_build_catalog	lcs165	Catalog for modeltool build
modeltool_build_radius	12	Radius in degrees for modeltool build