clean up workflow image sims

docs/source/pipeline_canfar.md

@@ -404,7 +404,7 @@ this requies a job with a large memory, e.g. with `mem=380000`
 
 
 ```bash
-[squeue] python ~/astro/repositories/github/sp_validation/notebooks/extract_info.py
+[squeue] python ~/astro/repositories/github/sp_validation/scripts/calibrate/extract_info.py
 ```
 
 This creates a patch-wise comprehensive catalogue.

profiles/candide/config.yaml

@@ -0,0 +1,58 @@
+# Snakemake profile for the ShapePipe workflow on candide (IAP). Selected by
+# `SP_PROFILE=candide workflow/bin/sp run`. Mirrors profiles/nibi/config.yaml;
+# only the cluster facts differ, and each is stated where it is set.
+executor: slurm
+# candide is a shared ~25-node cluster: cap concurrent submissions well below
+# what nibi takes.
+jobs: 150
+default-resources:
+  mem_mb: 2000
+  runtime: 120                   # minutes
+  slurm_account: cusers
+  # comp nodes have TmpDisk=0, so the tile_shape group (--tmp, see tile.smk's
+  # TILE_SLURM_EXTRA) lands on pscomp; everything else may use either.
+  slurm_partition: "comp,pscomp"
+  # n09/n17/n36: excluded in the sp_validation candide profile; n23: slurm jobs
+  # failed there (sp_validation 2b40c99).
+  slurm_extra: "'--exclude=n09,n17,n23,n36'"
+software-deployment-method: [apptainer]
+# /local/scratch is where tile.smk puts the node-local tile store (TILE_LOCAL,
+# ~5.6 GB per real-data tile). candide has no /local, and its node-local /scratch
+# is writable only per job (/scratch/$USER/$SLURM_JOB_ID, made by the prolog),
+# which this line cannot name: snakemake passes it through literally, and
+# --cleanenv drops SLURM_JOB_ID inside the container. So the node's /tmp (local
+# disk, ~29 GB free, sticky) is bound there instead. That holds ~5 concurrent
+# tile stores per node, too few for dense image-sim tiles: such a campaign sets
+# `tile_store_root:` in its run config, and bin/sp replaces this bind's source in
+# the launch snapshot (see workflow/config.yaml). The PYTHONPATH pin is rewritten by bin/sp to the
+# launch snapshot's src/; the value here only has to be a checkout's src/.
+apptainer-args: "--cleanenv --env OMP_NUM_THREADS=1 --env MALLOC_ARENA_MAX=2 --env PYTHONPATH=/n17data/mkilbing/astro/repositories/github/shapepipe/src --bind /home,/automnt,/n17data,/n23data1,/n09data --bind /tmp:/local/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
+# Transient node-level failures are the common case on a shared cluster:
+# a fused tile_shape group died at joblib's fork_exec with BlockingIOError
+# [Errno 11] because the node was out of process slots, killing one tile of
+# 39 outright. Without this, attempt is always 1, so every
+# `mem_mb = lambda wc, attempt: N * attempt` in tile.smk is dead code and a
+# one-off OOM or EAGAIN is fatal. Safe for the fused group: a retry
+# resubmits the same member set, tile_vignets included (tile.smk's
+# "one sharp edge" note). tile_ngmix keeps its own rule-level retries: 2.
+retries: 2
+rerun-incomplete: true           # re-do jobs left incomplete by an unclean death
+show-failed-logs: true
+printshellcmds: true
+rerun-triggers: [mtime, params, code, software-env]
+# The tile_shape group (tile.smk) sets its own slurm_extra (TILE_SLURM_EXTRA,
+# the --tmp floor), which REPLACES the default above and so drops the node
+# excludes: its jobs landed on n09 and n17. Restate both, identically on all
+# four members (a group's string resources must agree). Only slurm_extra is
+# overridden; the rules keep their own threads and attempt-scaled mem_mb.
+set-resources:
+  tile_vignets:
+    slurm_extra: "'--tmp=16000 --exclude=n09,n17,n23,n36'"
+  tile_ngmix:
+    slurm_extra: "'--tmp=16000 --exclude=n09,n17,n23,n36'"
+  tile_merge_cats:
+    slurm_extra: "'--tmp=16000 --exclude=n09,n17,n23,n36'"
+  tile_make_cat:
+    slurm_extra: "'--tmp=16000 --exclude=n09,n17,n23,n36'"

scripts/python/create_final_cat.py

@@ -27,6 +27,56 @@
 from cs_util import logging
 
 
