Metadata-Version: 2.5
Name: nenuflow
Version: 0.4
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.25` 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.

### 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.

## Compute profiles

### Local

Use `-profile local`. DP3 and WSClean jobs declare `params.cpus`; Nextflow
limits concurrency according to the CPUs 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, time, memory, and cluster options in
`nextflow.config` for the local site:

```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
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.

## Restarting a partially completed run

Some calibration steps modify MeasurementSets in place, so Nextflow `-resume`
cannot reliably identify their state. 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
```

The wide-field check after the self-cal DI round and the deep image after DD
subtraction are optional diagnostics. They each image the full MS collection.
For throughput-focused runs, disable either with `--wsclean_di_img_check false`
or `--wsclean_dd_deep_check 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` |
| `do_subtract` | `true` | Run DD calibration and source subtraction |
| `wsclean_di_img_check` | `true` | Make the post-DI diagnostic image in selfcal mode |
| `wsclean_dd_deep_check` | `true` | Make the post-DD deep diagnostic image |
| `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 requested by WSClean and pipeline processes |
| `mem_per_job` | `40.GB` | Memory used for executor admission control |
| `number_of_clusters` | `10` | Number of DD calibration directions |
| `solint` | `30` | DI solution interval in time steps |
| `solint_target` | `30` | DD solution interval in time steps |
| `ntimesteps_to_average` | `3` | Time averaging before DD calibration |
| `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) |

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 five scenarios: `fast_track` with and without subtraction,
and `selfcal` with subtraction, with `skip_di`, and without the optional
check images.

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
```
