DynaCLR misc: CSV metadata sidecar, data-path profiling, DDP logging

CLAUDE.md

@@ -55,6 +55,10 @@ Invariant: `#SBATCH --ntasks-per-node=N` must equal `trainer.devices` in the YAM
 
 The dynacell launcher (`applications/dynacell/tools/submit_benchmark_job.py`) already emits `--ntasks-per-node` correctly; this note is for hand-written scripts (e.g., `applications/cytoland/examples/configs/*/run_*.slurm`).
 
+### Running jobs on Reef/Kelp (CoreWeave) vs Bruno
+
+Our SLURM scripts above target Bruno (home institution, not preemptible). Reef/Kelp is a separate, preemptible CoreWeave cluster with different partitions, QOS, filesystem paths, and job-launch tooling (`slurm_run`) — see [docs/clusters/reef.md](./docs/clusters/reef.md) before adapting or writing a `.slurm`/`sbatch` script for Reef.
+
 ### Job monitoring and inspection
 
 **Process state ≠ training completeness.** Wandb's `state: finished` only means `wandb.finish()` was called — Lightning calls it on clean SIGTERM teardown via `SLURMEnvironment`, so a `scancel`'d run shows `finished` identically to one that hit `max_epochs`. Always cross-check.

applications/airtable/src/airtable_utils/prepare.py

@@ -11,6 +11,8 @@
 from iohub import open_ome_zarr
 from pydantic import BaseModel, Field
 
+from viscy_data.meta_csv import read_meta_rows_csv
+
 logger = logging.getLogger(__name__)
 
 # ---------------------------------------------------------------------------
@@ -58,6 +60,7 @@ class SlurmStageConfig(BaseModel):
     time: str = "06:00:00"
     gres: str | None = None
     constraint: str | None = None
+    qos: str | None = None
 
 
 class SlurmConfig(BaseModel):
@@ -92,6 +95,8 @@ class PrepareConfig(BaseModel):
     nfs_root: Path = Path("/hpc/projects/intracellular_dashboard/organelle_dynamics")
     vast_root: Path = Path("/hpc/projects/organelle_phenotyping/datasets")
     workspace_dir: Path = Path("/hpc/mydata/eduardo.hirata/repos/viscy")
+    csv_dir: Path | None = None
+    jobs_dir: Path | None = None
     concatenate: ConcatenateConfig = Field(default_factory=ConcatenateConfig)
     qc: QCParams = Field(default_factory=QCParams)
     preprocess: PreprocessParams = Field(default_factory=PreprocessParams)
@@ -195,19 +200,28 @@ def check_zarr_version(zarr_path: Path) -> dict[str, int | str | None]:
     return result
 
 
-def check_preprocessed(zarr_path: Path) -> bool:
-    """Check if normalization metadata has been written to the zarr store.
+def check_preprocessed(zarr_path: Path, csv_dir: Path | None = None) -> bool:
+    """Check if normalization metadata has been written for the zarr store.
 
     Parameters
     ----------
     zarr_path : Path
         Path to the zarr store root.
+    csv_dir : Path or None
+        If given, check the per-store CSV sidecar under this directory
+        (written by ``viscy preprocess --csv_dir ...``) instead of the
+        store's ``.zattrs``/``zarr.json`` — for datasets prepared without
+        write access.
 
     Returns
     -------
     bool
         True if normalization stats are present.
     """
+    if csv_dir is not None:
+        sidecar = read_meta_rows_csv(csv_dir, zarr_path)
+        return sidecar is not None and bool((sidecar["field_name"] == "normalization").any())
+
     zarr_json = zarr_path / "zarr.json"
     zattrs = zarr_path / ".zattrs"
 
@@ -223,6 +237,34 @@ def check_preprocessed(zarr_path: Path) -> bool:
     return False
 
 
+# ---------------------------------------------------------------------------
+# Batch discovery (filesystem glob, no NFS/Airtable dependency)
+# ---------------------------------------------------------------------------
+
+
+def discover_zarr_stores(data_root: Path) -> list[Path]:
+    """Find dataset zarr stores under a root laid out like VAST's convention.
+
+    Matches ``{data_root}/{dataset_name}/{dataset_name}.zarr`` exactly (the
+    zarr's own basename must match its containing directory's name) — this
+    excludes sibling ``tracking.zarr`` dirs as well as any other variant
+    zarr stores (e.g. ``{dataset_name}_chunked.zarr``,
+    ``{dataset_name}_sharded.zarr``) that may sit alongside the canonical
+    store for rechunking/testing purposes.
+
+    Parameters
+    ----------
+    data_root : Path
+        Root directory containing one subdirectory per dataset.
+
+    Returns
+    -------
+    list[Path]
+        Sorted paths to each dataset's zarr store.
+    """
+    return sorted(p for p in Path(data_root).glob("*/*.zarr") if p.stem == p.parent.name)
+
+
 # ---------------------------------------------------------------------------
 # Discovery (reads NFS zarr via iohub)
 # ---------------------------------------------------------------------------
@@ -339,7 +381,7 @@ def generate_crop_concat_config(
     }
 
 
-def generate_qc_config(data_path: Path, qc_params: QCParams) -> dict:
+def generate_qc_config(data_path: Path, qc_params: QCParams, csv_dir: Path | None = None) -> dict:
     """Build a QC config dict compatible with ``qc run -c``.
 
     Parameters
