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