+def params_from_run_config(params, defaults):
+    """Fill unset paths from a workflow run config.
+
+    The workflow already knows where a campaign writes, so a manual merge
+    should not have to restate it. Resolution goes through the workflow's own
+    resolver (workflow/scripts/run_config.py), layering the run config on
+    workflow/config.yaml and then the machines: table, so what lands here is
+    what the rules would have used.
+
+    Only values still at their default are filled -- an explicit flag always
+    wins. Nothing is derived for the data path: its patch naming differs and
+    is not this function's business.
+    """
+    repo = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    sys.path.insert(0, os.path.join(repo, "workflow", "scripts"))
+    import run_config as _rc
+
+    cfg = _rc.load(os.path.join(repo, "workflow", "config.yaml"),
+                   params["run_config"])
+    if cfg.get("input_type") != "image_sims":
+        raise ValueError(
+            f"run config {params['run_config']} has input_type="
+            f"{cfg.get('input_type', 'data')!r}; -c derives paths for "
+            "image_sims only")
+
+    outputs = cfg.get("outputs") or {}
+    products = outputs.get("products_dir") or outputs.get("run_dir")
+    if not products:
+        raise ValueError(f"run config {params['run_config']} sets neither "
+                         "outputs.products_dir nor outputs.run_dir")
+
+    # The same derivation as the workflow's merge_final_cats rule: the patch
+    # dir is the branch dir holding product/tiles, and -i is its parent.
+    patch_dir = os.path.dirname(os.path.normpath(products))
+    patch = os.path.basename(patch_dir)
+    derived = {
+        "image_sims": True,
+        "input_root_dir": os.path.dirname(patch_dir),
+        "patch": patch,
+        "merged_cat_path": os.path.join(products, f"final_cat_{patch}.hdf5"),
+        "output_summary": os.path.join(products, "n_tiles_final.txt"),
+        "param_path": os.path.join(repo, "workflow", "config",
+                                   "cfis_image_sims", "final_cat.param"),
+    }
+    for key, value in derived.items():
+        if params.get(key) == defaults.get(key):
+            params[key] = value
+    return params
+
+
 def set_params_from_command_line(args):                               
     """Set Params From Command Line.                                        
 
@@ -48,6 +98,12 @@ def set_params_from_command_line(args):
         _params[key] = options[key]                           
 
     del options                                                             
+
+    # A run config fills in whatever is still at its default
+    # (see the docstring); explicit flags always win.
+    if _params.get("run_config"):
+        _defaults, _, _, _ = params_default()
+        _params = params_from_run_config(_params, _defaults)
 
     # Save calling command                                                  
     logging.log_command(args)
@@ -72,6 +128,7 @@ def params_default():
         "ID": None,
         "single_op": None,
         "image_sims": False,
+        "run_config": None,
     }
     _short_options = {
         "input_root_dir": "-i",
@@ -82,6 +139,7 @@ def params_default():
         "output_summary": "-o",
         "single_op": "-s",
         "image_sims": "-I",
+        "run_config": "-c",
     }
     _types = {
         "hdu_num": "int",
@@ -98,6 +156,9 @@ def params_default():
         "ID": "ID for single-ID operation, default={}",
         "single_op": "single ID operation, allowed are 'check', 'add', 'remove'; default={}",
         "image_sims": "image simulations mode (different dir layout and run prefix), default={}",
+        "run_config": "workflow run config (e.g. sp_1p2z_grid_1.yaml); fills"
+                      " in the paths below that were not given explicitly,"
+                      " default={}",
     }
 
     return _params, _short_options, _types, _help_strings
@@ -373,9 +434,17 @@ def collect_tile_ids_image_sims(patch_path):
     list of (tile_id, tile_path) tuples
     """
     id_pattern = re.compile(r"^\d+\.\d+$")
-    tiles_root = os.path.join(patch_path, "tiles")
+    # Two layouts. The Gen-2 / native runs keep tiles under the patch dir
+    # itself; the unified workflow (shapepipe #891) publishes to a separate
+    # products root, and clean_tile deletes the run-dir copies -- so on a
+    # reclaimed campaign <patch>/product/tiles holds the ONLY catalogues.
     result = []
-    if not os.path.isdir(tiles_root):
+    for sub in ("tiles", os.path.join("product", "tiles"),
+                os.path.join("products", "tiles")):
+        tiles_root = os.path.join(patch_path, sub)
+        if os.path.isdir(tiles_root):
+            break
+    else:
         return result
     for prefix in os.listdir(tiles_root):
         prefix_path = os.path.join(tiles_root, prefix)
@@ -387,11 +456,33 @@ def collect_tile_ids_image_sims(patch_path):
     return result
 
 
