From 15b34031c6b81f67fb37a4420204f1b01424be73 Mon Sep 17 00:00:00 2001 From: GABRIELA CORDOVA <100548769@alumnos.uc3m.es> Date: Thu, 6 Aug 2026 09:41:10 +0200 Subject: [PATCH] feat(scripts): pack, verify and document the KITScenes corpus for local training Training KITScenes outside the cluster is harder than it should be. The archives are ~3 GB each, so extracting the 533-scene train split needs well over 2 TB and the archives themselves are ~1.6 TB; streaming the download does not help, since each archive still lands on disk before packing. Mistakes surface late, too: train_il validates a packed corpus against the frozen manifest and aborts on any mismatch, but only after every shard has been scanned. And what a contributor needs before the first epoch is spread across issue comments, a Dockerfile and a couple of error messages. pack_kitscenes_corpus.py packs one scene at a time and deletes as it goes. With --fetch it downloads each archive too, so nothing accumulates: peak disk is one archive plus one extraction, roughly 8 GB for the whole corpus instead of 1.6 TB. Measured over the complete train split (533 partitions, 404 of them non-empty, 42,667 samples), the packed output is 252 KB per sample and 10.2 GB in total. Resumable; stops on its own after three consecutive failures or below a free-space floor. verify_kitscenes_corpus.py runs the eight provenance checks train_il performs, before the run rather than during it, reporting each one separately. Tar headers and manifest.json only; no camera payload decoded. Neither tool names a manifest or a version. Both resolve the frozen split through KITSCENES_TRAINING_POLICY.validation_manifest, and the packer takes its dataset version from that file, so they follow the snapshot the trainer is actually validating against. A checker pinned to one snapshot keeps passing a corpus that training has already started rejecting, which is the failure it exists to catch. Docs/training_on_a_local_machine.md, linked from the README, covers the prerequisites that are not in requirements.txt (the KITScenes SDK, its numpy<2.0 constraint, opencv, kubernetes being needed at import time), dataset access, getting and packing the corpus, which two filters drop scenes and the log line reporting all three counts, generating the audit, running the training, and what to report so results line up. Two behaviours are encoded in the scripts because they are silent when wrong: -C is positional in GNU tar, so trailing it returns 0 while extracting into the working directory; and data_processing defaults to DATASET_PACK_VERSION while the KITScenes navigation path uses KITSCENES_NAVIGATION_DATASET_VERSION, which is what the frozen manifest carries. Verified on the complete corpus. On a laptop that started with nothing installed and no data, following the page added here: 533 partitions fetched one archive at a time and packed by this script over 63 hours, then checked by this script against kitscenes_train_dev_v3.json on all eight comparisons -- 533 partitions, 129 empty, 404 eligible groups, 42,667 samples, v3.3, and the group, sample and contract digests -- with validation resolving to the 40 scenes and 3,820 samples behind group_digest 903fec7d. The checker was also exercised against a corpus with one damaged scene, where it reported five failures naming the discrepancy. The page then carried on through the audit, local S3 and MLflow, and a training epoch over that corpus which completed in 2 h 20 min and uploaded its checkpoint -- using the companion change for the BEV grid, since the 6 GB card here cannot hold the default one. 16 new tests cover the -C trap, a propagated tar failure, partition scanning, each frozen-manifest check failing independently, and that both tools track the policy's manifest rather than a snapshot of it. They come with a conftest that puts Model/ on the import path: Model/tests gets that from its own pytest.ini and __init__.py, Platform/tests has neither, and without it a bare `pytest Platform/tests` from the repository root fails on `training.*`. Signed-off-by: GABRIELA CORDOVA <100548769@alumnos.uc3m.es> --- Docs/training_on_a_local_machine.md | 793 ++++++++++++++++++ Platform/scripts/pack_kitscenes_corpus.py | 328 ++++++++ Platform/scripts/verify_kitscenes_corpus.py | 235 ++++++ Platform/tests/conftest.py | 22 + Platform/tests/test_kitscenes_corpus_tools.py | 221 +++++ README.md | 1 + 6 files changed, 1600 insertions(+) create mode 100644 Docs/training_on_a_local_machine.md create mode 100644 Platform/scripts/pack_kitscenes_corpus.py create mode 100644 Platform/scripts/verify_kitscenes_corpus.py create mode 100644 Platform/tests/conftest.py create mode 100644 Platform/tests/test_kitscenes_corpus_tools.py diff --git a/Docs/training_on_a_local_machine.md b/Docs/training_on_a_local_machine.md new file mode 100644 index 000000000..ba036bef5 --- /dev/null +++ b/Docs/training_on_a_local_machine.md @@ -0,0 +1,793 @@ +# Training the Reactive branch on your own machine + +A practical guide for running KITScenes training outside the cluster, so that the +results are comparable with everyone else's. + +The pipeline assumes a cluster in a few places. None of those assumptions is a bug, but +together they are what stands between a new contributor and a first epoch using only the +Reactive branch. This page collects them, along with the corpus handling that makes a +local run possible at all. + +**If you read one section, read the last one.** +[Reporting results so they can be compared](#reporting-results-so-they-can-be-compared) +is where an ADE turns into a result: the three lines from your own log that say which +scenes, which training policy and which metric contract produced it, and the score the +same validation set gets from a model with no cameras at all. Everything before it +exists so that those four things are obtainable — the corpus tooling included. A number +without them cannot be compared with anyone else's, which is the problem this page was +written for. + +Every step below has been run on a laptop that started with nothing installed and no +data: environment, dataset access, the full 533-scene corpus, verification against the +frozen split, the navigation audit, local S3 and MLflow, and a training epoch whose +checkpoint uploaded. Every wall described here was hit on that machine, and every fix +was applied there — including the training epoch, which needed the BEV grid edit under +[GPU memory](#gpu-memory) because the card was a 6 GB one. + +**If your GPU has less than 12 GB, read [GPU memory](#gpu-memory) before you start.** +There is a code edit you want to make before step 3, not after. + +--- + +## The commands, in order + +Copy-paste sequence. Each step names the section that explains it; read that one if +the step fails or if you want to know why it is there. Replace `/data` and `` +with your own paths. + + + + +```bash +# 1. Environment. The SDK is not on PyPI, needs numpy<2.0, and the pipelines do +# not import without kubernetes. -> "Prerequisites" +# Build it OUTSIDE the repo: neither the venv nor the SDK clone is gitignored, +# and a 7 GB .venv/ in `git status` is a mistake waiting to be committed. +mkdir -p /data/env && cd /data/env +python3 -m venv venv && source venv/bin/activate # python3: `python` may not exist yet +python -m pip install --upgrade pip # the bundled pip cannot parse the SDK's extra +git lfs install +GIT_LFS_SKIP_SMUDGE=1 git clone https://github.com/KIT-MRT/kitscenes.git +git -C kitscenes checkout 7765cdec5490894266070ab46e23724b58b3da42 +git -C kitscenes lfs pull +cd kitscenes && pip install ".[map]" && cd .. # from INSIDE: the wheels are relative +# torchvision and mlflow are on the training path but NOT in requirements.txt, +# and requirements.txt pins numpy==2.2.6 which breaks the SDK. Do not apply it. +pip install torch torchvision timm webdataset boto3 pyproj mlflow \ + "flytekit==1.14.9" kubernetes "opencv-python-headless<4.10" +pip install "numpy<2" +# Check the environment before spending days on a corpus: NumPy must be 1.x, and +# torch must actually see the GPU -- `torch` is unpinned, so pip picks a CUDA build +# that your driver may be too old for, and a CPU-only install trains silently. +env -u PYTHONPATH python -c "import numpy, torch, kitscenes, flytekit, kubernetes, \ +webdataset, boto3, pyproj, mlflow; print(numpy.__version__, torch.__version__, \ +torch.cuda.is_available())" + +# 2. Dataset access. Accept the terms on the dataset page first, then log in. +# Check the second line: an expired token reports a permissions error. -> "Prerequisites" +hf auth login +python -c "from huggingface_hub import whoami; print(whoami()['name'])" + +# On a card under 12 GB, make the BEV grid edit NOW -> "GPU memory". Packing +# re-imports workflows.py in worker processes, so editing it mid-pack can kill a +# multi-day run; before it starts or after it ends are the safe moments. + +# 3. Corpus. Downloads, packs and deletes one scene at a time: peak disk is about +# 8 GB, and the packed result is 10.2 GB. The full split took 63 h on a laptop, +# so budget two and a half days. Run it detached, give it its own checkout, and +# do not train on this machine meanwhile. Resumable. -> "Getting the corpus" +# The dataset version is read from the frozen manifest; do not pass one. +cd +export PYTHONPATH=/Model: +setsid nohup python Platform/scripts/pack_kitscenes_corpus.py \ + --fetch --work-root /data/_staging --out-root /data/_shards \ + > /data/pack.log 2>&1 < /dev/null & +tail -f /data/pack.log # the header must print: # version: + +# 4. Check the corpus BEFORE training, not during. Eight checks; exit 0 means your +# corpus matches the frozen split and your numbers will compare. -> "Check the corpus" +python Platform/scripts/verify_kitscenes_corpus.py --shards-root /data/_shards + +# 5. Navigation audit. Mandatory: train_il refuses to run on KITScenes without it, +# and it must cover exactly the shards you train on. -> "Running the training" +# shards_index_verified.json is written by step 4, and only when all eight +# checks pass -- so if it is missing, go back to step 4. +SHARDS=$(python -c "import json,sys;print(json.dumps(json.load(open(sys.argv[1]))['packed']))" \ + /data/_shards/shards_index_verified.json) +pyflyte run Platform/pipelines/workflows.py \ + audit_kitscenes_navigation_quality --shards "$SHARDS" +cp /data/_shards/audit.json + +# 6. Local services. Checkpoint upload assumes S3 and MLflow needs a real backend; +# a file:// tracking URI is rejected by MLflow 3.x. -> "Cluster assumptions" +export AWS_ENDPOINT_URL=http://localhost:9000 +export AUTO_E2E_CHECKPOINT_BUCKET= +export MLFLOW_TRACKING_URI=sqlite:///$HOME/mlflow.db + +# 7. Train. One epoch first: it is 2 h 20 min on a 6 GB card, and it is where a +# wrong audit or a missing bucket shows up. Raise --epochs once it has run +# through -- at that rate ten is about a day. -> "Running the training" +pyflyte run Platform/pipelines/workflows.py wf_train_il \ + --shards "$SHARDS" \ + --navigation_quality_audit /data/_shards/audit.json \ + --dataset "KIT-MRT/KITScenes-Multimodal" \ + --validation_scope full \ + --backbone swin_v2_tiny --epochs 1 \ + --batch_size 1 --grad_accum_steps 4 --lr 1e-4 \ + --val_fraction 0.1 --num_workers 4 --training_seed 149 + +# 8. Report your numbers together with the three lines that make them comparable: +# the split digest, the training policy and the epoch result. Do not stop here -- +# the section explains what each line rules out, and how to find out what your +# validation set scores with no perception at all. -> "Reporting results" +grep -E "group_digest=|Dataset training policy:|Epoch [0-9]+/" /data/train.log +``` + +The training step is where a small GPU stops — see [GPU memory](#gpu-memory). + +--- + +## GPU memory + +This is the one thing the page does not solve. The KITScenes geometry fixes the camera +BEV grid at 256x256 — 65,536 queries — and `train_il` exposes no way to change it. +Measured on a 6 GB card: out of memory at `batch_size 1`, and +`PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True` does not close the gap. A 12 GB card +is reported to fit with that flag; the threshold in between is not measured here. CPU +completes a smoke epoch and is far too slow for a real run. + +If your card is too small, the workaround is a code change. It is an *addition*, not a +value to find and replace: `train_il` does not spell the grid out, it asks the +geometry for it. Find where `view_fusion_kwargs` is built for KITScenes in +`Platform/pipelines/workflows.py` and override the two keys straight after: + +```python +view_fusion_kwargs = ( + DEFAULT_NAVIGATION_GEOMETRY.camera_bev_kwargs() +) +view_fusion_kwargs["bev_h"] = 64 +view_fusion_kwargs["bev_w"] = 64 +``` + +Leave `pc_range` alone, so the BEV still covers the same ground with coarser cells +rather than a smaller patch of it. Those two lines are the whole change. (Searching +the repository instead leads to `matching_bev_h`/`matching_bev_w` in +`Model/navigation/geometry.py`, which is the wrong place: the geometry is validated +against the map raster and rejects a smaller grid on construction.) + +At 64 the grid fits, but not comfortably: a full-corpus epoch on that 6 GB card peaked +at 5.7 GB of the 6.1 available, and the allocator logged recoverable +`memory allocation failed with OOM` warnings on the way. They are warnings, not a +crash — it frees cache and retries — but there is little headroom left. + +**Make this edit before step 3, not between steps.** Packing re-imports +`Platform/pipelines/workflows.py` in worker processes, so editing that file while a +multi-day pack is running can kill it. Before the pack or after it, never during. + +That edit is outside the scope of this page, which documents the pipeline as it +stands; making the grid a parameter is a separate change. + +--- + +## Prerequisites + +Everything below was hit in order on a fresh machine. None of it is written down +anywhere a contributor would look, so it is collected here first. + +**Put the environment outside the repository.** Neither `.venv/` nor a `kitscenes/` +clone is in `.gitignore`, so building them in the checkout puts several gigabytes into +`git status` where they can be committed by accident. Everything below assumes a +directory of your own next to the data, not inside the repo. + +Create it with `python3`, not `python`: on Debian and Ubuntu there is no bare `python` +unless `python-is-python3` is installed, so the very first command fails on a machine +that has nothing set up — which is the machine this page is for. Every later `python` +in this page is the one the activated venv provides, and those are fine. + +**The KITScenes SDK is not in `requirements.txt`, and not on PyPI.** It is a git +repository with LFS assets, pinned in `Platform/docker/data-prep/Dockerfile`: + +```bash +git lfs install +GIT_LFS_SKIP_SMUDGE=1 git clone https://github.com/KIT-MRT/kitscenes.git +git -C kitscenes checkout 7765cdec5490894266070ab46e23724b58b3da42 +git -C kitscenes lfs pull +cd kitscenes && pip install ".[map]" && cd .. +``` + +The `map` extra ships prebuilt Lanelet2 wheels for CPython 3.8 to 3.12, so pick an +interpreter in that range. + +**The `map` extra declares its wheels as *relative* file URLs**, and that one fact +breaks the install twice, in this order. + +*First, old pip cannot parse them.* The pip that `python3 -m venv` bundles on Ubuntu +22.04 (22.0.2) rejects `lanelet2 @ file:res/ml_converter_wheels/...whl` during +dependency resolution, with a traceback ending in: + +``` +pip._vendor.pkg_resources.RequirementParseError: Invalid URL given +``` + +It names neither pip nor the SDK, and reads like a corrupt package. Verified on the +same machine: pip 22.0.2 rejects that requirement string and pip 26.2.1 accepts it. +`python -m pip install --upgrade pip` first is the whole fix. + +*Then, with a pip that parses them, the path is resolved from the wrong place.* Being +relative, those URLs resolve against the current working directory rather than against +the package being built, so `pip install "./kitscenes[map]"` gets as far as the wheel +and stops: + +``` +ERROR: Could not install packages due to an OSError: [Errno 2] No such file or +directory: 'res/ml_converter_wheels/lanelet2-1.2.2-cp310-...whl' +``` + +The file is there; pip is looking one directory too high. Install from inside the +clone — `Platform/docker/data-prep/Dockerfile` does +`cd /tmp/kitscenes && pip install ".[map]"` for exactly this reason. + +**Use a dedicated virtual environment, and do not simply apply `requirements.txt` on +top.** The SDK requires `numpy<2.0`; `requirements.txt` pins `numpy==2.2.6`. The two +cannot both be satisfied, and installing the requirements after the SDK silently +upgrades NumPy and breaks it — with an error that points at the SDK rather than at +the version. Install the runtime dependencies explicitly instead, and pin NumPy last: + +```bash +pip install torch torchvision timm webdataset boto3 pyproj mlflow \ + "flytekit==1.14.9" kubernetes "opencv-python-headless<4.10" +pip install "numpy<2" +``` + +`torchvision` and `mlflow` are imported by the training path but are not in +`requirements.txt`, so neither an editable install nor the requirements file gets you +there. + +OpenCV is not in `requirements.txt` either, and packing does not survive without it: +`Model/data_parsing/kit_scenes/map.py` imports `cv2`, so every partition fails with a +bare `ModuleNotFoundError` — after its 3 GB archive has already been downloaded — and +three consecutive failures abort the run. The `<4.10` bound is the one +`Platform/docker/data-prep/Dockerfile` uses, and it is what keeps OpenCV from pulling +NumPy 2 back in on top of the SDK. + +**If you have ROS sourced, unset `PYTHONPATH` first.** ROS puts its own +`site-packages` on `PYTHONPATH`, and that leaks into a virtual environment and can +shadow packages — NumPy in particular. Either `unset PYTHONPATH` or prefix commands +with `env -u PYTHONPATH`. This is easy to miss precisely because the environment +looks activated. + +**`kubernetes` is what makes the import fail first**, if you wonder why it is in that +list: `overlay_tasks` builds a pod template at import time, so +`Platform.pipelines.workflows` will not import without it even for a purely local run. + +**Dataset access.** `KIT-MRT/KITScenes-Multimodal` is gated: the file listing is +public but downloads are not. Accept the terms on the dataset page, then +authenticate: + +```bash +hf auth login +python -c "from huggingface_hub import whoami; print(whoami()['name'])" +``` + +Worth checking that second line, because an **expired** token reports +`Access denied. This repository requires approval` on download — which points at the +terms rather than at the token, and sends you to the wrong place. + +## Getting the archives + +You need the KITScenes archives on disk first. `data_ingest` in +`Platform/pipelines/workflows.py` is the pipeline's own ingest path, but note that +`PinnedKITScenesDownloader.download()` downloads **and extracts** each scene, so +running it over the whole train split needs the same 2 TB-plus this page exists to +avoid. On a workstation, fetch the archives without extracting them: + +```bash +hf download KIT-MRT/KITScenes-Multimodal \ + --repo-type dataset --revision 6fde0034446669e2ed7235e4c7fe323cd23d599d \ + --include "data/train/*.tar" --local-dir /data/KITScenes-Multimodal +``` + +(`huggingface-cli download` was the older spelling and no longer works on current +`huggingface_hub`.) + +That revision is the one `data_processing` pins; packing against a different one is +rejected. The archives are around 3 GB each and roughly 1.6 TB in total, and they +stay compressed — only one is extracted at a time by the next step. + +## The split the code enforces today + +`KITSCENES_TRAINING_POLICY.validation_manifest` in `Model/training/dataset_policy.py` +names the frozen train/dev split, and `train_il` checks the packed corpus against it. +This section describes what that manifest contains and what happens if a corpus does +not match it; which split the working group ultimately adopts is a separate +discussion, in #168. + +Read the filename from the policy rather than memorising it. The snapshot is re-cut +as the packed data changes, and each re-cut carries a new dataset version and a new +contract digest — a corpus packed against the previous one no longer validates. + +The manifest declares: + +| Field | Value | Stable across re-cuts? | +|---|---|---| +| `official_split` | `train` | yes | +| `available_scene_count` | 533 | yes | +| `excluded_empty_scene_count` | 129 | yes | +| `eligible_group_count` | 404 | yes | +| `validation_fraction` | 0.1 | yes | +| `validation_group_count` | 40 | yes | +| `validation_sample_count` | 3820 | yes | +| `dataset_version` | changes | **no** | +| `packed_contract_digest` | changes | **no** | + +The counts and the holdout have survived every re-cut so far; the version and the +contract digest are exactly what a re-cut changes. Print the current ones rather than +copying them from here: + +```bash +python - <<'EOF' +import json, pathlib +from training.dataset_policy import KITSCENES_TRAINING_POLICY as policy +path = pathlib.Path("Model/training") / policy.validation_manifest +frozen = json.loads(path.read_text()) +print(path.name, frozen["dataset_version"], frozen["packed_contract_digest"][:16]) +EOF +``` + +With `--validation_scope full`, `train_il` compares your packed corpus against that +manifest — partition counts, sample counts and two SHA-256 digests — and aborts on +any mismatch. In practice that means packing the 533 `data/train/` archives and +letting the code select the holdout; anything else stops the run rather than +silently training on a different corpus. + +Use `--validation_scope subset` only for a bring-up run on a partial corpus. It +relaxes the *counts*, accepting a proper subset of the 533 scenes, but still requires +the provenance triple (source revision, dataset version, packed contract digest) to +match. Metrics from a subset run are **not** comparable with anyone else's. + +## Getting the corpus onto a workstation + +**If your card is under 12 GB, make the BEV grid edit from [GPU memory](#gpu-memory) +before you start this step.** Nothing in the corpus depends on it — it is only step 7 +that cares — but packing re-imports `Platform/pipelines/workflows.py` in worker +processes, so editing that file mid-run can kill a multi-day job. Before the pack or +after it, never during. + +The archives are around 3 GB each, so extracting the whole train split at once needs +well over 2 TB. Streaming the download does not help, because each archive still +lands on disk before it can be packed. + +Stream the *packing* instead — extract one archive, pack it, delete the extracted +copy, move on: + +```bash +export PYTHONPATH=/Model: +python Platform/scripts/pack_kitscenes_corpus.py \ + --tar-src /data/KITScenes-Multimodal/data/train \ + --work-root /data/_staging \ + --out-root /data/_shards +``` + +Peak disk stays at roughly one scene instead of the whole corpus, because the packed +output is a small fraction of the raw archives. Measured over the complete train +split — 533 partitions, 404 of them non-empty, 42,667 samples — it comes to **252 KB +per sample and 10.2 GB in total**. + +Do not size a disk from a handful of scenes: per-scene cost ranges from 114 to 553 KB +per sample, so a small subset can be off by a factor of two in either direction. + +The run is resumable — a partition that already has a `manifest.json` is skipped — so +it can be interrupted and restarted. + +**If the archives themselves do not fit either, let it fetch them one at a time:** + +```bash +python Platform/scripts/pack_kitscenes_corpus.py \ + --fetch --work-root /data/_staging --out-root /data/_shards +``` + +Each archive is downloaded, unpacked, deleted, and the extracted copy deleted after +packing, so peak usage is one archive plus one extraction — roughly 8 GB for the +whole 533-scene corpus rather than the 1.6 TB the archives would occupy together. + +Budget time generously. The full 533-scene split took **63 hours** on a laptop over a +USB disk, fetch to packed: 7.1 minutes per scene on average, a median of 5.1, and a +spread of 1.3 to 40 minutes that tracks how long each scene is. Plan for two and a +half days, unattended. + +Do not extrapolate from a handful. The first two scenes we timed averaged 3.2 minutes +and would have predicted one day; sixteen consecutive scenes gave a 6.2-minute median +and predicted three. Only the whole run settles it. + +**Run it detached, because two and a half days is longer than a terminal session:** + +```bash +setsid nohup python Platform/scripts/pack_kitscenes_corpus.py \ + --fetch --work-root /data/_staging --out-root /data/_shards \ + > /data/pack.log 2>&1 < /dev/null & +``` + +Write the log outside the checkout — `pack.log` is not in `.gitignore` either. +`setsid` and `< /dev/null` are what make it survive the shell exiting or an SSH drop; +`nohup` alone is not always enough. Follow it with `tail -f /data/pack.log`. To stop it, take +the PID from `pgrep -f "[p]ack_kitscenes_corpus"` and `kill` that — a bare +`pkill -f pack_kitscenes_corpus` also matches the shell you typed it in. + +**Do not train on the same machine while it packs.** Packing is CPU and IO heavy and +holds a scene in memory; a training run alongside it exhausted RAM on a 14 GB laptop +and killed both. The pack resumes, but the hours in flight are lost. + +**Do not switch branches in the checkout it is running from.** Packing spawns worker +processes, and each one re-imports the script by its path; if the file has moved or +disappeared in the meantime, every worker dies with `BrokenProcessPool` and the run +aborts after three consecutive failures. The traceback names `multiprocessing` and +reads like a resource problem, but the cause is that the script is no longer on disk. +Give a multi-day pack its own checkout — `git worktree add` is enough — and leave that +tree alone until it finishes. + +Note that `shards_index.json` is written when the run finishes, so after an +interruption it lags behind what is on disk until you resume and let it complete. + +Clear the staging root after killing a run, **and only while nothing is running.** +Cleanup happens per scene in a `finally`, which a killed process never reaches, so the +scene that was in flight stays extracted — a few gigabytes that no later run will +remove, since each one only cleans up after itself. + +Deleting it later, with a pack in progress, is how you get a corpus that fails +verification for no visible reason. A resumed run walks the scenes in the same order, +so it reaches that same directory again and starts extracting into it; a concurrent +`rm -rf` removes files bottom-up, takes out a camera directory, and fails on the +parent with *"directory not empty"* — which reads like the delete did nothing. The +scene then packs as `missing cameras [...]. Skipping.` and lands as an empty +partition, one line among hundreds. Nothing else is affected, and the counts come up +one short at the end of the run. Wait for the run to end. + +Two things the script encodes, because both are silent when wrong: + +- **One scene per partition.** Calibration and map state are scene-scoped, so + `data_processing` raises *"KITScenes partition contains scenes with different + calibration; pack one scene per partition"* when a partition mixes them. + `train_il` takes the resulting list of shard directories directly. +- **Pack with the KITScenes dataset version.** `data_processing` defaults to + `DATASET_PACK_VERSION`, but the KITScenes navigation path uses + `KITSCENES_NAVIGATION_DATASET_VERSION`, and that is what the frozen manifest + carries. Packing with the default and then training fails with *"validation + manifest dataset version does not match packed shards"*. The script reads the + version out of the frozen manifest and prints it in its header, so leave + `--dataset-version` alone unless you are deliberately packing against an older + snapshot. + +## Check the corpus before you train, not during + +`train_il` validates the corpus at start-up, after scanning every shard. On a corpus +that took hours to pack, that is a late place to discover a wrong dataset version. +Run the same checks first: + +```bash +python Platform/scripts/verify_kitscenes_corpus.py --shards-root /data/_shards +``` + +It reports the eight comparisons individually and exits non-zero if any fails. On +success it writes `shards_index_verified.json` next to the shards — that file is the +input to the next two steps, so a missing one means the corpus did not pass. + +### When the frozen snapshot is re-cut + +It will be, and it does not announce itself. What you see is a corpus that passed +last week failing two checks and only two: + +``` + [FAIL] dataset version v3.0 expected v3.3 + [FAIL] contract digest c81a5746a365246f... expected 6fb9d857d877e570... +``` + +Counts and group digests still pass, which is the confusing part: the data is fine, +the contract it was packed under is not. `data_processing` stamps both at pack time, +so there is nothing to edit — **a re-cut corpus has to be repacked, not re-checked.** + +Pull first and check before you spend days on it. If a pack is already running when +the re-cut lands, let it finish and repack afterwards rather than resuming into it: +resuming skips partitions that already have a `manifest.json`, so the old ones keep +the old version and you end up with a corpus that can never pass. Packing into a +fresh `--out-root` keeps the previous one intact while you decide. + +## What happens to your scenes + +Two filters remove data, and they are easy to confuse. + +**Packing drops scenes that are too short.** A sample needs 64 history steps plus 64 +future steps plus one, so a scene needs at least 129 usable frames — 12.9 s at 10 Hz +(`MIN_ROWS` in `Model/data_parsing/kit_scenes/egomotion.py`). "Usable" is the minimum +across poses, reference timestamps and the *contiguous prefix* of each camera, so one +early camera gap truncates a scene that otherwise looks long enough. This is where +`excluded_empty_scene_count: 129` comes from. + +**The navigation quality audit trims further, but only for the optimizer.** It is +mandatory for KITScenes — `train_il` refuses to run without one — and removes +partitions from training. The validation groups are built from the unfiltered +non-empty partitions, so two contributors whose audits accept different scenes still +validate on the same samples. + +One line in the training log gives you all three numbers: + +``` +Selected N/M non-empty partition(s) for the optimizer (skipped_empty=K, ...) +``` + +`M` is what survived packing, `K` what packing dropped, `N` what the audit kept. + +## Running the training + +`train_il` refuses to run on KITScenes without a navigation quality audit, and the +audit's coverage must match the shards you pass **exactly** — reusing one generated +over a different set of partitions fails with "navigation quality audit partition +coverage differs from shards". So generate it over the corpus you are about to train +on: + +```bash +SHARDS=$(python -c "import json,sys;print(json.dumps(json.load(open(sys.argv[1]))['packed']))" \ + /data/_shards/shards_index_verified.json) + +pyflyte run Platform/pipelines/workflows.py \ + audit_kitscenes_navigation_quality --shards "$SHARDS" +# copy the resulting file next to the shards, e.g. /data/_shards/audit.json +``` + +Over 400 partitions this takes a while and says nothing until it is done, then prints +one line and the report path: + +``` +KITScenes navigation quality: accepted=329 excluded=75 report=/tmp/navigation-quality-.../navigation_quality_audit.json +``` + +**Copy it out of `/tmp` before you reboot.** That is the only copy, and regenerating it +means running the audit over every partition again. + +Keep `$SHARDS` in the same shell you train from, or rebuild it there: it is a JSON +array of a few hundred absolute paths, and the training command needs the same one the +audit was generated over. + +Then train: + +```bash +pyflyte run Platform/pipelines/workflows.py wf_train_il \ + --shards "$SHARDS" \ + --navigation_quality_audit /data/_shards/audit.json \ + --dataset "KIT-MRT/KITScenes-Multimodal" \ + --validation_scope full \ + --backbone swin_v2_tiny --epochs 1 \ + --batch_size 1 --grad_accum_steps 4 --lr 1e-4 \ + --val_fraction 0.1 \ + --num_workers 4 --training_seed 149 +``` + +Without `--remote`, `pyflyte` runs locally. The CI uses `--remote` for the cluster. + +`pyflyte` opens with a screenful of `UserWarning: The parameter --x is used more than +once`, once per boolean flag, before anything of yours runs. It is click reporting on +the generated CLI, not on your command. Ignore it; the first line that is about your +run is `Running Execution on local.` + +**Nothing is printed between epochs.** There is no per-step counter, and the MLflow +metrics are written when an epoch ends, so on a long epoch the run is silent for hours +and an interrupted one leaves no trace of how far it got. To tell a working run from a +hung one, watch the GPU rather than the log: + +```bash +nvidia-smi --query-gpu=utilization.gpu,memory.used --format=csv,noheader +``` + +Sustained non-zero utilisation means it is training. A rate well below 100% is normal +here: at `batch_size 1` the JPEG decode in the dataloader, not the GPU, is usually the +limit. + +**What an epoch costs.** On an RTX 3060 Laptop (6 GB) with the BEV grid at 64 and +`num_workers 4`, over the 329 partitions a navigation audit accepted out of 404: one +epoch took **2 h 20 min**, at 3.49 samples per second and 0.87 optimizer steps per +second. Ten epochs is therefore about a day on that machine, and the run reports both +rates itself, so you can extrapolate from your own first epoch rather than from this +one. + +When the epoch ends it prints a single line with everything in it: + +``` +Epoch 1/1 loss=0.2037 traj=0.2037 route=0.0000 jepa=0.0000 reason=0.0000 +val_ADE=1.5738 val_FDE=4.3911 score_improved=True trajectory_improved=True +samples_per_second=3.492 optimizer_steps_per_second=0.873 bad_epochs=0 +checkpoint=s3:///imitation-learning//epoch-0001.pt +``` + +`val_ADE` and `val_FDE` are **at 3 seconds**, not over the full prediction horizon — +see [Reporting results](#reporting-results-so-they-can-be-compared) before comparing +them with anything. + +**On GPU memory**, see [GPU memory](#gpu-memory): the 256x256 grid does not fit in +6 GB, and the workaround is the `bev_h`/`bev_w` edit described there. A deformable +map fusion, which removes the quadratic term in the map-to-BEV attention at +production resolution, is a separate lever and is already available as +`map_fusion_mode`. + +One caveat if you have more than one GPU: `train_il` writes its local epoch +checkpoints under a shared directory, so two trainings on the same machine can +delete each other's files and one dies mid-run with a `FileNotFoundError`. Run them +one at a time unless you are on a revision that qualifies that path per run. + +## Cluster assumptions to work around + +**Checkpoint storage assumes AWS, and needs to be disk-backed.** `train_il` resolves +the bucket through STS `get_caller_identity` and uploads with `put_object`, so +without AWS credentials it stops before the first epoch. Two variables avoid it: + +```bash +export AUTO_E2E_CHECKPOINT_BUCKET= # skips the STS lookup +export AWS_ENDPOINT_URL=http://localhost:9000 +``` + +The upload path is plain `put_object`, so any S3-compatible server works — but an +**in-memory** mock is not enough. An epoch checkpoint is 650 to 850 MB depending on +the BEV grid — 663 MB at 64, where the BEV query embedding is small — and an +in-memory server starts returning 500s after a few of them. A disk-backed server +runs a full training without trouble: verified here with MinIO writing to a USB disk, +where an epoch over the full corpus uploaded and the run went on to register the +checkpoint under its `best`, `best_trajectory` and `final` roles. + +**`MLFLOW_TRACKING_URI` has no default.** It is read as `os.environ[...]`, so an +unset variable is a bare `KeyError`. Point it at a SQLite file rather than a +directory: + +```bash +export MLFLOW_TRACKING_URI=sqlite:///$HOME/mlflow.db +``` + +A `file://` URI used to work and no longer does: MLflow 3.x refuses the filesystem +tracking backend outright — *"in maintenance mode and will not receive further +updates... migrate to a database backend"* — unless `MLFLOW_ALLOW_FILE_STORE=true` +is set. SQLite needs no server and is what MLflow recommends. + +The SQLite URI covers the tracking database, not the artifacts: MLflow still writes an +`mlruns/` tree into the working directory, and that directory is not gitignored +either. Training from the repository root therefore leaves it in `git status`. Run +from elsewhere, or delete it afterwards. + +**`num_workers` changes results, not just speed.** It defaults to 0. Running the same +configuration three times with each value — identical data, identical validation +groups, only that parameter differing — the runs with 0 converged worse and far more +erratically: the spread across seeds was several times larger and the training loss +stayed higher. Treat `num_workers > 0` as a correctness setting rather than a +performance one. + +## Reporting results so they can be compared + +**Paste your `group_digest`.** When `--validation_scope full` succeeds, training +prints: + +``` +Validation split: ... groups=40 group_digest= +``` + +An ADE on its own is not a result. What makes it one is the three lines below it that +say which scenes it was measured on, under which training policy, and under which +metric contract — because all three have changed at least once, and a number produced +before a change looks exactly like one produced after. + +### Pull the report out of your own log + +```bash +grep -E "group_digest=|Dataset training policy:|Epoch [0-9]+/" /data/train.log +``` + +That is the whole report. On the run this page was written from it prints: + +``` +Validation split: strategy=exact_group_fraction split_id=kitscenes_train_dev_v1 + groups=40 group_digest=903fec7dd3ca779875ed634ea4fae3c0bddd0bee58c39b3fc2573eb2ca1c1685 +Dataset training policy: auto_e2e_timesteps=64 temporal_decay=0.99 + temporal_weight_normalization=mean_one acceleration_scale=0.778 curvature_scale=0.035 +Epoch 1/1 loss=0.2037 val_ADE=1.5738 val_FDE=4.3911 samples_per_second=3.492 ... +``` + +Post those three lines together and anyone can check your numbers are commensurable +with theirs instead of assuming it. + +- **`group_digest`** — same digest, same 40 scenes and 3,820 samples. A different one + means a different exam. +- **The policy line** — not yours to choose, it comes from `KITSCENES_TRAINING_POLICY`, + but it is revised from time to time and it changes what the loss weights. Two runs + on the same scenes under different policies are not the same experiment. +- **`val_ADE`/`val_FDE`** — the mean and the last-step error **over 3 seconds**, not + over the full prediction horizon. The metric contract that produced them travels + with the validation result: + + ``` + "metric_contract": {"version": "control_rollout_validation_v2", + "horizon_seconds": 3.0, "horizon_steps": 30, + "aggregation": "sample_mean"} + ``` + + `prediction_steps` stays at 64, so the head still predicts 6.4 s; only the scored + window is shorter. Earlier numbers in the thread quote 6.4 s aggregates that are no + longer computed, and nothing warns you — both print as `val_ADE`. The per-horizon + breakdown is reported alongside under `horizons` if you want to know whether an + error invisible at 3 s grows later; the head emits (acceleration, curvature) and the + metric integrates them twice, so it can. + +### Two things the log will not tell you + +**How much of your number is seed noise.** Repeating one fixed configuration across +seeds produces a spread wide enough that a single run per variant cannot separate a +real effect from initialisation luck. Run two or three seeds and report the range, not +a point. + +**What the same validation set scores with no perception at all.** The metric +integrates ego dynamics, so much of it is predictable from the egomotion history +alone: holding the last observed acceleration and curvature already produces a +trajectory. Until you know that number, an ADE cannot be read as good or bad — and it +is not small. The repository does not compute it, but the packed corpus has everything +needed, and the recipe is short enough to state in full: + +Run this from the repository root, with `PYTHONPATH` set as in step 3. It reads only +the packed corpus you already have — no GPU, no SDK, no original archives — and takes +a few minutes: + +```python +import glob +import json +import tarfile +from pathlib import Path + +import numpy as np + +from evaluation.metrics import integrate_trajectory +from training.dataset_policy import KITSCENES_TRAINING_POLICY as policy + +SHARDS = "/data/_shards" + +frozen = json.loads( + (Path("Model/training") / policy.validation_manifest).read_text() +) +validation = set(frozen["validation_group_uids"]) + +ade, fde = [], [] +for tar_path in sorted(glob.glob(f"{SHARDS}/*/*.tar")): + with tarfile.open(tar_path) as archive: + egos, groups = {}, {} + for member in archive: + # Despite the name, .ego.npy is a raw buffer, not a .npy container: + # np.load rejects it. float32[384] = history(64,4) + future(64,2), + # history columns 0 speed, 1 acceleration, 3 curvature. + if member.name.endswith(".ego.npy"): + uid = member.name[: -len(".ego.npy")] + egos[uid] = np.frombuffer( + archive.extractfile(member).read(), dtype=np.float32 + ) + elif member.name.endswith(".meta.json"): + uid = member.name[: -len(".meta.json")] + meta = json.loads(archive.extractfile(member).read()) + groups[uid] = meta.get("split_group_uid") + for uid, ego in egos.items(): + if groups.get(uid) not in validation: # score the holdout only + continue + hist = ego[: 64 * 4].reshape(64, 4) + fut = ego[64 * 4:].reshape(64, 2) # the logged target signals + v0 = float(hist[-1, 0]) # the v0 the evaluator uses + gt = integrate_trajectory(fut[:, 0], fut[:, 1], v0) + hold = integrate_trajectory(np.full(64, hist[-1, 1]), + np.full(64, hist[-1, 3]), v0) + err = np.linalg.norm(hold - gt, axis=1) + ade.append(err[:30].mean()) # the contract's window + fde.append(err[29]) + +print(f"samples {len(ade)} ADE@3s {np.mean(ade):.3f} FDE@3s {np.mean(fde):.3f}") +``` + +On the frozen split at the time of writing it prints: + +``` +samples 3820 ADE@3s 0.843 FDE@3s 2.516 +``` + +That is the bar. A model that scores above it on the same 40 scenes is doing worse +than ignoring its cameras and holding the last observed control, which is worth +knowing before reporting the number as an improvement. diff --git a/Platform/scripts/pack_kitscenes_corpus.py b/Platform/scripts/pack_kitscenes_corpus.py new file mode 100644 index 000000000..2ac4349e9 --- /dev/null +++ b/Platform/scripts/pack_kitscenes_corpus.py @@ -0,0 +1,328 @@ +#!/usr/bin/env python3 +"""Pack the KITScenes train corpus one scene at a time, without staging it all. + +Motivation. `data_processing` needs its `raw_data` to hold extracted scenes, and +the KITScenes archives are ~3 GB each: extracting the 533-scene train split at +once needs well over 2 TB, which does not fit on a typical workstation. Streaming +the *download* does not help, because each archive still lands on disk before it +can be packed. + +The packed output, on the other hand, is small. This script therefore streams the +*packing*: extract one archive, pack it, delete the extracted copy, move to the +next. Peak disk stays at roughly one scene instead of the whole corpus, so the +frozen split becomes reachable on a machine that cannot hold the raw data. + +One scene per partition is the intended shape, not a workaround: calibration and +map state are scene-scoped, and `data_processing` raises "KITScenes partition +contains scenes with different calibration; pack one scene per partition" when a +partition mixes them. + +With `--fetch` it also downloads each archive and deletes it once packed, so the +archives never all exist at once either. That matters: streaming the packing alone +still needs the ~1.6 TB of compressed archives on disk, which is more than a laptop +has. Fetching one at a time brings peak usage down to a single archive plus a single +extraction — roughly 8 GB — for the whole 533-scene corpus. + +Resumable: a scene whose output already carries a `manifest.json` is skipped, so +the run can be interrupted and restarted. + +Usage: + export PYTHONPATH=/Model: + python Platform/scripts/pack_kitscenes_corpus.py \\ + --tar-src /data/KITScenes-Multimodal/data/train \\ + --work-root /data/_staging \\ + --out-root /data/_shards + +Validate the result before training with `verify_kitscenes_corpus.py`. +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +import sys +import time +import traceback +from pathlib import Path + +# Pinned by `data_processing`, which raises if the packed revision differs. +KITSCENES_SOURCE_REVISION = "6fde0034446669e2ed7235e4c7fe323cd23d599d" + +# `data_processing` defaults to DATASET_PACK_VERSION, but the KITScenes navigation +# path is versioned separately as KITSCENES_NAVIGATION_DATASET_VERSION, and it is +# that one the frozen validation manifest carries. Packing with the default and +# then training fails with "validation manifest dataset version does not match +# packed shards". +# +# Read from the manifest rather than pinned here: the contract is re-cut as the +# packed data changes, and a corpus packed against a stale version has to be +# repacked, not re-checked. +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def frozen_dataset_version() -> str: + """The packed dataset version the current frozen split expects.""" + from training.dataset_policy import KITSCENES_TRAINING_POLICY + + manifest = REPO_ROOT / "Model" / "training" / ( + KITSCENES_TRAINING_POLICY.validation_manifest + ) + return str(json.loads(manifest.read_text())["dataset_version"]) + +# Abort after this many CONSECUTIVE failures. One bad archive should not end a +# multi-hour run, but a systematic breakage should stop it immediately. +MAX_CONSECUTIVE_FAILURES = 3 + + +def free_gb(path: Path) -> float: + """Free space, in GiB, on the filesystem holding ``path``.""" + return shutil.disk_usage(path).free / 1024**3 + + +def extract_scene(tar_path: Path, dest_dir: Path) -> str: + """Extract one scene archive under ``dest_dir`` and return its scene ID. + + ``-C`` is passed BEFORE the archive: in GNU tar it is positional and applies + to the operands that follow it, so trailing it silently extracts into the + current working directory while still returning 0. + + tar's stderr is propagated rather than swallowed: a failure that returns a + non-zero code with no message is indistinguishable from a successful run that + produced nothing. + """ + dest_dir.mkdir(parents=True, exist_ok=True) + result = subprocess.run( + ["tar", "-xf", str(tar_path), "-C", str(dest_dir)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"tar exited {result.returncode}: {result.stderr.strip()[:300]}" + ) + scene_id = tar_path.stem + if not (dest_dir / scene_id).is_dir(): + produced = sorted(p.name for p in dest_dir.iterdir() if p.is_dir()) + raise RuntimeError( + f"archive did not produce {scene_id}/, found {produced[:5]}" + ) + return scene_id + + +def list_scene_archives(repo_id: str, revision: str, split: str = "train") -> list[str]: + """Repository paths of every scene archive in ``split``, sorted.""" + from huggingface_hub import HfApi + + info = HfApi().dataset_info(repo_id, revision=revision, files_metadata=False) + prefix = f"data/{split}/" + return sorted( + sibling.rfilename for sibling in info.siblings + if sibling.rfilename.startswith(prefix) + and sibling.rfilename.endswith(".tar") + ) + + +def fetch_archive(repo_id: str, revision: str, filename: str, dest_dir: Path) -> Path: + """Download one archive into ``dest_dir`` and return its path. + + ``local_dir`` keeps the file out of the shared HF cache, so deleting it after + packing actually reclaims the space instead of leaving a cached copy behind. + """ + from huggingface_hub import hf_hub_download + + path = hf_hub_download( + repo_id=repo_id, + filename=filename, + repo_type="dataset", + revision=revision, + local_dir=str(dest_dir), + ) + return Path(path) + + +def build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--tar-src", + help="Directory holding already-downloaded archives. Omit with " + "--fetch to download them one at a time instead") + ap.add_argument("--fetch", action="store_true", + help="Download each archive, pack it, then delete it. Keeps peak " + "disk at one archive plus one extraction") + ap.add_argument("--repo-id", default="KIT-MRT/KITScenes-Multimodal") + ap.add_argument("--source-revision", default=KITSCENES_SOURCE_REVISION, + help="Dataset revision to fetch; must match what packing pins") + ap.add_argument("--work-root", required=True, + help="Staging root; emptied after each scene") + ap.add_argument("--out-root", required=True, + help="Destination for the packed partitions") + ap.add_argument("--image-size", type=int, default=256) + ap.add_argument("--dataset-version", default=None, + help="Must match the frozen validation manifest " + "(default: read from it)") + ap.add_argument("--limit", type=int, default=0, + help="Pack only the first N archives (0 = all)") + ap.add_argument("--min-free-gb", type=float, default=30.0, + help="Stop when free space falls below this") + ap.add_argument("--keep-extracted", action="store_true", + help="Do not delete extracted scenes (disables streaming)") + return ap + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + # Validate the arguments BEFORE importing the pipeline: those imports are heavy + # and need PYTHONPATH set, so doing them first turns a plain "you forgot an + # argument" into an ImportError that points somewhere else entirely. + if not args.fetch and not args.tar_src: + print("error: pass --tar-src, or --fetch to download the archives", + file=sys.stderr) + return 1 + + from flytekit.types.directory import FlyteDirectory + + from Platform.pipelines.workflows import Dataset, data_processing + + if args.dataset_version is None: + args.dataset_version = frozen_dataset_version() + + if args.fetch: + scene_sources = list_scene_archives(args.repo_id, args.source_revision) + origin = f"{args.repo_id}@{args.source_revision[:7]}" + else: + tar_src = Path(args.tar_src) + if not tar_src.is_dir(): + print(f"error: {tar_src} is not a directory", file=sys.stderr) + return 1 + scene_sources = [str(path) for path in sorted(tar_src.glob("*.tar"))] + origin = str(tar_src) + if args.limit: + scene_sources = scene_sources[: args.limit] + if not scene_sources: + print(f"error: no .tar archives found in {origin}", file=sys.stderr) + return 1 + + work_root = Path(args.work_root) + out_root = Path(args.out_root) + # `data_processing` builds KitScenesDataset(data_root=raw, split="train"), + # so the staging tree must expose exactly that layout. + stage_scenes = work_root / "data" / "train" + stage_scenes.mkdir(parents=True, exist_ok=True) + out_root.mkdir(parents=True, exist_ok=True) + + print(f"# {len(scene_sources)} archive(s) from {origin}" + + (" (fetching one at a time)" if args.fetch else "")) + print(f"# staging: {stage_scenes} (emptied after each scene)") + print(f"# output : {out_root}") + print(f"# version: {args.dataset_version}") + print(f"# free : {free_gb(out_root):.0f} GiB\n") + + packed: list[str] = [] + failed: list[dict] = [] + consecutive = 0 + started = time.time() + + archive_dir = work_root / "archives" + for index, source in enumerate(scene_sources, 1): + scene_id = Path(source).stem + dest = out_root / scene_id + if (dest / "manifest.json").exists(): + packed.append(str(dest)) + print(f" [{index}/{len(scene_sources)}] {scene_id[:8]} already packed, skipping") + continue + + if free_gb(work_root) < args.min_free_gb: + print( + f"\naborting: {free_gb(out_root):.0f} GiB free, below the " + f"{args.min_free_gb:.0f} GiB floor. Free space and rerun; the " + "run resumes where it stopped.", + file=sys.stderr, + ) + break + + done = index - 1 + eta = "" + if done: + rate = (time.time() - started) / done + eta = f" ETA {(len(scene_sources) - done) * rate / 3600:.1f} h" + print(f" [{index}/{len(scene_sources)}] {scene_id[:8]} " + + ("fetching ..." if args.fetch else "extracting ..."), + end="", flush=True) + + staged = stage_scenes / scene_id + archive: Path | None = None + try: + if args.fetch: + archive = fetch_archive( + args.repo_id, args.source_revision, source, archive_dir + ) + print(" extracting ...", end="", flush=True) + else: + archive = Path(source) + extract_scene(archive, stage_scenes) + if args.fetch: + # Delete the archive as soon as it is unpacked: holding it until the + # end of the scene would double peak usage for no reason. + archive.unlink(missing_ok=True) + archive = None + print(" packing ...", end="", flush=True) + out = data_processing.task_function( + raw_data=FlyteDirectory(str(work_root)), + dataset=Dataset.KITSCENES, + source_revision=KITSCENES_SOURCE_REVISION, + dataset_version=args.dataset_version, + hz=10, + image_size=args.image_size, + episodes=1, + world_model=False, + group_ids=[scene_id], + ) + src = Path(str(getattr(out, "path", out))) + if dest.exists(): + shutil.rmtree(dest) + shutil.copytree(src, dest) + samples = json.loads( + (dest / "manifest.json").read_text() + ).get("total_samples", "?") + packed.append(str(dest)) + consecutive = 0 + print(f" ok ({samples} samples){eta}") + except Exception as error: # noqa: BLE001 - reported, then counted + consecutive += 1 + failed.append( + {"scene": scene_id, "error": f"{type(error).__name__}: {error}"[:400]} + ) + print(f" FAILED {type(error).__name__}: {str(error)[:120]}") + traceback.print_exc(limit=3) + if consecutive >= MAX_CONSECUTIVE_FAILURES: + print( + f"\naborting: {consecutive} consecutive failures, which " + "suggests a systematic problem rather than a bad archive.", + file=sys.stderr, + ) + break + finally: + # These deletes are what keep peak disk at one scene. + if not args.keep_extracted and staged.exists(): + shutil.rmtree(staged, ignore_errors=True) + if args.fetch and archive is not None: + archive.unlink(missing_ok=True) + + index_path = out_root / "shards_index.json" + index_path.write_text(json.dumps({"packed": packed, "failed": failed}, indent=2)) + print(f"\n# packed {len(packed)} · failed {len(failed)}") + print(f"# index -> {index_path}") + for item in failed[:5]: + print(f"# {item['scene'][:8]} {item['error'][:110]}") + if args.fetch: + shutil.rmtree(archive_dir, ignore_errors=True) + print("\n# Validate before training:") + print(f"# python Platform/scripts/verify_kitscenes_corpus.py " + f"--shards-root {out_root}") + return 0 if packed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/Platform/scripts/verify_kitscenes_corpus.py b/Platform/scripts/verify_kitscenes_corpus.py new file mode 100644 index 000000000..a6ea802e7 --- /dev/null +++ b/Platform/scripts/verify_kitscenes_corpus.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +"""Check a packed KITScenes corpus against the frozen split, before training. + +`train_il` already validates the packed corpus against the frozen split named by +`KITSCENES_TRAINING_POLICY.validation_manifest` and aborts on any mismatch, but it +does so after start-up, once the shards have been scanned. On a corpus that took +hours to pack that is a late and expensive place to find out. + +The manifest is read from the policy rather than named here, because the frozen +contract is re-cut as the packed data changes: a checker pinned to one snapshot +would keep passing a corpus that training has already started rejecting. + +This runs the same comparison beforehand and reports every check, so a corpus +can be fixed before a run rather than during one. It reads tar headers and +`manifest.json` members only; no camera payload is decoded. + +Checks performed, all against the frozen manifest: + + packed partitions == available_scene_count + empty partitions == excluded_empty_scene_count + eligible groups == eligible_group_count + group UID digest == eligible_group_uid_digest + samples == eligible_sample_count + sample UID digest == eligible_sample_uid_digest + packed dataset version == dataset_version + packed contract digest == packed_contract_digest + +When they all pass, `--validation_scope full` will select the frozen holdout, so +the resulting ADE/FDE can be compared with anyone else's run over the same split. + +Usage: + export PYTHONPATH=/Model: + python Platform/scripts/verify_kitscenes_corpus.py --shards-root /data/_shards + +Exit codes: 0 all checks pass · 1 corpus does not match · 2 nothing to check. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any, NamedTuple + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def frozen_manifest_path() -> Path: + """Locate the split manifest ``train_il`` will validate this corpus against. + + Resolved through the training policy, so the checker and the trainer cannot + disagree about which snapshot is current. + """ + from training.dataset_policy import KITSCENES_TRAINING_POLICY + + return REPO_ROOT / "Model" / "training" / ( + KITSCENES_TRAINING_POLICY.validation_manifest + ) + + +class Check(NamedTuple): + """One comparison between the packed corpus and the frozen manifest.""" + + name: str + actual: Any + expected: Any + + @property + def ok(self) -> bool: + return self.actual == self.expected + + +def read_partitions(root: Path) -> tuple[dict[str, dict], list[str], int, set[str]]: + """Scan partition directories exactly as ``train_il`` does at start-up. + + Returns the manifests keyed by directory, the non-empty directories, the + number of empty ones, and the set of packed dataset versions. + """ + manifests: dict[str, dict] = {} + non_empty: list[str] = [] + empty = 0 + versions: set[str] = set() + for path in sorted(root.iterdir()): + manifest_path = path / "manifest.json" + if not (path.is_dir() and manifest_path.exists()): + continue + manifest = json.loads(manifest_path.read_text()) + manifests[str(path)] = manifest + version = manifest.get("dataset_version") + if version: + versions.add(str(version)) + if int(manifest.get("total_samples", 0)) <= 0: + empty += 1 + else: + non_empty.append(str(path)) + return manifests, non_empty, empty, versions + + +def build_checks( + frozen: dict, + *, + partition_count: int, + empty_count: int, + group_uids: tuple[str, ...], + group_digest: str, + sample_count: int, + sample_digest: str, + dataset_version: str, + contract_digest: str, +) -> list[Check]: + """Pair every packed quantity with what the frozen manifest requires.""" + return [ + Check("packed partitions", partition_count, + frozen["available_scene_count"]), + Check("empty partitions", empty_count, + frozen["excluded_empty_scene_count"]), + Check("eligible groups", len(group_uids), + frozen["eligible_group_count"]), + Check("group UID digest", group_digest, + frozen["eligible_group_uid_digest"]), + Check("samples", sample_count, frozen["eligible_sample_count"]), + Check("sample UID digest", sample_digest, + frozen["eligible_sample_uid_digest"]), + Check("dataset version", dataset_version, frozen["dataset_version"]), + Check("contract digest", contract_digest, + frozen["packed_contract_digest"]), + ] + + +def format_check(check: Check) -> str: + actual, expected = str(check.actual), str(check.expected) + if len(actual) > 20: + actual, expected = actual[:16] + "...", expected[:16] + "..." + mark = "PASS" if check.ok else "FAIL" + return f" [{mark}] {check.name:<24} {actual:>20} expected {expected}" + + +def build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--shards-root", required=True, + help="Directory holding the packed partitions") + ap.add_argument("--manifest", default=None, + help="Frozen split manifest to compare against " + "(default: the one the training policy names)") + return ap + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + from data_parsing.pre_extracted import discover_split_inventory + from training.dataset_policy import group_uid_digest + + from Platform.pipelines.training_checkpoint import stable_digest + + root = Path(args.shards_root) + if not root.is_dir(): + print(f"error: {root} is not a directory") + return 2 + manifest_path = ( + Path(args.manifest) if args.manifest else frozen_manifest_path() + ) + frozen = json.loads(manifest_path.read_text()) + + manifests, non_empty, empty, versions = read_partitions(root) + if not manifests: + print(f"error: no partitions with a manifest.json under {root}") + return 2 + + print(f"corpus : {root}") + print(f" partitions with a manifest : {len(manifests)}") + print(f" empty : {empty}") + print(f" non-empty : {len(non_empty)}") + print(f" dataset version : {sorted(versions) or ['(none)']}") + if not non_empty: + print("error: every partition is empty") + return 2 + + print("\nscanning shard headers (meta.json members only) ...", flush=True) + try: + inventory = discover_split_inventory(non_empty) + except ValueError as error: + # Expected while a corpus is still being packed, e.g. "requires metadata + # for at least two split groups". Not a failure of the corpus itself. + missing = frozen["available_scene_count"] - len(manifests) + print(f"\n not analysable yet: {error}") + print(f" {missing} of {frozen['available_scene_count']} partitions still " + "missing; keep packing and run this again.") + return 1 + + contract_digests = { + stable_digest(manifest.get("contracts")) for manifest in manifests.values() + } + checks = build_checks( + frozen, + partition_count=len(manifests), + empty_count=empty, + group_uids=inventory.group_uids, + group_digest=group_uid_digest(inventory.group_uids), + sample_count=inventory.sample_count, + sample_digest=inventory.sample_uid_digest, + dataset_version=next(iter(versions), ""), + contract_digest=( + next(iter(contract_digests)) if len(contract_digests) == 1 + else f"AMBIGUOUS({len(contract_digests)})" + ), + ) + + print(f"\n== against {manifest_path.name} ==") + for check in checks: + print(format_check(check)) + + failures = [check for check in checks if not check.ok] + if failures: + print(f"\n== not comparable yet: {len(failures)} check(s) failed.") + missing = frozen["available_scene_count"] - len(manifests) + if missing > 0: + print(f" {missing} partition(s) still to pack.") + print(" Until they all pass, --validation_scope full will abort and the " + "resulting metrics are not comparable with the frozen split.") + return 1 + + print("\n== corpus matches the frozen split.") + print(f" Holdout: {frozen['validation_group_count']} scenes / " + f"{frozen['validation_sample_count']} samples, group digest " + f"{frozen['validation_group_uid_digest'][:16]}...") + index_path = root / "shards_index_verified.json" + index_path.write_text(json.dumps( + {"packed": sorted(manifests)}, indent=2)) + print(f" Shard list -> {index_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/Platform/tests/conftest.py b/Platform/tests/conftest.py new file mode 100644 index 000000000..7b99d28da --- /dev/null +++ b/Platform/tests/conftest.py @@ -0,0 +1,22 @@ +"""Put ``Model/`` on the import path for tests under ``Platform/tests``. + +The pipelines import model packages by their top-level name — ``training.*``, +``evaluation.*`` — which resolves because ``Model/`` is on ``PYTHONPATH`` at run +time and, for ``Model/tests``, because ``Model/pytest.ini`` and +``Model/tests/__init__.py`` make pytest insert ``Model/`` itself. + +Neither applies here, so a test that reaches into the model packages fails with +``ModuleNotFoundError: No module named 'training'`` unless the caller happened to +export ``PYTHONPATH`` first. Doing it here keeps these tests runnable with a bare +``pytest Platform/tests`` from the repository root. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +MODEL_ROOT = Path(__file__).resolve().parents[2] / "Model" + +if str(MODEL_ROOT) not in sys.path: + sys.path.insert(0, str(MODEL_ROOT)) diff --git a/Platform/tests/test_kitscenes_corpus_tools.py b/Platform/tests/test_kitscenes_corpus_tools.py new file mode 100644 index 000000000..b03b61578 --- /dev/null +++ b/Platform/tests/test_kitscenes_corpus_tools.py @@ -0,0 +1,221 @@ +import io +import json +import subprocess +import tarfile +from pathlib import Path + +import pytest + +from Platform.scripts.pack_kitscenes_corpus import extract_scene +from Platform.scripts.verify_kitscenes_corpus import ( + build_checks, + read_partitions, +) + + +FROZEN = { + "available_scene_count": 3, + "excluded_empty_scene_count": 1, + "eligible_group_count": 2, + "eligible_group_uid_digest": "a" * 64, + "eligible_sample_count": 5, + "eligible_sample_uid_digest": "b" * 64, + "dataset_version": "v3.0", + "packed_contract_digest": "c" * 64, +} + + +def _partition(root, name, samples, *, version="v3.0"): + path = root / name + path.mkdir() + (path / "manifest.json").write_text(json.dumps({ + "total_samples": samples, + "dataset_version": version, + "partition_id": name, + "contracts": {"schema": "test"}, + })) + if samples: + with tarfile.open(path / "shard-000.tar", "w") as archive: + for index in range(samples): + uid = f"kitscenes-{name}-{index}" + payload = json.dumps({ + "split_group_uid": f"kitscenes-group-{name}", + "sample_uid": uid, + }).encode() + info = tarfile.TarInfo(f"{uid}.meta.json") + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + return path + + +class TestExtractScene: + def test_extracts_into_the_destination_not_the_cwd(self, tmp_path, monkeypatch): + """`-C` is positional in GNU tar. + + Passing it after the archive returns 0 and extracts into the current + working directory instead, which looks like success while leaving the + destination empty. + """ + source = tmp_path / "src" + (source / "abc12345").mkdir(parents=True) + (source / "abc12345" / "poses.txt").write_text("x") + archive = tmp_path / "abc12345.tar" + subprocess.run( + ["tar", "-cf", str(archive), "-C", str(source), "abc12345"], + check=True, + ) + + elsewhere = tmp_path / "cwd" + elsewhere.mkdir() + monkeypatch.chdir(elsewhere) + + dest = tmp_path / "dest" + scene_id = extract_scene(archive, dest) + + assert scene_id == "abc12345" + assert (dest / "abc12345" / "poses.txt").exists() + assert not (elsewhere / "abc12345").exists() + + def test_missing_archive_raises_instead_of_reporting_success(self, tmp_path): + """Swallowing tar's stderr makes a failed extraction look like an empty one.""" + with pytest.raises(RuntimeError, match="tar exited"): + extract_scene(tmp_path / "absent.tar", tmp_path / "dest") + + +class TestReadPartitions: + def test_separates_empty_partitions_and_collects_versions(self, tmp_path): + _partition(tmp_path, "scene-a", 3) + _partition(tmp_path, "scene-b", 2) + _partition(tmp_path, "scene-c", 0) + (tmp_path / "not-a-partition").mkdir() + + manifests, non_empty, empty, versions = read_partitions(tmp_path) + + assert len(manifests) == 3 + assert len(non_empty) == 2 + assert empty == 1 + assert versions == {"v3.0"} + + def test_mixed_versions_are_all_reported(self, tmp_path): + _partition(tmp_path, "scene-a", 1) + _partition(tmp_path, "scene-b", 1, version="v2.2") + _, _, _, versions = read_partitions(tmp_path) + assert versions == {"v3.0", "v2.2"} + + +class TestBuildChecks: + def _checks(self, **overrides): + values = { + "partition_count": 3, + "empty_count": 1, + "group_uids": ("g1", "g2"), + "group_digest": "a" * 64, + "sample_count": 5, + "sample_digest": "b" * 64, + "dataset_version": "v3.0", + "contract_digest": "c" * 64, + } + values.update(overrides) + return build_checks(FROZEN, **values) + + def test_a_matching_corpus_passes_every_check(self): + assert all(check.ok for check in self._checks()) + + @pytest.mark.parametrize("field,value,expected_name", [ + ("partition_count", 2, "packed partitions"), + ("empty_count", 0, "empty partitions"), + ("group_uids", ("g1",), "eligible groups"), + ("group_digest", "z" * 64, "group UID digest"), + ("sample_count", 4, "samples"), + ("sample_digest", "z" * 64, "sample UID digest"), + ("dataset_version", "v2.2", "dataset version"), + ("contract_digest", "z" * 64, "contract digest"), + ]) + def test_each_deviation_fails_its_own_check(self, field, value, expected_name): + """Every quantity the manifest pins must be caught on its own. + + A corpus that differs in one field only is the case this tool exists for: + it is the one that otherwise reaches `train_il` and aborts hours later. + """ + failed = [check.name for check in self._checks(**{field: value}) + if not check.ok] + assert failed == [expected_name] + + +class TestFetchMode: + def test_requires_a_source(self, capsys): + """Neither --tar-src nor --fetch must fail with a usable message.""" + from Platform.scripts.pack_kitscenes_corpus import main + + rc = main(["--work-root", "/tmp/w", "--out-root", "/tmp/o"]) + assert rc == 1 + assert "--fetch" in capsys.readouterr().err + + def test_lists_only_train_archives_sorted(self, monkeypatch): + """The scene order must be deterministic, and confined to the split. + + A run that packs scenes in a different order than another run produces the + same corpus, but only if the set is identical -- picking up `data/val/` or a + stray file would silently change what 'the train split' means. + """ + import Platform.scripts.pack_kitscenes_corpus as mod + + class _Sibling: + def __init__(self, name): + self.rfilename = name + + class _Api: + def dataset_info(self, repo_id, revision=None, files_metadata=False): + names = [ + "data/train/b.tar", "data/train/a.tar", + "data/val/z.tar", "data/train/notes.txt", "README.md", + ] + return type("I", (), {"siblings": [_Sibling(n) for n in names]})() + + monkeypatch.setattr(mod, "HfApi", _Api, raising=False) + monkeypatch.setitem( + __import__("sys").modules, "huggingface_hub", + type("M", (), {"HfApi": _Api})(), + ) + assert mod.list_scene_archives("r", "rev") == [ + "data/train/a.tar", "data/train/b.tar", + ] + + +class TestTheFrozenContractIsReadNotRemembered: + """Both tools must follow the policy's manifest, not a snapshot of it. + + The frozen split is re-cut as the packed data changes, and every re-cut + carries a new dataset version and contract digest. A tool that names one + snapshot keeps passing a corpus that `train_il` has started rejecting, which + is the exact failure these tools exist to catch early. + """ + + def test_the_checker_reads_the_manifest_the_policy_names(self): + from training.dataset_policy import KITSCENES_TRAINING_POLICY + + from Platform.scripts.verify_kitscenes_corpus import frozen_manifest_path + + path = frozen_manifest_path() + assert path.is_file() + assert path.name == Path( + KITSCENES_TRAINING_POLICY.validation_manifest + ).name + + def test_the_packer_writes_the_version_that_manifest_carries(self): + from Platform.scripts.pack_kitscenes_corpus import frozen_dataset_version + from Platform.scripts.verify_kitscenes_corpus import frozen_manifest_path + + frozen = json.loads(frozen_manifest_path().read_text()) + assert frozen_dataset_version() == frozen["dataset_version"] + + def test_that_version_is_the_navigation_one_not_the_default(self): + """The trap the comment in the packer describes, asserted.""" + from Platform.pipelines.workflows import ( + DATASET_PACK_VERSION, + KITSCENES_NAVIGATION_DATASET_VERSION, + ) + from Platform.scripts.pack_kitscenes_corpus import frozen_dataset_version + + assert frozen_dataset_version() == KITSCENES_NAVIGATION_DATASET_VERSION + assert frozen_dataset_version() != DATASET_PACK_VERSION diff --git a/README.md b/README.md index 02a3ee132..9c25bb5c6 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,7 @@ Review our academic paper, access our knowledge base and read through our work o ### Next steps - Explore the [Model](./Model/) folder for the model components, training and inference. - Follow the [Trial Guide](./TRIAL.md) to run the inference test on AWS EC2. +- Read [Training on a local machine](./Docs/training_on_a_local_machine.md) to run KITScenes training outside the cluster. ## Architecture at a glance