Metadata-Version: 2.5
Name: nenuflow
Version: 0.6
Summary: Nextflow calibration pipeline for NenuFAR Cosmic Dawn data
Project-URL: Homepage, https://gitlab.com/nenufar-cosmic-dawn/nenuflow
Project-URL: Repository, https://gitlab.com/nenufar-cosmic-dawn/nenuflow
Project-URL: Issues, https://gitlab.com/nenufar-cosmic-dawn/nenuflow/-/issues
Author-email: Florent Mertens <florent.mertens@gmail.com>, Kariuki Chege <chege@astro.rug.nl>, Satyapan Munshi <satyapan.iiserm@gmail.com>
License-Expression: GPL-3.0-or-later
License-File: LICENSE
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: <4.0,>=3.10
Requires-Dist: click>=8.0
Provides-Extra: dev
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff==0.16.1; extra == 'dev'
Description-Content-Type: text/markdown

# NenuFlow

NenuFlow calibrates NenuFAR Cosmic Dawn L2 MeasurementSets. It starts from an
MS list prepared with [NenuCal](https://nenucal-cd.readthedocs.io), performs DI
and optional DD calibration and subtraction, and writes calibrated data,
solutions, sky models, and quality-control products.

For raw L1 observations, begin with
[NenuPrePipe](https://gitlab.com/nenufar-cosmic-dawn/nenuprepipe). It creates
the bandpass-corrected, quality-checked, station-flagged L2 data expected here.

```text
NenuPrePipe L2 data
    → create one MS list per night and spectral window
    → choose fast-track or self-calibration
    → run NenuFlow
    → inspect calibration and AOQuality products
    → import any newly detected bad stations
```

## Requirements

- Nextflow 23 or newer and Java 11 or newer (Java 17+ recommended)
- Singularity and a `dawntools.sif` image
- `nenucal >= 0.26` on the host, providing `nenudata`, `modeltool`, and
  `aostats`

DP3, WSClean, and AOQuality run inside the container. NenuCal commands used to
prepare the MS list and manage station flags run on the host.

Check the host tools before setup:

```bash
java -version
nextflow -version
nenudata --help
```

If Nextflow is not installed, follow the
[official installation guide](https://www.nextflow.io/docs/latest/install.html).

## Quick start

### 1. Install and initialize

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

cd ~/nenufar-ops/my-field
nenuflow init
```

`nenuflow init` creates `nextflow.config` in the current directory. It can be
run in the same deployment directory previously created by `prepipe init`.
Set the `container` parameter to the absolute path of `dawntools.sif`, then
review the execution profile, resources, calibration mode, and output path.

### 2. Create the MS list

Use NenuCal's data handler to select one observation and spectral window:

```bash
nenudata make_ms_list L2_BP "20231208_NT04:SW03" \
    -c data_handler.toml --target ms_lists -C
```

Use the level actually produced by NenuPrePipe and registered in your
`data_handler.toml`; deployments using `L2` rather than `L2_BP` should pass
`L2` here.

The generated name must follow:

```text
{obs_id}_{level}_{sw}
```

For example, `20231208_NT04_L2_BP_SW03`. NenuFlow reads the observation,
processing level, and spectral window from this name. See
[NenuCal data access](https://nenucal-cd.readthedocs.io/en/latest/nenudata/data.html)
for selection patterns, altitude cuts, and per-MS lists.

### 3. Choose a calibration mode

Set `params.mode` in `nextflow.config`.

#### `fast_track`

Uses a catalog-based intrinsic sky model and does not run WSClean imaging:

```text
build channels → DI calibration and A-team subtraction
               → DD calibration and subtraction (optional)
               → post-processing and diagnostics
```

Use this for the standard fast calibration path. NenuFlow builds the catalog
and clustered models automatically unless `intrinsic_catalog_model` or
`intrinsic_catalog_model_cluster` points to a fixed model.

#### `selfcal`

Builds the sky model iteratively with WSClean:

```text
build channels → initial DI calibration
               → WSClean/self-calibration cycles
               → DD calibration and subtraction (optional)
               → post-processing and diagnostics
```

Set `n_selfcal_cycles` to the desired number of cycles (up to five). The final
WSClean model is clustered for DD calibration.

In both modes, the subtractions and the DI apply interpolate the solutions
linearly in time and frequency (`soltool unwrap` first unwraps their phases, as
DP3 interpolates the stored values): nearest interpolation would leave steps
between frequency solution intervals in the data, and power at the matching
k_par in the power spectrum. The published solutions are unchanged.

### 4. Run the pipeline

On one machine:

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 -profile local
```

`nenuflow run` forwards additional arguments to Nextflow, so parameters can be
overridden for one run:

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 \
    -profile local --mode fast_track --do_subtract true
```

Run this command from the directory containing `nextflow.config`. Each MS list
gets its own work directory and result directory.

Several MS lists, e.g. one per night, get one Nextflow run each, four at a time
(`--jobs`), or one after another with `-profile local`, as the runs would share
the machine without knowing it:

```bash
nenuflow run ms_lists/2025011*_NT04_L2_BP_SW03 -profile slurm --mode fast_track
```

The output of each run goes to `work/<name>/console.log` and
`work/<name>/nextflow.log`; a failed run does not stop the others, and the
command lists them at the end. `nenuflow run` does not add
`-resume` (see [Restarting a partially completed run](#restarting-a-partially-completed-run)).

## Compute profiles

Every process declares its CPUs and memory in `nextflow.config`, sized on a
33-MS NT04 run. DP3 calibration and WSClean imaging request `cpus` and
`mem_per_job` (WSClean `-j` and DP3 `numthreads` follow); the other processes
request 1 to 8 CPUs and 2 to 16 GB, capped at `cpus` and `mem_per_job`, so that
they fit on a node next to the heavy ones. Choose `cpus` and `mem_per_job` so
that a whole number of heavy jobs fills a node, e.g. `cpus = 12` and
`mem_per_job = 43.GB` for two jobs on a 24-core, 90 GB node. A DP3 calibration
peaks at about 23 GB on 33 NT04 MSs.

### Local

Use `-profile local`. Nextflow limits concurrency according to the CPUs and
memory available to the local executor.

### HyperQueue

Start one server and one worker on each compute node, then use `-profile hq`:

```bash
hq server start &
hq worker start --cpus=auto --resource "mem=bytes:128000000000" &
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 -profile hq
```

Set `mem_per_job` so the intended number of jobs fits safely in each node. The
repository also contains `scripts/hq_cluster.py` and an example YAML file for
starting and stopping a declared worker pool.

### SLURM

Use `-profile slurm` and adjust the queue and cluster options in
`nextflow.config` for the local site. A `memory`, `cpus` or `time` set for all
processes in the profile is ignored, as the resources per process take
precedence. Each process has a time limit of about five times its longest run,
multiplied by the attempt number, so that a job stuck on a faulty node is killed and run
elsewhere; jobs cancelled or lost by the scheduler are retried up to three
times:

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 -profile slurm
```

## Results and quality control

By default each run writes to `results/<mslist-name>/`:

```text
models/       catalog, clustered, intrinsic, and apparent sky models; the
              clustered-model plots show the beam half-power and first-null
              contours at the band edges, at mid-observation; the spectra
              (*_spectra.png) and VLSSr cross-match (*_xmatch.png, .ecsv, .log)
              of the intrinsic catalogue and self-calibration models
solutions/    combined calibration solutions and plots
aoquality/    combined statistics, plots, and bad-station JSON files
wsclean/      self-calibration images and logs
report.html   Nextflow execution report
versions.txt  versions of nenuflow, Nextflow, nenucal, DP3, WSClean... and the
              container image
timeline.html execution timeline
trace.html    process trace
```

Per-MS logs and some per-MS calibration products are stored beside the input
MeasurementSets. Review both the solution plots and the AOQuality plots before
accepting the run. The
[AOQuality documentation](https://pyaoquality.readthedocs.io/en/latest/aostats.html)
explains the plotted statistics and bad-station detection.

Import reviewed NenuFlow detections into the data-handler registry with:

```bash
nenudata bad-stations import \
    "results/*/aoquality/bad_stations_l2a.json" \
    --note "reviewed NenuFlow detections" \
    -c data_handler.toml
nenudata bad-stations show-history -c data_handler.toml
```

See [NenuCal bad stations](https://nenucal-cd.readthedocs.io/en/latest/nenudata/bad_stations.html)
for reviewing, adding, removing, and propagating station decisions.

## Sky model from several nights

`nenuflow skymodel` builds a sky model from one WSClean image of several nights,
calibrated beforehand (`CORRECTED_DATA_L2_A`, e.g. after a `selfcal` run of each
night; `skymodel_data_column` otherwise). It takes one MS list per night and
runs the `skymodel` mode, with the same profiles as `nenuflow run`:

```bash
nenuflow skymodel ms_lists/2025011*_NT04_L2_BP_SW03 -profile slurm
```

As in the `selfcal` mode, it images the data with `wsclean_selfcal_config`,
keeps the sources within `selfcal_model_radius`, deattenuates them with the
beam averaged over the observed time of all the nights (7 times per night,
`deattenuate_n_times`), and clusters the result. The merged MS list and the
products go to `results/<level>_<sw>_<n>nights/` (`--name` otherwise):

```text
mslist.txt  the MSs of all the nights
wsclean/    images and WSClean log
models/     skymodel_intrinsic.skymodel and skymodel_clustered.skymodel, the
            plot of the clusters, the spectra and VLSSr cross-match of the
            intrinsic model
```

To calibrate every night with this model, set `intrinsic_catalog_model` and
`intrinsic_catalog_model_cluster` to the two models and run the `fast_track`
mode on the same MS lists:

```bash
nenuflow run ms_lists/2025011*_NT04_L2_BP_SW03 -profile slurm --mode fast_track \
    --intrinsic_catalog_model results/L2_BP_SW03_5nights/models/skymodel_intrinsic.skymodel \
    --intrinsic_catalog_model_cluster results/L2_BP_SW03_5nights/models/skymodel_clustered.skymodel
```

The `skymodel` mode only reads the MSs, so `-resume` can rerun the steps after a
failure without redoing the image, as long as the MSs were not calibrated again
in between.

## Restarting a partially completed run

Some calibration steps modify MeasurementSets in place, so Nextflow `-resume`
cannot reliably identify their state: a cached task could be reused after its MS
has changed. `nenuflow run` therefore does not add it; use the explicit stage
flags instead:

| Flag | Skip when this product already exists |
|---|---|
| `--skip_di true` | First DI calibration and `CORRECTED_DATA_L2_A` |
| `--skip_wsclean true` | Self-calibration imaging and final intrinsic model |
| `--skip_di_img true` | Second DI calibration on the self-calibration model |
| `--skip_dd true` | DD calibration/subtraction and `SUBTRACTED_DATA_L3` |

Set only the flags matching stages whose outputs have been checked on disk:

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 \
    -profile local --skip_di true --skip_wsclean true
```

Three WSClean check images are optional diagnostics, published to `wsclean/`.
Each images the full MS collection, which takes as long as a WSClean run:

| Flag | Image |
|---|---|
| `wsclean_check_before_di` | `<stage>_check_before`: wide-field dirty image of the uncalibrated data, before the first DI round |
| `wsclean_check_after_di` | `<stage>_check_after`: the same after the final DI round, A-team subtracted (`l2a` in fast-track mode, `selfcal_final` in self-calibration mode) |
| `wsclean_check_after_dd` | `l3_check_deep`: deep clean after DD subtraction |

For throughput-focused runs, disable them with e.g.
`--wsclean_check_before_di false`. These switches do not skip the WSClean image
that builds the self-calibration sky model.

## Main configuration parameters

| Parameter | Default | Purpose |
|---|---:|---|
| `mode` | `selfcal` | `fast_track` or `selfcal` (`skymodel` is set by `nenuflow skymodel`) |
| `do_subtract` | `true` | Run DD calibration and source subtraction |
| `wsclean_check_before_di` | `true` | Check image of the uncalibrated data |
| `wsclean_check_after_di` | `true` | Check image after the final DI round |
| `wsclean_check_after_dd` | `true` | Deep check image after DD subtraction |
| `sky_model_diagnostics` | `true` | Plot the spectra of the intrinsic sky models and cross-match them with VLSSr |
| `xmatch_resolution` | `12` | Resolution of the sky models in arcmin, for the VLSSr cross-match |
| `cpus` | `10` | CPUs of a DP3 calibration or WSClean job, and cap of the other processes |
| `mem_per_job` | `40.GB` | Memory of a DP3 calibration or WSClean job, and cap of the other processes |
| `number_of_clusters` | `10` | Number of DD calibration directions |
| `solint` | `30` | DI solution interval in time steps |
| `solint_target` | `10` | DD solution interval in averaged time steps (6 min at 3 × 12 s) |
| `ntimesteps_to_average` | `3` | Time averaging before DD calibration, in the calibration job (not kept on disk) |
| `sols_per_dir_min_flux`, `sols_per_dir_max_flux` | `30`, `50` | Flux range in Jy of the faintest direction, which sets the solutions per direction |
| `sols_per_dir_freq_mhz` | `66.5` | Frequency at which the direction fluxes are compared |
| `di_smoothness_main`, `di_smoothness_other` | `4e6`, `1e6` | DI frequency smoothing kernel in Hz of Main and of the other directions (`null` or `0`: the parset value for all) |
| `dd_solutions_per_direction` | `false` | Also give the DD directions several solutions per interval, from their flux |
| `dd_min_beam_cut` | `0.1` | Remove from the apparent DD model of each MS the components whose beam is below this value at any frequency (0: keep all) |
| `bad_stations_stat` | `DStd` | Statistic used to find bad stations |
| `modeltool_build_catalog` | `lcs165` | Catalog used for automatic sky-model construction |
| `modeltool_build_radius` | `12` | Catalog-model radius in degrees |
| `selfcal_model_radius` | `12` | Radius in degrees of the sources kept from the WSClean self-calibration model (selfcal mode) |
| `selfcal_min_beam_cut` | `0.1` | Remove from the self-calibration model the components whose beam is below this value at any frequency (0: keep all) |

DP3 thread counts are also set in the bundled parsets and should remain
consistent with the requested resources.

## Testing pipeline changes

For development, lint the pipeline and run the stub suite. Both also run in CI.

`nextflow lint` follows the `include` statements from the entrypoint, so it
covers every module. It needs Nextflow 25.02 or newer:

```bash
nextflow lint nenuflow/main.nf
```

Stubs test pipeline wiring without a container or real data:

```bash
cd test/stub
./setup.sh
./run.sh
```

The suite covers six scenarios: `fast_track` with and without subtraction,
`selfcal` with subtraction, with `skip_di`, and without the optional check
images, and `skymodel` on two nights whose MSs share their names. Besides the
exit status, it checks that the DD solutions, the check images and the sky
model products are published, or not, as the flags of each scenario request.

Read every boolean parameter through `flag('<name>')` (from
`nenuflow/functions.nf`, which holds its default): on the command line,
`--<name> false` gives the string `'false'`, which Groovy treats as true, and an
older `nextflow.config` may not define the parameter.

Every new process must provide a matching `stub:` block with the outputs
declared by that process.

## End-to-end: process one night of data

This example follows one target night, `20231208_NT04`, and one spectral
window, `SW03`, from the raw archive through NenuFlow. Replace the project
code, observation ID, level, paths, and execution profile with the values for
your deployment.

### 1. Create one shared deployment

Install both pipelines, create the field directory, and initialize the
NenuPrePipe data handler:

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

mkdir -p ~/nenufar-ops/my-field
cd ~/nenufar-ops/my-field
prepipe init --project-code RP3A
```

Edit `project.toml` and `data_handler.toml`. At minimum, verify the L1 archive,
L2 output paths, processing nodes, date range, target/calibrator patterns,
bandpass-solution path, and DP3 parsets.

### 2. Populate the data handler

Preview and then register observations and N2 night identifiers:

```bash
prepipe update project.toml --dry-run
prepipe update project.toml
```

Confirm that the requested night and its calibrator resolve correctly:

```bash
prepipe run project.toml "20231208_NT04" SW03 --dry-run
```

### 3. Run NenuPrePipe calibration and quality collection

```bash
prepipe run project.toml "20231208_NT04" SW03
```

This prepares the calibrator, derives the bandpass, creates the target L2
MeasurementSets, collects AOQuality statistics, proposes bad stations, applies
the registered station flags, and transfers the L2 data when `push_l2` is
enabled.

If processing moves to another server, run the `nenudata import-n2` command
printed by `push_l2` there before continuing.

### 4. Check and finalize antenna flags

Inspect the plots under `quality_l2_cal/` and `quality_l2_bp/`. Check the
registered stations and its history:

```bash
nenudata bad-stations show "20231208_NT04" -c data_handler.toml
nenudata bad-stations show-history -c data_handler.toml
```

Record any reviewed corrections with `nenudata bad-stations add`, `remove`, or
`clear`. Then apply the current registry to the prepared target data:

```bash
nenudata bad-stations flag L2_BP "20231208_NT04:SW03" \
    -c data_handler.toml --only-n2
```

The command restores the pre-station-flag state before applying a changed
registry, so removed stations do not remain flagged from an earlier decision.

### 5. Set up NenuFlow

In the same directory on the NenuFlow host:

```bash
nenuflow init
```

Edit `nextflow.config` and set the absolute `container` path. Choose
`fast_track` or `selfcal`, review `do_subtract`, resource settings, model
settings, and the selected compute profile.

### 6. Create the one-night MS list

```bash
nenudata make_ms_list L2_BP "20231208_NT04:SW03" \
    -c data_handler.toml --target ms_lists -C
```

Open `ms_lists/20231208_NT04_L2_BP_SW03` and verify that it contains exactly
the intended MeasurementSets before launching calibration.

### 7. Run NenuFlow

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 -profile local
```

For a cluster, replace `local` with the configured `hq` or `slurm` profile.
Monitor the Nextflow output and the per-MS logs; use the explicit `skip_*`
flags only after confirming which stage products are complete.

### 8. Validate and record the result

Review:

1. `results/20231208_NT04_L2_BP_SW03/solutions/` for calibration behavior;
2. `models/` and, for self-calibration, `wsclean/` for sky-model quality. In
   `*_xmatch.png`, the model / VLSSr flux ratio should stay near 1 with the
   distance from the centre: a drop away from it points to the beam
   correction. In `*_spectra.png`, spectral indices far flatter than the usual
   -0.7 point to a spectral tilt of the calibrated data;
3. `aoquality/` for time, frequency, baseline, and antenna diagnostics;
4. `report.html`, `timeline.html`, and failed-process logs for execution issues.

Import any reviewed NenuFlow station detections into the data handler used to
create the MS list, which keeps the audit trail with the night's data. No import
is needed when no new stations were found.

```bash
nenudata bad-stations import \
    "results/20231208_NT04_L2_BP_SW03/aoquality/bad_stations_l2a.json" \
    --note "reviewed after NenuFlow 20231208_NT04 SW03" \
    -c data_handler.toml
nenudata bad-stations show-history -c data_handler.toml
```

If NenuFlow runs on another server, also copy the reviewed JSON back and import
it into the original NenuPrePipe data handler, so the two station registries
stay consistent.

If the import adds stations, apply the updated registry to the L2
MeasurementSets. This restores the flags saved before the first bad-station
flagging, then flags the current list:

```bash
nenudata bad-stations flag L2_BP "20231208_NT04:SW03" \
    -c data_handler.toml --only-n2
```

Then run NenuFlow again from the first calibration stage, so the new flags are
included throughout the result, and review the new products before accepting
the night:

```bash
nenuflow run ms_lists/20231208_NT04_L2_BP_SW03 -profile local
```