@@ -348,6 +390,9 @@ def generate_qc_config(data_path: Path, qc_params: QCParams) -> dict:
         Path to the VAST zarr (target of QC).
     qc_params : QCParams
         Focus-slice QC parameters.
+    csv_dir : Path or None
+        If given, ``qc run`` writes results to a per-store CSV sidecar
+        under this directory (read-only store) instead of ``.zattrs``.
 
     Returns
     -------
@@ -357,6 +402,7 @@ def generate_qc_config(data_path: Path, qc_params: QCParams) -> dict:
     return {
         "data_path": str(data_path),
         "num_workers": qc_params.num_workers,
+        "csv_dir": str(csv_dir) if csv_dir else None,
         "focus_slice": {
             "channel_names": qc_params.channel_names,
             "NA_det": qc_params.NA_det,
@@ -412,6 +458,8 @@ def _slurm_header(job_name: str, output_dir: Path, cfg: SlurmStageConfig) -> str
         lines.append(f"#SBATCH --gres={cfg.gres}")
     if cfg.constraint:
         lines.append(f'#SBATCH --constraint="{cfg.constraint}"')
+    if cfg.qos:
+        lines.append(f"#SBATCH --qos={cfg.qos}")
     return "\n".join(lines)
 
 
@@ -532,7 +580,7 @@ def generate_qc_slurm(
         export PYTHONNOUSERSITE=1
 
         echo "=== QC: focus slice detection ==="
-        uv run --project "{workspace_dir}" --package qc \
+        uv run --project "{workspace_dir}" --package viscy-qc \
             qc run -c "{qc_config_path}"
         echo "QC complete."
     """)
@@ -546,6 +594,7 @@ def generate_preprocess_slurm(
     workspace_dir: Path,
     preprocess_params: PreprocessParams,
     slurm_cfg: SlurmStageConfig,
+    csv_dir: Path | None = None,
 ) -> str:
     """Generate SLURM script for normalization preprocessing (CPU only).
 
@@ -563,6 +612,10 @@ def generate_preprocess_slurm(
         Normalization preprocessing parameters.
     slurm_cfg : SlurmStageConfig
         SLURM resource parameters.
+    csv_dir : Path or None
+        If given, ``viscy preprocess`` writes results to a per-store CSV
+        sidecar under this directory (read-only store) instead of
+        ``.zattrs``.
 
     Returns
     -------
@@ -576,6 +629,7 @@ def generate_preprocess_slurm(
         ch_flag = f"--channel_names={ch_arg}"
     else:
         ch_flag = " ".join(f"--channel_names={c}" for c in ch_arg)
+    csv_dir_flag = f' --csv_dir "{csv_dir}"' if csv_dir else ""
 
     body = dedent(f"""\
 
@@ -586,7 +640,7 @@ def generate_preprocess_slurm(
         uv run --project "{workspace_dir}" --package dynaclr \
             viscy preprocess --data_path "{vast_zarr_path}" \
             {ch_flag} --num_workers {preprocess_params.num_workers} \
-            --block_size {preprocess_params.block_size}
+            --block_size {preprocess_params.block_size}{csv_dir_flag}
         echo "Preprocess complete."
     """)
     return header + "\n" + body
@@ -597,7 +651,9 @@ def generate_preprocess_slurm(
 # ---------------------------------------------------------------------------
 
 
-def check_dataset_status(dataset_name: str, nfs_root: Path, vast_root: Path) -> dict[str, str]:
+def check_dataset_status(
+    dataset_name: str, nfs_root: Path, vast_root: Path, csv_dir: Path | None = None
+) -> dict[str, str]:
     """Check existence and version info for a dataset across NFS and VAST.
 
     Parameters
@@ -608,6 +664,9 @@ def check_dataset_status(dataset_name: str, nfs_root: Path, vast_root: Path) ->
         NFS root directory.
     vast_root : Path
         VAST root directory.
+    csv_dir : Path or None
+        If given, check the ``preprocessed`` status via the CSV sidecar
+        under this directory instead of ``.zattrs`` (see ``check_preprocessed``).
 
     Returns
     -------
@@ -629,7 +688,7 @@ def check_dataset_status(dataset_name: str, nfs_root: Path, vast_root: Path) ->
         ver = check_zarr_version(vast["zarr"])
         zarr_fmt = str(ver["zarr_format"]) if ver["zarr_format"] else "?"
         ome_ver = str(ver["ome_version"]) if ver["ome_version"] else "?"
-        preprocessed = "yes" if check_preprocessed(vast["zarr"]) else "no"
+        preprocessed = "yes" if check_preprocessed(vast["zarr"], csv_dir=csv_dir) else "no"
 
     return {
         "dataset": dataset_name,

applications/airtable/src/airtable_utils/prepare_cli.py

@@ -7,6 +7,7 @@
 import subprocess
 
 import click
+from viscy_data.meta_csv import backfill_zattrs_to_csv
 
 from airtable_utils.prepare import (
     PrepareConfig,
@@ -15,6 +16,7 @@
     check_zarr_version,
     discover_channels,
     discover_wells,
+    discover_zarr_stores,
     filter_raw_channels,
     format_status_table,
     generate_concatenate_script,
@@ -100,7 +102,7 @@ def run(dataset_name: str, config_path: str, dry_run: bool, force: bool) -> None
         ver = check_zarr_version(vast["zarr"])
         is_v3 = ver["zarr_format"] == 3
         is_ome05 = ver["ome_version"] == "0.5"
-        is_preprocessed = check_preprocessed(vast["zarr"])
+        is_preprocessed = check_preprocessed(vast["zarr"], csv_dir=cfg.csv_dir)
 
         if is_v3 and is_ome05 and is_preprocessed:
             click.echo(
@@ -149,7 +151,7 @@ def run(dataset_name: str, config_path: str, dry_run: bool, force: bool) -> None
     click.echo(f"  Wrote: {crop_concat_path}")
 
     # 8. Generate qc_config.yml
-    qc_cfg = generate_qc_config(vast["zarr"], cfg.qc)
+    qc_cfg = generate_qc_config(vast["zarr"], cfg.qc, csv_dir=cfg.csv_dir)
     qc_config_path = vast["output_dir"] / "qc_config.yml"
     write_yaml(qc_cfg, qc_config_path)
     click.echo(f"  Wrote: {qc_config_path}")
@@ -192,6 +194,7 @@ def run(dataset_name: str, config_path: str, dry_run: bool, force: bool) -> None
         workspace_dir=cfg.workspace_dir,
         preprocess_params=cfg.preprocess,
         slurm_cfg=cfg.slurm.preprocess,
+        csv_dir=cfg.csv_dir,
     )
     preprocess_script_path = vast["output_dir"] / "03_preprocess.sh"
     preprocess_script_path.write_text(preprocess_script)
@@ -246,10 +249,100 @@ def status(dataset_names: tuple[str, ...], config_path: str) -> None:
     """Check NFS/VAST existence and version status for one or more datasets."""
     cfg = _load_prepare_config(config_path)
 
-    rows = [check_dataset_status(name, cfg.nfs_root, cfg.vast_root) for name in dataset_names]
+    rows = [check_dataset_status(name, cfg.nfs_root, cfg.vast_root, csv_dir=cfg.csv_dir) for name in dataset_names]
     click.echo(format_status_table(rows))
 
 
+@prepare.command("batch-preprocess")
+@click.argument("data_root", type=click.Path(exists=True, file_okay=False))
+@click.option(
+    "-c",
+    "--config",
+    "config_path",
+    required=True,
+    type=click.Path(exists=True),
+    help="Path to prepare config YAML.",
+)
+@click.option("--dry-run", is_flag=True, help="Generate configs/scripts without submitting SLURM jobs.")
+def batch_preprocess(data_root: str, config_path: str, dry_run: bool) -> None:
+    """Preprocess every not-yet-preprocessed dataset zarr under DATA_ROOT.
+
+    Discovers `{name}/{name}.zarr` stores under DATA_ROOT, skips any already
+    marked preprocessed in the CSV sidecar, mirrors metadata that's already
+    sitting in a store's own `.zattrs` (e.g. computed elsewhere with write
+    access, then copied here) into the CSV sidecar for free, and only
+    generates + submits QC/normalization SLURM jobs for stores with neither.
+    Safe to re-run repeatedly as new datasets appear — anything already
+    covered (via CSV or zattrs) is always skipped, so this never redoes work.
+    """
+    cfg = _load_prepare_config(config_path)
+    if cfg.csv_dir is None:
+        raise click.ClickException(
+            "batch-preprocess requires 'csv_dir' in the prepare config — datasets under "
+            "DATA_ROOT may be owned by teammates without write access to their .zattrs."
+        )
+    jobs_root = cfg.jobs_dir or cfg.csv_dir
+
+    stores = discover_zarr_stores(data_root)
+    click.echo(f"Discovered {len(stores)} dataset(s) under {data_root}.")
+
+    pending = [store for store in stores if not check_preprocessed(store, csv_dir=cfg.csv_dir)]
+
+    backfilled = []
+    todo = []
+    for store in pending:
+        if backfill_zattrs_to_csv(store, cfg.csv_dir):
+            backfilled.append(store)
+        else:
+            todo.append(store)
+
+    click.echo(
+        f"  {len(stores) - len(pending)} already in CSV sidecar, "
+        f"{len(backfilled)} backfilled from existing zattrs, {len(todo)} need SLURM jobs."
+    )
+
+    for store in todo:
+        dataset_name = store.stem
+        job_dir = jobs_root / "_jobs" / dataset_name
+        job_dir.mkdir(parents=True, exist_ok=True)
+
+        qc_config_path = job_dir / "qc_config.yml"
+        write_yaml(generate_qc_config(store, cfg.qc, csv_dir=cfg.csv_dir), qc_config_path)
+
+        qc_script_path = job_dir / "02_qc.sh"
+        qc_script_path.write_text(
+            generate_qc_slurm(dataset_name, job_dir, qc_config_path, cfg.workspace_dir, cfg.slurm.qc)
+        )
+
+        preprocess_script_path = job_dir / "03_preprocess.sh"
+        preprocess_script_path.write_text(
+            generate_preprocess_slurm(
+                dataset_name,
+                job_dir,
+                store,
+                cfg.workspace_dir,
+                cfg.preprocess,
+                cfg.slurm.preprocess,
+                csv_dir=cfg.csv_dir,
+            )
+        )
+        click.echo(f"  Wrote: {job_dir}")
+
+        if dry_run:
+            continue
+
+        result_qc = subprocess.run(["sbatch", str(qc_script_path)], capture_output=True, text=True, check=True)
+        qc_job_id = _parse_slurm_job_id(result_qc.stdout)
+        result_pp = subprocess.run(["sbatch", str(preprocess_script_path)], capture_output=True, text=True, check=True)
+        pp_job_id = _parse_slurm_job_id(result_pp.stdout)
+        click.echo(f"  {dataset_name}: QC job {qc_job_id}, preprocess job {pp_job_id}")
+
+    if dry_run:
+        click.echo("\n--dry-run: configs and scripts generated, nothing submitted.")
+    else:
+        click.echo(f"\nSubmitted jobs for {len(todo)} dataset(s).")
+
+
 def main() -> None:
     """Entry point for the prepare CLI."""
     prepare()