diff --git a/CLAUDE.md b/CLAUDE.md index 190152c..5c5805d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -9,6 +9,7 @@ uv sync --extra cpu # install dependencies with CPU-only torch (s uv sync --extra cuda # install dependencies with CUDA 11.8 torch uv sync --extra cpu --extra dev # add dev extras (pytest, ruff, ty, bump-my-version, git-cliff, + all runtime extras) uv sync --extra cpu --extra geometry # add scikit-learn for the geometry oracle (giant rollout) +uv sync --extra cpu --extra workflow # add b2luigi for `giant workflow` pipeline orchestration pytest # run tests giant new-run --hidden-dim 512 --lr 3e-4 # scaffold a config.toml + run dir ahead of training giant train path/to/steps.parquet # train (defaults: stage 1 flow, stage 2 wgan + autoregressive) @@ -18,8 +19,9 @@ giant train path/to/steps.parquet --router --router-type energy # MoE routing t giant model summary --config config.toml # build-only: parameter counts + which config keys actually bite giant predict path/to/steps.parquet --checkpoint ckpt/best.pt # per-step predictions giant rollout path/to/steps.parquet --checkpoint ckpt/best.pt --geometry oracle.pkl # full showers -giant analyze submit rollout.yaml --accounting-group cms # parallel rollout-vs-reference analysis on HTCondor -giant analyze render --gallery # render PDFs + HTML gallery (run_dir from prep/submit) +giant workflow run spec.toml --batch --workers 20 # whole pipeline (cache-warm -> train -> rollout -> analysis) +giant analyze prep rollout.yaml --chunks 32 # lay out an analysis run dir (compute jobs come from the workflow) +giant analyze render --gallery # render PDFs + HTML gallery (run_dir from prep) giant analyze metrics # training-progress plots from metrics.csv dwarf --help # dataset/tooling CLI: convert, migrate, bump-gen, # bump-schema, status, update-manifest, create-manifest, @@ -51,7 +53,7 @@ Work on this repo happens across three kinds of machine: - **Local dev machines** (laptop + desktop, identical): repo at `~/Programming/giant`, no access to `/ceph` — datasets, training results, and models aren't reachable here. - **Portal machines** (`portal1`, `deepthought`, `deepthought2`, `bms1`, `bms2`, `bms3`): repo lives under `/work`, and `/ceph` holds ROOT/parquet files and trained models. **These are shared with other users** — stay strictly within `/work/lbogner` and `/ceph/lbogner`, and keep resource usage to roughly a quarter of CPU/RAM and a single GPU so as not to disturb other users' jobs. -- **HTCondor worker nodes**: never run or SSH onto these directly — the only sanctioned path is submitting jobs through condor (`giant analyze submit`, and the in-progress remote-GPU train/rollout submission on `condor-gpu-train-rollout`). `/ceph` is available there; `/work` is only sometimes mounted, depending on the node. +- **HTCondor worker nodes**: never run or SSH onto these directly — the only sanctioned path is `giant workflow run --batch` (b2luigi, see the Workflow section), which submits and polls every job. `/ceph` is available there; `/work` is only sometimes mounted, depending on the node. ## Architecture @@ -99,7 +101,9 @@ Secondary energies are a **stick-breaking partition of the `e_sec` budget** from **Analysis** (`giant/analysis/`, `giant analyze` CLI): a lean, streaming rollout-vs-reference plotting pipeline that compares one or more autoregressive `giant rollout` runs against a single held-out miniCaloSim reference steps file shared by all of them, and produces publication-styled PDFs assembled into an HTML gallery — one distinctly colored series per rollout, one reference line/panel. It exploits the fact that rollout output and a raw reference file share a world-frame physical column subset under identical names (`pre_*`/`post_*`/`edep`/`step_length`/`pdg`/`material`/`event_id`), so no ALR/local-frame decode is needed — everything is world-frame mm/MeV. Structure: `sources.py` (canonical LazyFrames + `RolloutSpec`/`Side` — a rollout's opened frames + per-checkpoint diagnostic inputs — + synthetic-termination-row filtering + the secondary view, which is `generation>0 & step_no==0` rollout tracks vs exploded `sec_*_list` reference columns), `variables.py` (the per-step value expressions shared by range sizing and the plot registry), `reduce.py` (the streaming primitives — a single `hist1d` `group_by([group,bin]).len()` pass, per-event scalars, edep-weighted depth/transverse profiles, species share, leakage), `grouping.py`/`context.py` (fixed bin edges + energy-quantile/pdg/material group sets resolved once by `prep` into `shared.json` over the union of the reference and every rollout, so every compute job is one pass with no range scan), `reduced.py` (`Partial`/`Reduced` — the compact self-describing JSON a compute job emits), `catalog.py` (the declarative `PlotSpec` registry — marginals × {overall,energy,pdg,material}, per-event totals, shower profiles/containment, species/leakage, secondaries, distance/confusion summaries, router and type-embedding diagnostics; `giant analyze list` prints every id), `runtime_estimate.py` (per-(plot, chunk) walltime estimates for the submit description), and `render.py` (the only module importing ETPlot's `plotstyle`/LaTeX; dispatches on `Reduced.kind`, writes PDFs + `metadata.yaml`; each rollout gets a stable `ps.get_color(i)` slot by its position in `series`, the reference always draws in one fixed dashed-ink style). `Bundle.rollouts` is a name-keyed dict of `Side`, and every `compute_partial`/`finalize` builds a `Reduced.payload["series"]` dict keyed the same way, with `payload["reference"]` as the one distinguished non-rollout entry. The heatmap-shaped specs (`marginal_distance_summary`, `n_sec_confusion`) and the checkpoint-bound diagnostics (`router_gating.py`, `type_embedding_distance.py`) are inherently one-matrix/one-checkpoint per rollout, so they render as one panel per rollout instead of one line/bar per rollout. -**Input is one or more `giant rollout` YAML sidecars** (`condor.py:load_rollout_yamls`): each YAML's `output`/`dataset` keys name its rollout parquet and seed file (= the reference truth); every supplied YAML must resolve to the same `dataset`, checked up front with a clear error otherwise (the premise is "N candidates vs one ground truth"). Each rollout's series name comes from a repeated `--label` CLI flag, else the YAML stem (N>1), else `"rollout"` (a single YAML). `prep` creates a **run directory** (`/analysis_runs/analysis_/` by default, `--run-dir` to override) holding `shared.json`, `run_meta.json` (`RunMeta.rollouts: list[{name,path,plot_meta}]`, insertion order = CLI order = every plot's series order), `reduced_partial/`, `reduced/`, `plots/`. **Compute/merge/render split:** `giant analyze submit a.yaml [b.yaml ...] --chunks N` runs `prep` (recording `N` in `run_meta.json`) then submits one HTCondor job per (plot, chunk) pair (`compute-one --id --chunk --run-dir`, polars/numpy only — no LaTeX on workers), each streaming over an `event_id`-disjoint slice (`event_id % N == chunk`) of the reference **and every rollout** and writing a small `reduced_partial/__.json`; every `PlotSpec` splits into a `compute_partial`/`finalize` pair so chunks can be summed/concatenated back per rollout (`chunkable=False` specs — the checkpoint-bound diagnostics, already bounded/subsampled — always run as a single chunk). The local `giant analyze render ` first joins every plot's chunk partials into `reduced/.json` (`merge_all`, a no-op join when `N=1`; `merge-one` does a single plot for debugging), then turns those into the styled PDF/gallery tree. `giant analyze metrics ` is a separate, unrelated entry point: training-progress plots straight from a run's `metrics.csv`. +**Input is one or more `giant rollout` YAML sidecars** (`run.py:load_rollout_yamls`): each YAML's `output`/`dataset` keys name its rollout parquet and seed file (= the reference truth); every supplied YAML must resolve to the same `dataset`, checked up front with a clear error otherwise (the premise is "N candidates vs one ground truth"). Each rollout's series name comes from a repeated `--label` CLI flag, else the YAML stem (N>1), else `"rollout"` (a single YAML). `prep` creates a **run directory** (`/analysis_runs/analysis_/` by default, `--run-dir` to override) holding `shared.json`, `run_meta.json` (`RunMeta.rollouts: list[{name,path,plot_meta}]`, insertion order = CLI order = every plot's series order), `reduced_partial/`, `reduced/`, `plots/`. **Compute/merge/render split:** `giant analyze prep a.yaml [b.yaml ...] --chunks N` records `N` in `run_meta.json`, and the workflow's `AnalysisComputeTask` runs one HTCondor job per (plot, chunk) pair (`compute-one --id --chunk --run-dir`, polars/numpy only — no LaTeX on workers), each streaming over an `event_id`-disjoint slice (`event_id % N == chunk`) of the reference **and every rollout** and writing a small `reduced_partial/__.json`; every `PlotSpec` splits into a `compute_partial`/`finalize` pair so chunks can be summed/concatenated back per rollout (`chunkable=False` specs — the checkpoint-bound diagnostics, already bounded/subsampled — always run as a single chunk). The local `giant analyze render ` first joins every plot's chunk partials into `reduced/.json` (`merge_all`, a no-op join when `N=1`; `merge-one` does a single plot for debugging), then turns those into the styled PDF/gallery tree. `giant analyze metrics ` is a separate, unrelated entry point: training-progress plots straight from a run's `metrics.csv`. + +**Workflow orchestration** (`giant/workflow/`, `giant workflow run` CLI): b2luigi is the **only sanctioned way to run a multi-step pipeline**; `giant`/`dwarf` are single-step primitives the tasks invoke. One workflow TOML (`configs/workflow_example.toml`) parameterises a whole experiment — `[workflow]`/`[condor]`/`[dataset]`/`[geometry]` plus repeated `[[train]]`/`[[rollout]]`/`[[analysis]]` tables, each cross-referenced by name — and `spec.py` parses it into frozen dataclasses, rejecting unknown keys and dangling references. Every task's output directory is `//name=/spec_hash=/…`, where the 8-hex `spec_hash` covers that task's resolved sub-spec **and its transitive parents**, so an edited spec re-runs exactly the affected subtree instead of silently reusing stale outputs. The DAG (`tasks.py`): `DatasetTask` (external, fails fast if `/ceph` isn't mounted) → `WarmCacheTask` / `GeometryOracleTask` → `TrainEpochTask(name, milestone)` → `TrainTask` → `RolloutTask` → `AnalysisPrepTask` → `AnalysisComputeTask(name, plot_id, chunk)` → `AnalysisRenderTask` → `WorkflowTask`. Training is fanned out into **one short GPU job per epoch** (`epochs_per_job` trades queue waits back), chained by `--resume` on the previous job's `last.pt` — the loop already handles that unchanged — and `TrainTask` republishes `best.pt`/`last.pt`/a concatenated `metrics.csv` so nothing downstream sees the fan-out. `StreamingStepsDataset.set_epoch` (called per epoch by `training/loop.py`) seeds shuffling from `(seed, epoch, worker_id)` so epoch *k*'s batch order is the same either way. `AnalysisRenderTask` is always local (the only step importing plotstyle/LaTeX); `htcondor.py` holds the CPU/GPU submit settings, with the GPU requirement strings (`TARGET.ProvidesEtpCeph` + device/memory pins) ported from the `condor-gpu-train-rollout` branch. `run.py` is the script b2luigi re-executes on workers (`--spec` forwarded via `task_cmd_additional_args`, so a worker resolves the identical graph); `giant workflow run` is a thin exec of it. Needs `uv sync --extra cpu --extra workflow`. **Shower rollout** (`giant/rollout.py`, `giant rollout` CLI): autoregressively steps the two-stage model into a full shower, advancing tracks breadth-first (every sweep steps all active tracks once, in `batch_size` chunks, so many tracks share each forward pass). Each primary post-step becomes the next pre-step, secondaries are pushed as new tracks, and per-step `material`/`layer_id` come from a `GeometryOracle` (`giant/geometry.py`, built via `dwarf build-geometry-oracle`) that learns position → (material, layer_id) from data and flags detector escape by nearest-neighbour distance. Tracks terminate on one of the `TERM_*` reasons in `constants.py` (energy cutoff, max steps, escape, natural end, unknown pdg, max tracks); energy is deposited locally on every stop except escape (leakage), so showers conserve energy by construction. `giant/checkpoint_io.py` is the shared checkpoint → ready-to-run-models path used by both `predict` and `rollout`. @@ -121,4 +125,4 @@ v0.2 configs and checkpoints are auto-migrated (`config.migrate_config`, `model. A sampling-calorimeter (multi-material) dataset track is still open and unblocked, not yet started. See the knowledge base (`/home/lars/knowledge-base/meta/roadmap.md`). -**Condor-submitted GPU training/rollout (in progress, `condor-gpu-train-rollout` branch, not yet merged):** moves `giant train`/`giant rollout` off the shared portal GPU dev machines (see Compute environment) onto remote-GPU HTCondor submission on TOpAS/NEMO2. Partway between "needs major features" and feature-complete — not ready to merge yet. +**Condor-submitted GPU training/rollout (`condor-gpu-train-rollout` branch, superseded):** its goal — moving `giant train`/`giant rollout` off the shared portal GPU dev machines onto remote-GPU HTCondor submission — is now met by the b2luigi workflow above. Its `train-submit`/`rollout-submit` commands are deliberately **not** ported and must not be revived when that branch is eventually merged; the only part that survived is `_gpu_requirements`, which moved into `giant/workflow/htcondor.py`. diff --git a/README.md b/README.md index 2f224d0..2f6e2b9 100644 --- a/README.md +++ b/README.md @@ -110,10 +110,15 @@ giant/ │ │ ├── reduced.py # Partial/Reduced — the compact JSON a compute job emits │ │ ├── catalog.py # declarative PlotSpec registry (`giant analyze list`) │ │ ├── router_gating.py / type_embedding_distance.py # checkpoint-bound diagnostics -│ │ ├── runtime_estimate.py # per-(plot, chunk) walltime estimates for submit -│ │ ├── condor.py # prep / compute-one / merge / submit-description plumbing +│ │ ├── runtime_estimate.py # per-(plot, chunk) walltime estimates for the job requests +│ │ ├── run.py # prep / compute-one / merge plumbing │ │ └── render.py # PDFs + HTML gallery (only module importing plotstyle/LaTeX) -│ └── cli.py # `giant train` / `new-run` / `model summary` / `predict` / `rollout` / `analyze` +│ ├── workflow/ # b2luigi pipeline orchestration (`giant workflow run spec.toml`) +│ │ ├── spec.py # workflow TOML -> frozen dataclasses, validation, per-task spec hashes +│ │ ├── htcondor.py # CPU/GPU submit settings (docker image, +RemoteJob, GPU requirements) +│ │ ├── tasks.py # the task graph: cache-warm -> train (one job/epoch) -> rollout -> analysis +│ │ └── run.py # the script b2luigi re-executes on every worker +│ └── cli.py # `giant train` / `new-run` / `model summary` / `predict` / `rollout` / `analyze` / `workflow` ├── giant/tools/ # dataset/tooling logic, unified under the `dwarf` CLI (`dwarf --help`) │ ├── dwarf.py # Typer app: convert, migrate, bump-gen, bump-schema, status, │ │ # update-manifest, create-manifest, make-root, @@ -140,9 +145,10 @@ uv sync --extra cpu --extra geometry # add scikit-learn, for `dwarf build-geome uv sync --extra cpu --extra analysis # matplotlib/polars/plotstyle, for `giant analyze render` uv sync --extra cpu --extra convert # uproot/awkward/polars, for `dwarf convert` uv sync --extra cpu --extra wandb # W&B logging (`giant train --wandb`) +uv sync --extra cpu --extra workflow # b2luigi, for `giant workflow run` ``` -The `dev` extra pulls in `convert`, `analysis`, `geometry` and `wandb` as well. +The `dev` extra pulls in `convert`, `analysis`, `geometry`, `wandb` and `workflow` as well. `cpu` and `cuda` are mutually exclusive — pick one to select the torch build (pinned to 2.3.x). Plain `uv sync` installs no torch at all. See `CLAUDE.md` for details. @@ -179,17 +185,29 @@ Config-file-only knobs (no CLI flag — use `--config config.toml`): `stage2_mod - `giant analyze` — deeper rollout-vs-reference diagnostics (marginals by energy/pdg/material, per-event totals, shower profiles, species share, leakage, secondaries): ```bash -giant analyze submit rollout.yaml --accounting-group cms # prep + one HTCondor job per plot × chunk (compute only) -giant analyze submit a.yaml b.yaml --accounting-group cms --label flow --label wgan # N rollouts vs one shared reference +giant analyze prep rollout.yaml --chunks 8 # lay out the run directory +giant analyze prep a.yaml b.yaml --label flow --label wgan # N rollouts vs one shared reference giant analyze render --gallery # local: merge chunks, then styled PDFs + HTML gallery (needs LaTeX) giant analyze list # every catalog plot id -giant analyze prep rollout.yaml --chunks 8 # just the run directory, no submission giant analyze compute-one --id marginal_edep --run-dir --chunk 0 # what a condor job runs giant analyze merge-one --id marginal_edep --run-dir # merge one plot's chunks (debugging) ``` -`` defaults to `/analysis_runs/analysis_` (`--run-dir` overrides it; `prep`/`submit` print it). Multiple rollout YAMLs must all name the same reference (`dataset`) file; each renders as its own colored series against one reference line/panel. Compute jobs are polars/numpy only; only `render` needs LaTeX, so it always runs locally. +The per-(plot, chunk) compute jobs themselves are submitted by the workflow (below), not by `giant analyze` — these commands are the single-step primitives it calls. `` defaults to `/analysis_runs/analysis_` (`--run-dir` overrides it; `prep` prints it). Multiple rollout YAMLs must all name the same reference (`dataset`) file; each renders as its own colored series against one reference line/panel. Compute jobs are polars/numpy only; only `render` needs LaTeX, so it always runs locally. + +## Workflow orchestration + +Multi-step pipelines run through [b2luigi](https://github.com/belle2/b2luigi) — one spec file describes a whole experiment, and every step's outputs are files on `/ceph` that are only recomputed when their spec (or an upstream one) changes: + +```bash +uv sync --extra cpu --extra workflow +giant workflow run configs/workflow_example.toml --mode dry-run # what would run +giant workflow run configs/workflow_example.toml --mode show-output # where every output goes +giant workflow run configs/workflow_example.toml --batch --workers 20 # submit to HTCondor and wait +``` + +The spec holds `[workflow]`/`[condor]`/`[dataset]`/`[geometry]` plus repeated `[[train]]`, `[[rollout]]` and `[[analysis]]` tables cross-referenced by name (see `configs/workflow_example.toml`). The task graph is `DatasetTask → WarmCacheTask/GeometryOracleTask → TrainEpochTask… → TrainTask → RolloutTask → AnalysisPrepTask → AnalysisComputeTask(plot, chunk) → AnalysisRenderTask`. Training is split into one short GPU job per epoch (chained by `--resume`), which schedules better on a busy farm and survives preemption; `TrainTask` then publishes one `best.pt`/`last.pt`/`metrics.csv` for everything downstream. Rendering always runs locally, since it is the only step that needs LaTeX. Separately, `giant analyze metrics ` renders training-progress plots (loss/lr/accuracy/grad-norm/router/wgan/throughput) straight from a training run's `metrics.csv`. diff --git a/giant/analysis/__init__.py b/giant/analysis/__init__.py index 1056d85..a33d168 100644 --- a/giant/analysis/__init__.py +++ b/giant/analysis/__init__.py @@ -4,7 +4,7 @@ Compares one or more autoregressive ``giant rollout`` runs against a single held-out miniCaloSim reference file shared by all of them, producing publication-styled comparison plots (one colored series per rollout, one reference line) generated in parallel on HTCondor (one job per plot x data -chunk, compute/merge/render split). +chunk, compute/merge/render split) — orchestrated by ``giant/workflow``. Only ``render`` (and the ``render`` CLI path) imports plotstyle/LaTeX; everything re-exported here is plotstyle-free so it runs on a compute worker. Import @@ -12,10 +12,9 @@ re-exported here is plotstyle-free so it runs on a compute worker. Import """ from giant.analysis.catalog import build_catalog, catalog_ids, get_spec -from giant.analysis.condor import ( +from giant.analysis.run import ( LoadedRollout, RunMeta, - SubmitConfig, compute_one, compute_reduced, derive_run_dir, @@ -24,7 +23,6 @@ from giant.analysis.condor import ( merge_all, merge_one, prep, - write_submit, ) from giant.analysis.context import Context, build_context from giant.analysis.reduced import Partial, Reduced @@ -37,7 +35,6 @@ __all__ = [ "get_spec", "LoadedRollout", "RunMeta", - "SubmitConfig", "compute_one", "compute_reduced", "derive_run_dir", @@ -46,7 +43,6 @@ __all__ = [ "merge_all", "merge_one", "prep", - "write_submit", "Context", "build_context", "Partial", diff --git a/giant/analysis/render.py b/giant/analysis/render.py index d4ae5c4..46a684d 100644 --- a/giant/analysis/render.py +++ b/giant/analysis/render.py @@ -535,7 +535,7 @@ def render_run(run_dir: str | Path, *, run_gallery: bool = False) -> list[Path]: (checkpoint, paths, cutoffs) from ``run_meta.json`` into every plot's gallery metadata and renders. """ - from giant.analysis.condor import RunMeta, merge_all + from giant.analysis.run import RunMeta, merge_all run_dir = Path(run_dir) merge_all(run_dir) diff --git a/giant/analysis/condor.py b/giant/analysis/run.py similarity index 72% rename from giant/analysis/condor.py rename to giant/analysis/run.py index 09f2f68..0fcdd9e 100644 --- a/giant/analysis/condor.py +++ b/giant/analysis/run.py @@ -1,4 +1,6 @@ -"""HTCondor orchestration driven by one or more ``giant rollout`` YAML sidecars. +"""Analysis run directories: prep, per-(plot, chunk) compute, and merge. + +Driven by one or more ``giant rollout`` YAML sidecars. A rollout writes a YAML sidecar (``giant/cli.py:_write_prediction_ref`` + rollout extras) that already names both files we need and carries the run's @@ -22,13 +24,16 @@ everything out under it: /reduced/.json merged, per plot /plots//.pdf rendered locally -Job model (one condor job per (plot, chunk), compute/merge/render split): +Job model (one job per (plot, chunk), compute/merge/render split). Job +submission itself is b2luigi's (``giant/workflow/tasks.py`` — ``AnalysisPrepTask`` +/ ``AnalysisComputeTask`` / ``AnalysisRenderTask``); this module only provides +the three steps they call: -1. ``prep`` runs once on the submit node — reads the YAML, resolves the shared +1. ``prep`` runs once locally — reads the YAML, resolves the shared context from a subsample, writes ``shared.json`` + ``run_meta.json`` (including the run's configured ``n_chunks``). 2. one job per catalog id x chunk index runs ``giant analyze compute-one - --run-dir`` on a worker — a single streaming pass over that + --run-dir`` (or ``compute_one`` in-process) on a worker — a single streaming pass over that ``event_id``-disjoint chunk, writing ``reduced_partial/__.json`` (polars/numpy only, no LaTeX). Specs marked ``chunkable=False`` (``PlotSpec``, ``catalog.py``) always run as a single chunk. @@ -38,15 +43,15 @@ Job model (one condor job per (plot, chunk), compute/merge/render split): ``reduced/.json``, then renders those into the styled PDF + gallery tree (that step imports plotstyle/LaTeX). -Files on ``/ceph`` or ``/work`` are reached via ``ProvidesETPResources``; no -HTCondor file transfer of the multi-GB inputs. +Files on ``/ceph`` or ``/work`` are reached directly (see +``giant/workflow/htcondor.py``); no HTCondor file transfer of the multi-GB +inputs. """ from __future__ import annotations import json import shutil -import sys from collections.abc import Sequence from dataclasses import dataclass, field from pathlib import Path @@ -57,7 +62,6 @@ import yaml from giant.analysis.catalog import Bundle, catalog_ids, get_spec from giant.analysis.context import Context, build_context from giant.analysis.reduced import Partial -from giant.analysis.runtime_estimate import estimate_runtime_s from giant.analysis.sources import RolloutSpec, Side, open_side # Keys copied verbatim from a rollout YAML into each plot's gallery metadata. @@ -429,136 +433,3 @@ def merge_one(spec_id: str, run_dir: str | Path) -> Path: def merge_all(run_dir: str | Path) -> list[Path]: """Merge every catalog plot's chunk partials into ``reduced/.json``.""" return [merge_one(spec_id, run_dir) for spec_id in catalog_ids()] - - -# --------------------------------------------------------------------------- -# submit description -# --------------------------------------------------------------------------- - - -@dataclass -class SubmitConfig: - run_dir: Path - accounting_group: str - repo_dir: Path - docker_image: str = "cverstege/alma9-gridjob" - request_memory_mb: int = 8192 - request_cpus: int = 1 - remote: bool = False # +RemoteJob (grid I/O) vs ProvidesETPResources (local files) - n_chunks: int = 1 # per-plot data chunks; ignored for chunkable=False specs - - -_WRAPPER = """#!/bin/bash -set -euo pipefail -cd {repo_dir} -exec {giant_exe} analyze compute-one --id "$1" --chunk "$2" --run-dir {run_dir} -""" - - -def _submit_description(cfg: SubmitConfig, wrapper: Path, jobs_file: Path) -> str: - reqs_attrs = "+RemoteJob = True\n" if cfg.remote else "requirements = TARGET.ProvidesETPResources\n" - return ( - "universe = docker\n" - f"docker_image = {cfg.docker_image}\n" - f"executable = {wrapper}\n" - "arguments = $(plotid) $(chunk)\n" - "should_transfer_files = YES\n" - "when_to_transfer_output = ON_EXIT\n" - f"request_memory = {cfg.request_memory_mb}\n" - f"request_cpus = {cfg.request_cpus}\n" - "+RequestWalltime = $(walltime)\n" - f"accounting_group = {cfg.accounting_group}\n" - f"{reqs_attrs}" - f"output = {cfg.run_dir}/logs/$(plotid)__$(chunk).out\n" - f"error = {cfg.run_dir}/logs/$(plotid)__$(chunk).err\n" - f"log = {cfg.run_dir}/logs/condor.log\n" - f"queue plotid,chunk,walltime from {jobs_file}\n" - ) - - -def _job_walltimes(run_dir: Path, ids: list[str], n_chunks: int) -> list[tuple[str, int, int]]: - """``(spec_id, chunk, walltime_s)`` for every job, sized from ``run_meta.json``. - - Row counts come from ``prep``'s ``RunMeta.rows_per_chunk``/``total_rows``; - ``chunkable=False`` specs (router diagnostics) always use the dataset - total since they run as a single job regardless of ``n_chunks``. - """ - meta = RunMeta.load(run_dir / "run_meta.json") - jobs: list[tuple[str, int, int]] = [] - for spec_id in ids: - chunkable = get_spec(spec_id).chunkable - chunks = range(n_chunks) if chunkable else [0] - for chunk in chunks: - n_rows = meta.rows_per_chunk[chunk] if chunkable else meta.total_rows - jobs.append((spec_id, chunk, estimate_runtime_s(spec_id, n_rows))) - return jobs - - -def _resolve_giant_executable(repo_dir: Path) -> Path: - """Path to the ``giant`` entry point to bake into the condor wrapper script. - - Prefers the venv currently running this process (``sys.executable``'s - sibling ``giant``) so a submit from a non-default venv (e.g. ``--extra - cuda`` on a dev box) doesn't silently pick up a different one; falls back - to ``repo_dir/.venv/bin/giant`` for the case this is invoked from outside - any venv (e.g. a system Python). - """ - active = Path(sys.executable).parent / "giant" - if active.exists(): - return active - venv_giant = repo_dir / ".venv" / "bin" / "giant" - if not venv_giant.exists(): - raise FileNotFoundError( - f"no `giant` executable found next to {sys.executable} or at " - f"{venv_giant} — condor jobs run it directly (no `uv` on the " - f"worker image), so run `uv sync --extra cpu` in {repo_dir} " - "before submitting." - ) - return venv_giant - - -def write_submit(cfg: SubmitConfig, ids: list[str] | None = None) -> Path: - """Write the wrapper script, (plot, chunk) job list, and HTCondor submit - description. - - Each catalog id gets ``cfg.n_chunks`` jobs, except ``chunkable=False`` - specs (the router diagnostics), which always get exactly one regardless of - ``cfg.n_chunks``. Every job's ``+RequestWalltime`` is estimated from its - chunk's row count (``runtime_estimate.estimate_runtime_s``, requires - ``run_meta.json`` from ``prep`` to already carry ``rows_per_chunk``). - Returns the submit description path (``/analyze.sub``). Does not - submit — call ``condor_submit`` on the returned file. - - ``cfg.n_chunks`` and the run directory's own ``RunMeta.n_chunks`` (fixed by - ``prep``, and what ``RunMeta.rows_per_chunk`` was sized against) are two - independent values — checked equal up front so a mismatch is a clear error - here rather than an ``IndexError`` out of ``_job_walltimes``. - """ - giant_exe = _resolve_giant_executable(cfg.repo_dir) - - ids = ids or catalog_ids() - run_dir = cfg.run_dir - meta = RunMeta.load(run_dir / "run_meta.json") - if cfg.n_chunks != meta.n_chunks: - raise ValueError( - f"SubmitConfig.n_chunks={cfg.n_chunks} does not match the " - f"n_chunks this run directory was prepped with " - f"(RunMeta.n_chunks={meta.n_chunks} in {run_dir}/run_meta.json) — " - "re-run `prep` with the desired n_chunks, or fix cfg.n_chunks to " - "match it." - ) - (run_dir / "logs").mkdir(parents=True, exist_ok=True) - (run_dir / "reduced").mkdir(parents=True, exist_ok=True) - (run_dir / "reduced_partial").mkdir(parents=True, exist_ok=True) - - wrapper = run_dir / "run_compute.sh" - wrapper.write_text(_WRAPPER.format(repo_dir=cfg.repo_dir, giant_exe=giant_exe, run_dir=run_dir)) - wrapper.chmod(0o755) - - jobs = _job_walltimes(run_dir, ids, cfg.n_chunks) - jobs_file = run_dir / "jobs.txt" - jobs_file.write_text("\n".join(f"{i},{k},{w}" for i, k, w in jobs) + "\n") - - sub = run_dir / "analyze.sub" - sub.write_text(_submit_description(cfg, wrapper, jobs_file)) - return sub diff --git a/giant/analysis/runtime_estimate.py b/giant/analysis/runtime_estimate.py index 5f639c6..3d9a4bb 100644 --- a/giant/analysis/runtime_estimate.py +++ b/giant/analysis/runtime_estimate.py @@ -1,4 +1,4 @@ -"""Per-(plot, chunk) HTCondor walltime estimates for `giant analyze submit`. +"""Per-(plot, chunk) HTCondor walltime estimates for the analysis compute jobs. Each catalog spec's compute cost is close to linear in the number of input rows a `compute-one` job streams over — every spec is one (or a couple of) diff --git a/giant/cli.py b/giant/cli.py index 3c47d8f..db1a2d0 100644 --- a/giant/cli.py +++ b/giant/cli.py @@ -1732,83 +1732,5 @@ def analyze_metrics( typer.echo(f"rendered {len(paths)} plots -> {paths[0].parent if paths else '(nothing to render)'}") -@analyze_app.command("submit") -def analyze_submit( - rollout_yamls: Annotated[ - list[Path], - typer.Argument( - help="giant rollout YAML sidecar(s). Multiple compare N rollouts against one " - "shared reference — every YAML must name the same `dataset`." - ), - ], - accounting_group: Annotated[str, typer.Option("--accounting-group")], - label: Annotated[ - Optional[list[str]], - typer.Option( - "--label", - help="Series name for a rollout YAML, positionally matched to it — give none, " - 'or exactly one per YAML. Defaults to the YAML stem (or "rollout" for a ' - "single YAML).", - ), - ] = None, - run_dir: Annotated[ - Optional[Path], - typer.Option( - "--run-dir", - "-o", - help="Override the run directory (default: /analysis_runs/analysis_)", - ), - ] = None, - docker_image: Annotated[str, typer.Option("--docker-image")] = "cverstege/alma9-gridjob", - request_memory: Annotated[int, typer.Option("--request-memory", help="MB")] = 8192, - remote: Annotated[ - bool, - typer.Option("--remote/--local", help="+RemoteJob vs ProvidesETPResources"), - ] = False, - chunks: Annotated[ - int, - typer.Option( - "--chunks", - help="Split each plot's data into this many event_id chunks/jobs", - ), - ] = 1, - n_energy_bins: Annotated[int, typer.Option("--energy-bins")] = 4, - n_marginal_bins: Annotated[int, typer.Option("--bins")] = 50, - top_k_pdg: Annotated[int, typer.Option("--top-pdg")] = 6, - dry_run: Annotated[bool, typer.Option("--dry-run", help="Write files but don't condor_submit")] = False, -) -> None: - """prep + write the HTCondor submit description (one job per plot x chunk), then submit.""" - import subprocess - - from giant.analysis import SubmitConfig, prep, write_submit - - path = prep( - rollout_yamls, - run_dir, - n_chunks=chunks, - default_base=Path.cwd() / "analysis_runs", - labels=label, - n_energy_bins=n_energy_bins, - n_marginal_bins=n_marginal_bins, - top_k_pdg=top_k_pdg, - ) - cfg = SubmitConfig( - run_dir=path, - accounting_group=accounting_group, - repo_dir=Path.cwd(), - docker_image=docker_image, - request_memory_mb=request_memory, - remote=remote, - n_chunks=chunks, - ) - sub = write_submit(cfg) - typer.echo(f"run directory: {path}") - typer.echo(f"wrote submit description: {sub}") - if dry_run: - typer.echo("dry-run: not submitting") - return - subprocess.run(["condor_submit", str(sub)], check=True) - - if __name__ == "__main__": app() diff --git a/giant/tools/profile_analysis_costs.py b/giant/tools/profile_analysis_costs.py index 28dc1af..b7534f9 100644 --- a/giant/tools/profile_analysis_costs.py +++ b/giant/tools/profile_analysis_costs.py @@ -26,7 +26,7 @@ import numpy as np import polars as pl from giant.analysis.catalog import catalog_ids, get_spec -from giant.analysis.condor import compute_reduced +from giant.analysis.run import compute_reduced from giant.analysis.context import build_context from giant.analysis.sources import RolloutSpec diff --git a/giant/training/plots.py b/giant/training/plots.py index b15ef17..fa3bb34 100644 --- a/giant/training/plots.py +++ b/giant/training/plots.py @@ -115,7 +115,7 @@ def derive_metrics_dir( Precedence: an explicit `out_dir` always wins. Otherwise `default_base / f"metrics_{run_dir.name}"` (the CLI passes the repo's - gitignored `analysis_runs/`, matching `giant.analysis.condor.derive_run_dir`'s + gitignored `analysis_runs/`, matching `giant.analysis.run.derive_run_dir`'s convention) — training-progress plots live alongside rollout-vs-reference analysis runs, not inside the training run directory itself. """ diff --git a/giant/workflow/htcondor.py b/giant/workflow/htcondor.py index e0cde5c..ce8317d 100644 --- a/giant/workflow/htcondor.py +++ b/giant/workflow/htcondor.py @@ -5,7 +5,7 @@ that job's submit description, so these helpers are just the ETP-specific resource/requirement conventions in one place: * **CPU jobs** (setup cache, geometry oracle, analysis compute) keep what - ``giant analyze submit`` used: ``+RemoteJob`` for grid I/O, or + the deleted ``giant analyze submit`` used: ``+RemoteJob`` for grid I/O, or ``TARGET.ProvidesETPResources`` when the files are local to the cluster. * **GPU jobs** (training epochs, rollout) are remote-only, so they always carry ``+RemoteJob`` and reach ``/ceph`` through diff --git a/tests/test_condor.py b/tests/test_analysis_run.py similarity index 72% rename from tests/test_condor.py rename to tests/test_analysis_run.py index e35356f..9a5b6fe 100644 --- a/tests/test_condor.py +++ b/tests/test_analysis_run.py @@ -1,8 +1,7 @@ -"""Tests for the rollout-YAML(s) → run-directory flow, compute, and submit.""" +"""Tests for the rollout-YAML(s) → run-directory flow, compute, and merge.""" from __future__ import annotations -import sys from pathlib import Path import pyarrow.parquet as pq @@ -11,8 +10,6 @@ import yaml from giant.analysis import ( RunMeta, - SubmitConfig, - catalog_ids, compute_one, compute_reduced, derive_run_dir, @@ -20,10 +17,8 @@ from giant.analysis import ( load_rollout_yamls, merge_one, prep, - write_submit, ) -from giant.analysis.catalog import get_spec -from giant.analysis.condor import Context +from giant.analysis.run import Context from giant.analysis.reduced import Partial, Reduced from giant.constants import PREDICT_COORD_METADATA_KEY, ROLLOUT_COORD_VALUE from tests.test_analysis_reduce import _reference_frame, _rollout_frame @@ -86,14 +81,6 @@ def _write_two_inputs(tmp_path: Path) -> tuple[Path, Path]: return paths[0], paths[1] -def _fake_venv(repo_dir: Path) -> None: - """Stand in for a `uv sync`'d venv: write_submit checks `.venv/bin/giant` exists.""" - giant = repo_dir / ".venv" / "bin" / "giant" - giant.parent.mkdir(parents=True, exist_ok=True) - giant.write_text("#!/bin/bash\n") - giant.chmod(0o755) - - def _prep(rollout_yamls, run_dir: str | Path | None = None, chunks: int = 1, labels=None) -> Path: """``prep`` with small test-sized context bins/sampling.""" return prep( @@ -313,73 +300,6 @@ def test_compute_reduced_rejects_out_of_range_chunk(tmp_path: Path): compute_one("marginal_edep", run_dir, chunk_index=1) -def test_write_submit_description(tmp_path: Path): - run_dir = _prep([_write_inputs(tmp_path)]) - _fake_venv(tmp_path) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path) - txt = write_submit(cfg).read_text() - assert "universe = docker" in txt - assert "docker_image = cverstege/alma9-gridjob" in txt - assert "requirements = TARGET.ProvidesETPResources" in txt - assert "accounting_group = cms" in txt - assert "+RequestWalltime = $(walltime)" in txt - assert "queue plotid,chunk,walltime from" in txt - jobs = [line.split(",") for line in (run_dir / "jobs.txt").read_text().split()] - assert [i for i, _, _ in jobs] == catalog_ids() - assert all(k == "0" for _, k, _ in jobs) # n_chunks=1 default - assert all(int(w) > 0 for _, _, w in jobs) - wrapper = run_dir / "run_compute.sh" - assert wrapper.exists() and (wrapper.stat().st_mode & 0o111) - body = wrapper.read_text() - assert "giant analyze compute-one --id" in body - assert "--chunk" in body and "--run-dir" in body - - -def test_write_submit_requires_synced_venv(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): - run_dir = _prep([_write_inputs(tmp_path)]) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path) - # No `giant` next to the (fake) active interpreter, so this falls through - # to repo_dir/.venv/bin/giant, which _write_inputs/_prep also didn't create. - monkeypatch.setattr(sys, "executable", str(tmp_path / "not-a-venv" / "bin" / "python")) - with pytest.raises(FileNotFoundError, match="uv sync"): - write_submit(cfg) - - -def test_write_submit_remote_flag(tmp_path: Path): - run_dir = _prep([_write_inputs(tmp_path)]) - _fake_venv(tmp_path) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path, remote=True) - txt = write_submit(cfg).read_text() - assert "+RemoteJob = True" in txt - assert "ProvidesETPResources" not in txt - - -def test_write_submit_chunks_respect_chunkable(tmp_path: Path): - assert get_spec("router_gating").chunkable is False - run_dir = _prep([_write_inputs(tmp_path)], chunks=4) - _fake_venv(tmp_path) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path, n_chunks=4) - write_submit(cfg) - jobs = [line.split(",") for line in (run_dir / "jobs.txt").read_text().split()] - counts: dict[str, int] = {} - for spec_id, _, _ in jobs: - counts[spec_id] = counts.get(spec_id, 0) + 1 - assert counts["marginal_edep"] == 4 - assert counts["router_gating"] == 1 # chunkable=False, ignores n_chunks - - -def test_write_submit_rejects_n_chunks_mismatch_with_run_meta(tmp_path: Path): - """cfg.n_chunks must match the n_chunks the run_dir was actually prepped - with — RunMeta.rows_per_chunk is sized to the prepped value, so a - mismatch would otherwise surface as a confusing IndexError deep inside - _job_walltimes instead of a clear error here.""" - run_dir = _prep([_write_inputs(tmp_path)], chunks=2) - _fake_venv(tmp_path) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path, n_chunks=4) - with pytest.raises(ValueError, match="n_chunks"): - write_submit(cfg) - - def test_estimate_runtime_s_scales_with_rows_and_margin(): from giant.analysis import RUNTIME_SAFETY_MARGIN, estimate_runtime_s from giant.analysis.runtime_estimate import _FIXED_OVERHEAD_S @@ -389,18 +309,3 @@ def test_estimate_runtime_s_scales_with_rows_and_margin(): large = estimate_runtime_s("marginal_edep", 100_000_000) assert small >= (1 + RUNTIME_SAFETY_MARGIN) * _FIXED_OVERHEAD_S assert large > small # bigger chunk -> longer estimate - - -def test_write_submit_walltime_grows_with_chunk_rows(tmp_path: Path): - """A chunked run's later job walltimes track that chunk's row count.""" - from giant.analysis.runtime_estimate import estimate_runtime_s - - run_dir = _prep([_write_inputs(tmp_path)], chunks=2) - meta = RunMeta.load(run_dir / "run_meta.json") - _fake_venv(tmp_path) - cfg = SubmitConfig(run_dir=run_dir, accounting_group="cms", repo_dir=tmp_path, n_chunks=2) - write_submit(cfg) - jobs = {(i, int(k)): int(w) for i, k, w in (line.split(",") for line in (run_dir / "jobs.txt").read_text().split())} - for chunk in range(2): - expected = estimate_runtime_s("marginal_edep", meta.rows_per_chunk[chunk]) - assert jobs[("marginal_edep", chunk)] == expected diff --git a/tests/test_render.py b/tests/test_render.py index da580aa..af77b3c 100644 --- a/tests/test_render.py +++ b/tests/test_render.py @@ -206,21 +206,21 @@ def test_render_all_run_gallery_invokes_subprocess(tmp_path: Path, monkeypatch): assert kwargs == {"check": True} -def test_render_run_glues_condor_run_meta_into_render_all(tmp_path: Path, monkeypatch): - from giant.analysis import condor as condor_mod +def test_render_run_glues_run_meta_into_render_all(tmp_path: Path, monkeypatch): + from giant.analysis import run as run_mod run_dir = tmp_path / "run" (run_dir / "reduced").mkdir(parents=True) merge_calls = [] - monkeypatch.setattr(condor_mod, "merge_all", lambda rd: merge_calls.append(Path(rd))) - meta = condor_mod.RunMeta( + monkeypatch.setattr(run_mod, "merge_all", lambda rd: merge_calls.append(Path(rd))) + meta = run_mod.RunMeta( rollouts=[{"name": "rollout", "path": "rollout.parquet", "plot_meta": {"checkpoint": "ckpt/best.pt"}}], reference="reference.parquet", run_dir=str(run_dir), title="my-run", ) - monkeypatch.setattr(condor_mod.RunMeta, "load", classmethod(lambda cls, p: meta)) + monkeypatch.setattr(run_mod.RunMeta, "load", classmethod(lambda cls, p: meta)) Reduced("s", "species", "single_hist", "Single", "x", {"edges": [0, 1], "series": {"rollout": [1]}}).save( run_dir / "reduced" / "s.json"