+def find_final_cat(id, id_path, run_prefix):
+    """Path of this tile's final catalogue, or None.
+
+    Flat layout first: the unified workflow copies one catalogue per tile to
+    <products>/tiles/<shard>/<tile>/final_cat-<tile>.fits, in DOT form and with
+    no run sub-tree. Then the legacy layout, where the catalogue sits under the
+    tile's own shapepipe run dir in DASH form, newest run wins.
+    """
+    flat = os.path.join(id_path, f"final_cat-{id}.fits")
+    if os.path.exists(flat):
+        return flat
+
+    base_pattern = os.path.join(id_path, "output", run_prefix)
+    all_matches = [d for d in glob.glob(base_pattern) if os.path.isdir(d)]
+    if not all_matches:
+        return None
+    newest_dir = max(all_matches, key=os.path.getmtime)
+    id_dash = re.sub(r"\.", "-", id)
+    legacy = f"{newest_dir}/make_cat_runner/output/final_cat-{id_dash}.fits"
+    return legacy if os.path.exists(legacy) else None
+
+
 def process(params):
 
     if params["image_sims"]:
         patch_name = params["patch"]
-        run_prefix = "run_sp_tile_Mc_*"
+        run_prefix = "run_sp_tile_Mc*"
     else:
         patch_name = rf"P{params['patch']}"
         run_prefix = "run_sp_tile_Mc_*"
@@ -446,22 +537,11 @@ def process(params):
                         print(f"Skipping {id} (already processed)")
                     continue
 
-                base_pattern = os.path.join(id_path, "output", run_prefix)
-                all_matches = [d for d in glob.glob(base_pattern) if os.path.isdir(d)]
-                if not all_matches:
+                fits_file = find_final_cat(id, id_path, run_prefix)
+                if fits_file is None:
                     if params["verbose"]:
                         print(f"Final cat for {id} not found, continuing")
                     continue
-                newest_dir = max(all_matches, key=os.path.getmtime)
-
-                id_dash = re.sub(r"\.", "-", id)
-                fits_file = f"{newest_dir}/make_cat_runner/output/final_cat-{id_dash}.fits"
-
-                # Exclude unsuccessful run without output FITS file
-                if not os.path.exists(fits_file):
-                    if params["verbose"]:
-                        print(f"Run without output file found for {id}, skipping")
-                    continue
 
                 extracted_data, dtype = read_data(fits_file, params)
 

src/shapepipe/modules/fake_interp_runner.py

@@ -0,0 +1,44 @@
+"""FAKE INTERP RUNNER.
+
+Module runner for ``fake_psf`` under the ``<psf>_interp_runner`` name.
+
+The Snakemake workflow's configs read the galaxy PSF from
+``${SP_PSF}_interp_runner`` for every PSF model, so with ``psf_model: fake``
+(image simulations, true PSF) this runner stands where ``psfex_interp_runner``
+and ``mccd_interp_runner`` stand for the real data. It writes the same
+``galaxy_psf`` SqliteDict, taken from the simulation's PSF dictionary instead
+of a fitted model. ``fake_psf_runner`` is the same module under its original
+name, used by the legacy bash job scripts.
+
+:Author: Martin Kilbinger <martin.kilbinger@cea.fr>
+
+"""
+
+from shapepipe.modules.fake_psf_package import fake_psf
+from shapepipe.modules.module_decorator import module_runner
+
+
+@module_runner(
+    version="1.0",
+    file_pattern=["sexcat"],
+    file_ext=".fits",
+    depends=["numpy", "astropy", "sqlitedict"],
+    numbering_scheme="-000-000",
+)
+def fake_interp_runner(
+    input_file_list,
+    run_dirs,
+    file_number_string,
+    config,
+    module_config_sec,
+    w_log,
+):
+    """Define The Fake Interp Runner."""
+    sexcat_path = input_file_list[0]
+    psf_dict_path = config.getexpanded(module_config_sec, "PSF_DICT_PATH")
+    output_path = f'{run_dirs["output"]}/galaxy_psf{file_number_string}.sqlite'
+
+    inst = fake_psf.FakePsf(sexcat_path, psf_dict_path, output_path, w_log)
+    inst.process()
+
+    return None, None

src/shapepipe/modules/get_images_runner.py

