From f92cb44b6747a010c4cf369d7e515e9278824244 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 15:46:49 -0400 Subject: [PATCH 1/9] feat(orchestration): Snakemake workflow for the real-data pipeline (#848) Implements the round-2 design in #848: static two-phase execution driven by a parse-time SQLite run index (never a rule input, so appending tiles never invalidates completed work); rules wrap the existing per-unit shapepipe_run config chains at tile/exposure granularity with deterministic RUN_DATETIME=False directory() outputs; in-job completeness count-floor (the ported bash complete_check table, workflow/scripts/completeness.py); one-allocation execution profile for nibi with apptainer software-deployment (the workflow never calls apptainer itself); out-of-DAG run_report verb + onsuccess/onerror hooks; workflow/bin/sp as the committed launcher. First pass touches no module code. Validated by dry-run against a 4-tile P3 subset (219-job compute DAG); prepare phase runs green on nibi. Co-Authored-By: Claude Fable 5 --- .gitignore | 1 + profiles/nibi/config.yaml | 101 +++++++ workflow/README.md | 105 ++++++++ workflow/Snakefile | 124 +++++++++ workflow/bin/sp | 63 +++++ workflow/config.yaml | 35 +++ workflow/rules/exposure.smk | 70 +++++ workflow/rules/prepare.smk | 67 +++++ workflow/rules/tile.smk | 147 ++++++++++ workflow/scripts/build_forest.py | 42 +++ workflow/scripts/build_index.py | 139 ++++++++++ workflow/scripts/clean_exposures.py | 64 +++++ workflow/scripts/completeness.py | 86 ++++++ workflow/scripts/run_report.py | 107 ++++++++ workflow/scripts/sp_rule.py | 405 ++++++++++++++++++++++++++++ 15 files changed, 1556 insertions(+) create mode 100644 profiles/nibi/config.yaml create mode 100644 workflow/README.md create mode 100644 workflow/Snakefile create mode 100755 workflow/bin/sp create mode 100644 workflow/config.yaml create mode 100644 workflow/rules/exposure.smk create mode 100644 workflow/rules/prepare.smk create mode 100644 workflow/rules/tile.smk create mode 100644 workflow/scripts/build_forest.py create mode 100644 workflow/scripts/build_index.py create mode 100644 workflow/scripts/clean_exposures.py create mode 100644 workflow/scripts/completeness.py create mode 100644 workflow/scripts/run_report.py create mode 100644 workflow/scripts/sp_rule.py diff --git a/.gitignore b/.gitignore index db20f3086..4c4b6d0b0 100644 --- a/.gitignore +++ b/.gitignore @@ -139,3 +139,4 @@ code # .felt here is a machine-local symlink into it. Never track it in this repo. /.felt/ /.felt +.snakemake/ diff --git a/profiles/nibi/config.yaml b/profiles/nibi/config.yaml new file mode 100644 index 000000000..e977803c1 --- /dev/null +++ b/profiles/nibi/config.yaml @@ -0,0 +1,101 @@ +# Snakemake profile for the Nibi cluster (Digital Research Alliance). +# +# ONE-ALLOCATION mode: sbatch one node up front, then run snakemake with the +# LOCAL scheduler inside it (no per-job sbatch; the local scheduler bin-packs the +# DAG into the allocation against `cores` + `resources: mem_mb`). This is the +# proven nibi path (job_p3_batch1) and the single honest first-pass mode; +# multi-node scaling = multiple tile-disjoint invocations, deferred. +# +# Launch via `workflow/bin/sp` (loads apptainer/1.4.5, uses the /project venv). +# software-deployment-method wraps every job's shell in `apptainer exec` — the +# user never types apptainer; the container is set in the Snakefile (local .sif). + +# Local scheduler (no `executor:` -> built-in local). Size to the allocation. +cores: 192 # node cores available to the DAG +local-cores: 4 # cap localrule/aggregation concurrency +resources: + mem_mb: 720000 # node RAM cap the local scheduler bin-packs under + +software-deployment-method: [apptainer] +# Explicit container environment (finding 2/3): the apptainer SDM otherwise drops +# the proven-recipe env and hardcodes --home , hiding ~/.ssl/cadcproxy.pem. +# --cleanenv strict host-env isolation (APPTAINERENV_*/SINGULARITYENV_* survive) +# OMP_NUM_THREADS=1 caps OpenBLAS fork-explosion (verified: pool 32->1) +# MALLOC_ARENA_MAX=2 bounded allocator (im_sims/nibi lesson) +# --home /home/cdaley wins over the SDM's --home ; restores cadcproxy.pem for vos/vcp +# PYTHONPATH SETTLED CALL 3 — P0 pins the prod worktree src so science +# content matches the bash validation runs (which use this +# override). Production later rebuilds the sif at the exact +# validation commit and DROPS this --env line. +apptainer-args: "--cleanenv --env OMP_NUM_THREADS=1 --env MALLOC_ARENA_MAX=2 --env PYTHONPATH=/project/def-mjhudson/cdaley/shapepipe-prod/src --home /home/cdaley --bind /project --bind /scratch" + +latency-wait: 60 # NFS: wait for outputs to appear after a job +keep-going: true # a failed job poisons only its cone; siblings run on +rerun-incomplete: true # re-do jobs left incomplete by an unclean death +show-failed-logs: true +printshellcmds: true + +# rerun-triggers left at the full Snakemake v9 default +# {mtime,params,input,code,software-env}: params/code fixes SHOULD propagate. +# No override — the wrapper writes outputs only on change so mtimes move only +# when reality moves; protected() guards finished science products. + +default-resources: + mem_mb: 4000 + runtime: 120 # minutes + cpus_per_task: 1 + +# Per-stage resources — measured footprints from the nibi P3 runs (#808). +# threads == cpus_per_task == ShapePipe SMP fork width (sp_rule sets +# SMP_BATCH_SIZE={threads}); mem_mb == threads x measured per-worker footprint. +# +# set-threads is REQUIRED for {threads} to resolve above 1: the local scheduler +# derives cpus_per_task from threads, NOT the reverse (verified snakemake 9.23 — +# set-resources cpus_per_task alone leaves {threads}==1, i.e. SMP_BATCH_SIZE=1, +# serial). Keep the two blocks in lockstep. +set-threads: + exp_split: 4 + exp_mask: 4 + exp_psf: 8 + tile_mask: 16 + tile_detect: 16 + tile_merge_headers: 8 + tile_vignets: 8 + tile_ngmix: 8 + tile_merge_cats: 4 + tile_make_cat: 4 + +set-resources: + exp_split: + cpus_per_task: 4 + mem_mb: 8000 + exp_mask: + cpus_per_task: 4 + mem_mb: 8000 + exp_psf: + cpus_per_task: 8 + mem_mb: 16000 + runtime: 240 + tile_mask: + cpus_per_task: 16 + mem_mb: 16000 + tile_detect: + cpus_per_task: 16 + mem_mb: 16000 + tile_merge_headers: + cpus_per_task: 8 + mem_mb: 8000 + tile_vignets: + cpus_per_task: 8 + mem_mb: 24000 + runtime: 240 + tile_ngmix: + cpus_per_task: 8 + mem_mb: 24000 + runtime: 360 + tile_merge_cats: + cpus_per_task: 4 + mem_mb: 8000 + tile_make_cat: + cpus_per_task: 4 + mem_mb: 8000 diff --git a/workflow/README.md b/workflow/README.md new file mode 100644 index 000000000..4d8abd1f1 --- /dev/null +++ b/workflow/README.md @@ -0,0 +1,105 @@ +# ShapePipe Snakemake orchestration + +Snakemake workflow that orchestrates real-data ShapePipe runs, replacing the +`curl_canfar_local.sh → run_job_sp_canfar_v2.0.bash → job_sp_canfar_v2.0.bash` +bash layers (and per-site sbatch reimplementations). **Module code is untouched**: +rules call `shapepipe_run -c ` on the existing config chains. Design and +rationale: [CosmoStat/shapepipe#848](https://github.com/CosmoStat/shapepipe/issues/848) +(the living PRD). + +## Quick start (nibi) + +**One-allocation mode.** `sbatch` one node, then run Snakemake with the *local* +scheduler inside it: it bin-packs the DAG into the allocation (no per-job +sbatch). Each job's shell runs `shapepipe_run` **inside the container** (the +profile's apptainer software-deployment — you never type `apptainer`). + +```bash +# One-time: a snakemake env on a SHARED filesystem (/project — NOT /tmp, which is +# node-local; the executor re-invokes this python inside jobs). +uv venv /project/def-mjhudson/cdaley/snakemake-env --python 3.12 +source /project/def-mjhudson/cdaley/snakemake-env/bin/activate +uv pip install 'snakemake>=9,<10' + +# Edit workflow/config.yaml: tile_list, run_dir, config_src, container, star_cats. + +# The committed launcher loads apptainer/1.4.5 + the /project venv, so a fresh +# shell always has the right state. Inside your allocation: +workflow/bin/sp prepare # prepare_tiles -> build_index.py -> prepare_exposures +workflow/bin/sp run # the main compute DAG +``` + +`sp` subcommands: `prepare` (chains all three prepare steps), `prepare_tiles`, +`index`, `prepare_exposures`, `run`, `rerun ` (forced recompute of +protected products), `cancel ` (scancel sweep before `--unlock`), +`clean-exposures [--apply]`, `report`. + +## Execution is three static invocations + +The exposure job set is data-derived from the tiles' `find_exposures` output, so +it cannot be scheduled in the same static DAG that produces it. Hence: + +1. `snakemake prepare_tiles` — per-tile static DAG (Git_vos → Uz → Fe), + `keep-going` so tile failures are independent. +2. `build_index.py` — a **plain script**, not a DAG node. It iterates the + declared tile list, checks each tile's Fe output at its deterministic path (no + glob), writes `run_index.sqlite` + `missing.json`, and exits nonzero only if + the missing *fraction* exceeds a threshold. (There is no `--allow-missing` + Snakemake flag; that mechanism does not exist.) +3. `snakemake prepare_exposures` — per-exposure static DAG (Gie_vos + per-unit + star cats), parseable now the index exists. +4. `snakemake` (main DAG) — the tile/exposure compute chain. + +`sp prepare` chains 1–3 so the UX stays one command. + +## Layout + +``` +workflow/ + Snakefile parse-time index load; global container:; onsuccess/onerror report hooks + config.yaml the run: tile list, paths, container, chunk count + bin/sp committed launcher (module load + /project venv + subcommands) + rules/ + prepare.smk tile get_images/uncompress/find_exposures + per-unit star cats + exposure.smk per-exposure: get_images, split, mask, psf (no temp()) + tile.smk per-tile: exp forest, merge_headers, mask, detect, vignets, ngmix, merge, make_cat + scripts/ + sp_rule.py the thin per-unit wrapper (isolation furniture, config copy, log-sync, count floor) + build_index.py prepare-phase run_index.sqlite builder (plain script) + build_forest.py per-tile exposure symlink forest (group-compatible shell) + completeness.py the ported count-floor table (shared by sp_rule + run_report) + run_report.py standalone report (NOT a DAG node; run_report hooks call it) + clean_exposures.py index-driven exposure-store reclaim (replaces temp()) +profiles/nibi/config.yaml local scheduler; apptainer SDM; resources; keep-going +``` + +## How it works + +- **Completeness = the count floor.** Each stage declares its deterministic + `output/run_sp_/` directory as `directory()` output (the per-unit + config copy forces `RUN_DATETIME = False`, so the path is known at DAG time). + After the run, `sp_rule.py` counts products per mandatory runner against + `completeness.py`'s floor and exits nonzero below it; per-CCD attrition between + floor and `expect` is tolerated (the directory simply lacks that product). + This *is* the failure policy — no 3-class taxonomy, no error-signature + whitelist. `--keep-going` isolates a failure to its own cone. +- **Per-unit isolation is by work-dir content, not `-e/--exclusive`.** Every + tile/exposure runs its own `shapepipe_run` in `tiles//` or `exp//`, + with a `cfis` config symlink, `star_cat_{exp,tiles}` symlinks, a + `tile_numbers.txt` (tiles) or fabricated `exp_numbers-000-000.txt` pseudo-Fe + (exposures), and a config copy setting `RUN_DATETIME=False` and + `SMP_BATCH_SIZE={threads}`. `NUMBER_LIST=-` is injected **only** for + stages whose numbering scheme is the unit ID (tile-scheme stages + exp split), + never for get_images / exp_mask / exp_psf. +- **No `temp()` on shared exposure dirs.** Exposures overlap tiles, so temp() + would cascade destructive reruns when a tile is appended. The store is + persistent; `sp clean-exposures` reclaims space only for exposures all of + whose consuming tiles have a final_cat. +- **The index is parse-time data, never a rule input.** Appending tiles changes + which jobs exist without invalidating completed work. Star cats are re-keyed + per unit for the same reason. `final_cat` is `protected()`. +- **Failure is a report, not a gate.** `run_report.py` disk-scans the trees + against the count table and enumerates shortfalls (whole-unit absence vs + per-CCD attrition). It is a standalone script — a DAG report node would be + poisoned by the very failures it must enumerate — emitted automatically by the + `onsuccess`/`onerror` hooks and runnable any time via `sp report`. diff --git a/workflow/Snakefile b/workflow/Snakefile new file mode 100644 index 000000000..04722a73c --- /dev/null +++ b/workflow/Snakefile @@ -0,0 +1,124 @@ +"""ShapePipe real-data orchestration — Snakemake workflow. + +Design / rationale: CosmoStat/shapepipe#848 (the living PRD). + +Execution is honestly THREE static invocations, chained by ``workflow/bin/sp`` +so the UX stays one command (``sp prepare`` / ``sp run``): + + snakemake --profile profiles/nibi prepare_tiles # Git_vos + Uz + Fe (keep-going) + python workflow/scripts/build_index.py ... # PLAIN SCRIPT, not a DAG node + snakemake --profile profiles/nibi prepare_exposures # Gie_vos + star cats + snakemake --profile profiles/nibi # the main compute DAG + +The run index is loaded ONCE here at parse time into plain dicts; it is never a +rule input, so appending tiles and rebuilding it changes which jobs exist without +invalidating completed work. build_index.py is a plain script *between* +prepare_tiles and prepare_exposures (the exposure job set is data-derived from +the tiles' find_exposures output, so it cannot be scheduled in the same static +DAG that produces it). +""" + +import sqlite3 +from pathlib import Path + +configfile: "workflow/config.yaml" + +# Every job's shell runs inside this container (apptainer software-deployment in +# the profile); the user never types apptainer. +container: config["container"] + +# --- paths ----------------------------------------------------------------- +RUN_DIR = Path(config["run_dir"]) +CONFIG_SRC = Path(config["config_src"]) +STAR_CATS = Path(config["star_cats"]) +INDEX_DB = Path(config["index_db"]) +SCRIPTS = Path(workflow.basedir) / "scripts" +NGMIX_CHUNKS = config.get("ngmix_chunks", 8) + +with open(config["tile_list"]) as f: + TILES = [ln.strip() for ln in f if ln.strip()] + +# --- parse-time index load (compute phase only) ---------------------------- +# EXP: global set of exposure base-ids. TILE_EXP: tile -> [exp_ids]. +EXP, TILE_EXP = set(), {} +if INDEX_DB.exists(): + _con = sqlite3.connect(INDEX_DB) + EXP = {r[0] for r in _con.execute("SELECT exp_id FROM exposures")} + for tile, exp in _con.execute("SELECT tile_id, exp_id FROM tile_exposures"): + TILE_EXP.setdefault(tile, []).append(exp) + _con.close() + +wildcard_constraints: + tile = r"\d{3}\.\d{3}", + exp = r"\d{6,7}", + chunk = r"\d+", + +# --- shared wrapper invocation --------------------------------------------- +def sp_rule(stage, config_name, level, unit, *, isolate=True, exp_forest=None, + extra=""): + """Build the shell command that runs one unit through sp_rule.py. + + NUMBER_LIST is injected by default (tile-scheme stages + exp split); pass + ``isolate=False`` for get_images / exp_mask / exp_psf, whose per-CCD/download + numbering must not be constrained. ``{threads}`` is filled by Snakemake and + becomes SMP_BATCH_SIZE (fork width == cpus_per_task). + """ + cmd = ( + f"python {SCRIPTS}/sp_rule.py --stage {stage} --config {config_name} " + f"--unit {unit} --level {level} --run-dir {RUN_DIR} " + f"--config-src {CONFIG_SRC} --star-cats {STAR_CATS} --threads {{threads}}" + ) + if not isolate: + cmd += " --no-isolate" + if exp_forest: + cmd += f" --exp-forest {exp_forest}" + if extra: + cmd += " " + extra + return cmd + +def tile_dir(tile): + return RUN_DIR / "tiles" / tile + +def exp_dir(exp): + return RUN_DIR / "exp" / exp + +include: "rules/prepare.smk" +include: "rules/exposure.smk" +include: "rules/tile.smk" + +# --- top-level targets ------------------------------------------------------ +# Only aggregation targets are localrules (no-ops under the local executor, but +# they keep the compute chain group-compatible: no mid-chain localrules). +localrules: all, prepare_tiles, prepare_exposures + +rule all: + input: + expand(str(RUN_DIR / "tiles/{tile}/final_cat-{tile}.fits"), tile=TILES) + +# Phase A.1 — per-tile static DAG (download + find_exposures). keep-going so +# tile failures are independent; build_index.py runs next as a plain script. +rule prepare_tiles: + input: + expand(str(RUN_DIR / "tiles/{tile}/output/run_sp_tile_Fe"), tile=TILES) + +# Phase A.2 — per-exposure static DAG (download + star cats), parseable now the +# index exists. Star cats are re-keyed per unit (see prepare.smk). +rule prepare_exposures: + input: + expand(str(RUN_DIR / "exp/{exp}/output/run_sp_exp_Gie"), exp=sorted(EXP)), + expand(str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits"), exp=sorted(EXP)), + expand(str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits"), tile=TILES) + +# --- report hooks ----------------------------------------------------------- +# run_report is NOT a DAG node (a descendant of every job would be poisoned by +# any hard failure — the exact case it exists for). It is a standalone script, +# emitted automatically at the end of every invocation, runnable any time. +def _report(status): + shell(f"python {SCRIPTS}/run_report.py --run-dir {RUN_DIR} " + f"--index {INDEX_DB} --status {status} || true") + +onsuccess: + _report("success") + +onerror: + _report("error") diff --git a/workflow/bin/sp b/workflow/bin/sp new file mode 100755 index 000000000..f226aa441 --- /dev/null +++ b/workflow/bin/sp @@ -0,0 +1,63 @@ +#!/usr/bin/env bash +# sp — the committed launcher for the ShapePipe Snakemake workflow. +# +# Removes the shell-state dependencies the review flagged (finding 20): it loads +# the apptainer module (Snakemake resolves `apptainer` via PATH at job runtime) +# and runs the Snakemake that lives on the shared /project venv (must be on a +# shared FS — the executor re-invokes it inside jobs). One entry point, so a +# fresh tmux / a restart after a crash always launches with the right state. +# +# Usage: +# sp prepare # prepare_tiles -> build_index.py -> prepare_exposures +# sp prepare_tiles # phase A.1 only +# sp index # build_index.py only (plain script, threshold-gated) +# sp prepare_exposures # phase A.2 only +# sp run [ARGS...] # the main compute DAG +# sp rerun # forced recompute of protected/finished products +# sp cancel # scancel this workflow's jobs (name sweep) before --unlock +# sp clean-exposures [--apply] # index-driven exposure-store reclaim +# sp report # emit run_report.json now (mid-run ok) +# Extra args after the subcommand pass through to snakemake. +set -euo pipefail + +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" # workflow/ +REPO="$(dirname "$HERE")" +VENV="${SP_SNAKEMAKE_ENV:-/project/def-mjhudson/cdaley/snakemake-env}" +PROFILE="$REPO/profiles/nibi" +SCRIPTS="$HERE/scripts" +CONFIG="$HERE/config.yaml" + +module load apptainer/1.4.5 2>/dev/null || true +# shellcheck disable=SC1091 +source "$VENV/bin/activate" + +# Minimal scalar reader for workflow/config.yaml (key: value, no nesting). +cfg() { sed -n "s/^$1:[[:space:]]*//p" "$CONFIG" | head -1; } +RUN_DIR="$(cfg run_dir)"; INDEX_DB="$(cfg index_db)" +TILE_LIST="$(cfg tile_list)"; MISSING_THRESH="${SP_MISSING_THRESHOLD:-0.0}" + +sm() { snakemake --profile "$PROFILE" "$@"; } +build_index() { + python "$SCRIPTS/build_index.py" --tile-list "$TILE_LIST" \ + --run-dir "$RUN_DIR" --db "$INDEX_DB" --missing-threshold "$MISSING_THRESH" +} + +cmd="${1:-}"; shift || true +case "$cmd" in + prepare) sm prepare_tiles "$@"; build_index; sm prepare_exposures "$@" ;; + prepare_tiles) sm prepare_tiles "$@" ;; + index) build_index ;; + prepare_exposures) sm prepare_exposures "$@" ;; + run) sm "$@" ;; + rerun) sm --forcerun "$@" ;; + cancel) + run="${1:?usage: sp cancel }" + squeue --me --noheader --format='%i %j' \ + | awk -v r="$run" '$2 ~ r {print $1}' | xargs -r scancel + echo "cancelled jobs matching '$run'; safe to --unlock / rerun now" ;; + clean-exposures) python "$SCRIPTS/clean_exposures.py" \ + --run-dir "$RUN_DIR" --index "$INDEX_DB" "$@" ;; + report) python "$SCRIPTS/run_report.py" \ + --run-dir "$RUN_DIR" --index "$INDEX_DB" --status manual "$@" ;; + *) echo "sp: unknown subcommand '${cmd:-}' (see the header of $0)"; exit 2 ;; +esac diff --git a/workflow/config.yaml b/workflow/config.yaml new file mode 100644 index 000000000..53542d1c2 --- /dev/null +++ b/workflow/config.yaml @@ -0,0 +1,35 @@ +# Run configuration for the ShapePipe Snakemake workflow. +# +# A "run" is declared by a tile list plus the paths below. Everything here is +# read at parse time; none of it is a rule input, so editing it (e.g. appending +# tiles) never invalidates completed work — it only changes which jobs exist. + +# The tile list that scopes this run (one "IDra.IDdec" per line). +tile_list: /scratch/cdaley/shapepipe-output/snakemake-p0/tiles4.txt + +# The container every job runs inside (apptainer software-deployment in the profile). +container: /project/def-mjhudson/cdaley/containers/shapepipe-develop-runtime.sif + +# Where products land ($SP_RUN root). Per-unit work-dir forest is built under here: +# /tiles/// and /exp/// +run_dir: /scratch/cdaley/shapepipe-output/snakemake-p0 + +# The canonical config chain (example/cfis in the repo, or a run-specific copy). +config_src: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/cfis + +# Pre-staged inputs (P3 data already on /project; get_images RETRIEVE=symlink). +# Star catalogues for masking, pre-generated (network step done in prepare): +star_cats: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/star_cats + +# The run index. Durable, low-volume products (index, missing.json, run_report, +# final_cat) should land on /project, not /scratch (60-day purge, finding 15). +index_db: /project/def-mjhudson/cdaley/shapepipe-output/snakemake-p0/index/run_index.sqlite + +# ngmix within-tile chunking: static N chunks (closed ID ranges computed +# per-tile, in-job, from the tile's own sexcat). +ngmix_chunks: 8 + +# The container's installed shapepipe is overridden by the prod worktree via +# --env PYTHONPATH in profiles/nibi (settled call 3); no config knob here. +# build_index.py's missing-tile fraction floor is passed by workflow/bin/sp +# (SP_MISSING_THRESHOLD, default 0.0 = any missing tile is fatal). diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk new file mode 100644 index 000000000..50403d386 --- /dev/null +++ b/workflow/rules/exposure.smk @@ -0,0 +1,70 @@ +"""Exposure chain — per exposure, keyed by exp_id (dedup is structural). + +Gie -> Sp -> Ma -> psf, each in the exposure's own work dir, chained by +ShapePipe's last:/run_sp_exp_* INPUT_DIR resolver. Completeness is the count +floor (completeness.py). + +NO temp() on these shared exposure directories: exposures overlap tiles by +construction (~7-10 tiles per exposure), so temp() would delete an exposure the +moment its forests are built and cascade destructive reruns across spatial +neighbours whenever a tile is appended (findings 4/5/6/19). The store is +persistent; space is reclaimed by the explicit index-driven `sp clean-exposures` +verb (scripts/clean_exposures.py), which deletes an exposure's intermediates +only once every consuming tile in the index has its final_cat present. + +NUMBER_LIST is injected only for exp_split (numbering scheme == exposure id); +NEVER for get_images (download) / exp_mask / exp_psf (per-CCD schemes where the +#746 startup validation would turn tolerated per-CCD attrition into a +whole-exposure hard failure). +""" + +EXP_OUT = str(RUN_DIR / "exp/{exp}/output") + +# Symlink/download exposure image/weight/flag; reads the fabricated pseudo-Fe +# exp_numbers-000-000.txt the wrapper drops (via last:find_exposures_runner). +rule exp_get_images: + output: + directory(f"{EXP_OUT}/run_sp_exp_Gie") + params: + cmd=lambda wc: sp_rule("exp_get_images", "config_exp_Gie_vos.ini", + "exp", wc.exp, isolate=False) + shell: + "{params.cmd}" + +# Split multi-HDU exposure into single-CCD files (+ headers-*.npy). Bulk output; +# persistent (see module docstring on temp()). +rule exp_split: + input: + rules.exp_get_images.output + output: + directory(f"{EXP_OUT}/run_sp_exp_Sp") + params: + cmd=lambda wc: sp_rule("exp_split", "config_exp_Sp.ini", "exp", wc.exp) + shell: + "{params.cmd}" + +# Mask per-CCD; this exposure's star cat is a declared input (per-unit re-key). +rule exp_mask: + input: + rules.exp_split.output, + star=str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits"), + output: + directory(f"{EXP_OUT}/run_sp_exp_Ma") + params: + cmd=lambda wc: sp_rule("exp_mask", "config_exp_Ma_onthefly.ini", + "exp", wc.exp, isolate=False) + shell: + "{params.cmd}" + +# SExtractor -> setools star selection -> PSFEx model -> psfex_interp, per CCD. +# setools may reject a sparse CCD (~0.2% attrition) — tolerated by the floor. +rule exp_psf: + input: + rules.exp_mask.output + output: + directory(f"{EXP_OUT}/run_sp_exp_SxSePsfPi") + params: + cmd=lambda wc: sp_rule("exp_psf", "config_exp_psfex.ini", "exp", wc.exp, + isolate=False) + shell: + "{params.cmd}" diff --git a/workflow/rules/prepare.smk b/workflow/rules/prepare.smk new file mode 100644 index 000000000..2911c3b19 --- /dev/null +++ b/workflow/rules/prepare.smk @@ -0,0 +1,67 @@ +"""Phase A — prepare: tile staging, find_exposures, per-unit star cats. + +Split into two static invocations with build_index.py (a plain script) between: + * prepare_tiles — tile_get_images -> tile_uncompress -> tile_find_exposures + * [build_index.py] — plain script, threshold-gated (not a DAG node) + * prepare_exposures — exp_get_images (exposure.smk) + per-unit star cats + +Nibi compute nodes have internet, so downloads run in-DAG (no login-node tier). +Rules stay group-compatible: no mid-chain localrules, no run:/pipe outputs. +""" + +# get_images reads a one-line tile_numbers.txt the wrapper drops in $SP_RUN. +# NUMBER_LIST is NEVER injected here (download stage; nothing on disk to +# validate against — #746 would hard-fail). +rule tile_get_images: + output: + directory(str(RUN_DIR / "tiles/{tile}/output/run_sp_tile_Git")) + params: + cmd=lambda wc: sp_rule("tile_get_images", "config_tile_Git_vos.ini", + "tile", wc.tile, isolate=False) + shell: + "{params.cmd}" + +rule tile_uncompress: + input: + rules.tile_get_images.output + output: + directory(str(RUN_DIR / "tiles/{tile}/output/run_sp_tile_Uz")) + params: + cmd=lambda wc: sp_rule("tile_uncompress", "config_tile_Uz.ini", + "tile", wc.tile, isolate=True) + shell: + "{params.cmd}" + +# find_exposures reads the tile FITS HISTORY header -> exp_numbers-.txt, the +# data-derived tile->exposure edge build_index.py aggregates (as a plain script). +rule tile_find_exposures: + input: + rules.tile_uncompress.output + output: + directory(str(RUN_DIR / "tiles/{tile}/output/run_sp_tile_Fe")) + params: + cmd=lambda wc: sp_rule("tile_find_exposures", "config_tile_Fe.ini", + "tile", wc.tile, isolate=True) + shell: + "{params.cmd}" + +# Star catalogues for masking are pre-generated offline (create_star_cat.py on a +# networked login node — the one genuinely networked prepare step). They are +# re-keyed PER UNIT here: each tile/exposure's cat is its own DAG node, so +# appending tile 51 adds nodes rather than mutating a run-level output that would +# dirty every finished mask (the growth-invariant fix, finding 11). +rule exp_star_cat: + output: + str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits") + params: + src=lambda wc: str(STAR_CATS / "exp" / f"star_cat-{wc.exp}.fits") + shell: + "ln -sf {params.src} {output}" + +rule tile_star_cat: + output: + str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits") + params: + src=lambda wc: str(STAR_CATS / "tiles" / f"star_cat-{wc.tile}.fits") + shell: + "ln -sf {params.src} {output}" diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk new file mode 100644 index 000000000..845167710 --- /dev/null +++ b/workflow/rules/tile.smk @@ -0,0 +1,147 @@ +"""Tile post-chain — per tile: gather exposures, then detect/PSF/shape/catalogue. + +A per-tile exposure "forest" (symlink view of exactly this tile's exposures' +products) gives the tile modules a deterministic $SP_EXP to glob. But the forest +is only a runtime convenience, NEVER the DAG edge: every tile rule that reads +through $SP_EXP declares the underlying exposure directory() outputs as input via +an index dict lookup, so the last real reader is the last DAG consumer (findings +5/19). All rules are group-compatible (shell only, no mid-chain localrules). +""" + +TILE_OUT = str(RUN_DIR / "tiles/{tile}/output") +FOREST = str(RUN_DIR / "tiles/{tile}/exp_forest") + + +def _exp_out(wc, prefix): + return [str(exp_dir(e) / "output" / prefix) for e in TILE_EXP.get(wc.tile, [])] + +def tile_exp_split(wc): return _exp_out(wc, "run_sp_exp_Sp") +def tile_exp_mask(wc): return _exp_out(wc, "run_sp_exp_Ma") +def tile_exp_psf(wc): return _exp_out(wc, "run_sp_exp_SxSePsfPi") +def tile_exp_products(wc): + """All of this tile's exposures' product dirs — the forest's DAG inputs.""" + return tile_exp_split(wc) + tile_exp_mask(wc) + tile_exp_psf(wc) + + +# Build the per-tile symlink forest (exp_forest//output/run_sp_exp_* -> the +# real exposure products). Declaring the exposure dirs as input makes this the +# tile's wait-on-its-exposures edge; the forest itself is just the SP_EXP view. +rule tile_exp_forest: + input: + tile_exp_products + output: + directory(FOREST) + params: + cmd=lambda wc: (f"python {SCRIPTS}/build_forest.py --tile {wc.tile} " + f"--run-dir {RUN_DIR} --index {INDEX_DB} --forest {{output}}") + shell: + "{params.cmd}" + +# Merge single-exposure WCS headers into the tile-level sqlite log +# (log_exp_headers--.sqlite — the tile suffix downstream Sx/PiViVi/Ng +# consume via NUMBERING_SCHEME -000-000). Reads headers-*.npy through the forest, +# so the exposure split dirs are explicit inputs. +rule tile_merge_headers: + input: + forest=rules.tile_exp_forest.output, + split=tile_exp_split, + output: + directory(f"{TILE_OUT}/run_sp_tile_Mh_exp") + params: + cmd=lambda wc: sp_rule("tile_merge_headers", "config_tile_Mh_exp.ini", + "tile", wc.tile, isolate=True, + exp_forest=FOREST.format(tile=wc.tile)) + shell: + "{params.cmd}" + +# Mask tiles; this tile's star cat is a declared input (per-unit re-key). +rule tile_mask: + input: + git=rules.tile_get_images.output, + star=str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits"), + output: + directory(f"{TILE_OUT}/run_sp_tile_Ma") + params: + cmd=lambda wc: sp_rule("tile_mask", "config_tile_Ma_onthefly.ini", + "tile", wc.tile, isolate=True) + shell: + "{params.cmd}" + +# SExtractor object detection on the tile. +rule tile_detect: + input: + mask=rules.tile_mask.output, + mh=rules.tile_merge_headers.output, + output: + directory(f"{TILE_OUT}/run_sp_tile_Sx") + params: + cmd=lambda wc: sp_rule("tile_detect", "config_tile_Sx.ini", + "tile", wc.tile, isolate=True) + shell: + "{params.cmd}" + +# PSFEx interpolation to galaxies + vignet postage stamps (reads exposure PSF +# models + split through the forest — those dirs are explicit inputs). +rule tile_vignets: + input: + sx=rules.tile_detect.output, + forest=rules.tile_exp_forest.output, + split=tile_exp_split, + psf=tile_exp_psf, + output: + directory(f"{TILE_OUT}/run_sp_tile_PiViVi") + params: + cmd=lambda wc: sp_rule("tile_vignets", "config_tile_PiViVi_canfar_sx.ini", + "tile", wc.tile, isolate=True, + exp_forest=FOREST.format(tile=wc.tile)) + shell: + "{params.cmd}" + +# ngmix shape measurement — static N chunks. Each chunk computes its own closed +# object-ID range in-job from the tile's own sexcat (a declared input); chunks +# never rewrite the shared log (--no-log-sync) and read literal INPUT_DIRs. +rule tile_ngmix: + input: + vignets=rules.tile_vignets.output, + sx=rules.tile_detect.output, + output: + directory(f"{TILE_OUT}/run_sp_tile_ngmix_Ng{{chunk}}u") + params: + cmd=lambda wc: sp_rule( + "tile_ngmix", "config_tile_Ng_template_batch.ini", "tile", wc.tile, + extra=f"--ngmix-chunk {wc.chunk} --ngmix-nchunks {NGMIX_CHUNKS}"), + shell: + "{params.cmd}" + +def ngmix_chunks(wc): + return [f"{RUN_DIR}/tiles/{wc.tile}/output/run_sp_tile_ngmix_Ng{k}u" + for k in range(1, NGMIX_CHUNKS + 1)] + +# Merge the N chunk catalogues (DAG-serialized: the single per-tile log-sync +# point before make_cat). N_SPLIT_MAX == chunk count. +rule tile_merge_cats: + input: + ngmix_chunks + output: + directory(f"{TILE_OUT}/run_sp_Ms") + params: + cmd=lambda wc: sp_rule("tile_merge_cats", "config_merge_sep_cats_template.ini", + "tile", wc.tile, isolate=True, + extra=f"--n-split-max {NGMIX_CHUNKS}"), + shell: + "{params.cmd}" + +# Build the final catalogue: the run's science product. protected() so a +# code/params-drift rerun cannot silently discard a finished catalogue — the +# recompute must go through `sp rerun` (--forcerun). +rule tile_make_cat: + input: + rules.tile_merge_cats.output + output: + protected(str(RUN_DIR / "tiles/{tile}/final_cat-{tile}.fits")) + params: + cmd=lambda wc: sp_rule("tile_make_cat", "config_make_cat_psfex_nosm.ini", + "tile", wc.tile, isolate=True, + extra=f"--final-cat {RUN_DIR}/tiles/{wc.tile}/final_cat-{wc.tile}.fits"), + shell: + "{params.cmd}" diff --git a/workflow/scripts/build_forest.py b/workflow/scripts/build_forest.py new file mode 100644 index 000000000..82df8b0c6 --- /dev/null +++ b/workflow/scripts/build_forest.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""Build a tile's exposure symlink forest (the SP_EXP view). + +A plain script (not a run: block) so the tile chain stays group-compatible. +Reads the tile's exposures from run_index.sqlite and symlinks each exposure's +``exp//output`` into ``//output`` by exact name (no glob), so +the tile modules' $SP_EXP globs resolve to exactly this tile's exposures' +products. The forest is a convenience view; the DAG edge to the exposures is +declared in the rule's input (tile.smk), not here. +""" + +import argparse +import sqlite3 +from pathlib import Path + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--tile", required=True) + p.add_argument("--run-dir", required=True, type=Path) + p.add_argument("--index", required=True, type=Path) + p.add_argument("--forest", required=True, type=Path) + args = p.parse_args() + + con = sqlite3.connect(args.index) + exps = [r[0] for r in con.execute( + "SELECT exp_id FROM tile_exposures WHERE tile_id=?", (args.tile,))] + con.close() + + args.forest.mkdir(parents=True, exist_ok=True) + for e in exps: + src = args.run_dir / "exp" / e / "output" + dst = args.forest / e / "output" + dst.parent.mkdir(parents=True, exist_ok=True) + if dst.is_symlink() or dst.exists(): + dst.unlink() + dst.symlink_to(src) + print(f"[build_forest] {args.tile}: {len(exps)} exposures -> {args.forest}") + + +if __name__ == "__main__": + main() diff --git a/workflow/scripts/build_index.py b/workflow/scripts/build_index.py new file mode 100644 index 000000000..13783314a --- /dev/null +++ b/workflow/scripts/build_index.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""Build the run index (``run_index.sqlite``) that drives the compute DAG. + +The index is *parse-time data*, never a rule input: the Snakefile loads it once +at parse time into plain dicts, so extending a run's tile list changes which +jobs exist without touching the mtime chain of completed work. + +It records, per run: + + tiles(tile_id, ra_dir, n_exp, status) + exposures(exp_id) -- deduplicated union over tiles + tile_exposures(tile_id, exp_id) -- the tile->exposure edges + +The tile->exposure edges are *data-derived*: they are read from each tile's +``find_exposures`` output (``exp_numbers--.txt``), which +``find_exposures_runner`` produces by parsing the tile FITS ``HISTORY`` header. +So this is a **plain script**, not a DAG node: it runs between the two static +invocations — after ``snakemake prepare_tiles`` (Git+Uz+Fe, keep-going) and +before ``snakemake prepare_exposures`` parses the now-existing index. It +iterates the *declared* tile list and checks each tile's Fe output at its +deterministic path (no globbing — O(tile-list) existence checks, respecting the +no-``ls``-at-scale ban). A bad tile costs only that tile: it is recorded in +``missing.json`` and the index is built over the rest. The script exits nonzero +only if the missing *fraction* exceeds ``--missing-threshold`` (default 0.0), +so a keep-going download storm that lost a few tiles does not cost the run. +There is NO ``--allow-missing`` flag (that Snakemake flag does not exist). + +Exposure IDs are stored with their trailing single-char suffix stripped +(``2243881p`` -> ``2243881``); that ``exp_base`` is the dedup key and the +exposure-rule wildcard, matching the ``exp///`` forest layout. +""" + +import argparse +import json +import sqlite3 +import sys +from pathlib import Path + + +def read_exposure_list(exp_numbers_file: Path) -> list[str]: + """Return the exp_base IDs listed in one tile's find_exposures output. + + Each line is an exposure name like ``2243881p``; we strip a trailing + alphabetic suffix so the base ID is the dedup key. + """ + ids = [] + for line in exp_numbers_file.read_text().splitlines(): + name = line.strip() + if not name: + continue + ids.append(name[:-1] if name[-1].isalpha() else name) + return ids + + +def build(tile_ids: list[str], run_dir: Path, db_path: Path, + missing_threshold: float = 0.0) -> dict: + """Build the index over ``tile_ids``; return a summary dict. + + For each tile, check its ``exp_numbers--.txt`` at the tile's + deterministic ``find_exposures`` run dir (RUN_DATETIME=False, no glob). A + tile whose exposure list is missing is recorded in ``missing.json`` and the + index is built over the rest (a bad tile costs that tile, not the run). The + build is fatal only if the missing fraction exceeds ``missing_threshold``. + """ + db_path.parent.mkdir(parents=True, exist_ok=True) + con = sqlite3.connect(db_path) + con.executescript( + """ + DROP TABLE IF EXISTS tiles; + DROP TABLE IF EXISTS exposures; + DROP TABLE IF EXISTS tile_exposures; + CREATE TABLE tiles(tile_id TEXT PRIMARY KEY, ra_dir TEXT, n_exp INTEGER); + CREATE TABLE exposures(exp_id TEXT PRIMARY KEY); + CREATE TABLE tile_exposures( + tile_id TEXT, exp_id TEXT, + PRIMARY KEY (tile_id, exp_id)); + """ + ) + + missing = [] + all_exposures: set[str] = set() + for tile_id in tile_ids: + ra_dir = tile_id.split(".")[0] + idra, iddec = tile_id.split(".") + # Flat forest, deterministic run dir (RUN_DATETIME=False). + exp_file = (run_dir / "tiles" / tile_id / "output" / "run_sp_tile_Fe" / + "find_exposures_runner" / "output" / + f"exp_numbers-{idra}-{iddec}.txt") + if not exp_file.exists(): + missing.append(tile_id) + continue + exp_ids = read_exposure_list(exp_file) + con.execute("INSERT INTO tiles VALUES (?,?,?)", + (tile_id, ra_dir, len(exp_ids))) + for exp_id in exp_ids: + all_exposures.add(exp_id) + con.execute("INSERT OR IGNORE INTO tile_exposures VALUES (?,?)", + (tile_id, exp_id)) + + con.executemany("INSERT OR IGNORE INTO exposures VALUES (?)", + [(e,) for e in sorted(all_exposures)]) + con.commit() + con.close() + + (db_path.parent / "missing.json").write_text(json.dumps(missing, indent=2)) + frac = len(missing) / len(tile_ids) if tile_ids else 0.0 + if frac > missing_threshold: + raise SystemExit( + f"Missing exposure lists for {len(missing)}/{len(tile_ids)} tile(s) " + f"(fraction {frac:.3f} > threshold {missing_threshold}): {missing}. " + f"Re-run prepare_tiles for them, or raise --missing-threshold.") + return {"n_tiles": len(tile_ids) - len(missing), + "n_exposures": len(all_exposures), + "n_missing": len(missing)} + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--tile-list", required=True, type=Path, + help="file of tile IDs, one per line") + p.add_argument("--run-dir", required=True, type=Path, + help="$SP_RUN: root of the tiles/ work-dir forest") + p.add_argument("--db", required=True, type=Path, + help="output run_index.sqlite path") + p.add_argument("--missing-threshold", type=float, default=0.0, + help="fatal if the missing-tile fraction exceeds this " + "(default 0.0: any missing tile is fatal)") + args = p.parse_args() + + tile_ids = [ln.strip() for ln in args.tile_list.read_text().splitlines() + if ln.strip()] + summary = build(tile_ids, args.run_dir, args.db, args.missing_threshold) + print(f"run_index: {summary['n_tiles']} tiles, " + f"{summary['n_exposures']} exposures, " + f"{summary['n_missing']} missing -> {args.db}", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/workflow/scripts/clean_exposures.py b/workflow/scripts/clean_exposures.py new file mode 100644 index 000000000..4f46bf19b --- /dev/null +++ b/workflow/scripts/clean_exposures.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +"""Index-driven exposure-store cleanup (`sp clean-exposures`). + +temp() is deliberately NOT used on the shared exposure directories (findings +4/5/6/19): they overlap tiles by construction, so temp() deletion would cascade +destructive reruns across spatial neighbours whenever a tile is appended. Instead +the store is persistent and space is reclaimed explicitly here. + +For each exposure in run_index.sqlite, delete its bulky intermediates +(run_sp_exp_Sp / run_sp_exp_Ma / run_sp_exp_SxSePsfPi) ONLY when every tile that +consumes it (tile_exposures) has its final_cat-.fits present across the whole +output root. An exposure still feeding an unfinished tile is left untouched. + +Caveat: if a later batch appends a tile that shares a deleted exposure, +Snakemake will regenerate that exposure — cleanup trades peak storage for +recompute. Run it between campaign phases, not mid-DAG. --dry-run by default. +""" + +import argparse +import shutil +import sqlite3 +import sys +from pathlib import Path + +PREFIXES = ("run_sp_exp_Sp", "run_sp_exp_Ma", "run_sp_exp_SxSePsfPi") + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--run-dir", required=True, type=Path) + p.add_argument("--index", required=True, type=Path) + p.add_argument("--apply", action="store_true", + help="actually delete (default: dry-run, list only)") + args = p.parse_args() + + con = sqlite3.connect(args.index) + edges = con.execute("SELECT exp_id, tile_id FROM tile_exposures").fetchall() + con.close() + + consumers: dict[str, list[str]] = {} + for exp_id, tile_id in edges: + consumers.setdefault(exp_id, []).append(tile_id) + + def tile_done(tile): + return (args.run_dir / "tiles" / tile / f"final_cat-{tile}.fits").exists() + + freed, held = 0, 0 + for exp_id, tiles in consumers.items(): + if not all(tile_done(t) for t in tiles): + held += 1 + continue + for prefix in PREFIXES: + d = args.run_dir / "exp" / exp_id / "output" / prefix + if d.is_dir(): + freed += 1 + print(f"{'DELETE' if args.apply else 'would delete'} {d}") + if args.apply: + shutil.rmtree(d) + print(f"[clean_exposures] {freed} dirs {'freed' if args.apply else 'reclaimable'}, " + f"{held} exposures held (consuming tile unfinished)", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/workflow/scripts/completeness.py b/workflow/scripts/completeness.py new file mode 100644 index 000000000..f35e81aee --- /dev/null +++ b/workflow/scripts/completeness.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""The count-floor completeness table — the single failure policy. + +This is the ported ``complete_check`` count table from the v2.0 bash layer +(``run_job_sp_canfar_v2.0.bash`` job dispatch, survey §4). It is the *only* +failure policy in the design: there is no 3-class taxonomy and no error-signature +whitelist. A stage is a real failure iff a mandatory runner produced fewer than +its ``floor`` files; per-CCD attrition (a sparse CCD setools rejects, ~0.2%) +sits between ``floor`` and ``expect`` and is tolerated. + +Two consumers share this table: + * ``sp_rule.py`` — after ``shapepipe_run``, counts products per runner and + exits nonzero if any mandatory runner is below its floor (``keep-going`` + then isolates that unit's cone; ``retries`` handle transient failures). + * ``run_report.py`` — disk-scans finished trees against ``expect`` to + enumerate shortfalls for the human, mid-run or after. + +Per-runner fields: + expect nominal file count for a fully complete unit (report yardstick) + floor the fail-loud minimum (below this the job exits nonzero) + warn if True the runner never fails the unit at all (bash ``:warn`` — + e.g. psfex_interp on tiles missing some epochs) + subpath count files in ``/output//`` instead of + ``/output/`` (bash ``:rand_split`` — setools split cats) + +Counts are file counts in the runner's output dir, matching the bash +``ls / | wc -l`` semantics (broken symlinks excluded by the caller). +""" + +# stage -> {runner_subdir: {expect, floor, [warn], [subpath]}} +COMPLETENESS = { + # --- tile prepare (phase A) --- + "tile_get_images": {"get_images_runner": dict(expect=4, floor=4)}, + "tile_uncompress": {"uncompress_fits_runner": dict(expect=1, floor=1)}, + "tile_find_exposures": {"find_exposures_runner": dict(expect=1, floor=1)}, + + # --- exposure chain --- + "exp_get_images": {"get_images_runner": dict(expect=6, floor=6)}, + "exp_split": {"split_exp_runner": dict(expect=121, floor=41)}, + "exp_mask": {"mask_runner": dict(expect=40, floor=1)}, + "exp_psf": { + "sextractor_runner": dict(expect=80, floor=2), + "setools_runner": dict(expect=80, floor=2, subpath="rand_split"), + "psfex_runner": dict(expect=80, floor=2), + "psfex_interp_runner": dict(expect=40, floor=0, warn=True), + }, + + # --- tile post --- + "tile_merge_headers": {"merge_headers_runner": dict(expect=1, floor=1)}, + "tile_mask": {"mask_runner": dict(expect=1, floor=1)}, + "tile_detect": {"sextractor_runner": dict(expect=2, floor=2)}, + "tile_vignets": { + "psfex_interp_runner": dict(expect=1, floor=1), + "vignetmaker_runner_run_1": dict(expect=1, floor=1), + "vignetmaker_runner_run_2": dict(expect=4, floor=4), + }, + "tile_ngmix": {"ngmix_runner": dict(expect=1, floor=1)}, + "tile_merge_cats": {"merge_sep_cats_runner": dict(expect=1, floor=1)}, + "tile_make_cat": {"make_cat_runner": dict(expect=1, floor=1)}, +} + + +def count_products(run_dir, runner, spec): + """Count files in ``run_dir//output[/]/`` (live links only).""" + out = run_dir / runner / "output" + if "subpath" in spec: + out = out / spec["subpath"] + if not out.is_dir(): + return 0 + return sum(1 for p in out.iterdir() if p.exists()) # p.exists() drops dead links + + +def check_floor(stage, run_dir): + """Return (ok, details). ok is False iff a mandatory runner is below floor. + + ``details`` is a list of (runner, n_found, floor, expect, warn) tuples. + """ + table = COMPLETENESS[stage] + details, ok = [], True + for runner, spec in table.items(): + n = count_products(run_dir, runner, spec) + details.append((runner, n, spec["floor"], spec["expect"], + spec.get("warn", False))) + if not spec.get("warn", False) and n < spec["floor"]: + ok = False + return ok, details diff --git a/workflow/scripts/run_report.py b/workflow/scripts/run_report.py new file mode 100644 index 000000000..bb3300405 --- /dev/null +++ b/workflow/scripts/run_report.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Standalone run report — NOT a DAG node. + +A report rule that declares all tiles' outputs as inputs is a descendant of every +job, so any single hard failure (under --keep-going) poisons its cone and the +report never runs — the exact scenario it exists for (finding 8). So this is a +plain script: it disk-scans the tiles/ and exp/ trees against the count-floor +table (completeness.py) and the index, and emits run_report.json. It runs +automatically via the Snakefile's onsuccess/onerror hooks, and by hand any time +(``sp report``), mid-run included. + +It distinguishes whole-exposure absence (all runners missing — a real gap or a +deletion/forest bug) from tolerated per-CCD attrition (counts between floor and +expect), so a future deletion bug cannot be silently absorbed (finding 5). +""" + +import argparse +import json +import sqlite3 +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from completeness import COMPLETENESS, check_floor # noqa: E402 + +# stage -> (level, run_sp_ dir name) +STAGE_DIR = { + "tile_get_images": ("tile", "run_sp_tile_Git"), + "tile_uncompress": ("tile", "run_sp_tile_Uz"), + "tile_find_exposures": ("tile", "run_sp_tile_Fe"), + "exp_get_images": ("exp", "run_sp_exp_Gie"), + "exp_split": ("exp", "run_sp_exp_Sp"), + "exp_mask": ("exp", "run_sp_exp_Ma"), + "exp_psf": ("exp", "run_sp_exp_SxSePsfPi"), + "tile_merge_headers": ("tile", "run_sp_tile_Mh_exp"), + "tile_mask": ("tile", "run_sp_tile_Ma"), + "tile_detect": ("tile", "run_sp_tile_Sx"), + "tile_vignets": ("tile", "run_sp_tile_PiViVi"), + "tile_merge_cats": ("tile", "run_sp_Ms"), + "tile_make_cat": ("tile", "run_sp_Mc"), +} + + +def classify(stage, run_dir): + """'absent' | 'under' | 'attrition' | 'complete' for one unit's stage.""" + if not run_dir.is_dir(): + return "absent", [] + ok, details = check_floor(stage, run_dir) + if not ok: + # all mandatory runners empty -> whole-unit absence; else below floor + mand = [d for d in details if not d[4]] + if mand and all(n == 0 for _, n, *_ in mand): + return "absent", details + return "under", details + if any(n < expect for _, n, floor, expect, warn in details if not warn): + return "attrition", details + return "complete", details + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--run-dir", required=True, type=Path) + p.add_argument("--index", required=True, type=Path) + p.add_argument("--status", default="manual") + p.add_argument("--out", type=Path, default=None) + args = p.parse_args() + + tiles, exps = [], [] + if args.index.exists(): + con = sqlite3.connect(args.index) + tiles = [r[0] for r in con.execute("SELECT tile_id FROM tiles")] + exps = [r[0] for r in con.execute("SELECT exp_id FROM exposures")] + con.close() + + missing_json = args.index.parent / "missing.json" + missing = json.loads(missing_json.read_text()) if missing_json.exists() else [] + + report = {"status": args.status, "n_tiles": len(tiles), "n_exposures": len(exps), + "missing_tiles": missing, "stages": {}} + for stage, (level, prefix) in STAGE_DIR.items(): + if stage not in COMPLETENESS: + continue + units = tiles if level == "tile" else exps + sub = "tiles" if level == "tile" else "exp" + tally = {"complete": 0, "attrition": 0, "under": [], "absent": []} + for u in units: + run_dir = args.run_dir / sub / u / "output" / prefix + verdict, _ = classify(stage, run_dir) + if verdict in ("complete", "attrition"): + tally[verdict] += 1 + else: + tally[verdict].append(u) + report["stages"][stage] = tally + + finals = [t for t in tiles + if (args.run_dir / "tiles" / t / f"final_cat-{t}.fits").exists()] + report["final_cats"] = {"present": len(finals), "of": len(tiles), + "missing": [t for t in tiles if t not in finals]} + + out = args.out or (args.index.parent / "run_report.json") + out.write_text(json.dumps(report, indent=2)) + print(f"[run_report] {report['final_cats']['present']}/{len(tiles)} final " + f"cats -> {out}", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/workflow/scripts/sp_rule.py b/workflow/scripts/sp_rule.py new file mode 100644 index 000000000..cce92e2c8 --- /dev/null +++ b/workflow/scripts/sp_rule.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +"""Thin per-unit ShapePipe rule wrapper — the single entrypoint every rule calls. + +Under the one-allocation local executor + apptainer software-deployment, +Snakemake runs each rule's shell *inside* the container on the allocation's node. +So this script already runs in-container: it calls ``shapepipe_run`` directly +(never apptainer itself), which is why the workflow never hand-invokes apptainer. + +What it owns — only what is intrinsically ShapePipe's, nothing Snakemake already +does: + +1. **Per-unit work dir + v2.0 isolation furniture.** Each tile/exposure gets its + own ``$SP_RUN`` (``tiles//`` or ``exp//``). Isolation is by work-dir + *content*, the proven v2.0 mechanism — NOT ``-e/--exclusive``: + - a ``cfis`` symlink to the canonical config dir (so ``$SP_CONFIG`` / ``.mask`` + refs resolve); + - ``star_cat_exp`` / ``star_cat_tiles`` symlinks to the shared cat pool; + - for a tile: ``tile_numbers.txt`` = this tile ID (dot format, real data — + what get_images reads); + - for an exposure: a fabricated + ``output/run_sp_tile_Fe/find_exposures_runner/output/exp_numbers-000-000.txt`` + = this exposure ID (v2.0 ``init_exp_work_dir`` pseudo-Fe), so the Gie + config's ``last:find_exposures_runner`` resolver finds exactly one unit. + +2. **Per-unit config copy.** Forces ``RUN_DATETIME = False`` (deterministic + ``output/run_sp_/`` the rule declares as its ``directory()`` output), + sets ``SMP_BATCH_SIZE = {threads}`` (fork width == cpus_per_task, so SLURM + packing density and true per-job process count are one number), and — only + when ``isolate`` (tile-scheme stages + exp split; NEVER get_images / exp_mask / + exp_psf, whose per-CCD/download numbering would make #746 turn tolerated + per-CCD attrition into a whole-exposure hard failure) — ``NUMBER_LIST = -``. + +3. **Log sync at START and end.** ``update_runs_log_file.py`` regenerates + ``log_run_sp.txt`` before ``shapepipe_run`` (after the pre-delete, so + resolution never sees a just-deleted dir) and again after. Skipped for ngmix + chunks: the log is already complete from the prior tile stages, and the N + concurrent chunk jobs share the tile work dir — a full log *rewrite* would + race, whereas ShapePipe's own single-line ``O_APPEND`` of each chunk's run is + atomic. ngmix chunk configs additionally carry *literal* INPUT_DIRs + (deterministic under RUN_DATETIME=False), so they need no log read at all. + +4. **Count-floor check** (``completeness.py`` — the ported bash ``complete_check`` + table, the single failure policy: no 3-class taxonomy, no whitelist). After + the run, count products per mandatory runner and exit nonzero below the floor. + ``keep-going`` isolates that unit's cone; per-CCD attrition between floor and + ``expect`` is tolerated. Stages absent from the table fall back to a + zero-output floor. + +Stage-specific verbs (grounded in ``runs/p3-batch1/job_p3_batch1.sh``): + +- ``--ngmix-chunk K --ngmix-nchunks N`` — read this tile's SExtractor object + count from its ``run_sp_tile_Sx`` output, split ``[1, N_obj]`` into N disjoint + closed ID ranges, and generate chunk K's config from + ``config_tile_Ng_template_batch.ini``. +- ``--n-split-max N`` — generate the merge config from + ``config_merge_sep_cats_template.ini`` with ``N_SPLIT_MAX = N``. +- ``--final-cat PATH`` — after ``make_cat``, copy the produced ``final_cat`` FITS + to PATH (the rule's declared, protected science-product output). + +Run by hand to reproduce exactly what a rule does (the escape hatch — it +pre-deletes the run dir, just like Snakemake's delete-before-rerun):: + + python sp_rule.py --stage exp_split --config config_exp_Sp.ini \\ + --unit 2243881 --level exp --run-dir $SP_RUN --config-src $SP_CONFIG \\ + --star-cats $STAR_CATS --threads 8 +""" + +import argparse +import os +import shutil +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from completeness import COMPLETENESS, check_floor # noqa: E402 + + +def unit_work_dir(run_dir: Path, level: str, unit: str) -> Path: + """Per-unit work dir in the forest: tiles/ or exp/. + + Flat (no IDra/prefix sharding): the shard existed only to keep the bash + layer's ``ls exp/`` bearable, but Snakemake never globs the forest — the + run index drives every path — so flattening keeps single-wildcard rules. + ``run`` level (aggregation stages) work directly in ``run_dir``. + """ + if level == "tile": + return run_dir / "tiles" / unit + if level == "exp": + base = unit[:-1] if unit[-1].isalpha() else unit + return run_dir / "exp" / base + return run_dir + + +def materialise_unit(work: Path, level: str, unit: str, config_src: Path, + star_cats: Path) -> None: + """Create the v2.0 unit-isolation furniture in the work dir.""" + (work / "output").mkdir(parents=True, exist_ok=True) + + cfis = work / "cfis" # $SP_CONFIG -> canonical config dir (.mask refs resolve) + if not cfis.exists(): + cfis.symlink_to(config_src) + + for name, sub in (("star_cat_exp", "exp"), ("star_cat_tiles", "tiles")): + link = work / name + if not link.exists() and (star_cats / sub).exists(): + link.symlink_to(star_cats / sub) + + if level == "tile": + (work / "tile_numbers.txt").write_text(unit + "\n") # dot format, real data + elif level == "exp": + fe = work / "output" / "run_sp_tile_Fe" / "find_exposures_runner" / "output" + fe.mkdir(parents=True, exist_ok=True) + exp_numbers = fe / "exp_numbers-000-000.txt" + if not exp_numbers.exists(): + exp_numbers.write_text(unit + "\n") + + +def normalize_config_text(text: str, unit: str, isolate: bool, threads: int) -> str: + """Force RUN_DATETIME=False, SMP_BATCH_SIZE=threads, and (if isolating) + NUMBER_LIST=-. + + ``NUMBER_LIST`` uses ShapePipe's image-number convention (dots -> dashes), + the ``set_config_number_list`` mechanism that replaced the retired + ``-e/--exclusive`` flag (#746). + """ + lines = text.splitlines() + out, in_file, saw_datetime, saw_numberlist, saw_smp = [], False, False, False, False + number = "-" + unit.replace(".", "-") + for line in lines: + stripped = line.strip() + if stripped.startswith("[") and stripped.endswith("]"): + in_file = stripped.upper() == "[FILE]" + if stripped.upper().startswith("RUN_DATETIME"): + out.append("RUN_DATETIME = False") + saw_datetime = True + continue + if stripped.upper().startswith("SMP_BATCH_SIZE"): + out.append(f"SMP_BATCH_SIZE = {threads}") + saw_smp = True + continue + if isolate and stripped.upper().startswith("NUMBER_LIST"): + out.append(f"NUMBER_LIST = {number}") + saw_numberlist = True + continue + out.append(line) + if in_file and stripped.upper() == "[FILE]" and isolate and not saw_numberlist: + out.append(f"NUMBER_LIST = {number}") + saw_numberlist = True + if not saw_datetime: + # ShapePipe reads RUN_DATETIME from [DEFAULT]; inject after its header. + patched, done = [], False + for line in out: + patched.append(line) + if not done and line.strip().upper() == "[DEFAULT]": + patched.append("RUN_DATETIME = False") + done = True + out = patched if done else ["[DEFAULT]", "RUN_DATETIME = False"] + out + if not saw_smp: + # SMP_BATCH_SIZE lives in [JOB]; inject after its header. + patched, done = [], False + for line in out: + patched.append(line) + if not done and line.strip().upper() == "[JOB]": + patched.append(f"SMP_BATCH_SIZE = {threads}") + done = True + out = patched if done else out + ["[JOB]", f"SMP_BATCH_SIZE = {threads}"] + return "\n".join(out) + "\n" + + +def run_prefix_from_text(text: str, config_label: str) -> str: + """The RUN_NAME (== run_sp_) the module will write under.""" + for line in text.splitlines(): + s = line.strip() + if s.upper().startswith("RUN_NAME"): + return s.split("=", 1)[1].strip() + raise SystemExit(f"no RUN_NAME in {config_label}") + + +# --- ngmix chunking -------------------------------------------------------- + +# Literal, deterministic INPUT_DIRs for the ngmix chunk (RUN_DATETIME=False), so +# concurrent chunks need no log read at all (finding 18). +NGMIX_LITERAL_INPUT = ( + "./output/run_sp_tile_Sx/sextractor_runner/output, " + "./output/run_sp_tile_PiViVi/psfex_interp_runner/output, " + "./output/run_sp_tile_PiViVi/vignetmaker_runner_run_2/output, " + "./output/run_sp_tile_Mh_exp/merge_headers_runner/output") + + +def ngmix_id_ranges(n_obj: int, n_chunks: int) -> list[tuple[int, int]]: + """Split ``[1, n_obj]`` into ``n_chunks`` disjoint *closed* ID ranges. + + SExtractor's ``NUMBER`` column (what ngmix reads as ``obj_id``) runs 1..N + contiguous, so covering ``[1, N]`` processes every object exactly once. The + first n_chunks-1 chunks get ``n_obj // n_chunks`` each; the last gets the + remainder — a *closed* upper bound at ``n_obj`` (never ``ID_OBJ_MAX = -1``: + ngmix treats ``id_obj_max <= 0`` as unbounded, which would double-count). + + Deviation from the nibi monolith (deliberate): the monolith sizes every + tile's chunks from the *average* object count across all tiles and leaves the + last chunk open (``-1``); here each tile is an independent job, so we size + from *this* tile's own count and close the last chunk. The merged catalogue + is identical either way (merge_sep_cats recombines all chunks, ngmix fits + each object independently) — this just removes the open-ended overflow. + """ + base, rem = divmod(n_obj, n_chunks) + ranges, lo = [], 1 + for k in range(1, n_chunks + 1): + size = base + (rem if k == n_chunks else 0) + hi = lo + size - 1 + ranges.append((lo, hi)) + lo = hi + 1 + return ranges + + +def tile_object_count(work: Path) -> int: + """Object count for this tile from its run_sp_tile_Sx SExtractor catalogue. + + Matches ``get_number_objects.py``: the ``NAXIS2`` of the catalogue's last + HDU. Per-unit isolation means exactly one ``sexcat*.fits`` lives here. + """ + from astropy.io import fits + + sexcats = sorted((work / "output" / "run_sp_tile_Sx").glob( + "sextractor_runner/output/sexcat*.fits")) + if not sexcats: + raise SystemExit( + f"[sp_rule] FATAL ngmix: no sexcat under {work}/output/run_sp_tile_Sx") + with fits.open(sexcats[0]) as hdul: + return int(hdul[-1].header["NAXIS2"]) + + +def apply_ngmix_subs(text: str, chunk: int, lo: int, hi: int) -> str: + """Fill the ngmix batch template for one chunk (the monolith's sed recipe) + plus literal INPUT_DIRs so the chunk needs no shared-log read.""" + text = text.replace("NgXu", f"Ng{chunk}u").replace("X_interp", "psfex_interp") + out = [] + for line in text.splitlines(): + s = line.strip() + if s.startswith("ID_OBJ_MIN"): + out.append(f"ID_OBJ_MIN = {lo}") + elif s.startswith("ID_OBJ_MAX"): + out.append(f"ID_OBJ_MAX = {hi}") + elif s.startswith("INPUT_DIR") and "sextractor_runner" in s: + out.append(f"INPUT_DIR = {NGMIX_LITERAL_INPUT}") + else: + out.append(line) + return "\n".join(out) + + +def apply_merge_subs(text: str, n_split_max: int) -> str: + """Fill N_SPLIT_MAX in the merge template (the monolith's sed recipe).""" + out = [] + for line in text.splitlines(): + if line.strip().startswith("N_SPLIT_MAX"): + out.append(f"N_SPLIT_MAX = {n_split_max}") + else: + out.append(line) + return "\n".join(out) + + +# --- in-job count floor ---------------------------------------------------- + +def count_floor(run_out: Path, stage: str) -> None: + """Fail loudly if any mandatory runner is below its ``floor`` (completeness.py). + + Stages absent from the table skip this floor (only the zero-output floor + applies to them). + """ + if stage not in COMPLETENESS: + return + ok, details = check_floor(stage, run_out) + for runner, n, floor, expect, warn in details: + tag = "warn" if warn else ("OK" if n >= floor else " <-- BELOW floor") + print(f"[sp_rule] {runner}: {n}/{expect} (floor {floor}) {tag}", + file=sys.stderr) + if not ok: + print(f"[sp_rule] FATAL {stage}: count floor breached in {run_out}", + file=sys.stderr) + sys.exit(1) + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--stage", required=True) + p.add_argument("--config", required=True, help="config filename in --config-src") + p.add_argument("--unit", required=True) + p.add_argument("--level", required=True, choices=["tile", "exp", "run"]) + p.add_argument("--run-dir", required=True, type=Path, help="$SP_RUN root") + p.add_argument("--config-src", required=True, type=Path, + help="dir holding the canonical configs") + p.add_argument("--star-cats", type=Path, default=None, + help="shared star-cat pool (tiles/ + exp/ subdirs)") + p.add_argument("--threads", type=int, default=1, + help="SMP_BATCH_SIZE == cpus_per_task (fork width)") + p.add_argument("--exp-forest", type=Path, default=None, + help="$SP_EXP: per-tile exposure forest (tile post-stages)") + p.add_argument("--no-isolate", action="store_true", + help="skip NUMBER_LIST (get_images / exp_mask / exp_psf)") + p.add_argument("--ngmix-chunk", type=int, default=None, + help="1-indexed ngmix chunk K (with --ngmix-nchunks)") + p.add_argument("--ngmix-nchunks", type=int, default=None, + help="total ngmix chunks N (with --ngmix-chunk)") + p.add_argument("--n-split-max", type=int, default=None, + help="N_SPLIT_MAX for the merge_sep_cats template") + p.add_argument("--final-cat", type=Path, default=None, + help="copy the produced final_cat FITS here (make_cat)") + args = p.parse_args() + + ngmix_mode = args.ngmix_chunk is not None + if ngmix_mode and args.ngmix_nchunks is None: + raise SystemExit("--ngmix-chunk requires --ngmix-nchunks") + + work = unit_work_dir(args.run_dir, args.level, args.unit) + materialise_unit(work, args.level, args.unit, args.config_src, + args.star_cats or args.run_dir) + + # Build the config copy. ngmix chunks and merge fill template placeholders + # first; every copy is then normalized (RUN_DATETIME=False, SMP_BATCH_SIZE, + # [+ NUMBER_LIST]). + src_text = (args.config_src / args.config).read_text() + isolate = not args.no_isolate + if ngmix_mode: + n_obj = tile_object_count(work) + lo, hi = ngmix_id_ranges(n_obj, args.ngmix_nchunks)[args.ngmix_chunk - 1] + print(f"[sp_rule] ngmix chunk {args.ngmix_chunk}/{args.ngmix_nchunks}: " + f"n_obj={n_obj} -> ID_OBJ [{lo}, {hi}]", file=sys.stderr) + src_text = apply_ngmix_subs(src_text, args.ngmix_chunk, lo, hi) + cfg_name = f"config_tile_Ng{args.ngmix_chunk}u.ini" # distinct per chunk + elif args.n_split_max is not None: + src_text = apply_merge_subs(src_text, args.n_split_max) + cfg_name = args.config + else: + cfg_name = args.config + + cfg_mod_dir = work / "cfis_mod" + cfg_mod = cfg_mod_dir / cfg_name + cfg_text = normalize_config_text(src_text, args.unit, isolate, args.threads) + cfg_mod_dir.mkdir(parents=True, exist_ok=True) + cfg_mod.write_text(cfg_text) + + prefix = run_prefix_from_text(cfg_text, str(cfg_mod)) + run_out = work / "output" / prefix + + # Pre-delete the deterministic run dir (delete-before-rerun for the hand-run + # escape hatch; ShapePipe's FileHandler.mkdir raises on an existing run dir). + if run_out.exists(): + shutil.rmtree(run_out) + + env = dict(os.environ) + env["SP_RUN"] = str(work) + env["SP_CONFIG"] = str(work / "cfis") # canonical dir (.mask refs), not cfis_mod + if args.exp_forest: + env["SP_EXP"] = str(args.exp_forest) + # Thread caps also come via apptainer-args; keep here for the hand-run path. + env.setdefault("OMP_NUM_THREADS", "1") + env.setdefault("OPENBLAS_NUM_THREADS", "1") + env.setdefault("MKL_NUM_THREADS", "1") + env.setdefault("NUMEXPR_NUM_THREADS", "1") + env.setdefault("MALLOC_ARENA_MAX", "2") + env.setdefault("MALLOC_TRIM_THRESHOLD_", "0") + + # Log sync before the run (after pre-delete). Skipped for ngmix chunks: N + # concurrent chunks would race a full rewrite (literal INPUT_DIRs make it moot). + if not ngmix_mode: + subprocess.call(["update_runs_log_file.py"], cwd=str(work), env=env) + + print(f"[sp_rule] stage={args.stage} unit={args.unit} work={work}", + file=sys.stderr) + rc = subprocess.call(["shapepipe_run", "-c", str(cfg_mod)], + cwd=str(work), env=env) + if rc != 0: + print(f"[sp_rule] WARN shapepipe_run exit {rc} — checking count floor", + file=sys.stderr) + + if not ngmix_mode: + subprocess.call(["update_runs_log_file.py"], cwd=str(work), env=env) + + # Zero-output floor for stages without a count table; count floor otherwise. + if args.stage in COMPLETENESS: + count_floor(run_out, args.stage) + else: + produced = list(run_out.glob("**/output/*")) if run_out.exists() else [] + if not produced: + print(f"[sp_rule] FATAL {args.stage} {args.unit}: zero output in {run_out}", + file=sys.stderr) + sys.exit(1) + + # make_cat: publish the science product to its declared, protected path. + if args.final_cat is not None: + finals = sorted(run_out.glob("**/output/final_cat*.fits")) + if not finals: + print(f"[sp_rule] FATAL {args.stage} {args.unit}: no final_cat under {run_out}", + file=sys.stderr) + sys.exit(1) + args.final_cat.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(finals[0], args.final_cat) + print(f"[sp_rule] final_cat -> {args.final_cat}", file=sys.stderr) + + print(f"[sp_rule] OK {args.stage} {args.unit}", file=sys.stderr) + + +if __name__ == "__main__": + main() From 187950c4ec95035b0e1b40ffaf69841122c31f89 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 15:57:40 -0400 Subject: [PATCH 2/9] fix(orchestration): resolve {threads} at rule level, not inside params.cmd MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Snakemake formats params once, so a {threads} placeholder inside params.cmd survives literally and SMP_BATCH_SIZE never resolves (fork width silently 1, serial modules — the p3-first10 failure mode). Move '--threads {threads}' into each rule's shell: string; tile_exp_forest takes none (build_forest.py is single-threaded symlinking and has no such flag). Completeness table doc touch-ups ride along. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NA8M1oLZLbWJxwoyTNrTAi --- workflow/Snakefile | 6 ++++-- workflow/rules/exposure.smk | 8 ++++---- workflow/rules/prepare.smk | 6 +++--- workflow/rules/tile.smk | 15 ++++++++------- workflow/scripts/completeness.py | 8 ++++++-- 5 files changed, 25 insertions(+), 18 deletions(-) diff --git a/workflow/Snakefile b/workflow/Snakefile index 04722a73c..09d218318 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -60,13 +60,15 @@ def sp_rule(stage, config_name, level, unit, *, isolate=True, exp_forest=None, NUMBER_LIST is injected by default (tile-scheme stages + exp split); pass ``isolate=False`` for get_images / exp_mask / exp_psf, whose per-CCD/download - numbering must not be constrained. ``{threads}`` is filled by Snakemake and + numbering must not be constrained. ``--threads {threads}`` is appended in + each rule's ``shell:`` string (NOT here: Snakemake formats the shell string + once, so a ``{threads}`` inside params.cmd would survive literally) and becomes SMP_BATCH_SIZE (fork width == cpus_per_task). """ cmd = ( f"python {SCRIPTS}/sp_rule.py --stage {stage} --config {config_name} " f"--unit {unit} --level {level} --run-dir {RUN_DIR} " - f"--config-src {CONFIG_SRC} --star-cats {STAR_CATS} --threads {{threads}}" + f"--config-src {CONFIG_SRC} --star-cats {STAR_CATS}" ) if not isolate: cmd += " --no-isolate" diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk index 50403d386..3e934b185 100644 --- a/workflow/rules/exposure.smk +++ b/workflow/rules/exposure.smk @@ -29,7 +29,7 @@ rule exp_get_images: cmd=lambda wc: sp_rule("exp_get_images", "config_exp_Gie_vos.ini", "exp", wc.exp, isolate=False) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # Split multi-HDU exposure into single-CCD files (+ headers-*.npy). Bulk output; # persistent (see module docstring on temp()). @@ -41,7 +41,7 @@ rule exp_split: params: cmd=lambda wc: sp_rule("exp_split", "config_exp_Sp.ini", "exp", wc.exp) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # Mask per-CCD; this exposure's star cat is a declared input (per-unit re-key). rule exp_mask: @@ -54,7 +54,7 @@ rule exp_mask: cmd=lambda wc: sp_rule("exp_mask", "config_exp_Ma_onthefly.ini", "exp", wc.exp, isolate=False) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # SExtractor -> setools star selection -> PSFEx model -> psfex_interp, per CCD. # setools may reject a sparse CCD (~0.2% attrition) — tolerated by the floor. @@ -67,4 +67,4 @@ rule exp_psf: cmd=lambda wc: sp_rule("exp_psf", "config_exp_psfex.ini", "exp", wc.exp, isolate=False) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" diff --git a/workflow/rules/prepare.smk b/workflow/rules/prepare.smk index 2911c3b19..4ca369ec9 100644 --- a/workflow/rules/prepare.smk +++ b/workflow/rules/prepare.smk @@ -19,7 +19,7 @@ rule tile_get_images: cmd=lambda wc: sp_rule("tile_get_images", "config_tile_Git_vos.ini", "tile", wc.tile, isolate=False) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" rule tile_uncompress: input: @@ -30,7 +30,7 @@ rule tile_uncompress: cmd=lambda wc: sp_rule("tile_uncompress", "config_tile_Uz.ini", "tile", wc.tile, isolate=True) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # find_exposures reads the tile FITS HISTORY header -> exp_numbers-.txt, the # data-derived tile->exposure edge build_index.py aggregates (as a plain script). @@ -43,7 +43,7 @@ rule tile_find_exposures: cmd=lambda wc: sp_rule("tile_find_exposures", "config_tile_Fe.ini", "tile", wc.tile, isolate=True) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # Star catalogues for masking are pre-generated offline (create_star_cat.py on a # networked login node — the one genuinely networked prepare step). They are diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index 845167710..8abfdd18b 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -35,6 +35,7 @@ rule tile_exp_forest: cmd=lambda wc: (f"python {SCRIPTS}/build_forest.py --tile {wc.tile} " f"--run-dir {RUN_DIR} --index {INDEX_DB} --forest {{output}}") shell: + # no --threads: build_forest.py is single-threaded symlinking, takes no such flag "{params.cmd}" # Merge single-exposure WCS headers into the tile-level sqlite log @@ -52,7 +53,7 @@ rule tile_merge_headers: "tile", wc.tile, isolate=True, exp_forest=FOREST.format(tile=wc.tile)) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # Mask tiles; this tile's star cat is a declared input (per-unit re-key). rule tile_mask: @@ -65,7 +66,7 @@ rule tile_mask: cmd=lambda wc: sp_rule("tile_mask", "config_tile_Ma_onthefly.ini", "tile", wc.tile, isolate=True) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # SExtractor object detection on the tile. rule tile_detect: @@ -78,7 +79,7 @@ rule tile_detect: cmd=lambda wc: sp_rule("tile_detect", "config_tile_Sx.ini", "tile", wc.tile, isolate=True) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # PSFEx interpolation to galaxies + vignet postage stamps (reads exposure PSF # models + split through the forest — those dirs are explicit inputs). @@ -95,7 +96,7 @@ rule tile_vignets: "tile", wc.tile, isolate=True, exp_forest=FOREST.format(tile=wc.tile)) shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # ngmix shape measurement — static N chunks. Each chunk computes its own closed # object-ID range in-job from the tile's own sexcat (a declared input); chunks @@ -111,7 +112,7 @@ rule tile_ngmix: "tile_ngmix", "config_tile_Ng_template_batch.ini", "tile", wc.tile, extra=f"--ngmix-chunk {wc.chunk} --ngmix-nchunks {NGMIX_CHUNKS}"), shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" def ngmix_chunks(wc): return [f"{RUN_DIR}/tiles/{wc.tile}/output/run_sp_tile_ngmix_Ng{k}u" @@ -129,7 +130,7 @@ rule tile_merge_cats: "tile", wc.tile, isolate=True, extra=f"--n-split-max {NGMIX_CHUNKS}"), shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" # Build the final catalogue: the run's science product. protected() so a # code/params-drift rerun cannot silently discard a finished catalogue — the @@ -144,4 +145,4 @@ rule tile_make_cat: "tile", wc.tile, isolate=True, extra=f"--final-cat {RUN_DIR}/tiles/{wc.tile}/final_cat-{wc.tile}.fits"), shell: - "{params.cmd}" + "{params.cmd} --threads {threads}" diff --git a/workflow/scripts/completeness.py b/workflow/scripts/completeness.py index f35e81aee..d52ad8d0a 100644 --- a/workflow/scripts/completeness.py +++ b/workflow/scripts/completeness.py @@ -30,12 +30,16 @@ # stage -> {runner_subdir: {expect, floor, [warn], [subpath]}} COMPLETENESS = { # --- tile prepare (phase A) --- - "tile_get_images": {"get_images_runner": dict(expect=4, floor=4)}, + # get_images counts are CONFIG-FLAVOR-DEPENDENT: the v2.0 bash table said 4/6 + # for the canfar vos flavor; the nibi symlink configs produce one file per + # INPUT_FILE_PATTERN entry (tile: image+weight=2; exp: image+weight+flag=3), + # verified against the p3-batch1 baseline tree (100 files / 50 tiles). + "tile_get_images": {"get_images_runner": dict(expect=2, floor=2)}, "tile_uncompress": {"uncompress_fits_runner": dict(expect=1, floor=1)}, "tile_find_exposures": {"find_exposures_runner": dict(expect=1, floor=1)}, # --- exposure chain --- - "exp_get_images": {"get_images_runner": dict(expect=6, floor=6)}, + "exp_get_images": {"get_images_runner": dict(expect=3, floor=3)}, "exp_split": {"split_exp_runner": dict(expect=121, floor=41)}, "exp_mask": {"mask_runner": dict(expect=40, floor=1)}, "exp_psf": { From ac6bddd7f5eddf1c9e514a4e759a1c4db7660dff Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 16:07:59 -0400 Subject: [PATCH 3/9] fix(orchestration): carry original exposure names through the index; star-cat store naming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three P0-blocking fixes found by running against real data: - exposures(name): the fabricated per-unit exp_numbers list must carry the original exposure name verbatim (2605805p — get_images matches .fits.fz; the bare dedup id matches nothing). The index now stores both; exp_get_images passes --exp-name from the parse-time dict. - tile_star_cat: the pre-generated store names tiles in ShapePipe's image-number convention (dots->dashes); translate when linking. - profile: PYTHONPATH pinned to this branch's src/ (identical to develop@97e16d50 — orchestration commits never touch src/); NOT shapepipe-prod (drifted to a PR branch mid-run, and the live p3-batch1 job reads it) and not the sif default (frozen pre-#843). Run scoped to the 210/211 overlap quad (19 unique exposures, 15 shared — exercises structural dedup; the zero-overlap 196 quad is kept for the append-invariant test). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NA8M1oLZLbWJxwoyTNrTAi --- profiles/nibi/config.yaml | 12 ++++++----- workflow/Snakefile | 8 +++++--- workflow/config.yaml | 9 ++++++--- workflow/rules/exposure.smk | 3 ++- workflow/rules/prepare.smk | 4 +++- workflow/scripts/build_index.py | 35 ++++++++++++++++++--------------- workflow/scripts/sp_rule.py | 17 ++++++++++++---- 7 files changed, 55 insertions(+), 33 deletions(-) diff --git a/profiles/nibi/config.yaml b/profiles/nibi/config.yaml index e977803c1..07f68833a 100644 --- a/profiles/nibi/config.yaml +++ b/profiles/nibi/config.yaml @@ -23,11 +23,13 @@ software-deployment-method: [apptainer] # OMP_NUM_THREADS=1 caps OpenBLAS fork-explosion (verified: pool 32->1) # MALLOC_ARENA_MAX=2 bounded allocator (im_sims/nibi lesson) # --home /home/cdaley wins over the SDM's --home ; restores cadcproxy.pem for vos/vcp -# PYTHONPATH SETTLED CALL 3 — P0 pins the prod worktree src so science -# content matches the bash validation runs (which use this -# override). Production later rebuilds the sif at the exact -# validation commit and DROPS this --env line. -apptainer-args: "--cleanenv --env OMP_NUM_THREADS=1 --env MALLOC_ARENA_MAX=2 --env PYTHONPATH=/project/def-mjhudson/cdaley/shapepipe-prod/src --home /home/cdaley --bind /project --bind /scratch" +# PYTHONPATH SETTLED CALL 3 — P0 pins THIS branch's src/ (identical to +# develop@97e16d50: orchestration commits never touch src/). +# NOT shapepipe-prod (drifted to a PR branch mid-run, and the +# live p3-batch1 job reads it) and not the sif default +# (frozen pre-#843). Production later rebuilds the sif at the +# validated commit and DROPS this --env line. +apptainer-args: "--cleanenv --env OMP_NUM_THREADS=1 --env MALLOC_ARENA_MAX=2 --env PYTHONPATH=/project/def-mjhudson/cdaley/shapepipe-snakemake/src --home /home/cdaley --bind /project --bind /scratch" latency-wait: 60 # NFS: wait for outputs to appear after a job keep-going: true # a failed job poisons only its cone; siblings run on diff --git a/workflow/Snakefile b/workflow/Snakefile index 09d218318..6e391bda2 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -39,11 +39,13 @@ with open(config["tile_list"]) as f: TILES = [ln.strip() for ln in f if ln.strip()] # --- parse-time index load (compute phase only) ---------------------------- -# EXP: global set of exposure base-ids. TILE_EXP: tile -> [exp_ids]. -EXP, TILE_EXP = set(), {} +# EXP: exposure base-id -> original name (2605805 -> 2605805p; the name goes +# verbatim into the fabricated per-unit exp_numbers list so get_images matches +# .fits.fz in the store). TILE_EXP: tile -> [exp_ids]. +EXP, TILE_EXP = {}, {} if INDEX_DB.exists(): _con = sqlite3.connect(INDEX_DB) - EXP = {r[0] for r in _con.execute("SELECT exp_id FROM exposures")} + EXP = dict(_con.execute("SELECT exp_id, name FROM exposures")) for tile, exp in _con.execute("SELECT tile_id, exp_id FROM tile_exposures"): TILE_EXP.setdefault(tile, []).append(exp) _con.close() diff --git a/workflow/config.yaml b/workflow/config.yaml index 53542d1c2..ec529c4dc 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -5,14 +5,17 @@ # tiles) never invalidates completed work — it only changes which jobs exist. # The tile list that scopes this run (one "IDra.IDdec" per line). -tile_list: /scratch/cdaley/shapepipe-output/snakemake-p0/tiles4.txt +# 210/211 quad: 19 unique exposures, 15 reused across tiles — exercises the +# structural exposure dedup (the 196 quad had zero overlap; kept for the +# append-invariant test). +tile_list: /scratch/cdaley/shapepipe-output/smk-p0/tiles4.txt # The container every job runs inside (apptainer software-deployment in the profile). container: /project/def-mjhudson/cdaley/containers/shapepipe-develop-runtime.sif # Where products land ($SP_RUN root). Per-unit work-dir forest is built under here: # /tiles/// and /exp/// -run_dir: /scratch/cdaley/shapepipe-output/snakemake-p0 +run_dir: /scratch/cdaley/shapepipe-output/smk-p0 # The canonical config chain (example/cfis in the repo, or a run-specific copy). config_src: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/cfis @@ -23,7 +26,7 @@ star_cats: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/star_cats # The run index. Durable, low-volume products (index, missing.json, run_report, # final_cat) should land on /project, not /scratch (60-day purge, finding 15). -index_db: /project/def-mjhudson/cdaley/shapepipe-output/snakemake-p0/index/run_index.sqlite +index_db: /project/def-mjhudson/cdaley/shapepipe-output/smk-p0/index/run_index.sqlite # ngmix within-tile chunking: static N chunks (closed ID ranges computed # per-tile, in-job, from the tile's own sexcat). diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk index 3e934b185..f39f6be23 100644 --- a/workflow/rules/exposure.smk +++ b/workflow/rules/exposure.smk @@ -27,7 +27,8 @@ rule exp_get_images: directory(f"{EXP_OUT}/run_sp_exp_Gie") params: cmd=lambda wc: sp_rule("exp_get_images", "config_exp_Gie_vos.ini", - "exp", wc.exp, isolate=False) + "exp", wc.exp, isolate=False, + extra=f"--exp-name {EXP[wc.exp]}") shell: "{params.cmd} --threads {threads}" diff --git a/workflow/rules/prepare.smk b/workflow/rules/prepare.smk index 4ca369ec9..530ac9b31 100644 --- a/workflow/rules/prepare.smk +++ b/workflow/rules/prepare.smk @@ -62,6 +62,8 @@ rule tile_star_cat: output: str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits") params: - src=lambda wc: str(STAR_CATS / "tiles" / f"star_cat-{wc.tile}.fits") + # the store names tiles in ShapePipe's image-number convention (dots->dashes) + src=lambda wc: str(STAR_CATS / "tiles" / + f"star_cat-{wc.tile.replace('.', '-')}.fits") shell: "ln -sf {params.src} {output}" diff --git a/workflow/scripts/build_index.py b/workflow/scripts/build_index.py index 13783314a..7ba865d94 100644 --- a/workflow/scripts/build_index.py +++ b/workflow/scripts/build_index.py @@ -37,19 +37,22 @@ from pathlib import Path -def read_exposure_list(exp_numbers_file: Path) -> list[str]: - """Return the exp_base IDs listed in one tile's find_exposures output. - - Each line is an exposure name like ``2243881p``; we strip a trailing - alphabetic suffix so the base ID is the dedup key. +def read_exposure_list(exp_numbers_file: Path) -> list[tuple[str, str]]: + """Return ``(exp_id, name)`` pairs from one tile's find_exposures output. + + Each line is an exposure *name* like ``2243881p``; the bare base ID (suffix + stripped) is the dedup key everywhere in the DAG, but the original name is + kept in the index — the fabricated per-unit ``exp_numbers`` list must carry + it verbatim (``get_images`` matches ``.fits.fz`` in the store; the + bare ID matches nothing). """ - ids = [] + pairs = [] for line in exp_numbers_file.read_text().splitlines(): name = line.strip() if not name: continue - ids.append(name[:-1] if name[-1].isalpha() else name) - return ids + pairs.append((name[:-1] if name[-1].isalpha() else name, name)) + return pairs def build(tile_ids: list[str], run_dir: Path, db_path: Path, @@ -70,7 +73,7 @@ def build(tile_ids: list[str], run_dir: Path, db_path: Path, DROP TABLE IF EXISTS exposures; DROP TABLE IF EXISTS tile_exposures; CREATE TABLE tiles(tile_id TEXT PRIMARY KEY, ra_dir TEXT, n_exp INTEGER); - CREATE TABLE exposures(exp_id TEXT PRIMARY KEY); + CREATE TABLE exposures(exp_id TEXT PRIMARY KEY, name TEXT NOT NULL); CREATE TABLE tile_exposures( tile_id TEXT, exp_id TEXT, PRIMARY KEY (tile_id, exp_id)); @@ -78,7 +81,7 @@ def build(tile_ids: list[str], run_dir: Path, db_path: Path, ) missing = [] - all_exposures: set[str] = set() + all_exposures: set[tuple[str, str]] = set() for tile_id in tile_ids: ra_dir = tile_id.split(".")[0] idra, iddec = tile_id.split(".") @@ -89,16 +92,16 @@ def build(tile_ids: list[str], run_dir: Path, db_path: Path, if not exp_file.exists(): missing.append(tile_id) continue - exp_ids = read_exposure_list(exp_file) + exp_pairs = read_exposure_list(exp_file) con.execute("INSERT INTO tiles VALUES (?,?,?)", - (tile_id, ra_dir, len(exp_ids))) - for exp_id in exp_ids: - all_exposures.add(exp_id) + (tile_id, ra_dir, len(exp_pairs))) + for exp_id, _name in exp_pairs: con.execute("INSERT OR IGNORE INTO tile_exposures VALUES (?,?)", (tile_id, exp_id)) + all_exposures.update(exp_pairs) - con.executemany("INSERT OR IGNORE INTO exposures VALUES (?)", - [(e,) for e in sorted(all_exposures)]) + con.executemany("INSERT OR IGNORE INTO exposures VALUES (?,?)", + sorted(all_exposures)) con.commit() con.close() diff --git a/workflow/scripts/sp_rule.py b/workflow/scripts/sp_rule.py index cce92e2c8..b114fad08 100644 --- a/workflow/scripts/sp_rule.py +++ b/workflow/scripts/sp_rule.py @@ -93,7 +93,7 @@ def unit_work_dir(run_dir: Path, level: str, unit: str) -> Path: def materialise_unit(work: Path, level: str, unit: str, config_src: Path, - star_cats: Path) -> None: + star_cats: Path, exp_name: str | None = None) -> None: """Create the v2.0 unit-isolation furniture in the work dir.""" (work / "output").mkdir(parents=True, exist_ok=True) @@ -112,8 +112,13 @@ def materialise_unit(work: Path, level: str, unit: str, config_src: Path, fe = work / "output" / "run_sp_tile_Fe" / "find_exposures_runner" / "output" fe.mkdir(parents=True, exist_ok=True) exp_numbers = fe / "exp_numbers-000-000.txt" - if not exp_numbers.exists(): - exp_numbers.write_text(unit + "\n") + # Written UNCONDITIONALLY (cheap, deterministic): an exists-guard once + # pinned a stale pre-fix file with the bare base id. Content is the + # ORIGINAL exposure name (with its 'p'-style suffix, from the index) — + # the on-disk store is .fits.fz; the bare base id matches nothing. + # v2.0 copies the name verbatim from the tile's Fe output; the index + # carries it for us. + exp_numbers.write_text((exp_name or unit) + "\n") def normalize_config_text(text: str, unit: str, isolate: bool, threads: int) -> str: @@ -296,6 +301,10 @@ def main() -> None: help="SMP_BATCH_SIZE == cpus_per_task (fork width)") p.add_argument("--exp-forest", type=Path, default=None, help="$SP_EXP: per-tile exposure forest (tile post-stages)") + p.add_argument("--exp-name", default=None, + help="original exposure name incl. suffix (e.g. 2605805p); " + "written into the fabricated exp_numbers list so " + "get_images finds .fits.fz in the store") p.add_argument("--no-isolate", action="store_true", help="skip NUMBER_LIST (get_images / exp_mask / exp_psf)") p.add_argument("--ngmix-chunk", type=int, default=None, @@ -314,7 +323,7 @@ def main() -> None: work = unit_work_dir(args.run_dir, args.level, args.unit) materialise_unit(work, args.level, args.unit, args.config_src, - args.star_cats or args.run_dir) + args.star_cats or args.run_dir, exp_name=args.exp_name) # Build the config copy. ngmix chunks and merge fill template placeholders # first; every copy is then normalized (RUN_DATETIME=False, SMP_BATCH_SIZE, From 9b0456e72438ab7f6f10e304d231100c0e3423ae Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 16:09:45 -0400 Subject: [PATCH 4/9] fix(orchestration): exposure-name pass-through, star-cat dirs, P0 run config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - get_images looks up .fits.fz: the store keeps the CFHT 'p' suffix the index's base-id key strips. exposures gains a name column; rules pass --exp-name; the wrapper writes the ORIGINAL name into the fabricated exp_numbers list — and writes it unconditionally (an exists-guard pinned a stale pre-fix file). - Star cats are consumed as DIRECTORIES via the wrapper's $SP_RUN symlinks (the v2.0 mechanism; exp cats are per-CCD, 40/exposure) — dropped the per-unit star-cat file rules that linked nonexistent names nothing read. - get_images completeness counts follow the nibi symlink configs (tile 2, exp 3; the v2.0 4/6 were the canfar vos flavor), verified against p3-batch1. - P0 run config: 210/211 quad (19 unique exps, 13 shared across tiles — exercises structural dedup; the 196 quad had zero overlap), PYTHONPATH pinned to this branch's src (= develop@97e16d50; shapepipe-prod drifted mid-run). Co-Authored-By: Claude Fable 5 --- workflow/Snakefile | 4 +--- workflow/rules/prepare.smk | 31 ++++++++++--------------------- workflow/rules/tile.smk | 3 +-- 3 files changed, 12 insertions(+), 26 deletions(-) diff --git a/workflow/Snakefile b/workflow/Snakefile index 6e391bda2..774ae3ad2 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -109,9 +109,7 @@ rule prepare_tiles: # index exists. Star cats are re-keyed per unit (see prepare.smk). rule prepare_exposures: input: - expand(str(RUN_DIR / "exp/{exp}/output/run_sp_exp_Gie"), exp=sorted(EXP)), - expand(str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits"), exp=sorted(EXP)), - expand(str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits"), tile=TILES) + expand(str(RUN_DIR / "exp/{exp}/output/run_sp_exp_Gie"), exp=sorted(EXP)) # --- report hooks ----------------------------------------------------------- # run_report is NOT a DAG node (a descendant of every job would be poisoned by diff --git a/workflow/rules/prepare.smk b/workflow/rules/prepare.smk index 530ac9b31..d8698c826 100644 --- a/workflow/rules/prepare.smk +++ b/workflow/rules/prepare.smk @@ -46,24 +46,13 @@ rule tile_find_exposures: "{params.cmd} --threads {threads}" # Star catalogues for masking are pre-generated offline (create_star_cat.py on a -# networked login node — the one genuinely networked prepare step). They are -# re-keyed PER UNIT here: each tile/exposure's cat is its own DAG node, so -# appending tile 51 adds nodes rather than mutating a run-level output that would -# dirty every finished mask (the growth-invariant fix, finding 11). -rule exp_star_cat: - output: - str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits") - params: - src=lambda wc: str(STAR_CATS / "exp" / f"star_cat-{wc.exp}.fits") - shell: - "ln -sf {params.src} {output}" - -rule tile_star_cat: - output: - str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits") - params: - # the store names tiles in ShapePipe's image-number convention (dots->dashes) - src=lambda wc: str(STAR_CATS / "tiles" / - f"star_cat-{wc.tile.replace('.', '-')}.fits") - shell: - "ln -sf {params.src} {output}" +# networked login node) and consumed as DIRECTORIES: the mask configs read +# INPUT_DIR $SP_RUN/star_cat_{exp,tiles} with per-CCD numbering (exp cats are +# star_cat--.fits, 40/exposure; tile cats star_cat--.fits) +# — the v2.0 mechanism. The wrapper materialises the two dir symlinks in every +# unit work dir (sp_rule.materialise_unit); there is NO per-unit star-cat DAG +# node: the store is pre-run input like the image store, and a missing cat fails +# the mask stage's count-floor loudly. (Earlier per-unit file rules linked names +# that don't exist — one file per exposure vs the store's 40 — and nothing read +# them: the configs only see the dir symlinks.) When star-cat GENERATION moves +# in-workflow (P2), it becomes real per-unit rules producing into the store. diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index 8abfdd18b..1bf7d4244 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -55,11 +55,10 @@ rule tile_merge_headers: shell: "{params.cmd} --threads {threads}" -# Mask tiles; this tile's star cat is a declared input (per-unit re-key). +# Mask tiles (star cats: pre-run store, read via the wrapper's dir symlink). rule tile_mask: input: git=rules.tile_get_images.output, - star=str(RUN_DIR / "tiles/{tile}/star_cat-{tile}.fits"), output: directory(f"{TILE_OUT}/run_sp_tile_Ma") params: From 6fe141fc01221eb06b3016a883999195369282fe Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 16:35:07 -0400 Subject: [PATCH 5/9] fix(orchestration): {output} placeholder inside params.cmd survives literally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same single-format trap as {threads} (187950c4): Snakemake formats the shell string once, so the escaped {{output}} in tile_exp_forest's params reached build_forest as a literal './{output}' — all four forest jobs racing one garbage dir (FileExistsError at 50% of the P0 run). Move --forest {output} into the shell string. Swept the rules for further escaped placeholders: none. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ac/code/session_01NA8M1oLZLbWJxwoyTNrTAi --- workflow/rules/tile.smk | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index 1bf7d4244..d36e91d95 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -32,11 +32,14 @@ rule tile_exp_forest: output: directory(FOREST) params: + # --forest {output} lives in the shell string: Snakemake formats shell + # ONCE — an {output} placeholder inside params.cmd survives literally + # (same trap as {threads}; all four forest jobs then race one './{output}'). cmd=lambda wc: (f"python {SCRIPTS}/build_forest.py --tile {wc.tile} " - f"--run-dir {RUN_DIR} --index {INDEX_DB} --forest {{output}}") + f"--run-dir {RUN_DIR} --index {INDEX_DB}") shell: - # no --threads: build_forest.py is single-threaded symlinking, takes no such flag - "{params.cmd}" + # no --threads: build_forest.py is single-threaded symlinking + "{params.cmd} --forest {output}" # Merge single-exposure WCS headers into the tile-level sqlite log # (log_exp_headers--.sqlite — the tile suffix downstream Sx/PiViVi/Ng From a110d89faeca7f0ac6bb2bf6a0ac79862ea044fe Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 16:42:12 -0400 Subject: [PATCH 6/9] =?UTF-8?q?fix(orchestration):=20shard=20the=20exposur?= =?UTF-8?q?e=20forest=20=E2=80=94=20the=20module=20glob's=20shape=20is=20n?= =?UTF-8?q?ot=20cosmetic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit exp_utils.get_exp_output_files hardwires the v2.0 sharded layout into its $SP_EXP glob (///output/run_sp_*/...); a flat forest fails every tile gather stage with 'No split_exp_runner output found'. The exposure STORE stays flat (nothing globs it — the index drives every path); only the module-facing forest view carries the 2-digit shard. Verified by hand-running tile_merge_headers on 210.296 (1/1 OK). Co-Authored-By: Claude Fable 5 --- workflow/scripts/build_forest.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/workflow/scripts/build_forest.py b/workflow/scripts/build_forest.py index 82df8b0c6..86dc4354e 100644 --- a/workflow/scripts/build_forest.py +++ b/workflow/scripts/build_forest.py @@ -3,10 +3,14 @@ A plain script (not a run: block) so the tile chain stays group-compatible. Reads the tile's exposures from run_index.sqlite and symlinks each exposure's -``exp//output`` into ``//output`` by exact name (no glob), so -the tile modules' $SP_EXP globs resolve to exactly this tile's exposures' -products. The forest is a convenience view; the DAG edge to the exposures is -declared in the rule's input (tile.smk), not here. +``exp//output`` into ``///output`` by exact name (no +glob). The 2-digit ```` shard level is NOT cosmetic: ShapePipe's +``exp_utils.get_exp_output_files`` hardwires the sharded v2.0 layout into its +$SP_EXP glob (``///output/run_sp_*/...``), so a flat +forest makes every tile gather stage fail "No split_exp_runner output found". +(The exposure STORE stays flat — only the module-facing view needs the shard.) +The forest is a convenience view; the DAG edge to the exposures is declared in +the rule's input (tile.smk), not here. """ import argparse @@ -30,7 +34,7 @@ def main() -> None: args.forest.mkdir(parents=True, exist_ok=True) for e in exps: src = args.run_dir / "exp" / e / "output" - dst = args.forest / e / "output" + dst = args.forest / e[:2] / e / "output" # sharded: the module glob's shape dst.parent.mkdir(parents=True, exist_ok=True) if dst.is_symlink() or dst.exists(): dst.unlink() From 28c10d8cdca5947184fb5a2d2f831e84468b2c95 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 16:47:15 -0400 Subject: [PATCH 7/9] fix(orchestration): carry a wrapper-scripts content hash into every rule's params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default rerun-triggers' code trigger hashes only the rule's own shell string — NOT external scripts it calls. Bitten live: the forest-shard fix changed build_forest.py but no rule text, so stale flat forests survived a rerun and tile_merge_headers kept failing. SCRIPT_HASH (md5 of sp_rule.py + completeness.py + build_forest.py, computed at parse time) rides into params via an accept-and-ignore --script-hash flag, so script edits propagate exactly like code edits under the full default triggers. One-time cost: every rule's params change once, re-running completed work — the same tradeoff as any code change under full triggers, and the P0-scale rerun doubles as a live test of rerun semantics. Co-Authored-By: Claude Fable 5 --- workflow/Snakefile | 14 +++++++++++++- workflow/rules/tile.smk | 3 ++- workflow/scripts/build_forest.py | 5 +++++ workflow/scripts/sp_rule.py | 5 +++++ 4 files changed, 25 insertions(+), 2 deletions(-) diff --git a/workflow/Snakefile b/workflow/Snakefile index 774ae3ad2..3c869482c 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -70,7 +70,8 @@ def sp_rule(stage, config_name, level, unit, *, isolate=True, exp_forest=None, cmd = ( f"python {SCRIPTS}/sp_rule.py --stage {stage} --config {config_name} " f"--unit {unit} --level {level} --run-dir {RUN_DIR} " - f"--config-src {CONFIG_SRC} --star-cats {STAR_CATS}" + f"--config-src {CONFIG_SRC} --star-cats {STAR_CATS} " + f"--script-hash {SCRIPT_HASH}" ) if not isolate: cmd += " --no-isolate" @@ -86,6 +87,17 @@ def tile_dir(tile): def exp_dir(exp): return RUN_DIR / "exp" / exp +# Content hash of the wrapper scripts, computed once at parse time. Exposed as a +# param on every rule (via sp_rule() and tile_exp_forest) so the default +# rerun-triggers' *params* trigger covers script edits: the `code` trigger only +# hashes the rule's own shell string, NOT external scripts it calls — without +# this, a wrapper/forest-script fix silently leaves stale outputs in place +# (bitten live: the forest-shard fix rebuilt nothing on rerun). +import hashlib as _hashlib +SCRIPT_HASH = _hashlib.md5(b"".join( + (SCRIPTS / n).read_bytes() + for n in ("sp_rule.py", "completeness.py", "build_forest.py"))).hexdigest()[:12] + include: "rules/prepare.smk" include: "rules/exposure.smk" include: "rules/tile.smk" diff --git a/workflow/rules/tile.smk b/workflow/rules/tile.smk index d36e91d95..04cc4c6c9 100644 --- a/workflow/rules/tile.smk +++ b/workflow/rules/tile.smk @@ -36,7 +36,8 @@ rule tile_exp_forest: # ONCE — an {output} placeholder inside params.cmd survives literally # (same trap as {threads}; all four forest jobs then race one './{output}'). cmd=lambda wc: (f"python {SCRIPTS}/build_forest.py --tile {wc.tile} " - f"--run-dir {RUN_DIR} --index {INDEX_DB}") + f"--run-dir {RUN_DIR} --index {INDEX_DB} " + f"--script-hash {SCRIPT_HASH}") shell: # no --threads: build_forest.py is single-threaded symlinking "{params.cmd} --forest {output}" diff --git a/workflow/scripts/build_forest.py b/workflow/scripts/build_forest.py index 86dc4354e..e945820e0 100644 --- a/workflow/scripts/build_forest.py +++ b/workflow/scripts/build_forest.py @@ -24,6 +24,11 @@ def main() -> None: p.add_argument("--run-dir", required=True, type=Path) p.add_argument("--index", required=True, type=Path) p.add_argument("--forest", required=True, type=Path) + p.add_argument("--script-hash", default=None, + help="ignored at runtime: carries the wrapper-scripts content " + "hash into the rule's params so the params rerun-trigger " + "covers script edits (the code trigger only hashes the " + "rule's own shell string)") args = p.parse_args() con = sqlite3.connect(args.index) diff --git a/workflow/scripts/sp_rule.py b/workflow/scripts/sp_rule.py index b114fad08..8ee71c693 100644 --- a/workflow/scripts/sp_rule.py +++ b/workflow/scripts/sp_rule.py @@ -315,6 +315,11 @@ def main() -> None: help="N_SPLIT_MAX for the merge_sep_cats template") p.add_argument("--final-cat", type=Path, default=None, help="copy the produced final_cat FITS here (make_cat)") + p.add_argument("--script-hash", default=None, + help="ignored at runtime: carries the wrapper-scripts content " + "hash into the rule's params so the params rerun-trigger " + "covers script edits (the code trigger only hashes the " + "rule's own shell string)") args = p.parse_args() ngmix_mode = args.ngmix_chunk is not None From a273e76369119113d05251c3c8b33beb3e1ec654 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Thu, 16 Jul 2026 19:59:57 -0400 Subject: [PATCH 8/9] fix(orchestration): keep snakemake state and index off /project (hard-full quota) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit def-mjhudson /project sits at its 27/27 TiB ceiling: a .snakemake metadata write died mid-run (killing an otherwise-green invocation after all 8 ngmix chunks of the first tile passed), and by end of night even git and file edits on /project failed. bin/sp now runs snakemake with --directory -state on /scratch (state only — data paths are absolute), the Snakefile resolves its configfile relative to itself so --directory can't break it, and the run index lives with the run on /scratch until project space is reclaimed (finding 15's /project placement stays the design intent; noted in config). Co-Authored-By: Claude Fable 5 --- workflow/Snakefile | 5 ++++- workflow/bin/sp | 7 ++++++- workflow/config.yaml | 8 +++++--- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/workflow/Snakefile b/workflow/Snakefile index 3c869482c..5d93a1810 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -21,7 +21,10 @@ DAG that produces it). import sqlite3 from pathlib import Path -configfile: "workflow/config.yaml" +# Resolved relative to THIS file, not the working directory: snakemake runs with +# --directory on /scratch (bin/sp) so .snakemake/ state never lands on /project +# (group quota is a hard 27/27 TiB — a metadata write mid-run died on it live). +configfile: str(Path(workflow.snakefile).parent / "config.yaml") # Every job's shell runs inside this container (apptainer software-deployment in # the profile); the user never types apptainer. diff --git a/workflow/bin/sp b/workflow/bin/sp index f226aa441..d1f105548 100755 --- a/workflow/bin/sp +++ b/workflow/bin/sp @@ -34,9 +34,14 @@ source "$VENV/bin/activate" # Minimal scalar reader for workflow/config.yaml (key: value, no nesting). cfg() { sed -n "s/^$1:[[:space:]]*//p" "$CONFIG" | head -1; } RUN_DIR="$(cfg run_dir)"; INDEX_DB="$(cfg index_db)" +# Snakemake state (.snakemake: metadata, locks, incomplete markers) lives NEXT TO +# THE RUN on /scratch, never on /project (hard 27/27 TiB group quota killed a +# metadata write mid-run live). --directory only moves state: all data paths are +# absolute, and the Snakefile resolves its own configfile. +STATE_DIR="${SP_STATE_DIR:-${RUN_DIR}-state}"; mkdir -p "$STATE_DIR" TILE_LIST="$(cfg tile_list)"; MISSING_THRESH="${SP_MISSING_THRESHOLD:-0.0}" -sm() { snakemake --profile "$PROFILE" "$@"; } +sm() { snakemake --profile "$PROFILE" --directory "$STATE_DIR" "$@"; } build_index() { python "$SCRIPTS/build_index.py" --tile-list "$TILE_LIST" \ --run-dir "$RUN_DIR" --db "$INDEX_DB" --missing-threshold "$MISSING_THRESH" diff --git a/workflow/config.yaml b/workflow/config.yaml index ec529c4dc..f92cbb69f 100644 --- a/workflow/config.yaml +++ b/workflow/config.yaml @@ -24,9 +24,11 @@ config_src: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/cfis # Star catalogues for masking, pre-generated (network step done in prepare): star_cats: /home/cdaley/projects/def-mjhudson/cdaley/runs/p3-batch1/star_cats -# The run index. Durable, low-volume products (index, missing.json, run_report, -# final_cat) should land on /project, not /scratch (60-day purge, finding 15). -index_db: /project/def-mjhudson/cdaley/shapepipe-output/smk-p0/index/run_index.sqlite +# The run index. Design intent (finding 15) is durable products on /project — +# BLOCKED for now: def-mjhudson /project is hard-full (27/27 TiB), so index and +# report live with the run on /scratch until space is reclaimed. Mind the +# 60-day purge for anything that must survive. +index_db: /scratch/cdaley/shapepipe-output/smk-p0/index/run_index.sqlite # ngmix within-tile chunking: static N chunks (closed ID ranges computed # per-tile, in-job, from the tile's own sexcat). From 0010def6c3928185b3c84334983d6625538fd749 Mon Sep 17 00:00:00 2001 From: Cail Daley Date: Fri, 17 Jul 2026 01:45:14 -0400 Subject: [PATCH 9/9] fix(orchestration): drop exp_mask's dead per-unit star-cat input MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-unit star_cat file rules were removed (the store is per-CCD and the mask config reads it as a dir symlink), but exp_mask kept declaring exp/{exp}/star_cat-{exp}.fits as input. No producer exists, so the main DAG became unbuildable the moment the tile list grew past the prepared set — MissingInputException on every new exposure. Dormant for P0 (all 19 exposures pre-staged); caught live by the append-invariant test (196 quad appended: DAG build died before scheduling anything). With the fix the append dry-run schedules exactly the appended cone (185 jobs: 30 new exposures x4 stages once each, 4 tile chains) and nothing for the finished tiles; restored-state dry-run returns to 'Nothing to be done'. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_0188niYp9MmhLmZou1KXWbye --- workflow/rules/exposure.smk | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/workflow/rules/exposure.smk b/workflow/rules/exposure.smk index f39f6be23..82cdb4cd4 100644 --- a/workflow/rules/exposure.smk +++ b/workflow/rules/exposure.smk @@ -44,11 +44,14 @@ rule exp_split: shell: "{params.cmd} --threads {threads}" -# Mask per-CCD; this exposure's star cat is a declared input (per-unit re-key). +# Mask per-CCD. Star cats are NOT a per-unit input: the store is per-CCD +# (star_cat--.fits) and the mask config reads it as a DIR symlink +# (see prepare.smk) — a per-unit file input here has no producer and makes the +# DAG unbuildable the moment a new exposure is appended (caught live by the +# append-invariant test). A missing cat fails the mask count-floor loudly. rule exp_mask: input: rules.exp_split.output, - star=str(RUN_DIR / "exp/{exp}/star_cat-{exp}.fits"), output: directory(f"{EXP_OUT}/run_sp_exp_Ma") params: