diff --git a/.claude/skills/datapipe-examples/SKILL.md b/.claude/skills/datapipe-examples/SKILL.md new file mode 100644 index 00000000..7e43b7c7 --- /dev/null +++ b/.claude/skills/datapipe-examples/SKILL.md @@ -0,0 +1,65 @@ +--- +name: datapipe-examples +description: > + Use when working in the epoch8 datapipe repo with the examples/* folders, when asked to run or + debug any datapipe example pipeline on your own data, or when unsure which example or setup skill + applies. +--- + +# datapipe examples — router + +Every example here is meant to run on YOUR data; each bundled demo dataset is only a smoke-test. Pick the example, then its setup skill, then set that skill's "Run on YOUR data" knobs first. + +Each `examples/*` pipeline has a dedicated setup skill — pick by what it does: + +| Example | What it does | Skill | +|---|---|---| +| `embedder_fiftyone` | DINOv2/DINOv3 image embeddings → FiftyOne UMAP + similarity | **setup-embedder-fiftyone** | +| `e2e_template/image_detection` | YOLO detection + Label Studio human-in-the-loop → train → FiftyOne | **setup-e2e-template** | +| `e2e_template/image_keypoints` | YOLO-pose keypoints + Label Studio → train → FiftyOne | **setup-e2e-template** | +| `sam_cvat` | SAM3 text-prompt boxes+masks → CVAT pre-annotations | **setup-sam-cvat** | +| `detection_tags` | YOLO detection + **tags** (per-scenario metrics), no Label Studio / FiftyOne, GT injected | **setup-detection-tags** | + +## Ask first — don't assume (only the unresolved) +Clarify what's not obvious before acting — don't spin up services or pull data you don't need. Ask only the unresolved: +- **Demo or your own data?** · **Provision Postgres/services or reuse existing?** (e2e ships `docker compose`; embedder/sam need external Postgres) +- **Which Postgres + which database** for `DB_URL`? Never point it at an existing DB or drop in a `localhost` default without confirming. +- **Reuse an existing venv / `uv` env, or create a fresh one?** · **Which GPU?** (SAM3 >8 GB; DINOv2/YOLO ~8 GB) · **Annotation backend up?** (LS for e2e / external CVAT for sam) +- **Surface stage logs or run quiet?** · **Per-tag scenario metrics** (retrain new case, old vs new)? → the `detection_tags` example / **setup-detection-tags** + +## How to work +Read the setup, then propose a short plan and get a go-ahead before touching anything. Prepare `.env` and **pause for the user to verify it** before running. Run each stage with its logs shown and, after each, say what you did and what changed — don't run the pipeline silently. Trust the status table (`*_training_status`/`brain_status`), not the exit code. If a stage fails and the cause isn't clear from the normal logs, re-run it with `datapipe --debug … run` (or `--debug-sql` for SQL errors) sent to a file and `grep`ped, rather than dumping the verbose output inline. + +## Run on YOUR data (universal rule) +When you swap in your own subject, **align your class name across every place it appears** — text +prompt / label config, the label field, and any class filter. A mismatch runs fine but yields 0 +useful results. The per-example skill lists the exact knobs. + +## Universal prerequisites (every example) +1. **PostgreSQL** at `DB_URL`. embedder + sam_cvat need an EXTERNAL Postgres (they don't start one); + e2e_template bundles Postgres in its `docker compose`. Empty DB is fine; tables auto-create via + `datapipe db create-all`. `.env.example` default `...postgres:postgres@localhost:5432/postgres`. + External quick start: `docker run -d -e POSTGRES_PASSWORD=postgres -e POSTGRES_USER=postgres -e POSTGRES_DB=postgres -p 5432:5432 postgres:16` + (k8s pod / no docker → conda postgres, `initdb` as non-root with `--locale=C`.) +2. **A GPU big enough** (`nvidia-smi`): DINOv2/YOLO fit ~8 GB; SAM3 wants >8 GB (OOM'd on an 8 GB Pascal in our tests). +3. **Annotation backend, if used:** e2e_template ships Label Studio in its `docker compose`; sam_cvat + needs an external CVAT you provide (URL + creds + a project whose labels match its config). +4. **`uv` + Python ≥3.10,<3.13.** Each example has a `pyproject.toml` → `uv sync` (cu124 torch pinned, + CUDA OOTB; no manual venv/pip). Read-only/small `$HOME`: `export UV_CACHE_DIR=/tmp/uvcache HF_HOME=/tmp/hf`. +5. **Human-in-the-loop** (e2e_template, sam_cvat): a human reviews and marks tasks completed before + the pipeline advances. + +## HF auth (only for GATED models) +Needed for `dinov3_*` (embedder, off by default) and `sam3` (sam_cvat). Accept the model license on +its HF page, then `export HF_TOKEN=...` with **gated-repo read access**. A plain token → `403`, often +masked as `LocalEntryNotFoundError: check your connection`. Public models (dinov2, YOLO) need no token. + +## Universal run shape +```bash +cp .env.example .env # set DB_URL and the example's data + backend vars +uv sync +datapipe db create-all # auto-creates all tables +datapipe run # or: datapipe step --labels stage= run +``` +Datapipe caches by content: re-running does little work unless inputs or tracked config change. +Then follow the per-example skill — set its **Run on YOUR data** knobs first. diff --git a/.claude/skills/setup-detection-tags/SKILL.md b/.claude/skills/setup-detection-tags/SKILL.md new file mode 100644 index 00000000..088b899b --- /dev/null +++ b/.claude/skills/setup-detection-tags/SKILL.md @@ -0,0 +1,274 @@ +--- +name: setup-detection-tags +description: > + Use when setting up or running examples/detection_tags — a self-contained datapipe detection + demo built around tags (per-scenario metrics), split into two parts around a checkpoint (train a + baseline, then add a tagged TRAIN batch and retrain, watching the tag metric rise), with a + FiftyOne view and injected ground truth (no Label Studio). Also for "add a tagged batch, retrain, + watch the tag metric rise" and for rehearsing that retraining demo from a saved checkpoint. +--- + +# detection_tags (tags demo — two-part, FiftyOne, no Label Studio) + +Detection pipeline whose whole point is **tags**: train a baseline (model A), then add a **tagged +scenario TRAIN batch** and retrain (model B), and watch recall on that tag rise in a `tag_metrics` +table. Ground truth is **injected** (COCO labels, lowercase `cat`/`dog`) — no human annotation. + +The two-part / checkpoint split below is a **presentation device for the live demo only** — it lets +you prep part 1 ahead of time, present part 2, and rehearse it. It is NOT required to use the pipeline: +for real work you just run the stages end to end (the frozen-val data layout still matters — that's +about metric correctness, not the demo). See "Real data" at the end. + +The demo is deliberately **two parts around a checkpoint**, so you can prepare part 1 ahead of time +and present/rehearse part 2 (the retraining) as often as you like: + +- **Part 1 (prep, before the audience):** deploy → load `base-train` + `base-val` + `night-val` + (val is frozen up front) → train **model A** → compute all metrics → **take a checkpoint** → build + the FiftyOne view. Stop here. +- **Part 2 (the live demo):** show model A's metrics + FiftyOne → *ask "ready to retrain?"* → load + the tagged TRAIN batch `night-train` → retrain **model B** → show the metrics again: `night` recall + rises from A to B. Rehearse by resetting to the checkpoint. + +## Ask first — don't assume (only the unresolved) + +1. **Demo (test) or real data?** — the whole flow branches on this (see "Real data" at the end). +2. **Is an env/compose already up?** Check before deploying: `docker compose ls` / + `docker ps` on the target host, and whether a `.venv` already has torch+fiftyone. **Reuse it** + (skip `uv sync` / `compose up` / `db create-all` as appropriate) rather than redeploying — only + deploy the pieces that are missing. Don't drop a `localhost` DB default or target an existing DB + without confirming. +3. **GPU available?** Training is GPU. On a remote cluster (e.g. epoch8 gpu5) run the whole stack + there over SSH; note the host may have a **read-only home** and **no AVX2** (see Troubleshooting). +4. **Surface stage logs or run quiet?** Default: show each stage and report what changed. + +## Frozen val — why the data is loaded in this order (READ THIS) + +A metric on `val` is an **aggregate over the set of images currently in val**. Change that set and +the number moves *even if the model is byte-identical*. If you load the tagged batch as one blob and +let the random split scatter it across train/val at retrain time, model A's val (and tag/val) numbers +shift between the two measurements — an apples-to-oranges comparison that confuses everyone. + +Fix: **freeze val up front.** Load `base-val` and `night-val` (subset pinned to `val`) *before* +training model A, and only add `night-train` (pinned to `train`) later. Then val never changes; model +A is measured once on the full frozen val and its numbers are stable, and B is compared on the exact +same val. `add_request.py --subset train|val` pins a batch; the load step emits `image__subset_hint` +and the split step honors it (random split only fills images without a hint). + +Batches (COCO cat/dog; the pre-staged cache holds **500 images**, so keep the total ≤ 500 or expand +the cache — see Troubleshooting): + +| batch | n | offset | subset | tag | darken | when | +|-------|---|--------|--------|-----|--------|------| +| `base-train` | 325 | 0 | train | — | — | part 1 | +| `base-val` | 100 | 325 | val | — | — | part 1 | +| `night-val` | 25 | 425 | val | night | 0.25 | part 1 | +| `night-train`| 50 | 450 | train | night | 0.25 | part 2 | + +Validated on gpu5: model A → train recall ~0.95, night/val recall ~0.09 (blind in the dark); after +part 2, model B → night/train recall ~0.86. Keep night/val small but not tiny (25 imgs / ~45 GT) — +below ~20 the val payoff is pure ±1-box noise and can even go the wrong way. + +## Pre-flight — CHECK what's already there, then ASK (do this before deploying) + +Never assume a clean host. Before `docker compose up` / loading / training, check each resource and, +if it's occupied, ASK the user what to do (reuse / different port / different schema or DB / wipe) — +don't silently claim it or clobber someone else's work. + +```bash +# 1) ports free? (host) — postgres 5432, minio 9000/9001, mongo 27017, fiftyone 5151, app 8000 +ss -ltn | grep -E ':(5432|9000|9001|27017|5151|8000)\b' || echo "all free" +# 2) is a stack already up? +docker compose ls ; docker ps --format '{{.Names}}\t{{.Ports}}' +# 3) does the target DB/schema already hold this pipeline's tables? +psql "$DB_URL" -c "\dt $DB_SCHEMA.*" 2>/dev/null | grep -qi detection && echo "SCHEMA IN USE" || echo "schema clean" +# 4) is a venv already built (torch+fiftyone)? reuse vs fresh uv sync +``` + +Findings → questions to ask: a **busy port** means another stack (reuse it, or remap ports?); a +**populated schema** means a prior run (resume it, wipe it, or use a different `DB_SCHEMA`/DB?); a +**running compose** (reuse or `down`?). The same check applies on your laptop when a **local tunnel +port is busy** — pick a different left-hand port (`-L 5433:localhost:5432`), don't kill blindly. + +## Deploy from scratch (standard ports; skip pieces already up) + +```bash +cp .env.example .env && set -a && source .env && set +a # DB_URL, S3/MinIO, FIFTYONE_DATABASE_URI +docker compose up -d # postgres + minio + mongo (mongo is for FiftyOne). No Label Studio. +uv sync # cu124 torch + datapipe-ml[torch,fiftyone] + fiftyone + pi-heif +# On a pre-AVX2 host, force the lts polars to win (else the training subprocess SIGILLs): +uv pip uninstall polars polars-lts-cpu && uv pip install polars-lts-cpu==1.33.1 +cd detection +# DB_SCHEMA defaults to `public` (already exists). Only if you set a dedicated schema (to share the +# Postgres with other pipelines) create it first: psql "$DB_URL" -c "CREATE SCHEMA IF NOT EXISTS $DB_SCHEMA" +datapipe db create-all +``` + +## Part 1 — baseline to checkpoint + +Run `stage=train` as ONE step (it internally does split→freeze→train→inference); surface each stage +from its log and by querying the tables after — do NOT invoke `stage=train-prepare` separately before +`stage=train`, or the intermediate re-freeze retrains the model twice (see Troubleshooting). + +```bash +# from examples/detection_tags/detection, with .env sourced +python ../scripts/add_request.py --id base-train --n 325 --offset 0 --subset train +python ../scripts/add_request.py --id base-val --n 100 --offset 325 --subset val +python ../scripts/add_request.py --id night-val --n 25 --offset 425 --subset val --tag night --darken 0.25 +datapipe step --labels=stage=load run # 450 images; val frozen (100 base + 25 night) +# show the frozen split: SELECT subset_id, count(*) FROM image__subset_hint GROUP BY subset_id + +datapipe step --labels=stage=train run # split -> freeze -> train model A -> inference (SHOW every epoch) +datapipe step --labels=stage=count-metrics run # metrics_on_image/subset + tag_metrics +# count-metrics can print "Batches to process 0" right after training (it hasn't seen the fresh +# predictions yet) — RE-RUN it once; then the tables fill. (See Troubleshooting.) + +# SHOW THE FULL TABLES (see the Rule below). To rehearse part 2 later, snapshot the post-A state +# BEFORE the fiftyone stage (demo-only convenience; skip for real use): +docker exec pg_dump -U postgres -n "$DB_SCHEMA" postgres > /tmp/checkpoint.sql + +datapipe step --labels=stage=fiftyone run # download_images -> GT + model-A predictions +``` + +**Stop here (end of stage 1) and hand the baseline over to the user:** +- show the full training log (all epochs) and the **full** metric tables (`metrics_on_subset` + + `tag_metrics`) — `tag_metrics` for model A at `night/val` is **low** (baseline blind in the dark); +- bring up the **FiftyOne App** service so they can browse GT vs model-A predictions, and give them + its address / how to reach it (see "Let the user watch" — specifics depend on the setup); +- then ask whether to proceed to part 2 (retrain). That's the problem part 2 fixes. + +Surface every epoch from the training log (it streams to a file, e.g. `/tmp/train_A.log`): +```bash +grep -oE "[0-9]+/[0-9]+ +[0-9.]+G +[0-9.]+ +[0-9.]+ +[0-9.]+" /tmp/train_A.log # per-epoch box/cls/dfl loss +grep -E "^ +all +[0-9]+ +[0-9]+" /tmp/train_A.log # per-epoch val P/R/mAP50/mAP50-95 +``` + +## Part 2 — retrain and watch the tag metric rise + +```bash +python ../scripts/add_request.py --id night-train --n 50 --offset 450 --subset train --tag night --darken 0.25 +datapipe step --labels=stage=load run # 500 images total +datapipe step --labels=stage=train run # split (night -> TRAIN, val UNCHANGED) + train model B (SHOW every epoch) +# verify val stayed frozen after the split: +# SELECT subset_id, count(*) FILTER (WHERE image_name LIKE '%night%') night, count(*) total +# FROM image__subset s JOIN image__ground_truth USING(image_name) GROUP BY subset_id +datapipe step --labels=stage=count-metrics run # re-run if "0 batches" +datapipe step --labels=stage=fiftyone run # adds predictions_model_b +``` + +`night/val` recall rises from model A to model B — the payoff. Show the full tables again. + +## Rehearse part 2 from the snapshot (demo-only) + +Restore the post-A snapshot and wipe the FiftyOne db, then re-run Part 2 — no retraining of model A, +no image re-download: + +```bash +docker exec psql -U postgres -c "DROP SCHEMA IF EXISTS $DB_SCHEMA CASCADE; CREATE SCHEMA $DB_SCHEMA" +docker exec -i psql -U postgres < /tmp/checkpoint.sql +docker exec mongosh --quiet --eval "db.getSiblingDB('fiftyone').dropDatabase()" +``` + +## RULE: always print the FULL metrics table + +When you show metrics, dump the **whole** table, not a truncated view. Both of these, every time: + +```bash +docker exec psql -U postgres -x -c \ + "SELECT * FROM $DB_SCHEMA.detection_model_train__metrics_on_subset ORDER BY detection_model_id, subset_id" +docker exec psql -U postgres -x -c \ + "SELECT tm.*, t.tag_name FROM $DB_SCHEMA.tag_metrics tm JOIN $DB_SCHEMA.tag t USING(tag_id) \ + ORDER BY tm.detection_model_id, tm.tag_id, tm.subset_id" +``` + +`tag_id` is a numeric surrogate key (deterministic map in `config.TAG_IDS`, e.g. `night`→1); the +`tag` dimension holds `tag_name`/`tag_description`, so join it to show the readable name. + +The payoff comparison: `tag_metrics` at `tag=night, subset=val`, model A vs model B — recall rises. +The night/val set is small (25 images), so treat the rise as **directional**. + +## Let the user watch (offer it — work out the specifics per setup) + +The user can watch three things; make them available and hand over whatever access they need — don't +prescribe fixed commands here and don't leave a committed file behind (it would bake in one machine's +host/ports/paths and mislead on the next). Compute the specifics for the ACTUAL setup at runtime +(local vs remote host, which ports are free, tunnels if remote) and give the user the commands +directly in chat: + +- **tables** in DBeaver (Postgres → the `$DB_SCHEMA` schema): `tag_metrics`, + `detection_model_train__metrics_on_subset`, `detection_training_status`. +- **images** in the FiftyOne App: `ground_truth` vs `predictions_model_a` vs `predictions_model_b`, + filterable by `tag`/`subset`. +- **training progress**: the run streams to a log file — tail it; trust + `detection_training_status.status`, not exit codes. + +**When you START a training run, immediately tell the user how to follow it** (before it finishes), +so they can watch the epochs live — give the log path and the follow command, e.g. +`tail -f /tmp/train_A.log` (or the per-epoch `grep` from Part 1). Do this the moment training kicks +off, for model A and again for model B — not only after it completes. + +When the host is remote these are reached over SSH tunnels; if a local tunnel port is busy, pick a +different left-hand port rather than killing whatever holds it. + +## FiftyOne + +`stage=fiftyone` downloads images locally then publishes to one dataset (`FIFTYONE_DATASET_NAME`, +metadata in MongoDB): fields `ground_truth`, `predictions_model_a` (baseline), `predictions_model_b` +(retrained), plus sample fields `tag` and `subset`. Ported straight from +`examples/e2e_template/image_detection` (same `download_images` + `publish_to_fiftyone` + +`FiftyOneImagesDataTableStore`); the two model fields are assigned by sorted model-id (earliest = +`model_a`). Launch: `fiftyone app launch "$FIFTYONE_DATASET_NAME" --port 5151 --address 0.0.0.0`, +tunnel 5151 (local port MUST equal remote port — the App's WebSocket calls the same port). + +`tag` and `subset` are ordinary **sample fields** (set in-pipeline by `publish_gt_to_fiftyone`, no +scripts) — NOT FiftyOne's native "sample tags". Filter by the **fields** in the sidebar: pick +`tag = night` and `subset = val` — different fields AND together, so you get exactly the night-val +set. (Selecting multiple values inside FiftyOne's native TAGS panel ORs them, which is why we do NOT +mirror tag/subset into native tags.) That isolates where model A misses in the dark; after part 2, +`predictions_model_b` shows model B catching it. Fields stay `predictions_model_a` (baseline, +earliest model id) / `predictions_model_b` (retrained); the exact model ids are in the DB tables. + +## How to work + +Propose a short plan and get a go-ahead; show each stage's logs and report what changed. Trust the +`*_training_status` table, not the exit code. On an unclear failure re-run with `datapipe --debug … run` +to a file + `grep`, not inline. + +## Troubleshooting (verify against current files) + +- **Pre-staged cache is exactly 500 images** → `gt.json` in the cache dir caps the pool. An `offset` + past 500 yields an empty batch silently. Keep the batch total ≤ 500, or rebuild the cache larger + (or point `DATAPIPE_TAGS_CACHE_DIR` at a fresh dir to force a full COCO download). +- **`SIGILL` / `Illegal instruction` in training** → `polars` built for a newer CPU than the host + (pre-AVX2). The `polars-lts-cpu` pin isn't enough alone; the regular `polars` comes in transitively. + After `uv sync`: `uv pip uninstall polars polars-lts-cpu && uv pip install polars-lts-cpu==1.33.1`, + then verify `python -c "import polars; print(polars.__version__)"` is the lts one. +- **`No labels found` / every image "corrupt: No module named 'pi_heif'`** → reinstall `pi-heif`. +- **count-metrics prints "Batches to process 0" right after training** → datapipe hasn't propagated + the fresh predictions in the same pass. Re-run `count-metrics` once; then `metrics_on_image/subset` + and `tag_metrics` fill. +- **Two identical models get trained** → you ran `stage=train-prepare` and then `stage=train` + separately; the split between them changes `image__subset`, which re-freezes the dataset and + retrains. Run `stage=train` as one step (it includes prepare), and show the split with a query. +- **`kill -0 ` waiters, not `pgrep -f ""`** → a `pgrep`/`until` loop whose own + command line contains the pattern (e.g. `stage=load`, `uv sync`, `bin/datapipe step`) matches + ITSELF and never exits. Wait on the captured PID instead. +- **Read-only home on the cluster** (e.g. gpu5 `ml`) → put the repo, `HOME`, uv cache, and FiftyOne + local images under a writable path like `/var/tmp`; `/tmp` and `/var/tmp` are writable. +- **Flaky link to the cluster** (Moscow/RU hosts) → SSH may time out transiently; retry. +- **Metrics 0 on a trained model** → tiny/noisy val latches an early "best" checkpoint; use enough + data (~450 total) and the shipped epoch config. +- **Training exits 0 but no model** → datapipe swallows step errors; check `detection_training_status`. + +## Real data + +**Skip the two-part / checkpoint / rehearse machinery — that's demo-only.** For real data just run the +stages end to end (`load` → `train` → `count-metrics` → optionally `fiftyone`); no artificial stop, no +snapshot/restore. `--darken` is also demo-only (it synthesizes the low-light scenario from normal COCO +images — real data has real tagged images, so drop it). Still keep **val frozen** (pin `--subset`) — +that's about metric correctness, not the demo. + +Don't guess; gather up front and wire the loader to the real source instead of the COCO `load_batch`: +images + where boxes/labels come from (a labelled set to inject, or real annotation); real class +names (set `DETECTION_CLASSES` / GT labels to match exactly, casing!); the tag/scenario and which +images carry it; storage + DB (`DATAPIPE_TAGS_DIR`, `DB_URL`/`DB_SCHEMA`); GPU + enough data/epochs +that metrics are meaningful. diff --git a/.claude/skills/setup-e2e-template/SKILL.md b/.claude/skills/setup-e2e-template/SKILL.md new file mode 100644 index 00000000..112cc961 --- /dev/null +++ b/.claude/skills/setup-e2e-template/SKILL.md @@ -0,0 +1,75 @@ +--- +name: setup-e2e-template +description: > + Use when setting up or running examples/e2e_template (image_detection or image_keypoints) — + datapipe YOLO/Label Studio pipelines — or debugging their install, stages, training, or FiftyOne output. +--- + +# e2e_template (detection + keypoints YOLO/Label-Studio pipeline) + +This skill = run the YOLO/Label-Studio detection/keypoints pipeline on YOUR images. The COCO seed sample is just a smoke-test; the real goal is your own images — set the knobs below first. + +**Ask first — don't assume (only the unresolved):** demo or your own data? bring up the bundled `docker compose` services or reuse existing? **which Postgres + which database** for `DB_URL` — never point it at an existing DB without confirming; reuse an existing venv / `uv` env or create a fresh one? GPU? **annotate for real in Label Studio, or inject ready-made ground truth** (reuse existing labels / a labelled dataset) to skip the human `annotation` step? surface stage logs or run quiet? (per-tag scenario metrics live in a separate example → **setup-detection-tags**) + +**How to work:** read the setup, then propose a short plan and get a go-ahead before touching anything. Prepare `.env` and **pause for the user to verify it** before running. Run each stage with its logs shown and, after each, say what you did and what changed — don't run the pipeline silently. If a stage fails and the cause isn't clear from the normal logs, re-run it with `datapipe --debug … run` (or `--debug-sql` for SQL errors); debug is very verbose, so send it to a file and `grep` it (e.g. `datapipe --debug run > /tmp/dp_debug.log 2>&1; grep -nEi "error|traceback" /tmp/dp_debug.log`) rather than dumping it inline. + +## Run on YOUR data +- **Align your class everywhere** (mismatch → 0 useful results): `LABEL_CONFIG` names == `CLASSES_TO_KEEP`. Detection also `COCO_CLASSES`/`DETECTION_MODEL_CONFIG`; keypoints also `KEYPOINTS_LABELS` (order matters), `COCO_PERSON_KEYPOINT_FLIP_IDX`, `KEYPOINTS_MODEL_CONFIG`. +- **Where YOUR images go:** put them under `$DATAPIPE_E2E_DIR/images` (`DATAPIPE_E2E_DIR` defaults to `s3://datapipe-e2e`; the pipeline writes its own artifacts under `$DATAPIPE_E2E_DIR/datapipe`, a sibling — input and working dir don't overlap). Set `AWS_*`, `S3_ENDPOINT_URL`, and `S3_PUBLIC_URL` (browser-reachable — Label Studio loads images from it). Label Studio reaches S3 via its own `LABEL_STUDIO_S3_ENDPOINT_URL` (`minio:9000`), not `S3_ENDPOINT_URL`. + +## Prerequisites +- **Python 3.10–3.12** (hard pin `>=3.10,<3.13`). **Install:** `cd examples/e2e_template && uv sync` + (`pyproject.toml` declares editable workspace libs + cu124 `torch==2.6.0`). +- **Services:** `docker compose up` → Postgres 5432, MinIO (9000/9001, bucket `datapipe-e2e`, + anon-download for browser images), MongoDB 27017, Label Studio :8080. +- **Env:** `cp .env.example .env` then `set -a && source .env && set +a` before any `datapipe` command + (`config.py` raises if `DB_URL` unset). Detection/keypoints use separate schemas. +- **LS API token:** :8080 → Account & Settings → Access Token → `LABEL_STUDIO_API_KEY=...` in `.env` + (or `scripts/label_studio_token.py`). +- **GPU for training** (no CPU knob). **≥10 annotated images** to freeze a dataset (`min_delta=10`). +- Read-only/small `/home`: `export UV_CACHE_DIR=/tmp/uvcache HF_HOME=/tmp/hf` before `uv sync`. +- Stages: `annotation`, `ls-sync`, `train`, `fiftyone`. + +## Quick demo to verify setup +Skip if you have data: `uv run python scripts/seed_sample_data.py` downloads ~20 COCO images (10 cat/dog + 10 person keypoints; `--detection-limit`/`--keypoints-limit` to change) and uploads them to MinIO; then run §Run as-is. + +## Run (from the project subdir) +```bash +cd image_detection # or image_keypoints +source ../.venv/bin/activate # else prefix every command with `uv run` +set -a && source ../.env && set +a # config raises w/o DB_URL +datapipe db create-all +datapipe step --labels=stage=annotation run # LS tasks + pre-annotations +# → annotate ≥10 images in LS (:8080), mark completed → +datapipe step --labels=stage=ls-sync run # → image__ground_truth +datapipe step --labels=stage=train run # freeze + train YOLO + metrics + best +datapipe step --labels=stage=fiftyone run +fiftyone app launch datapipe_detection_e2e # or datapipe_keypoints_e2e +``` +Train uses `yolov8n*.pt` (imgsz 320, 30 ep); pre-annotation fallback `yolo11n*.pt`; best on `subset_id=val`. + +## Skip annotation — inject ground truth (unattended / demo) +No human in Label Studio? You can bypass `annotation`/`ls-sync` by writing `image__ground_truth` + +`image__subset` directly, then run `stage=train`. **The injected rows MUST follow the pipeline's own +conventions, or the freeze join silently yields nothing / classes don't match:** +- **`image_name` must equal what `list_s3_images` emits** — the key **relative to `INPUT_IMAGES_DIR` + (`$DATAPIPE_E2E_DIR/images`)**, i.e. the plain object name (`000000008458.jpg`). Don't invent a + prefixed form — if `s3_images` has `X` and your GT has `pfx___X`, the `GT ⋈ subset ⋈ s3_images` join + is empty → `freeze_dataset` fails with `No ground truth`. +- **`labels` must match `DETECTION_CLASSES` casing** (e.g. `Cat`/`Dog`, not COCO lowercase `cat`/`dog`) + — otherwise predictions and GT land in *different* classes and every metric is 0. +- **`bboxes`** = pixel `[x1,y1,x2,y2]`; assign `image__subset.subset_id` (`train`/`val`) yourself. +- Write via datapipe `DataStore`/`UpdateExternalTable` (keeps `*_meta` in sync) — not raw SQL `UPDATE` + on PK columns. Then `datapipe step --labels=stage=train run` (skip `annotation`/`ls-sync`). + +## Per-scenario tag metrics → see the dedicated example +Want to tag a scenario (e.g. dark-room pallets), add it to training, and measure the model on that +scenario separately (baseline vs retrained)? That lives as its own self-contained example — +`examples/detection_tags` (`tag`/`image__tag`/`tag_metrics` built into the pipeline, **no Label Studio +or FiftyOne**, ground truth injected). Use the **setup-detection-tags** skill. + +## Troubleshooting (may already be fixed — verify against current files) +- **No model after `train`, exit 0** → datapipe swallows step errors; check `detection_training_status`, not the exit code. +- **Demo pre-annotations are empty** → the fallback `DETECTION_MODEL_CONFIG` is a smoke model (`yolo11n`, `input_size:[16,16]`, `score_threshold:0.01`) used until a trained model exists, so it detects almost nothing. Expected for the demo; for useful pre-annotations set a real model + `input_size`/`score_threshold`. +- **Training metrics ~0** → not a config bug: the trained model is `yolov8n` (imgsz 320, 30 ep), and the seed sample (~20 images) is simply too small to learn from. Real metrics need enough annotated data. +- **`cv_pipeliner` keypoint pre-annotation is broken** (pinned rev): inferencer drops keypoints, LS parser doesn't apply them → keypoint `train` needs real keypoint GT injected. diff --git a/.claude/skills/setup-embedder-fiftyone/SKILL.md b/.claude/skills/setup-embedder-fiftyone/SKILL.md new file mode 100644 index 00000000..d526710c --- /dev/null +++ b/.claude/skills/setup-embedder-fiftyone/SKILL.md @@ -0,0 +1,68 @@ +--- +name: setup-embedder-fiftyone +description: > + Use when working in examples/embedder_fiftyone, or running / debugging the embedder→FiftyOne + datapipe example (DINOv2/DINOv3 image embeddings, UMAP, similarity search). +--- + +# embedder_fiftyone (DINO → FiftyOne) + +This skill = run the DINO→FiftyOne embedder on YOUR images. The FiftyOne-zoo dataset (caltech101) is just a smoke-test; the real goal is your own images — set the knobs below first. + +**Ask first — don't assume (only the unresolved):** demo (zoo) or your own data? **which Postgres + which database** for `DB_URL` — never point it at an existing DB or drop in a `localhost` default without confirming; reuse an existing venv / `uv` env or create a fresh one? GPU? surface stage logs or run quiet? + +**How to work:** read the setup, then propose a short plan and get a go-ahead before touching anything. Prepare `.env` and **pause for the user to verify it** before running. Run each stage with its logs shown and, after each, say what you did and what changed — don't run the pipeline silently. If a stage fails and the cause isn't clear from the normal logs, re-run it with `datapipe --debug … run` (or `--debug-sql` for SQL errors); debug is very verbose, so send it to a file and `grep` it (e.g. `datapipe --debug run > /tmp/dp_debug.log 2>&1; grep -nEi "error|traceback" /tmp/dp_debug.log`) rather than dumping it inline. + +## Run on YOUR data +- **Align `LABELS_JSON` keys to the file stem** (optional) — a mismatch silently yields an unlabeled + sample (`None`) / 0 useful results. +- **Your images:** `LOCAL_IMAGES_DIR=/path` (`.jpg/.jpeg/.png`; file stem → `image_name`). +- **Pick models:** `config.py` → `EMBEDDERS` (default dinov2_base + dinov2_large). + +## Prerequisites +Stages (labels match `app.py`): `source` (local folder or zoo fallback) → `embedder` → `fiftyone` +(publish samples) → `embeddings` (per-image `.npy`) → `fiftyone-brain` (UMAP + cosine sim per embedder). +- **External PostgreSQL** at `DB_URL` (SQLAlchemy URL). Not started by the example; an empty DB is fine — + `datapipe db create-all` auto-creates tables. `.env.example` default `...postgres@localhost:5432/postgres`. +- **FiftyOne must run on the SAME machine** — samples/brain live in local FiftyOne/MongoDB; remote → + empty panels / `no_matching_samples`. +- **GPU** ~8 GB plenty for dinov2_base/large (CPU works, slower). **`uv` + Python 3.10–3.12** → `uv sync` + (pins cu124 `torch==2.6.0`, `transformers`, `fiftyone`). +- **HF token only for DINOv3.** Defaults (dinov2_base/large) are public; the `dinov3_vitl16` entry is + **commented out** — to use it, uncomment, accept the `facebook/dinov3-*` license, set `HF_TOKEN` with + gated-repo read access (else the whole embeddings batch fails). + +## Quick demo to verify setup +Skip if you have data. Leave `LOCAL_IMAGES_DIR` unset → zoo fallback (`ZOO_DATASET`=caltech101, +`ZOO_NUM_CLASSES`=10, keeps the **last N** classes); run §Run as-is to confirm install/DB/GPU. First run +downloads it, then auto-deletes the source zoo dataset; `image_name` rewritten `/`→`__` (`steps.py:83`). + +## Run (from examples/embedder_fiftyone) +```bash +cp .env.example .env # DB_URL, LOCAL_IMAGES_DIR or ZOO_*, FIFTYONE_DATASET_NAME (+ FIFTYONE_DATABASE_DIR, see traps) +uv sync && source .venv/bin/activate # else prefix each command with `uv run` +datapipe db create-all && datapipe run +# one stage: datapipe step --labels=stage=source run (source|embedder|fiftyone|embeddings|fiftyone-brain) +``` + +## Caching +Embeddings cache as `.npy` in `EMBEDDINGS_DIR/{embedder_id}/` (default `./data/embeddings`); delete to recompute. + +## View in FiftyOne (same machine) +```bash +fiftyone app launch --remote --address 0.0.0.0 --port 5151 --wait -1 # venv active; SSH-forward 5151 +``` +Open `FIFTYONE_DATASET_NAME`. **Embeddings** panel → key `_umap` (single `_`), color by +`ground_truth`. **Similarity** → `__sim` (double `_`) — don't transpose. + +## Success criteria +`brain_status.status = 'ok'` per embedder — **trust the table, not the exit code**: datapipe swallows +step errors and still exits 0. Plus: Embeddings panel shows clusters, similarity returns neighbors. + +## Troubleshooting (may already be fixed — verify against current files) +- **`brain_status` empty / 0 images though run exited 0** → FiftyOne mongod failed to start (old + `~/.fiftyone` datadir: FCV mismatch → `Wrong mongod version` / `failed to bind to port`). Use a fresh + dir: `FIFTYONE_DATABASE_DIR=/tmp/fo_db`. +- **`no_matching_samples`** → no sample `image_name` matched an embedding key: check the `fiftyone` + stage published samples, pipeline + FiftyOne on the same machine, and (zoo) the `__`-rewrite. +- **DINOv3 403 / `LocalEntryNotFoundError`** → needs a license-accepted gated `HF_TOKEN` (see Prerequisites). diff --git a/.claude/skills/setup-sam-cvat/SKILL.md b/.claude/skills/setup-sam-cvat/SKILL.md new file mode 100644 index 00000000..9d01ec15 --- /dev/null +++ b/.claude/skills/setup-sam-cvat/SKILL.md @@ -0,0 +1,62 @@ +--- +name: setup-sam-cvat +description: > + Use when working in examples/sam_cvat, or when setting up / running / debugging the datapipe + SAM3→CVAT pre-annotation example, on the cats-n-dogs demo data or on your own images. +--- + +# sam_cvat (SAM3 + CVAT) + +This skill = run SAM3→CVAT pre-annotation on YOUR images. The HZMD/cats-n-dogs demo is just a smoke-test; the real goal is your own images — set the knobs below first. + +**Ask first — don't assume (only the unresolved):** demo (cats-n-dogs) or your own data? **which Postgres + which database** for `DB_URL` — never point it at an existing DB or drop in a default without confirming; external CVAT ready or provision/point at it? reuse an existing venv / `uv` env or create a fresh one? GPU >8 GB? surface stage logs or run quiet? + +**How to work:** read the setup, then propose a short plan and get a go-ahead before touching anything. Prepare `.env` and **pause for the user to verify it** before running. Run each stage with its logs shown and, after each, say what you did and what changed — don't run the pipeline silently. If a stage fails and the cause isn't clear from the normal logs, re-run it with `datapipe --debug … run` (or `--debug-sql` for SQL errors); debug is very verbose, so send it to a file and `grep` it (e.g. `datapipe --debug run > /tmp/dp_debug.log 2>&1; grep -nEi "error|traceback" /tmp/dp_debug.log`) rather than dumping it inline. + +## Run on YOUR data +- **Align across all three:** `SAM_TEXT_PROMPT` == the CVAT labels (`CVAT_BOX_LABEL` / `CVAT_POLYGON_LABEL`) == `HF_DATASET_LABEL` must all mean the same class — a mismatch runs clean but yields 0 useful results. +- **Point at your images:** `INPUT_DIR=/path` (`.jpg/.jpeg/.png`; stem → `image_id`). An existing-but-EMPTY `INPUT_DIR` yields 0 images (no HF fallback). +- **Levers:** `SAM_SCORE_THRESHOLD` (0.5), `SAM_MAX_DETECTIONS` (10). + +## Prerequisites +- **GPU >8 GB with FlashAttention.** *Field note:* ~5.5 GB on a 16 GB RTX 4060 Ti, but OOM on an 8 GB + GTX 1070 (Pascal: no FlashAttention → math-attention, O(n²) memory). +- **HF token — SAM3 is gated.** Accept the license at huggingface.co/facebook/sam3, set `HF_TOKEN` + with gated-repo read access (a non-gated token → download/load failure). +- **External PostgreSQL** at `DB_URL` (not started here; tables auto-create via `datapipe db create-all`). +- **External CVAT** (not started here): set `CVAT_URL`, `CVAT_USERNAME`, `CVAT_PASSWORD`, optional + `CVAT_ORGANIZATION`, and `CVAT_PROJECT_ID` for a project whose label names match `CVAT_BOX_LABEL` / + `CVAT_POLYGON_LABEL`. +- **`uv` + Python ≥3.10,<3.13** → `uv sync`. `pyproject.toml` pins cu124 torch and editable local libs + (`../../libs/datapipe-*`, resolves only inside the monorepo checkout), and builds `sam3` from a pinned + git source (needs git, a compiler, `nvcc`/`CUDA_HOME`). If `uv sync` fails, check the `sam3` rev in `[tool.uv.sources]`. +- Stages: `ingest` (local folder or HF-dataset fallback) → `sam` (text-prompt inference → boxes + mask + polygons) → `cvat` (upload images + pre-annotations; parse edits back into `image__annotations`). + +## Quick demo to verify setup +Skip if you have data. Leave `INPUT_DIR` unset → HF fallback; shipped `.env.example` is self-consistent +(`a cat` / `cat_box`/`cat_mask` / `HF_DATASET_LABEL=0`=cats) → first run on `HZMD/cats-n-dogs` yields +detections, confirming install/DB/GPU/CVAT. + +## Run +```bash +cp .env.example .env # DB_URL, HF_TOKEN, CVAT_*, CVAT_PROJECT_ID, SAM_TEXT_PROMPT, labels +uv sync && source .venv/bin/activate # else prefix each command with `uv run` +datapipe db create-all && datapipe run +# or by stage: datapipe step --labels stage=sam run +``` +Run from `examples/sam_cvat/` (`app.py` `load_dotenv()`s before importing config — wrong cwd breaks `.env`). + +## Caching gotcha +`SAM_TEXT_PROMPT` IS tracked (→ `sam_config` table) so changing it retriggers inference — **but +`CVATStep` does NOT push new pre-annotations to EXISTING tasks** (only on image add/remove/path change). +Treat the prompt as fixed per run; to change it, wipe the old CVAT tasks + datapipe CVAT tables, re-run. + +## Annotate (human-in-the-loop) +In CVAT fix preannotations, **mark the task completed**, re-run `datapipe run` → edits land in +`image__annotations`. CVAT tables (for wipes): `image_batches`, `cvat_task`, `cvat_images` (not "cvat_files"), `cvat_task_sync_table`. + +## Troubleshooting (may already be fixed — verify against current files) +- **0 detections** → `.env` class misaligned; re-align all three. +- **CVAT rejects label** → project needs labels named `CVAT_BOX_LABEL`/`CVAT_POLYGON_LABEL`. +- **Prompt change not in CVAT** → expected (see Caching). diff --git a/examples/detection_tags/.env.example b/examples/detection_tags/.env.example new file mode 100644 index 00000000..a10f4deb --- /dev/null +++ b/examples/detection_tags/.env.example @@ -0,0 +1,16 @@ +export AWS_ACCESS_KEY_ID=minioadmin +export AWS_SECRET_ACCESS_KEY=minioadmin +export AWS_REGION=us-east-1 +export S3_ENDPOINT_URL=http://localhost:9000 +export FSSPEC_S3_ENDPOINT_URL="${S3_ENDPOINT_URL}" +export S3_PUBLIC_URL=http://localhost:9000 +export DATAPIPE_TAGS_DIR=s3://datapipe-tags +export DATAPIPE_TAGS_TMP_DIR=/tmp/datapipe-tags +export DB_URL=postgresql://postgres:password@localhost:5432/postgres +# default schema; set a dedicated one (e.g. datapipe_tags) only if you SHARE this Postgres with +# other pipelines — then `db create-all` needs the schema created first (see README). +export DB_SCHEMA=public +# FiftyOne (stage=fiftyone): dataset metadata in MongoDB, images rendered from local files +export FIFTYONE_DATABASE_URI=mongodb://localhost:27017 +export FIFTYONE_DATASET_NAME=datapipe_detection_tags +export LOCAL_IMAGES_DIR=/tmp/datapipe-tags-fiftyone-images diff --git a/examples/detection_tags/README.md b/examples/detection_tags/README.md new file mode 100644 index 00000000..8c805639 --- /dev/null +++ b/examples/detection_tags/README.md @@ -0,0 +1,78 @@ +# detection_tags + +A self-contained datapipe detection example built around **tags** (per-scenario metrics), with a +**FiftyOne** view and **no Label Studio**. Ground truth is injected directly (COCO labels, lowercase +`cat`/`dog`), so it runs unattended from scratch. + +The point: train a baseline (model A), then add a **tagged scenario TRAIN batch** and retrain +(model B), and watch recall on that tag rise — visible in a dedicated `tag_metrics` table. The demo +is split into **two parts around a checkpoint** so you can prep part 1 and rehearse part 2. + +## Deploy from scratch +```bash +cp .env.example .env && set -a && source .env && set +a +docker compose up -d # postgres + minio + mongo (mongo backs FiftyOne) +uv sync # cu124 torch + datapipe-ml[torch,fiftyone] + fiftyone + pi-heif +# pre-AVX2 host only: force the lts polars to win (else training SIGILLs) +uv pip uninstall polars polars-lts-cpu && uv pip install polars-lts-cpu==1.33.1 +cd detection +# DB_SCHEMA defaults to `public` (exists already). For a dedicated schema (sharing the Postgres with +# other pipelines) create it first: psql "$DB_URL" -c "CREATE SCHEMA IF NOT EXISTS $DB_SCHEMA" +datapipe db create-all +``` + +## Frozen val (why the load order matters) +A `val` metric is an aggregate over whatever images are in val; if you add the tagged batch to val at +retrain time, model A's numbers move and the A-vs-B comparison is invalid. So **freeze val up front**: +load `base-val` + `night-val` (pinned `--subset val`) before training A, and add `night-train` +(`--subset train`) only for part 2. `add_request.py --subset` pins a batch (`image__subset_hint`), +the split step honors it. The pre-staged cache holds **500 images**, so keep the total ≤ 500. + +## Part 1 — baseline to checkpoint +```bash +# from examples/detection_tags/detection +python ../scripts/add_request.py --id base-train --n 325 --offset 0 --subset train +python ../scripts/add_request.py --id base-val --n 100 --offset 325 --subset val +python ../scripts/add_request.py --id night-val --n 25 --offset 425 --subset val --tag night --darken 0.25 +datapipe step --labels=stage=load run +datapipe step --labels=stage=train run # model A +datapipe step --labels=stage=count-metrics run # re-run once if it prints "Batches to process 0" +# (demo-only) snapshot the post-A state to rehearse part 2 later, before the fiftyone stage: +docker exec pg_dump -U postgres -n "$DB_SCHEMA" postgres > /tmp/checkpoint.sql +datapipe step --labels=stage=fiftyone run # GT + model-A predictions into FiftyOne +``` + +## Part 2 — retrain and watch the tag metric rise +```bash +python ../scripts/add_request.py --id night-train --n 50 --offset 450 --subset train --tag night --darken 0.25 +datapipe step --labels=stage=load run +datapipe step --labels=stage=train run # model B (night now in training) +datapipe step --labels=stage=count-metrics run # re-run once if "0 batches" +datapipe step --labels=stage=fiftyone run # adds predictions_model_b +``` +Rehearse (demo-only): restore the snapshot + wipe the FiftyOne db, then re-run part 2 — no retraining +of model A: +```bash +docker exec psql -U postgres -c "DROP SCHEMA IF EXISTS $DB_SCHEMA CASCADE; CREATE SCHEMA $DB_SCHEMA" +docker exec -i psql -U postgres < /tmp/checkpoint.sql +docker exec mongosh --quiet --eval "db.getSiblingDB('fiftyone').dropDatabase()" +``` + +## What you get +- `detection_model_train__metrics_on_subset` — overall metrics per (model, subset). +- **`tag_metrics`** — `(detection_model_id, tag_id, subset_id)` → precision/recall/f1. Compare model + A vs model B at `tag=night, subset=val`: recall rises once the tagged batch is in training. +- **FiftyOne** dataset `$FIFTYONE_DATASET_NAME`: `ground_truth`, `predictions_model_a`, + `predictions_model_b`, plus `tag`/`subset` sample fields. Launch: + `fiftyone app launch "$FIFTYONE_DATASET_NAME" --port 5151 --address 0.0.0.0` (tunnel that port if + remote — local port must equal the remote port; filter by the `tag`/`subset` fields). + +### Pre-staged cache (fast loads) +If `DATAPIPE_TAGS_CACHE_DIR/gt.json` + `DATAPIPE_TAGS_CACHE_DIR/images/.jpg` exist, the load +step reads from them instead of fetching COCO. Default dir `/tmp/datapipe-tags-cache`. It caps the +image pool at its size (500 here) — expand it or point at a fresh dir to download more from COCO. + +## Notes +- Classes are lowercase (`cat`/`dog`) to match COCO so injected GT and predictions align. +- Trust `detection_training_status.status`, not exit codes; `count-metrics` may need a second run. +- FiftyOne integration is ported from `examples/e2e_template/image_detection`. diff --git a/examples/detection_tags/detection/__init__.py b/examples/detection_tags/detection/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/examples/detection_tags/detection/app.py b/examples/detection_tags/detection/app.py new file mode 100644 index 00000000..07ead686 --- /dev/null +++ b/examples/detection_tags/detection/app.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +from datapipe.compute import DatapipeApp, Pipeline +from datapipe.datatable import DataStore +from datapipe.step.batch_transform import BatchTransform +from datapipe_ml.metrics.model_selection import FindBestModel +from datapipe_ml.tasks.detection.freeze import DetectionFreezeDataset +from datapipe_ml.tasks.detection.inference import Inference_DetectionModel +from datapipe_ml.tasks.detection.metrics import CountMetrics_Subset_DetectionModel +from datapipe_ml.tasks.detection.train.yolov8 import Train_YoloV8_DetectionModel, YoloV8_TrainingConfig +from datapipe_ml.training.specs import TrainingResumeConfig, TrainingSyncConfig + +import steps +from config import DATAPIPE_DIR, DBCONN, datapipe_tmp_folder +from data import catalog + +# Data is loaded via the `load` step: add a row to `load_request` (request_id, n, offset, tag, +# darken) and run `datapipe step --labels=stage=load run`. It downloads COCO cat/dog images, +# uploads them to object storage, and produces s3_images + ground truth (+ tag) directly — there +# is no Label Studio annotation stage. + +pipeline = Pipeline( + [ + BatchTransform( + func=steps.load_batch, + inputs=["load_request"], + outputs=["s3_images", "image__ground_truth", "tag", "image__tag", "image__subset_hint"], + transform_keys=["request_id"], + labels=[("stage", "load")], + ), + BatchTransform( + func=steps.split_df_train_val, + inputs=["image__ground_truth", "image__subset", "image__subset_hint"], + outputs=["image__subset"], + transform_keys=["image_name"], + kwargs=dict(primary_keys=["image_name"], val_perc=0.25, random_seed=42), + labels=[("stage", "train"), ("stage", "train-prepare")], + ), + DetectionFreezeDataset( # type: ignore[list-item] + input__image="s3_images", + input__image__ground_truth="image__ground_truth", + input__subset__has__image="image__subset", + output__detection_frozen_dataset="detection_frozen_dataset", + output__detection_frozen_dataset__has__image_gt="detection_frozen_dataset__has__image_gt", + working_dir=str(DATAPIPE_DIR), + min_within_time="1s", + min_delta=10, + primary_keys=["image_name"], + bbox_id__name=None, + image__image_path__name="image_url", + labels=[("stage", "train"), ("stage", "train-prepare")], + create_table=True, + ), + Train_YoloV8_DetectionModel( # type: ignore[list-item] + input__detection_frozen_dataset="detection_frozen_dataset", + input__detection_frozen_dataset__has__image_gt="detection_frozen_dataset__has__image_gt", + output__yolov8_train_config="yolov8_train_config", + output__detection_size_for_resize="detection_size_for_resize", + output__detection_frozen_dataset__resized_image_file="detection_frozen_dataset__resized_image_file", + output__detection_frozen_dataset__yolo_txt="detection_frozen_dataset__yolo_txt", + output__detection_model="detection_model_train", + output__detection_model_is_trained_on_detection_frozen_dataset=( + "detection_model_is_trained_on_detection_frozen_dataset" + ), + output__training_status="detection_training_status", + output__detection_frozen_dataset__class_names="detection_frozen_dataset__class_names", + max_within_time="1w", + working_dir=str(DATAPIPE_DIR), + tmp_folder=datapipe_tmp_folder(), + yolov8_train_configs=[ + YoloV8_TrainingConfig(model="yolov8n.pt", imgsz=320, batch=10, epochs=10, exist_ok=True) + ], + sync_config=TrainingSyncConfig(enabled=True, interval_s=30, retries=3, retry_sleep_s=30), + resume_config=TrainingResumeConfig( + continue_train_failed_models=True, min_completed_epochs=1, checkpoint="last", + max_attempts=10, reset_attempts_after="10m", lease_ttl_s=60, heartbeat_interval_s=10, + ), + primary_keys=["image_name"], + bbox_id__name=None, + labels=[("stage", "train")], + create_table=True, + allow_sample_size_mismatch=True, + model_suffix="_tags", + ), + Inference_DetectionModel( + input__image="s3_images", + input__detection_model="detection_model_train", + output__detection_prediction="detection_prediction_train", + primary_keys=["image_name"], + bbox_id__name=None, + image__image_path__name="image_url", + batch_size_default=1, + labels=[("stage", "train"), ("stage", "inference")], + create_table=True, + ), + CountMetrics_Subset_DetectionModel( + input__image__ground_truth="image__ground_truth", + input__subset__has__image="image__subset", + input__detection_prediction="detection_prediction_train", + output__detection_model__metrics__on__image="detection_model_train__metrics_on_image", + output__detection_model__metrics__on__subset="detection_model_train__metrics_on_subset", + primary_keys=["image_name"], + bbox_id__name=None, + minimum_iou=0.5, + labels=[("stage", "train"), ("stage", "count-metrics")], + create_table=True, + ), + FindBestModel( + input__model="detection_model_train", + input__model__metrics_on__subset="detection_model_train__metrics_on_subset", + output__attr__model__is_best="attr__detection_model__is_best", + output__best_model="best_detection_model", + subset_id="val", + is_best__name="detection_model__is_best", + primary_keys=["detection_model_id"], + metric__name="calc__f1_score", + func="max", + group_by=None, + labels=[("stage", "train"), ("stage", "count-metrics")], + ), + # tag arc: aggregate per-image metrics by (model, tag, subset) + BatchTransform( + func=steps.compute_tag_metrics, + inputs=["detection_model_train__metrics_on_image", "image__tag"], + outputs=["tag_metrics"], + transform_keys=["detection_model_id"], + labels=[("stage", "train"), ("stage", "count-metrics"), ("stage", "tag-metrics")], + ), + # --- FiftyOne (stage=fiftyone): browse GT + both models' predictions, filter by tag --- + BatchTransform( + func=steps.download_images, + inputs=["s3_images"], + outputs=["local_images"], + transform_keys=["image_name"], + kwargs=dict(image__image_path__name="image_url", image__local_image_path__name="local_path"), + labels=[("stage", "fiftyone")], + ), + BatchTransform( + func=steps.publish_gt_to_fiftyone, + inputs=["local_images", "image__ground_truth", "image__subset", "image__tag"], + outputs=["fiftyone_annotations"], + transform_keys=["image_name"], + kwargs=dict(primary_keys=["image_name"], image__image_path__name="local_path"), + labels=[("stage", "fiftyone")], + ), + BatchTransform( + func=steps.publish_predictions_to_fiftyone, + inputs=["local_images", "detection_prediction_train"], + outputs=["fiftyone_predictions_model_a"], + transform_keys=["image_name"], + kwargs=dict(slot="model_a", primary_keys=["image_name"], image__image_path__name="local_path"), + labels=[("stage", "fiftyone")], + ), + BatchTransform( + func=steps.publish_predictions_to_fiftyone, + inputs=["local_images", "detection_prediction_train"], + outputs=["fiftyone_predictions_model_b"], + transform_keys=["image_name"], + kwargs=dict(slot="model_b", primary_keys=["image_name"], image__image_path__name="local_path"), + labels=[("stage", "fiftyone")], + ), + ] +) + +ds = DataStore(DBCONN, create_meta_table=True) +app = DatapipeApp(ds, catalog, pipeline) diff --git a/examples/detection_tags/detection/coco_demo.py b/examples/detection_tags/detection/coco_demo.py new file mode 100644 index 00000000..fd24eb2b --- /dev/null +++ b/examples/detection_tags/detection/coco_demo.py @@ -0,0 +1,105 @@ +"""DEMO DATA SYNTHESIS — COCO cat/dog (replace this module for real data). + +Everything here exists only to fabricate the demo dataset: download the COCO cat/dog subset, inject +ground truth, and darken images to fake a low-light "night" scenario. None of it is part of the tag +pipeline itself — for REAL data, drop this module and feed the `load` step from your own source of +(image bytes, boxes, labels). The pipeline (steps.py / app.py) doesn't care where images come from. +""" +from __future__ import annotations + +import io +import json +import os +import random +import time +import zipfile +from pathlib import Path +from typing import Optional + +import numpy as np +import requests +from PIL import Image + +from config import COCO_CAT_IDS + +COCO_IMG_BASE = "http://images.cocodataset.org/train2017/" +COCO_ANN_URL = "http://images.cocodataset.org/annotations/annotations_trainval2017.zip" +CACHE = Path(os.environ.get("DATAPIPE_TAGS_CACHE_DIR", "/tmp/datapipe-tags-cache")) +RNG_SEED = 1234 + + +def _get(url: str, attempts: int = 4) -> requests.Response: + last: Optional[Exception] = None + for a in range(attempts): + try: + r = requests.get(url, timeout=60, stream=True) + r.raise_for_status() + return r + except requests.RequestException as e: + last = e + time.sleep(min(2 ** a, 20)) + assert last is not None + raise last + + +def _coco_annotations_by_image() -> tuple[dict, dict]: + """Return (id_to_image_meta, {image_id: [cat/dog annotations]}), cached on disk.""" + CACHE.mkdir(parents=True, exist_ok=True) + zip_path = CACHE / "annotations_trainval2017.zip" + if not zip_path.exists() or zip_path.stat().st_size < 1_000_000: + with _get(COCO_ANN_URL) as r, open(zip_path, "wb") as f: + for chunk in r.iter_content(1 << 20): + f.write(chunk) + with zipfile.ZipFile(zip_path) as zf, zf.open("annotations/instances_train2017.json") as h: + inst = json.load(h) + id_to_img = {im["id"]: im for im in inst["images"]} + anns: dict[int, list] = {} + for a in inst["annotations"]: + if a.get("category_id") in COCO_CAT_IDS and not a.get("iscrowd", 0) and a.get("bbox"): + anns.setdefault(a["image_id"], []).append(a) + return id_to_img, anns + + +def darken(raw: bytes, gamma: float) -> bytes: + """Gamma-darken JPEG bytes to fake a low-light scene (gamma < 1 darkens). Demo-only.""" + a = np.asarray(Image.open(io.BytesIO(raw)).convert("RGB")).astype(np.float32) / 255.0 + out = Image.fromarray((np.power(a, 1.0 / gamma) * 255).clip(0, 255).astype(np.uint8)) + buf = io.BytesIO() + out.save(buf, format="JPEG", quality=92) + return buf.getvalue() + + +class CocoDemoSource: + """Deterministic shuffled pool of COCO cat/dog filenames + per-file (bytes, boxes, labels). + + Prefers a pre-staged cache (`$DATAPIPE_TAGS_CACHE_DIR/gt.json` + `images/`) so loads are fast + and offline; falls back to fetching from COCO. Order is a fixed shuffle (RNG_SEED) so `offset`/`n` + slices are reproducible across runs. + """ + + def __init__(self) -> None: + cache_gt_path = CACHE / "gt.json" + self._use_cache = cache_gt_path.exists() and (CACHE / "images").is_dir() + if self._use_cache: + self._cache_gt = json.loads(cache_gt_path.read_text()) + pool = sorted(self._cache_gt.keys()) + else: + self._id_to_img, self._anns = _coco_annotations_by_image() + pool = [self._id_to_img[i]["file_name"] for i in sorted(self._anns.keys())] + self._fn_to_id = {self._id_to_img[i]["file_name"]: i for i in self._anns.keys()} + random.seed(RNG_SEED) + random.shuffle(pool) + self.pool = pool + + def fetch(self, fn: str) -> tuple[bytes, list, list]: + """Return (image_bytes, bboxes[[x1,y1,x2,y2]], labels[lowercase]) for a pool filename.""" + if self._use_cache: + data = (CACHE / "images" / fn).read_bytes() + g = self._cache_gt[fn] + return data, g["bboxes"], g["labels"] + raw = _get(COCO_IMG_BASE + fn).content + anns = self._anns[self._fn_to_id[fn]] + boxes = [[int(a["bbox"][0]), int(a["bbox"][1]), + int(a["bbox"][0] + a["bbox"][2]), int(a["bbox"][1] + a["bbox"][3])] for a in anns] + labels = [COCO_CAT_IDS[a["category_id"]] for a in anns] + return raw, boxes, labels diff --git a/examples/detection_tags/detection/config.py b/examples/detection_tags/detection/config.py new file mode 100644 index 00000000..610295fe --- /dev/null +++ b/examples/detection_tags/detection/config.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import os +from pathlib import Path +from typing import Optional + +import fsspec +from datapipe.store.database import DBConn +from pathy import Pathy + +# --- object storage (MinIO / S3) ------------------------------------------------ +AWS_KEY = os.environ.get("AWS_ACCESS_KEY_ID", "") +AWS_SECRET = os.environ.get("AWS_SECRET_ACCESS_KEY", "") +AWS_REGION = os.environ.get("AWS_REGION", "us-east-1") +S3_ENDPOINT_URL = os.environ.get("S3_ENDPOINT_URL") +S3_PUBLIC_URL = os.environ.get("S3_PUBLIC_URL", S3_ENDPOINT_URL or "http://localhost:9000") + +# --- classes (lowercase, as in COCO) ------------------------------------------- +# detection classes for this demo; keep lowercase to match COCO category names so +# injected ground truth and model predictions use identical label strings. +DETECTION_CLASSES = ["cat", "dog"] +COCO_CAT_IDS = {17: "cat", 18: "dog"} # COCO category_id -> class name + +# --- tags ----------------------------------------------------------------------- +# Numeric tag ids must be DETERMINISTIC (datapipe re-runs), so map known tag names to +# fixed ids; unknown names fall back to a stable crc32-derived number. +import zlib # noqa: E402 + +TAG_IDS = {"night": 1} +TAG_NAMES = {v: k for k, v in TAG_IDS.items()} + + +def tag_id_for(name: str) -> int: + return TAG_IDS.get(name) or (zlib.crc32(name.encode()) % 100000) + + +def tag_name_for(tag_id) -> str: + return TAG_NAMES.get(tag_id, str(tag_id)) + +# --- single storage root -------------------------------------------------------- +# Input images live under /images; the pipeline working_dir under /datapipe +# (siblings) so listing input images never re-ingests training artifacts. +DATAPIPE_DIR_ROOT = os.environ.get("DATAPIPE_TAGS_DIR", "s3://datapipe-tags").rstrip("/") +INPUT_IMAGES_DIR = f"{DATAPIPE_DIR_ROOT}/images" + + +def _is_cloud_path(path: str) -> bool: + protocol, _ = fsspec.core.split_protocol(path) + return protocol not in (None, "file") + + +def datapipe_working_dir() -> str: + if _is_cloud_path(DATAPIPE_DIR_ROOT): + return f"{DATAPIPE_DIR_ROOT}/datapipe" + local = Path(DATAPIPE_DIR_ROOT).resolve() / "datapipe" + local.mkdir(parents=True, exist_ok=True) + return str(local) + + +DATAPIPE_DIR = datapipe_working_dir() + + +def input_storage_options() -> dict: + if not _is_cloud_path(INPUT_IMAGES_DIR): + return {} + client_kwargs: dict = {"region_name": AWS_REGION} + if S3_ENDPOINT_URL: + client_kwargs["endpoint_url"] = S3_ENDPOINT_URL + return {"key": AWS_KEY, "secret": AWS_SECRET, "client_kwargs": client_kwargs} + + +def input_bucket() -> Optional[str]: + if not _is_cloud_path(DATAPIPE_DIR_ROOT): + return None + return Pathy(DATAPIPE_DIR_ROOT).root + + +def input_image_url(rel_key: str) -> str: + """Browser- and fsspec-readable URL for an input image relative to INPUT_IMAGES_DIR.""" + if _is_cloud_path(INPUT_IMAGES_DIR): + target = Pathy(INPUT_IMAGES_DIR) / rel_key + return f"{S3_PUBLIC_URL.rstrip('/')}/{target.root}/{target.key}" + return (Path(INPUT_IMAGES_DIR) / rel_key).resolve().as_uri() + + +def datapipe_tmp_folder() -> str: + if _is_cloud_path(DATAPIPE_DIR): + return str(Path(os.environ.get("DATAPIPE_TAGS_TMP_DIR", "/tmp/datapipe-tags")).resolve()) + return str(Path(DATAPIPE_DIR) / "tmp") + + +# --- database ------------------------------------------------------------------- +DB_URL = os.environ.get("DB_URL") +if not DB_URL: + raise RuntimeError( + "DB_URL is required. Copy examples/detection_tags/.env.example to .env, " + "start docker compose, and run: set -a && source .env && set +a" + ) + +DB_SCHEMA = os.environ.get("DB_SCHEMA", "public") +DBCONN = DBConn(DB_URL, DB_SCHEMA) + +# --- FiftyOne ------------------------------------------------------------------- +# FiftyOne stores dataset metadata in MongoDB (FIFTYONE_DATABASE_URI, brought up by +# docker compose) and renders samples from LOCAL image files, so the fiftyone stage +# downloads S3 images to LOCAL_IMAGES_DIR first. +FIFTYONE_DATASET_NAME = os.environ.get("FIFTYONE_DATASET_NAME", "datapipe_detection_tags") +LOCAL_IMAGES_DIR = Path(os.environ.get("LOCAL_IMAGES_DIR", "/tmp/datapipe-tags-fiftyone-images")) diff --git a/examples/detection_tags/detection/data.py b/examples/detection_tags/detection/data.py new file mode 100644 index 00000000..75834806 --- /dev/null +++ b/examples/detection_tags/detection/data.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from cv_pipeliner.utils.fiftyone import FifyOneSession +from datapipe.compute import Catalog, Table +from datapipe.store.database import TableStoreDB +from datapipe_ml.utils.image_data_stores import FiftyOneImagesDataTableStore +from sqlalchemy import Column, Float, Integer, JSON, String + +from config import DBCONN, FIFTYONE_DATASET_NAME + +# one FiftyOne session/dataset shared by the ground-truth + per-model prediction stores; +# each store writes its own field (ground_truth / predictions_model_a / predictions_model_b) +# into the SAME dataset, so a sample shows GT and both models side by side. +fo_session = FifyOneSession() + + +def _t(name, schema): + return Table(TableStoreDB(dbconn=DBCONN, name=name, data_sql_schema=schema, create_table=True)) + + +def _fo(field, additional_info_keys_in_sample=()): + return Table(FiftyOneImagesDataTableStore( + dataset=FIFTYONE_DATASET_NAME, + fo_session=fo_session, + fo_detections_label=field, + rm_only_fo_fields=True, + additional_info_keys_in_sample=list(additional_info_keys_in_sample), + primary_schema=[Column("image_name", String(255), primary_key=True)], + )) + + +catalog = Catalog( + { + # a load request: add a row here, then run `datapipe step --labels=stage=load run` + "load_request": _t("load_request", [ + Column("request_id", String, primary_key=True), + Column("n", Integer), + Column("offset", Integer), + Column("tag", String), + Column("darken", Float), + # pin every image of this batch to a subset (train/val); NULL = let the + # random split decide. Pinning is how the demo freezes val. + Column("subset", String), + ]), + # images uploaded by the load step (image_name = object basename) + "s3_images": _t("s3_images", [ + Column("image_name", String, primary_key=True), + Column("image_url", String), + ]), + # ground truth injected by the data-loading script (no Label Studio) + "image__ground_truth": _t("image__ground_truth", [ + Column("image_name", String, primary_key=True), + Column("bboxes", JSON), + Column("labels", JSON), + ]), + # train/val split + "image__subset": _t("image__subset", [ + Column("image_name", String, primary_key=True), + Column("subset_id", String, primary_key=True), + ]), + # per-batch pinned subset emitted by the load step (--subset); the split step + # honors these and only randomizes images without a hint. Freezes val. + "image__subset_hint": _t("image__subset_hint", [ + Column("image_name", String, primary_key=True), + Column("subset_id", String), + ]), + # --- tags ------------------------------------------------------------- + # tag dimension: numeric tag_id (surrogate key), tag_name (slug), tag_description (readable) + "tag": _t("tag", [ + Column("tag_id", Integer, primary_key=True), + Column("tag_name", String), + Column("tag_description", String), + ]), + "image__tag": _t("image__tag", [ + Column("image_name", String, primary_key=True), + Column("tag_id", Integer, primary_key=True), + ]), + # per-image metrics produced by the CountMetrics step (declared so the + # tag-metrics transform can read it as an input) + "detection_model_train__metrics_on_image": _t("detection_model_train__metrics_on_image", [ + Column("image_name", String, primary_key=True), + Column("detection_model_id", String, primary_key=True), + Column("subset_id", String, primary_key=True), + Column("calc__support", Integer), + Column("calc__TP", Integer), + Column("calc__FP", Integer), + Column("calc__FN", Integer), + Column("calc__iou_mean", Float), + ]), + # per-(model, tag, subset) metrics — the tag arc lives here + "tag_metrics": _t("tag_metrics", [ + Column("detection_model_id", String, primary_key=True), + Column("tag_id", Integer, primary_key=True), + Column("subset_id", String, primary_key=True), + Column("calc__images_support", Integer), + Column("calc__support", Integer), + Column("calc__TP", Integer), + Column("calc__FP", Integer), + Column("calc__FN", Integer), + Column("calc__precision", Float), + Column("calc__recall", Float), + Column("calc__f1_score", Float), + ]), + # --- FiftyOne (stage=fiftyone) --------------------------------------- + # images downloaded locally (FiftyOne renders from local files) + "local_images": _t("local_images", [ + Column("image_name", String(255), primary_key=True), + Column("local_path", String(1024)), + ]), + # GT boxes + per-sample subset/tag fields (filter tag=night in the FO App) + "fiftyone_annotations": _fo("ground_truth", additional_info_keys_in_sample=["subset", "tag"]), + # predictions of the baseline (model A) and the retrained model (model B) + "fiftyone_predictions_model_a": _fo("predictions_model_a"), + "fiftyone_predictions_model_b": _fo("predictions_model_b"), + } +) diff --git a/examples/detection_tags/detection/steps.py b/examples/detection_tags/detection/steps.py new file mode 100644 index 00000000..7eec59cb --- /dev/null +++ b/examples/detection_tags/detection/steps.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import os + +import fsspec +import numpy as np +import pandas as pd +from datapipe_ml.core.image_data import convert_df_with_bbox_to_df_with_image_data + +from coco_demo import CocoDemoSource, darken as demo_darken +from config import ( + LOCAL_IMAGES_DIR, + input_bucket, + input_image_url, + input_storage_options, + tag_id_for, + tag_name_for, +) + +# --- load step (demo: pulls images/GT from coco_demo; swap that module for real data) ----------- +def load_batch(df_request: pd.DataFrame): + """For each load_request row: pull images + ground truth from the demo source (coco_demo), + optionally darken (a tagged scenario), upload to object storage, and return (s3_images, + image__ground_truth, tag, image__tag, image__subset_hint). image_name is the object basename. + For REAL data, replace coco_demo — this step itself is source-agnostic plumbing.""" + s3_cols = ["image_name", "image_url"] + gt_cols = ["image_name", "bboxes", "labels"] + tag_cols = ["tag_id", "tag_name", "tag_description"] + it_cols = ["image_name", "tag_id"] + hint_cols = ["image_name", "subset_id"] + if df_request.empty: + return (pd.DataFrame(columns=s3_cols), pd.DataFrame(columns=gt_cols), + pd.DataFrame(columns=tag_cols), pd.DataFrame(columns=it_cols), + pd.DataFrame(columns=hint_cols)) + + source = CocoDemoSource() # demo data (COCO cat/dog); for real data swap coco_demo.py + fs = fsspec.filesystem("s3", **input_storage_options()) + bucket = input_bucket() + + s3_rows, gt_rows, tag_rows, tag_defs, hint_rows = [], [], [], {}, [] + for _, req in df_request.iterrows(): + n = int(req["n"]) + offset = int(req.get("offset") or 0) + # NB: when several requests load in one batch, a NULL in a numeric column arrives as NaN + # (not None) because pandas upcasts the column — so guard with pd.isna, else a base batch + # with darken=NULL would be "darkened" with gamma=NaN and produce garbage images. + tag = req.get("tag") + tag = None if pd.isna(tag) or str(tag) == "" else str(tag) + subset = req.get("subset") + subset = None if pd.isna(subset) or str(subset) == "" else str(subset) + darken = req.get("darken") + darken = None if pd.isna(darken) or str(darken) == "" else float(darken) + for fn in source.pool[offset: offset + n]: + stem, ext = os.path.splitext(fn) + raw, boxes, labels = source.fetch(fn) + if darken is not None: + name = f"{stem}__{tag or 'dark'}{ext}" + data = demo_darken(raw, darken) + else: + name, data = fn, raw + fs.pipe(f"{bucket}/images/{name}", data) + s3_rows.append({"image_name": name, "image_url": input_image_url(name)}) + gt_rows.append({"image_name": name, "bboxes": boxes, "labels": labels}) + if tag: + tid = tag_id_for(tag) + tag_rows.append({"image_name": name, "tag_id": tid}) + # tag_defs[name] = (numeric id, readable description); note darkening if any + desc = tag if darken is None else f"{tag} (low-light, gamma {darken})" + tag_defs[tag] = (tid, desc) + if subset: + hint_rows.append({"image_name": name, "subset_id": subset}) + + return ( + pd.DataFrame(s3_rows, columns=s3_cols), + pd.DataFrame(gt_rows, columns=gt_cols), + pd.DataFrame([{"tag_id": tid, "tag_name": nm, "tag_description": d} + for nm, (tid, d) in tag_defs.items()], columns=tag_cols), + pd.DataFrame(tag_rows, columns=it_cols), + pd.DataFrame(hint_rows, columns=hint_cols), + ) + + +# --- train/val split ------------------------------------------------------------ +def split_df_train_val(df: pd.DataFrame, subset_df: pd.DataFrame, hint_df: pd.DataFrame, + primary_keys: list[str], val_perc: float = 0.25, + random_seed: int = 42) -> pd.DataFrame: + """Assign a subset to every image that doesn't have one yet. An image whose subset the load + step pinned via ``image__subset_hint`` (batches loaded with ``--subset``) keeps that pinned + value; the rest are split train/val at random. Honoring the hint is what lets the demo FREEZE + val — load base-val + night-val up front (subset=val) and their assignment never changes when + train batches are added later, so a model's val metrics are stable across retrainings.""" + df__merged = pd.merge(df, subset_df, on=primary_keys, how="outer") + df__missing = df__merged[df__merged["subset_id"].isna()][primary_keys].copy() + if hint_df is not None and not hint_df.empty: + hint = hint_df[primary_keys + ["subset_id"]].drop_duplicates(subset=primary_keys) + df__missing = df__missing.merge(hint, on=primary_keys, how="left") + else: + df__missing["subset_id"] = np.nan + df__missing = df__missing.reset_index(drop=True) + unpinned = df__missing[df__missing["subset_id"].isna()] + df__val = unpinned.sample(frac=val_perc, random_state=random_seed) + df__missing.loc[unpinned.index, "subset_id"] = "train" + df__missing.loc[df__val.index, "subset_id"] = "val" + return pd.concat([subset_df, df__missing], ignore_index=True)[primary_keys + ["subset_id"]] + + +# --- tag metrics ---------------------------------------------------------------- +def compute_tag_metrics(df_metrics_on_image: pd.DataFrame, df_image_tag: pd.DataFrame) -> pd.DataFrame: + cols = ["detection_model_id", "tag_id", "subset_id", "calc__images_support", + "calc__support", "calc__TP", "calc__FP", "calc__FN", + "calc__precision", "calc__recall", "calc__f1_score"] + if df_metrics_on_image.empty or df_image_tag.empty: + return pd.DataFrame(columns=cols) + m = df_metrics_on_image.merge(df_image_tag[["image_name", "tag_id"]], on="image_name") + if m.empty: + return pd.DataFrame(columns=cols) + g = (m.groupby(["detection_model_id", "tag_id", "subset_id"], as_index=False) + .agg(images_support=("image_name", "count"), + tp=("calc__TP", "sum"), fp=("calc__FP", "sum"), fn=("calc__FN", "sum"))) + tp, fp, fn = g["tp"], g["fp"], g["fn"] + g["calc__precision"] = (tp / (tp + fp)).where((tp + fp) > 0, 0.0) + g["calc__recall"] = (tp / (tp + fn)).where((tp + fn) > 0, 0.0) + p, r = g["calc__precision"], g["calc__recall"] + g["calc__f1_score"] = (2 * p * r / (p + r)).where((p + r) > 0, 0.0) + g["calc__support"] = (tp + fn).astype(int) + g["calc__images_support"] = g["images_support"].astype(int) + g["calc__TP"] = tp.astype(int) + g["calc__FP"] = fp.astype(int) + g["calc__FN"] = fn.astype(int) + return g[cols] + + +# --- FiftyOne (stage=fiftyone) -------------------------------------------------- +# Same pattern as examples/e2e_template/image_detection: download images locally, +# then publish bbox rows into a FiftyOne dataset via FiftyOneImagesDataTableStore. +def download_images(s3_images_df: pd.DataFrame, image__image_path__name: str, + image__local_image_path__name: str) -> pd.DataFrame: + LOCAL_IMAGES_DIR.mkdir(parents=True, exist_ok=True) + local_paths = [] + for _, row in s3_images_df.iterrows(): + s3_path = row[image__image_path__name] + local_path = LOCAL_IMAGES_DIR / os.path.basename(str(s3_path)) + if not local_path.exists(): + # image_url is the public HTTP URL of the MinIO object (anonymous download), so open it + # plainly — passing S3 storage options here would hit the http filesystem with s3 kwargs. + with fsspec.open(s3_path, "rb") as f_src: + local_path.write_bytes(f_src.read()) + local_paths.append(str(local_path.resolve())) + s3_images_df[image__local_image_path__name] = local_paths + return s3_images_df[["image_name", image__local_image_path__name]] + + +def publish_gt_to_fiftyone(images_df: pd.DataFrame, gt_df: pd.DataFrame, subset_df: pd.DataFrame, + tag_df: pd.DataFrame, primary_keys: list[str], + image__image_path__name: str) -> pd.DataFrame: + """Publish ground-truth boxes and attach per-sample subset/tag (so you can filter tag=night).""" + pk = primary_keys[0] + df = convert_df_with_bbox_to_df_with_image_data( + df__with_bbox=pd.merge(gt_df, images_df, on=pk), + primary_keys=primary_keys, + image__image_path__name=image__image_path__name, + ) + subset_by_image = (subset_df.drop_duplicates(subset=[pk]).set_index(pk)["subset_id"].to_dict() + if not subset_df.empty else {}) + tagid_by_image = (tag_df.drop_duplicates(subset=[pk]).set_index(pk)["tag_id"].to_dict() + if not tag_df.empty else {}) + for _, row in df.iterrows(): + tid = tagid_by_image.get(row[pk]) + # FiftyOne sample fields must be strings; write the tag NAME (not the numeric id) + row["image_data"].additional_info = { + "subset": subset_by_image.get(row[pk]) or "none", + "tag": tag_name_for(tid) if tid is not None else "none", + } + return df + + +def publish_predictions_to_fiftyone(images_df: pd.DataFrame, predictions_df: pd.DataFrame, + slot: str, primary_keys: list[str], + image__image_path__name: str) -> pd.DataFrame: + """Publish one model's predictions into the slot's FiftyOne field. slot='model_a' is the + earliest-trained model (baseline), 'model_b' the next (retrained). Model ids are + timestamp-prefixed, so sorting them yields a stable A/B assignment across all images.""" + pk = primary_keys[0] + empty = pd.DataFrame(columns=primary_keys + ["image_data"]) + if predictions_df.empty: + return empty + ids = sorted(predictions_df["detection_model_id"].dropna().unique().tolist()) + if not ids: + return empty + slot_of = {mid: ("model_a" if i == 0 else "model_b") for i, mid in enumerate(ids)} + target_ids = [mid for mid, s in slot_of.items() if s == slot] + pred = predictions_df[predictions_df["detection_model_id"].isin(target_ids)] + if pred.empty: + return empty + return convert_df_with_bbox_to_df_with_image_data( + df__with_bbox=pd.merge(pred, images_df, on=pk), + primary_keys=primary_keys, + image__image_path__name=image__image_path__name, + ) diff --git a/examples/detection_tags/docker-compose.yml b/examples/detection_tags/docker-compose.yml new file mode 100644 index 00000000..a99d2d0e --- /dev/null +++ b/examples/detection_tags/docker-compose.yml @@ -0,0 +1,53 @@ +services: + postgres: + image: postgres:14 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: password + POSTGRES_DB: postgres + POSTGRES_HOST_AUTH_METHOD: trust + ports: + - "5432:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 10s + timeout: 5s + retries: 5 + + minio: + image: minio/minio:latest + command: server /data --console-address ":9001" + environment: + MINIO_ROOT_USER: minioadmin + MINIO_ROOT_PASSWORD: minioadmin + ports: + - "9000:9000" + - "9001:9001" + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + interval: 10s + timeout: 5s + retries: 5 + + minio-init: + image: minio/mc:latest + depends_on: + minio: + condition: service_healthy + entrypoint: > + /bin/sh -c " + mc alias set local http://minio:9000 minioadmin minioadmin && + mc mb --ignore-existing local/datapipe-tags && + mc anonymous set download local/datapipe-tags + " + + # FiftyOne dataset metadata store (the fiftyone stage). Point FIFTYONE_DATABASE_URI at it. + mongo: + image: mongo:7 + ports: + - "27017:27017" + healthcheck: + test: ["CMD", "mongosh", "--quiet", "--eval", "db.adminCommand('ping')"] + interval: 10s + timeout: 5s + retries: 5 diff --git a/examples/detection_tags/pyproject.toml b/examples/detection_tags/pyproject.toml new file mode 100644 index 00000000..a4b989f2 --- /dev/null +++ b/examples/detection_tags/pyproject.toml @@ -0,0 +1,34 @@ +[project] +name = "datapipe-detection-tags-example" +version = "0" +requires-python = ">=3.10,<3.13" +dependencies = [ + "datapipe-app", + "datapipe-ml[torch,fiftyone]", # fiftyone extra pulls FiftyOneImagesDataTableStore + cv_pipeliner.fiftyone + "torch==2.6.0", + "torchvision==0.21.0", + "stringzilla==4.4.0", # needed by albumentations + "polars-lts-cpu==1.33.1", # ultralytics/datapipe on pre-AVX2 CPUs (avoids SIGILL) + "pi-heif", # ultralytics image verification imports it + "fiftyone==1.17.0", # FiftyOne App + dataset (metadata in MongoDB) +] + +[[tool.uv.index]] +name = "pytorch-cu124" +url = "https://download.pytorch.org/whl/cu124" +explicit = true + +[tool.uv.sources] +torch = { index = "pytorch-cu124" } +torchvision = { index = "pytorch-cu124" } +datapipe-app = { path = "../../libs/datapipe-app", editable = true } +datapipe-ml = { path = "../../libs/datapipe-ml", editable = true } +datapipe-core = { path = "../../libs/datapipe-core", editable = true } + +[dependency-groups] +dev = [ + "pytest<8", + "requests", + "tqdm", + "boto3", +] diff --git a/examples/detection_tags/scripts/add_request.py b/examples/detection_tags/scripts/add_request.py new file mode 100644 index 00000000..790c3b4a --- /dev/null +++ b/examples/detection_tags/scripts/add_request.py @@ -0,0 +1,59 @@ +"""Add a load request, then run the load step: datapipe step --labels=stage=load run + +Fixed-val demo (freeze val up front, add the tagged train batch later): + + # loaded BEFORE training model A — val is frozen from the start + python ../scripts/add_request.py --id base-train --n 400 --offset 0 --subset train + python ../scripts/add_request.py --id base-val --n 100 --offset 400 --subset val + python ../scripts/add_request.py --id night-val --n 50 --offset 500 --subset val --tag night --darken 0.25 + # loaded BEFORE retraining model B (the tagged TRAIN batch — the fix) + python ../scripts/add_request.py --id night-train --n 50 --offset 550 --subset train --tag night --darken 0.25 + +--subset train|val pins every image of the batch to that subset so the random split never moves it. +Run from examples/detection_tags/detection (so `import config` resolves).""" +from __future__ import annotations + +import argparse +import sys + +import pandas as pd +from sqlalchemy import Column, Float, Integer, String + +sys.path.insert(0, ".") +import config # noqa: E402 +from datapipe.datatable import DataStore # noqa: E402 +from datapipe.store.database import TableStoreDB # noqa: E402 + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--id", required=True, help="request_id (unique per batch)") + ap.add_argument("--n", type=int, default=450) + ap.add_argument("--offset", type=int, default=0, help="skip the first OFFSET picked COCO images") + ap.add_argument("--tag", default=None) + ap.add_argument("--darken", type=float, default=None, help="gamma < 1 darkens (e.g. 0.1)") + ap.add_argument("--subset", default=None, choices=["train", "val"], + help="pin every image of this batch to a subset (freezes val); omit to random-split") + a = ap.parse_args() + + ds = DataStore(config.DBCONN, create_meta_table=True) + dt = ds.get_or_create_table("load_request", TableStoreDB( + dbconn=config.DBCONN, name="load_request", + data_sql_schema=[ + Column("request_id", String, primary_key=True), + Column("n", Integer), Column("offset", Integer), + Column("tag", String), Column("darken", Float), + Column("subset", String), + ], create_table=True)) + dt.store_chunk(pd.DataFrame([{ + "request_id": a.id, "n": a.n, "offset": a.offset, "tag": a.tag, + "darken": a.darken, "subset": a.subset, + }])) + print(f"added request {a.id}: n={a.n} offset={a.offset} tag={a.tag} " + f"darken={a.darken} subset={a.subset}") + print("now run: datapipe step --labels=stage=load run") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/e2e_template/README.md b/examples/e2e_template/README.md index cd1b5ed1..6ba40e87 100644 --- a/examples/e2e_template/README.md +++ b/examples/e2e_template/README.md @@ -1,5 +1,8 @@ # Datapipe E2E Template +**Claude Code skill:** `/setup-e2e-template` — auto-loads when relevant, or invoke it directly. It +carries the prerequisites, env knobs, and gotchas below. + This example contains end-to-end Datapipe pipeline templates: - `image_detection` for bbox detection. diff --git a/examples/e2e_template/image_detection/config.py b/examples/e2e_template/image_detection/config.py index f71e07a1..7e348998 100644 --- a/examples/e2e_template/image_detection/config.py +++ b/examples/e2e_template/image_detection/config.py @@ -24,8 +24,8 @@ - """ diff --git a/examples/e2e_template/pyproject.toml b/examples/e2e_template/pyproject.toml index f3a39ebd..02b9ca70 100644 --- a/examples/e2e_template/pyproject.toml +++ b/examples/e2e_template/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "stringzilla==4.4.0", # needed for albumentations "fiftyone==1.17.0", "polars-lts-cpu==1.33.1", # for ultralytics on old CPUs + "pi-heif", # ultralytics image verification imports it ] [[tool.uv.index]] diff --git a/examples/e2e_template/scripts/seed_sample_data.py b/examples/e2e_template/scripts/seed_sample_data.py index b25df7ef..f8d6b910 100644 --- a/examples/e2e_template/scripts/seed_sample_data.py +++ b/examples/e2e_template/scripts/seed_sample_data.py @@ -162,7 +162,7 @@ def upload_images(local_paths: list[Path]) -> None: def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Download sample COCO images and upload them to local MinIO.") - parser.add_argument("--detection-limit", type=int, default=10, help="Number of cat/dog images") + parser.add_argument("--detection-limit", type=int, default=120, help="Number of cat/dog images") parser.add_argument("--keypoints-limit", type=int, default=10, help="Number of person keypoint images") parser.add_argument("--skip-download", action="store_true", help="Only upload files already in sample_data/") parser.add_argument("--skip-upload", action="store_true", help="Only download images locally") diff --git a/examples/embedder_fiftyone/README.md b/examples/embedder_fiftyone/README.md index 3d3d5d20..2cbd3ce5 100644 --- a/examples/embedder_fiftyone/README.md +++ b/examples/embedder_fiftyone/README.md @@ -1,5 +1,7 @@ # Embedder + FiftyOne Pipeline +**Claude Code skill:** `/setup-embedder-fiftyone` — auto-loads when relevant, or invoke it directly. + Datapipe example for: - loading images + class labels (local folder or FiftyOne zoo fallback) - running multiple embedders diff --git a/examples/sam_cvat/README.md b/examples/sam_cvat/README.md index 61ea0556..5401f67e 100644 --- a/examples/sam_cvat/README.md +++ b/examples/sam_cvat/README.md @@ -1,5 +1,7 @@ # SAM3 + CVAT Pipeline +**Claude Code skill:** `/setup-sam-cvat` — auto-loads when relevant, or invoke it directly. + Datapipe example for: - loading images (local folder or Hugging Face dataset fallback)