Metadata-Version: 2.4
Name: nenuflow
Version: 0.2
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

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:

```bash
java -version
```

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

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

```bash
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):

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

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

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

Then run the pipeline:

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

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

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

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

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

```bash
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](#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` |