@@ -27,8 +27,11 @@ def get_images_runner(
     """Define The Get Images Runner."""
     # Read config file section
 
-    # Copy/download method
-    retrieve_method = config.get(module_config_sec, "RETRIEVE")
+    # Copy/download method. getexpanded, not get: the workflow's ini sets
+    # RETRIEVE = $SP_RETRIEVE (a run-config key since d64f88ed), and plain get
+    # returns the literal "$SP_RETRIEVE", which then fails the check below.
+    # RETRIEVE_OPTIONS a few lines down already reads the same way.
+    retrieve_method = config.getexpanded(module_config_sec, "RETRIEVE")
     retrieve_ok = ["vos", "symlink"]
     if retrieve_method not in retrieve_ok:
         raise ValueError(

src/shapepipe/modules/mccd_package/mccd_interpolation_script.py

@@ -442,7 +442,7 @@ def _interpolate_me(self):
 
         all_id = np.copy(cat.get_data()["NUMBER"])
         key_ne = "N_EPOCH"
-        if key_ne not in cat.get_data():
+        if key_ne not in cat.get_data().dtype.names:
             raise KeyError(
                 f"Key {key_ne} not found in input galaxy catalogue, needed for"
                 + " PSF interpolation to multi-epoch data; run previous module"

workflow/README.md

@@ -25,7 +25,8 @@ uv pip install 'snakemake>=9,<10' 'snakemake-executor-plugin-slurm>=2.7,<3'
 # Edit workflow/config.yaml: tile_list, inputs.tiles/exposures, outputs.run_dir,
 # outputs.products_dir/index_db, and container.
 
-# `psf_model` is `psfex` or `mccd`; mccd is wired but unvalidated here, while psfex is exercised by smk-g4 through smk-g6.
+# `psf_model` is `psfex` or `mccd`. psfex is exercised by smk-g4 through smk-g6; mccd has run the full chain on
+# an image-sim star tile (one focal-plane model per exposure, ~1.5 CPU-hours each).
 
 # The committed launcher loads apptainer/1.4.5 + the /project venv, so a
 # fresh shell always has the right state.
@@ -45,6 +46,66 @@ Anything other than `run`, `report`, `container`, `cancel` passes straight throu
 snakemake with the workflow's profile and state dir — the direct command path for
 `sp --unlock`, `sp --dag`, `sp exp_psf ...`.
 
+## Image simulations
+
+The same workflow runs the SKiLLS image simulations used to measure the shear
+multiplicative bias, so that m calibrates the pipeline that makes the real
+catalogue rather than a frozen copy of it. A simulation run sets
+`input_type: image_sims`, which points `$SP_CONFIG` at
+`config/cfis_image_sims/`. That directory holds real files only for the stages
+whose input naming differs (tile Git/Uz/Fe, exposure Gie/Sp) and for the true-PSF
+model; everything else is a symlink into `config/cfis/`, so a change to the
+real-data chain reaches the simulations with no second edit. Keep the diff of
+each overlay file to its `cfis/` original confined to input naming.
+
+`psf_model: fake` is the simulations' true PSF: the exposure stage runs only
+SExtractor (for the background maps the vignets read), and `tile_vignets` runs
+`fake_interp_runner`, which writes the `galaxy_psf` product from `psf_dict`.
+Simulations that contain stars can run `psfex` or `mccd` exactly as the data do.
+
+One campaign per shear branch, each with its own run config:
+
+```bash
+SP_PROFILE=candide workflow/bin/sp run -c /path/run_1p2z_grid_1.yaml
+```
+
+## Run configuration
+
+A run config passed with `-c/--config-file` is merged on top of
+`workflow/config.yaml` and snapshotted with the code. (`-c` is `sp`'s own flag;
+pass snakemake's cores as `--cores`/`-j`. `SP_RUN_CONFIG` still works and is what
+the jobs read.) `SP_PROFILE` (default `nibi`, or `machine:` in the run config, which must
+agree with it) and `input_type:` then select an entry of the `machines:` table, which supplies
+`tile_list`, `retrieve` (`symlink` or `vos`), `inputs`, `outputs` and
+`container` for any of these the run config leaves unset (`$base_dir` expands
+to that machine's `base_dir`, `$run` to the run config's `run:`). A value of `TBD` stops the run at parse time
+until it is set. A run config therefore only needs what differs, e.g. for one
+SKiLLS shear branch on candide:
+
+```yaml
+machine: candide
+input_type: image_sims
+psf_model: fake
+psf_dict: /home/hervas/fhervas/workdir_skills/input/psf_files/Full_psf_dict.pickle
+tile_list: /path/to/tiles.txt
+inputs:
+  tiles: /n09data/hervas/skills_out/1z2z_grid_3/images/SP_tiles
+  exposures: /n09data/hervas/skills_out/1z2z_grid_3/images/SP_exp
+outputs:
+  run_dir: /path/to/run
+  products_dir: /path/to/product
+  index_db: /path/to/run/index.sqlite
+```
+
+sp_validation's image-simulation workflow drives these campaigns and measures m
+from their final catalogues.
+
+On candide, the node-local tile store (bound from the node's 31 GB `/tmp`) does not
+hold several dense image-sim tiles at once. Set `tile_store_root:` in the run
+config to a shared directory; `sp run` binds it to `/local/scratch` for that
+campaign. The store names carry a per-campaign hash, so all branches can share one
+root.
+
 ## The container image
 
 `sp container` owns which image the jobs run inside. Two layers, and the second