From 4ff31557a890589d63d3ce1a76e4e3cfe0b7b9a9 Mon Sep 17 00:00:00 2001 From: Julian Cuni Date: Mon, 7 Sep 2026 14:34:13 +0200 Subject: [PATCH] =?UTF-8?q?feat(trainer):=20training=20from=20the=20collec?= =?UTF-8?q?tor=20UI=20=E2=80=94=20the=20trainer=20becomes=20a=20job=20serv?= =?UTF-8?q?ice,=20the=20review=20page=20gains=20a=20Training=20section?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Trainer: `parking-trainer serve` — a stdlib HTTP job API on the compose network (never published): /health, /readiness, /versions, /versions//report, /jobs. One job at a time; each job runs the CLI as a subprocess with its output captured, state + log persisted under /out/jobs/ so a restart keeps history. `publish` takes its URL from TRAINER_PUBLISH_URL. Dockerfile: CMD serve, EXPOSE 8091, healthcheck. Collector: COLLECTOR_TRAINER_URL + /api/training/{status,jobs,jobs/:id,versions/:v/report} — a reviewer-gated proxy that forwards a fixed set of paths and whitelisted knobs and passes the trainer's status codes through (409 while a job runs; 503 unconfigured, 502 unreachable). /review gains the Training section: labels per class vs the minimum with Train disabled until two classes clear it, mode / backbone / floor, the running job's live log, the versions with Report / Evaluate / Publish (publish confirms), and the reminder that pinning stays a git commit. Fixed on the way: an apostrophe in the page's inline script broke the whole page — a test now parses the script. Compose: `trainer` is a service (restart: unless-stopped, read-only data volume, its own trainer-out volume), the `train` profile and TRAINER_OUT are gone; the Docker-socket route was rejected (root on the host for a service booths upload to). Verified with both images running together: a Train started through the proxy finished, version and report came back, the page rendered. Wiki: bodytype-classifier-training (loop, running it, operating notes superseded), vision-review-outbox, fleet-deployment-komodo, log. Claude-Session: https://claude.ai/code/session_01FWncR69HgGPuei1dLrW3cU --- apps/collector/src/app.test.ts | 2 +- apps/collector/src/app.ts | 51 +++ apps/collector/src/config.ts | 4 + apps/collector/src/index.ts | 2 +- apps/collector/src/review-page.ts | 114 ++++++ apps/collector/src/training.test.ts | 126 ++++++ apps/trainer/Dockerfile | 20 +- apps/trainer/README.md | 12 +- apps/trainer/tests/test_server.py | 111 +++++ apps/trainer/trainer/cli.py | 28 +- apps/trainer/trainer/server.py | 382 ++++++++++++++++++ docker-compose.collector.yml | 29 +- komodo/resources.toml | 7 +- wiki/concepts/vision-review-outbox.md | 11 +- .../decisions/bodytype-classifier-training.md | 68 ++-- wiki/decisions/fleet-deployment-komodo.md | 9 +- wiki/log.md | 14 + 17 files changed, 915 insertions(+), 75 deletions(-) create mode 100644 apps/collector/src/training.test.ts create mode 100644 apps/trainer/tests/test_server.py create mode 100644 apps/trainer/trainer/server.py diff --git a/apps/collector/src/app.test.ts b/apps/collector/src/app.test.ts index 3635e98..7fe416d 100644 --- a/apps/collector/src/app.test.ts +++ b/apps/collector/src/app.test.ts @@ -16,7 +16,7 @@ const basic = "Basic " + Buffer.from(`${REVIEWER.user}:${REVIEWER.pass}`).toStri beforeEach(async () => { dir = await mkdtemp(path.join(tmpdir(), "collector-")); - app = await buildCollector({ host: "127.0.0.1", port: 0, dataDir: dir, boothTokens: TOKENS, reviewer: REVIEWER }, { dbFile: ":memory:" }); + app = await buildCollector({ host: "127.0.0.1", port: 0, dataDir: dir, boothTokens: TOKENS, reviewer: REVIEWER, trainerUrl: null }, { dbFile: ":memory:" }); await app.ready(); }); afterEach(async () => { diff --git a/apps/collector/src/app.ts b/apps/collector/src/app.ts index 739e489..73dc736 100644 --- a/apps/collector/src/app.ts +++ b/apps/collector/src/app.ts @@ -15,6 +15,8 @@ import { reviewPage } from "./review-page.js"; // /review + /api/* the reviewer's screen (HTTP Basic, one login) // GET /export/labels.csv the training set: reviewed, usable rows (crops sit beside it on // the volume, so the trainer on this host reads them directly) +// /api/training/* the Training section: a thin proxy to the trainer's job API on +// the compose network (never published), behind the reviewer login // It deliberately has no fleet features and no path back into a booth. /** The package's `meta` part, as the booth sends it (review-outbox.ts). */ @@ -230,6 +232,55 @@ export async function buildCollector(cfg: CollectorConfig, opts: { dbFile?: stri return reply.type("text/csv; charset=utf-8").header("content-disposition", 'attachment; filename="labels.csv"').send([head, ...lines].join("\n") + "\n"); }); + // --- Training (proxy to the trainer's job API) ---------------------------------------- + // The trainer is a sibling container reading the same volume; it is reachable only on the + // compose network, so the reviewer's login here is the only gate. The proxy forwards a + // fixed set of paths and passes the trainer's status codes through (409 = a job runs). + const trainer = cfg.trainerUrl; + async function viaTrainer(reply: FastifyReply, tpath: string, init?: RequestInit): Promise { + if (!trainer) return reply.code(503).send({ error: "trainer not configured" }); + let r: Response; + try { + r = await fetch(trainer + tpath, { ...init, signal: AbortSignal.timeout(15_000) }); + } catch (err) { + return reply.code(502).send({ error: `trainer unreachable: ${(err as Error).message}` }); + } + const ctype = r.headers.get("content-type") ?? "application/json"; + return reply.code(r.status).type(ctype).send(Buffer.from(await r.arrayBuffer())); + } + app.get("/api/training/status", { preHandler: requireReviewer }, async (_req, reply) => { + if (!trainer) return { configured: false }; + try { + const get = async (p: string) => { + const r = await fetch(trainer + p, { signal: AbortSignal.timeout(15_000) }); + if (!r.ok) throw new Error(`${p} → HTTP ${r.status}`); + return r.json() as Promise>; + }; + const [health, readiness, versions, jobs] = await Promise.all([get("/health"), get("/readiness"), get("/versions"), get("/jobs")]); + return { configured: true, reachable: true, health, readiness, versions: versions.versions, jobs: jobs.jobs, current: jobs.current }; + } catch (err) { + return reply.code(200).send({ configured: true, reachable: false, error: (err as Error).message }); + } + }); + app.post<{ Body: Record }>("/api/training/jobs", { preHandler: requireReviewer }, async (req, reply) => { + const b = req.body && typeof req.body === "object" ? req.body : {}; + const kind = b.kind; + if (kind !== "train" && kind !== "evaluate" && kind !== "publish") return reply.code(400).send({ error: "kind must be train, evaluate or publish" }); + // Only the knobs the UI offers cross over; the trainer validates their values. + const allowed = ["kind", "mode", "backbone", "minAccuracy", "minPerClass", "epochs", "version"]; + const body: Record = {}; + for (const k of allowed) if (b[k] !== undefined) body[k] = b[k]; + return viaTrainer(reply, "/jobs", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) }); + }); + app.get<{ Params: { id: string } }>("/api/training/jobs/:id", { preHandler: requireReviewer }, async (req, reply) => { + if (!ID_RE.test(req.params.id)) return reply.code(400).send({ error: "bad job id" }); + return viaTrainer(reply, `/jobs/${encodeURIComponent(req.params.id)}`); + }); + app.get<{ Params: { v: string } }>("/api/training/versions/:v/report", { preHandler: requireReviewer }, async (req, reply) => { + if (!ID_RE.test(req.params.v)) return reply.code(400).send({ error: "bad version" }); + return viaTrainer(reply, `/versions/${encodeURIComponent(req.params.v)}/report`); + }); + return app; } diff --git a/apps/collector/src/config.ts b/apps/collector/src/config.ts index 77e579b..78791c5 100644 --- a/apps/collector/src/config.ts +++ b/apps/collector/src/config.ts @@ -6,6 +6,9 @@ export interface CollectorConfig { readonly boothTokens: ReadonlyMap; /** The single reviewer login; null = review screen and export refuse (503). */ readonly reviewer: { readonly user: string; readonly pass: string } | null; + /** The trainer's job API on the compose network (http://trainer:8091); null = the + * Training section is hidden and /api/training/* answers 503. */ + readonly trainerUrl: string | null; } /** "booth-7:abc,booth-9:def" (commas, whitespace or newlines between pairs). */ @@ -32,5 +35,6 @@ export function configFromEnv(env: NodeJS.ProcessEnv = process.env): CollectorCo dataDir: env.COLLECTOR_DATA_DIR ?? "/data", boothTokens: parseBoothTokens(env.COLLECTOR_BOOTH_TOKENS ?? ""), reviewer: user && pass.length >= 8 ? { user, pass } : null, + trainerUrl: (env.COLLECTOR_TRAINER_URL ?? "").trim().replace(/\/+$/, "") || null, }; } diff --git a/apps/collector/src/index.ts b/apps/collector/src/index.ts index 512a57d..6592f30 100644 --- a/apps/collector/src/index.ts +++ b/apps/collector/src/index.ts @@ -5,7 +5,7 @@ const cfg = configFromEnv(); const app = await buildCollector(cfg); if (cfg.boothTokens.size === 0) app.log.warn("COLLECTOR_BOOTH_TOKENS is empty — no booth can ingest"); if (!cfg.reviewer) app.log.warn("COLLECTOR_REVIEWER_USER/PASS not set — the review screen and export refuse"); -app.log.info(`collector: ${cfg.boothTokens.size} booth token(s), data in ${cfg.dataDir}`); +app.log.info(`collector: ${cfg.boothTokens.size} booth token(s), data in ${cfg.dataDir}, trainer ${cfg.trainerUrl ?? "not configured"}`); await app.listen({ host: cfg.host, port: cfg.port }); const stop = async () => { diff --git a/apps/collector/src/review-page.ts b/apps/collector/src/review-page.ts index 9f9fd11..04656c0 100644 --- a/apps/collector/src/review-page.ts +++ b/apps/collector/src/review-page.ts @@ -35,6 +35,16 @@ export function reviewPage(): string { td, th { text-align:left; padding:.2rem .5rem; border-bottom:1px solid #2a2a2a; } th { color:var(--muted); font-weight:normal; font-size:.75rem; text-transform:uppercase; letter-spacing:.06em; } kbd { background:#2a2a2a; border:1px solid #444; border-radius:3px; padding:0 .3rem; font-size:.75rem; } + h2 { font-size:.8rem; letter-spacing:.08em; text-transform:uppercase; color:var(--amber); margin:0 0 .6rem; } + .row { display:flex; flex-wrap:wrap; gap:.6rem; align-items:center; } + select, input { background:#2a2a2a; color:var(--text); border:1px solid #444; border-radius:4px; padding:.4rem .5rem; font:inherit; } + input[type=number] { width:5rem; } + label { color:var(--muted); font-size:.8rem; } + pre { background:#0d0d0d; border:1px solid #2a2a2a; border-radius:4px; padding:.6rem; max-height:22rem; overflow:auto; font-size:.75rem; white-space:pre-wrap; margin:.6rem 0 0; } + .ok { color:var(--green); } + .bad { color:var(--red); } + button:disabled { opacity:.45; cursor:not-allowed; } + button.small { padding:.25rem .5rem; font-size:.75rem; } @@ -46,6 +56,10 @@ export function reviewPage(): string {
boothoperatorreviewedagreedisagreeunusable
+

Keys: 1–9, 0 pick a class in order · u unusable · s skip. Skipped items come back after a reload. Your verdict is the training label; the operator's pick is only compared against it.

`; diff --git a/apps/collector/src/training.test.ts b/apps/collector/src/training.test.ts new file mode 100644 index 0000000..3ead0d4 --- /dev/null +++ b/apps/collector/src/training.test.ts @@ -0,0 +1,126 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { buildCollector, type CollectorApp } from "./app.js"; + +// The Training section's proxy: reviewer-gated, forwards a fixed set of paths to the +// trainer's job API, passes its status codes through, and degrades cleanly when the trainer +// is not configured or not reachable. The trainer is faked with a bare node http server. + +const REVIEWER = { user: "julian", pass: "review-pass-123" }; +const basic = "Basic " + Buffer.from(`${REVIEWER.user}:${REVIEWER.pass}`).toString("base64"); + +let dir: string; +let fake: Server; +let fakeUrl: string; +let seen: { method: string; url: string; body: string }[]; +let app: CollectorApp; + +async function start(trainerUrl: string | null): Promise { + app = await buildCollector({ host: "127.0.0.1", port: 0, dataDir: dir, boothTokens: new Map(), reviewer: REVIEWER, trainerUrl }, { dbFile: ":memory:" }); + await app.ready(); +} + +beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), "collector-")); + seen = []; + fake = createServer((req: IncomingMessage, res: ServerResponse) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + seen.push({ method: req.method ?? "", url: req.url ?? "", body }); + const json = (code: number, obj: unknown) => { + res.writeHead(code, { "content-type": "application/json" }); + res.end(JSON.stringify(obj)); + }; + if (req.url === "/health") return json(200, { ok: true, busy: false }); + if (req.url === "/readiness") return json(200, { ready: false, labelled: { total: 3 } }); + if (req.url === "/versions") return json(200, { versions: [{ version: "v1", written: true }] }); + if (req.url === "/jobs" && req.method === "GET") return json(200, { jobs: [{ id: "j1" }], current: null }); + if (req.url === "/jobs" && req.method === "POST") return body.includes('"busy"') ? json(409, { error: "a job is already running" }) : json(202, { id: "j2", status: "running" }); + if (req.url === "/jobs/j1") return json(200, { id: "j1", status: "done", log: "ok" }); + if (req.url === "/versions/v1/report") { + res.writeHead(200, { "content-type": "text/markdown; charset=utf-8" }); + return res.end("# Body-type classifier v1\n"); + } + return json(404, { error: "not found" }); + }); + }); + await new Promise((r) => fake.listen(0, "127.0.0.1", r)); + const a = fake.address() as { port: number }; + fakeUrl = `http://127.0.0.1:${a.port}`; +}); +afterEach(async () => { + await app?.close(); + await new Promise((r) => fake.close(() => r())); + await rm(dir, { recursive: true, force: true }); +}); + +describe("review page script", () => { + it("parses as JavaScript (an apostrophe in a template literal once broke the whole page)", async () => { + const { reviewPage } = await import("./review-page.js"); + const html = reviewPage(); + const script = html.slice(html.indexOf("")); + expect(() => new Function(script)).not.toThrow(); + }); +}); + +describe("training proxy", () => { + it("is hidden when no trainer is configured", async () => { + await start(null); + const s = await app.inject({ method: "GET", url: "/api/training/status", headers: { authorization: basic } }); + expect(s.json()).toEqual({ configured: false }); + const j = await app.inject({ method: "POST", url: "/api/training/jobs", headers: { authorization: basic }, payload: { kind: "train" } }); + expect(j.statusCode).toBe(503); + }); + + it("aggregates status and forwards jobs and reports behind the reviewer login", async () => { + await start(fakeUrl); + expect((await app.inject({ method: "GET", url: "/api/training/status" })).statusCode).toBe(401); + const s = await app.inject({ method: "GET", url: "/api/training/status", headers: { authorization: basic } }); + expect(s.statusCode).toBe(200); + const body = s.json(); + expect(body.configured).toBe(true); + expect(body.reachable).toBe(true); + expect(body.readiness.labelled.total).toBe(3); + expect(body.versions[0].version).toBe("v1"); + expect(body.jobs[0].id).toBe("j1"); + + const j = await app.inject({ + method: "POST", + url: "/api/training/jobs", + headers: { authorization: basic }, + payload: { kind: "train", mode: "features", minAccuracy: 0.9, secret: "nope", version: "v2" }, + }); + expect(j.statusCode).toBe(202); + expect(j.json().id).toBe("j2"); + const posted = seen.find((r) => r.method === "POST")!; + expect(JSON.parse(posted.body)).toEqual({ kind: "train", mode: "features", minAccuracy: 0.9, version: "v2" }); // unknown keys dropped + + const busy = await app.inject({ method: "POST", url: "/api/training/jobs", headers: { authorization: basic }, payload: { kind: "evaluate", version: "busy" } }); + expect(busy.statusCode).toBe(409); // the trainer's answer passes through + + const bad = await app.inject({ method: "POST", url: "/api/training/jobs", headers: { authorization: basic }, payload: { kind: "rm-rf" } }); + expect(bad.statusCode).toBe(400); + + const one = await app.inject({ method: "GET", url: "/api/training/jobs/j1", headers: { authorization: basic } }); + expect(one.json().status).toBe("done"); + expect((await app.inject({ method: "GET", url: "/api/training/jobs/..%2Fx", headers: { authorization: basic } })).statusCode).toBe(400); + + const rep = await app.inject({ method: "GET", url: "/api/training/versions/v1/report", headers: { authorization: basic } }); + expect(rep.statusCode).toBe(200); + expect(rep.headers["content-type"]).toContain("text/markdown"); + expect(rep.body).toContain("# Body-type classifier v1"); + }); + + it("reports an unreachable trainer without failing the page", async () => { + await start("http://127.0.0.1:9"); // nothing listens on the discard port + const s = await app.inject({ method: "GET", url: "/api/training/status", headers: { authorization: basic } }); + expect(s.statusCode).toBe(200); + expect(s.json().reachable).toBe(false); + const j = await app.inject({ method: "POST", url: "/api/training/jobs", headers: { authorization: basic }, payload: { kind: "train" } }); + expect(j.statusCode).toBe(502); + }); +}); diff --git a/apps/trainer/Dockerfile b/apps/trainer/Dockerfile index 475177e..62d6814 100644 --- a/apps/trainer/Dockerfile +++ b/apps/trainer/Dockerfile @@ -1,9 +1,11 @@ # syntax=docker/dockerfile:1.7 -# Parking TRAINER image: the phase-B body-type classifier job. Build CONTEXT is apps/trainer -# (self-contained Python package). A ONE-OFF JOB on the reviewer's host (art-docker-station), -# never a booth service: it reads the wash collector's volume (collector.sqlite + crops/) -# and writes a versioned model folder. CPU-only PyTorch — the host has no usable GPU and a -# few thousand crops train in minutes/an hour on four Xeon cores. +# Parking TRAINER image: the phase-B body-type classifier. Build CONTEXT is apps/trainer +# (self-contained Python package). Runs on the reviewer's host (art-docker-station) beside +# the collector, never on a booth: by default it SERVES the job API the collector's Training +# section drives (`serve`); the same image runs the CLI one-off (`train`, `inspect`, …). It +# reads the wash collector's volume (collector.sqlite + crops/) and writes versioned model +# folders. CPU-only PyTorch — the host has no usable GPU and a few thousand crops train in +# minutes/an hour on four Xeon cores. # See wiki/decisions/bodytype-classifier-training.md. FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS base @@ -37,7 +39,11 @@ RUN useradd --system --create-home --uid 999 trainer \ USER trainer ENV TRAINER_DATA_DIR=/data \ - TRAINER_OUT_DIR=/out + TRAINER_OUT_DIR=/out \ + TRAINER_PORT=8091 VOLUME ["/out"] +EXPOSE 8091 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://localhost:8091/health').status==200 else 1)" || exit 1 ENTRYPOINT ["uv", "run", "--no-sync", "parking-trainer"] -CMD ["inspect"] +CMD ["serve"] diff --git a/apps/trainer/README.md b/apps/trainer/README.md index 4c2ad57..61fadd9 100644 --- a/apps/trainer/README.md +++ b/apps/trainer/README.md @@ -25,12 +25,12 @@ A passing run writes `//`: | `report.md` | the human report: accuracy, per-class recall/precision, confusion matrix, dropped classes, loss weights | | `metrics.json` | the same numbers, machine-readable | -On the reviewer's host (the `wash-collector` stack): - -``` -docker compose -f docker-compose.collector.yml --profile train run --rm trainer inspect -docker compose -f docker-compose.collector.yml --profile train run --rm trainer train --min-accuracy 0.85 -``` +On the reviewer's host the image runs `serve` as the `trainer` service of the +`wash-collector` stack: a job API (`/health`, `/readiness`, `/versions`, `/jobs`) on the compose +network that the collector's **Training section** (`/review`) drives — readiness, Train / +Evaluate / Publish, reports and logs. Jobs run as subprocesses of the CLI, one at a time; state +and logs persist under `/out/jobs/`. The CLI stays for debugging: +`docker compose -f docker-compose.collector.yml exec trainer parking-trainer inspect`. Local dev: `uv sync --extra train` (CPU torch, ~200 MB), `uv run pytest -q`. The test suite runs without the extra (torch tests skip), matching CI. diff --git a/apps/trainer/tests/test_server.py b/apps/trainer/tests/test_server.py new file mode 100644 index 0000000..75bfa60 --- /dev/null +++ b/apps/trainer/tests/test_server.py @@ -0,0 +1,111 @@ +"""The job API: readiness, one job at a time, subprocess jobs with persisted logs, versions.""" + +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer +from pathlib import Path + +import pytest + +from trainer.server import Handler, Jobs, readiness, versions, wait_idle + + +@pytest.fixture +def api(collector_dir: Path, tmp_path: Path): # type: ignore[no-untyped-def] + out = tmp_path / "out" + Handler.jobs = Jobs(collector_dir, out, "https://example.invalid/pkg", "tok") + httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + t = threading.Thread(target=httpd.serve_forever, daemon=True) + t.start() + base = f"http://127.0.0.1:{httpd.server_address[1]}" + + def call(method: str, path: str, body: dict | None = None): # type: ignore[no-untyped-def] + req = urllib.request.Request(base + path, method=method) + data = None + if body is not None: + data = json.dumps(body).encode() + req.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(req, data=data, timeout=10) as r: + raw = r.read() + return r.status, ( + json.loads(raw) if r.headers.get_content_type() == "application/json" else raw.decode() + ) + except urllib.error.HTTPError as e: + return e.code, json.loads(e.read() or b"{}") + + yield call, out + httpd.shutdown() + httpd.server_close() + + +def test_readiness_and_empty_versions(api) -> None: # type: ignore[no-untyped-def] + call, _ = api + code, r = call("GET", "/readiness") + assert code == 200 and r["ready"] is True and r["run"]["classes"] == ["sedan", "suv", "van"] + assert r["defaults"]["minAccuracy"] == 0.85 and "finetune" in r["modes"] + assert call("GET", "/versions") == (200, {"versions": []}) + assert call("GET", "/health")[1]["busy"] is False + assert readiness(Path("/nonexistent"))["ready"] is False + + +def test_evaluate_job_runs_as_a_subprocess_and_is_recorded(api) -> None: # type: ignore[no-untyped-def] + call, out = api + code, job = call("POST", "/jobs", {"kind": "evaluate", "version": "nope"}) + assert code == 202 and job["status"] == "running" and job["kind"] == "evaluate" + wait_idle(Handler.jobs) + code, j = call("GET", f"/jobs/{job['id']}") + assert code == 200 and j["status"] == "failed" and j["exitCode"] == 1 + assert "evaluate --data" in j["log"] and "nope" in j["log"] + assert (out / "jobs" / f"{job['id']}.json").is_file() and (out / "jobs" / f"{job['id']}.log").is_file() + code, lst = call("GET", "/jobs") + assert code == 200 and lst["jobs"][0]["id"] == job["id"] and lst["current"] is None + + +def test_bad_requests(api) -> None: # type: ignore[no-untyped-def] + call, _ = api + assert call("POST", "/jobs", {"kind": "nuke"})[0] == 400 + assert call("POST", "/jobs", {"kind": "train", "mode": "magic"})[0] == 400 + assert call("POST", "/jobs", {"kind": "evaluate", "version": "../etc"})[0] == 400 + assert call("POST", "/jobs", {"kind": "publish", "version": "v1", "url": "ftp://x"})[0] == 400 + assert call("GET", "/versions/../x/report")[0] == 400 + assert call("GET", "/versions/v9/report")[0] == 404 + assert call("GET", "/jobs/nope")[0] == 404 + assert call("GET", "/nothing")[0] == 404 + + +def test_train_job_then_versions_and_report(api) -> None: # type: ignore[no-untyped-def] + pytest.importorskip("torch") + call, out = api + body = {"kind": "train", "mode": "features", "minAccuracy": 0.0, "epochs": 100, "version": "vapi"} + # The test-only flags are not offered by the API; inject them via the CLI args the runner builds. + orig = Jobs._argv + + def patched(self, kind, a): # type: ignore[no-untyped-def] + argv = orig(self, kind, a) + return argv + ["--no-pretrained", "--input-size", "64", "--no-cache"] if kind == "train" else argv + + Jobs._argv = patched # type: ignore[method-assign] + try: + code, job = call("POST", "/jobs", body) + assert code == 202 + assert call("POST", "/jobs", {"kind": "evaluate", "version": "vapi"})[0] == 409 # one at a time + wait_idle(Handler.jobs, 120) + finally: + Jobs._argv = orig # type: ignore[method-assign] + code, j = call("GET", f"/jobs/{job['id']}") + assert j["status"] == "done" and "MODEL WRITTEN" in j["log"] + code, v = call("GET", "/versions") + assert code == 200 and v["versions"][0]["version"] == "vapi" and v["versions"][0]["written"] is True + assert v["versions"][0]["classes"] == ["sedan", "suv", "van"] and v["versions"][0]["accuracy"] >= 0.9 + code, report = call("GET", "/versions/vapi/report") + assert code == 200 and report.startswith("# Body-type classifier vapi") + assert versions(out)[0]["floor"] == 0.0 + # evaluate on the written model now succeeds + code, job2 = call("POST", "/jobs", {"kind": "evaluate", "version": "vapi"}) + wait_idle(Handler.jobs) + assert call("GET", f"/jobs/{job2['id']}")[1]["status"] == "done" diff --git a/apps/trainer/trainer/cli.py b/apps/trainer/trainer/cli.py index f5af89a..5f3d62e 100644 --- a/apps/trainer/trainer/cli.py +++ b/apps/trainer/trainer/cli.py @@ -272,6 +272,9 @@ def cmd_publish(a: argparse.Namespace) -> int: _log(f"{f} missing — nothing to publish (a run below the floor writes no model)") return 1 version = Sidecar.read(d / SIDECAR_FILE).version + if not a.url: + _log("no publish url: pass --url or set TRAINER_PUBLISH_URL") + return 1 token = a.token or os.environ.get("TRAINER_PUBLISH_TOKEN", "") if not token: _log("no token: pass --token or set TRAINER_PUBLISH_TOKEN") @@ -289,6 +292,20 @@ def cmd_publish(a: argparse.Namespace) -> int: return 0 +def cmd_serve(a: argparse.Namespace) -> int: + from .server import serve + + serve( + a.data, + a.out, + a.host, + a.port, + os.environ.get("TRAINER_PUBLISH_URL", ""), + os.environ.get("TRAINER_PUBLISH_TOKEN", ""), + ) + return 0 + + # ---------------------------------------------------------------------------------- @@ -366,10 +383,19 @@ def build_parser() -> argparse.ArgumentParser: u = sub.add_parser("publish", help="PUT a version folder to a Gitea generic package") u.add_argument("dir", type=Path, help="the / folder a passing run wrote") u.add_argument( - "--url", required=True, help="https:///api/packages//generic/parking-bodytype" + "--url", + default=os.environ.get("TRAINER_PUBLISH_URL", ""), + help="https:///api/packages//generic/parking-bodytype (or TRAINER_PUBLISH_URL)", ) u.add_argument("--token", default="", help="Gitea token with package:write (or TRAINER_PUBLISH_TOKEN)") u.set_defaults(fn=cmd_publish) + + s = sub.add_parser("serve", help="the job API the collector's Training section talks to") + data_args(s) + s.add_argument("--out", type=Path, default=Path(os.environ.get("TRAINER_OUT_DIR", "/out"))) + s.add_argument("--host", default=os.environ.get("TRAINER_HOST", "0.0.0.0")) + s.add_argument("--port", type=int, default=int(os.environ.get("TRAINER_PORT", "8091"))) + s.set_defaults(fn=cmd_serve) return p diff --git a/apps/trainer/trainer/server.py b/apps/trainer/trainer/server.py new file mode 100644 index 0000000..ecead02 --- /dev/null +++ b/apps/trainer/trainer/server.py @@ -0,0 +1,382 @@ +"""`parking-trainer serve` — the job API behind the collector's Training section. + +A tiny stdlib HTTP server (no framework, no extra deps) on the compose-internal network, +never published: the collector proxies to it behind the reviewer's login. One job at a +time; each job is the CLI run as a SUBPROCESS (`python -m trainer.cli …`) with its output +captured to a log file — torch's memory goes away with the process, and a crashing job +cannot take the service down. Job state + logs persist under `/jobs/` so a restart +still shows history. + + GET /health {ok, busy, version} + GET /readiness what `inspect` prints (+ the defaults the UI offers) + GET /versions every // folder: written?, metrics, sidecar + GET /versions//report report.md (text/markdown) + GET /jobs recent jobs, newest first + GET /jobs/ one job incl. the log tail + POST /jobs {kind: train|evaluate|publish, …args} → 202 {id} | 409 busy +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import threading +import time +import uuid +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +from .cli import METRICS_FILE, MODEL_FILE, REPORT_FILE, SIDECAR_FILE +from .data import load_labelled, make_split, summarise +from .preprocess import Sidecar + +_VERSION_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +BACKBONES = ("resnet18", "mobilenet_v3_small", "efficientnet_b0") +MODES = ("features", "finetune") +LOG_TAIL_BYTES = 16_000 + + +def _now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +class Jobs: + """The single-slot job runner. `start` refuses while one runs.""" + + def __init__(self, data_dir: Path, out_dir: Path, publish_url: str, publish_token: str) -> None: + self.data_dir = data_dir + self.out_dir = out_dir + self.publish_url = publish_url + self.publish_token = publish_token + self.jobs_dir = out_dir / "jobs" + self.jobs_dir.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._current: dict[str, Any] | None = None + self._proc: subprocess.Popen[bytes] | None = None + + # ---- state ----------------------------------------------------------------------- + def _write(self, job: dict[str, Any]) -> None: + (self.jobs_dir / f"{job['id']}.json").write_text(json.dumps(job, indent=2)) + + def _read(self, job_id: str) -> dict[str, Any] | None: + p = self.jobs_dir / f"{job_id}.json" + if not p.is_file(): + return None + return json.loads(p.read_text()) # type: ignore[no-any-return] + + def log_tail(self, job_id: str) -> str: + p = self.jobs_dir / f"{job_id}.log" + if not p.is_file(): + return "" + size = p.stat().st_size + with p.open("rb") as f: + if size > LOG_TAIL_BYTES: + f.seek(size - LOG_TAIL_BYTES) + return f.read().decode("utf-8", "replace") + + @property + def busy(self) -> bool: + return self._current is not None + + def current(self) -> dict[str, Any] | None: + return dict(self._current) if self._current else None + + def get(self, job_id: str) -> dict[str, Any] | None: + if self._current and self._current["id"] == job_id: + job = dict(self._current) + else: + job = self._read(job_id) or {} + if not job: + return None + job["log"] = self.log_tail(job_id) + return job + + def recent(self, limit: int = 20) -> list[dict[str, Any]]: + files = sorted(self.jobs_dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True) + out = [] + for p in files[:limit]: + try: + out.append(json.loads(p.read_text())) + except ValueError: + continue + if self._current and all(j["id"] != self._current["id"] for j in out): + out.insert(0, dict(self._current)) + out.sort(key=lambda j: j.get("startedAt", ""), reverse=True) + return out + + # ---- args → CLI ------------------------------------------------------------------ + def _argv(self, kind: str, a: dict[str, Any]) -> list[str]: + base = [sys.executable, "-m", "trainer.cli"] + if kind == "train": + mode = a.get("mode", "features") + backbone = a.get("backbone", "resnet18") + if mode not in MODES or backbone not in BACKBONES: + raise ValueError("bad mode/backbone") + floor = float(a.get("minAccuracy", 0.85)) + min_per = int(a.get("minPerClass", 20)) + epochs = int(a.get("epochs", 0)) + if not 0.0 <= floor <= 1.0 or min_per < 1 or epochs < 0: + raise ValueError("bad numbers") + argv = base + [ + "train", + "--data", + str(self.data_dir), + "--out", + str(self.out_dir), + "--mode", + mode, + "--backbone", + backbone, + "--min-accuracy", + str(floor), + "--min-per-class", + str(min_per), + "--epochs", + str(epochs), + ] + if a.get("version"): + argv += ["--version", self._version(a["version"])] + return argv + if kind == "evaluate": + v = self._version(a.get("version", "")) + return base + [ + "evaluate", + "--data", + str(self.data_dir), + "--model", + str(self.out_dir / v / MODEL_FILE), + ] + if kind == "publish": + v = self._version(a.get("version", "")) + url = str(a.get("url") or self.publish_url) + if not url.startswith("https://") and not url.startswith("http://"): + raise ValueError("bad publish url") + return base + ["publish", str(self.out_dir / v), "--url", url] + raise ValueError("kind must be train, evaluate or publish") + + @staticmethod + def _version(v: Any) -> str: + if not isinstance(v, str) or not _VERSION_RE.match(v) or v in ("cache", "jobs"): + raise ValueError("bad version") + return v + + # ---- run ------------------------------------------------------------------------- + def start(self, kind: str, args: dict[str, Any]) -> dict[str, Any]: + argv = self._argv(kind, args) + with self._lock: + if self._current is not None: + raise RuntimeError("busy") + job_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" + job: dict[str, Any] = { + "id": job_id, + "kind": kind, + "args": {k: v for k, v in args.items() if k != "token"}, + "status": "running", + "startedAt": _now(), + "finishedAt": None, + "exitCode": None, + } + env = dict(os.environ) + if kind == "publish": + env["TRAINER_PUBLISH_TOKEN"] = self.publish_token + log = (self.jobs_dir / f"{job_id}.log").open("wb") + log.write(f"$ {' '.join(argv[3:])}\n".encode()) + log.flush() + self._proc = subprocess.Popen(argv, stdout=log, stderr=subprocess.STDOUT, env=env) + self._current = job + self._write(job) + threading.Thread(target=self._wait, args=(job, log), daemon=True).start() + return dict(job) + + def _wait(self, job: dict[str, Any], log: Any) -> None: + assert self._proc is not None + code = self._proc.wait() + log.close() + with self._lock: + job["exitCode"] = code + job["finishedAt"] = _now() + job["status"] = "done" if code == 0 else ("refused" if code in (2, 3) else "failed") + self._write(job) + self._current = None + self._proc = None + + +# ---------------------------------------------------------------------------------- + + +def readiness(data_dir: Path, min_per_class: int = 20, val_fraction: float = 0.2) -> dict[str, Any]: + try: + samples, missing = load_labelled(data_dir) + except FileNotFoundError: + return { + "ready": False, + "labelled": {"total": 0, "byClass": {}}, + "missingCrops": 0, + "error": "no collector database yet", + } + split = make_split(samples, val_fraction, min_per_class) + return { + "ready": len(split.classes) >= 2, + "labelled": summarise(samples), + "missingCrops": missing, + "run": { + "classes": list(split.classes), + "train": split.counts("train"), + "val": split.counts("val"), + "dropped": split.dropped, + "minPerClass": min_per_class, + "valFraction": val_fraction, + }, + "defaults": { + "mode": "features", + "backbone": "resnet18", + "minAccuracy": 0.85, + "minPerClass": min_per_class, + }, + "modes": list(MODES), + "backbones": list(BACKBONES), + } + + +def versions(out_dir: Path) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + if not out_dir.is_dir(): + return out + for d in sorted(out_dir.iterdir(), key=lambda p: p.name, reverse=True): + if not d.is_dir() or d.name in ("cache", "jobs"): + continue + if not (d / REPORT_FILE).is_file() and not (d / METRICS_FILE).is_file(): + continue + entry: dict[str, Any] = { + "version": d.name, + "written": (d / MODEL_FILE).is_file() and (d / SIDECAR_FILE).is_file(), + "hasReport": (d / REPORT_FILE).is_file(), + "modifiedAt": datetime.fromtimestamp(d.stat().st_mtime, tz=timezone.utc) + .replace(microsecond=0) + .isoformat(), + } + try: + m = json.loads((d / METRICS_FILE).read_text()) + entry["accuracy"] = m.get("accuracy") + entry["macroRecall"] = m.get("macro_recall") + entry["n"] = m.get("n") + except (OSError, ValueError): + pass + if entry["written"]: + try: + side = Sidecar.read(d / SIDECAR_FILE) + entry["classes"] = side.classes + entry["mode"] = side.mode + entry["backbone"] = side.backbone + entry["trainedAt"] = side.trained_at + entry["labels"] = side.labels + entry["floor"] = side.metrics.get("floor") + except (OSError, ValueError): + pass + out.append(entry) + return out + + +# ---------------------------------------------------------------------------------- + + +class Handler(BaseHTTPRequestHandler): + jobs: Jobs # set on the class by serve() + server_version = "parking-trainer" + + def log_message(self, fmt: str, *args: Any) -> None: # quieter than the default + sys.stderr.write(f"[trainer.serve] {self.address_string()} {fmt % args}\n") + + def _json(self, code: int, body: Any) -> None: + raw = json.dumps(body).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def _text(self, code: int, body: str, ctype: str = "text/markdown; charset=utf-8") -> None: + raw = body.encode() + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_GET(self) -> None: # noqa: N802 + path = self.path.split("?", 1)[0] + j = self.jobs + if path == "/health": + self._json(200, {"ok": True, "busy": j.busy, "version": "parking-trainer"}) + elif path == "/readiness": + self._json(200, readiness(j.data_dir)) + elif path == "/versions": + self._json(200, {"versions": versions(j.out_dir)}) + elif path.startswith("/versions/") and path.endswith("/report"): + v = path[len("/versions/") : -len("/report")] + try: + p = j.out_dir / Jobs._version(v) / REPORT_FILE + except ValueError: + self._json(400, {"error": "bad version"}) + return + if not p.is_file(): + self._json(404, {"error": "no report"}) + else: + self._text(200, p.read_text()) + elif path == "/jobs": + self._json(200, {"jobs": j.recent(), "current": j.current()}) + elif path.startswith("/jobs/"): + job = j.get(path[len("/jobs/") :]) + self._json(200, job) if job else self._json(404, {"error": "no such job"}) + else: + self._json(404, {"error": "not found"}) + + def do_POST(self) -> None: # noqa: N802 + if self.path.split("?", 1)[0] != "/jobs": + self._json(404, {"error": "not found"}) + return + n = int(self.headers.get("Content-Length") or 0) + if n > 64_000: + self._json(413, {"error": "too large"}) + return + try: + body = json.loads(self.rfile.read(n) or b"{}") + if not isinstance(body, dict): + raise ValueError("object expected") + except ValueError as exc: + self._json(400, {"error": f"bad json: {exc}"}) + return + kind = str(body.pop("kind", "")) + try: + job = self.jobs.start(kind, body) + except ValueError as exc: + self._json(400, {"error": str(exc)}) + return + except RuntimeError: + self._json(409, {"error": "a job is already running", "current": self.jobs.current()}) + return + self._json(202, job) + + +def serve(data_dir: Path, out_dir: Path, host: str, port: int, publish_url: str, publish_token: str) -> None: + Handler.jobs = Jobs(data_dir, out_dir, publish_url, publish_token) + httpd = ThreadingHTTPServer((host, port), Handler) + sys.stderr.write(f"[trainer.serve] listening on {host}:{port}, data {data_dir}, out {out_dir}\n") + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass + finally: + httpd.server_close() + + +def wait_idle(jobs: Jobs, timeout: float = 60.0) -> None: + """Test helper: block until no job runs.""" + t0 = time.monotonic() + while jobs.busy and time.monotonic() - t0 < timeout: + time.sleep(0.1) diff --git a/docker-compose.collector.yml b/docker-compose.collector.yml index c08301d..5a4546b 100644 --- a/docker-compose.collector.yml +++ b/docker-compose.collector.yml @@ -22,29 +22,30 @@ services: COLLECTOR_REVIEWER_USER: ${COLLECTOR_REVIEWER_USER:-reviewer} COLLECTOR_REVIEWER_PASS: ${COLLECTOR_REVIEWER_PASS:?set COLLECTOR_REVIEWER_PASS in the stack env} LOG_LEVEL: ${LOG_LEVEL:-info} + # The trainer's job API (sibling service above). Unset = no Training section. + COLLECTOR_TRAINER_URL: ${COLLECTOR_TRAINER_URL-http://trainer:8091} volumes: - collector-data:/data - # Phase B trainer — a ONE-OFF JOB on this host's CPU, not a service (profile "train": it - # only runs when asked). Reads the collector's SQLite + crops straight off the same volume - # (read-only), writes a versioned model folder under TRAINER_OUT on the host. CPU-only - # PyTorch: the Xeon E3-1225 v5 trains a few thousand crops in minutes (features mode) to an - # hour (full fine-tune) — see wiki/decisions/bodytype-classifier-training.md. If a modern GPU - # ever lands in the host, add an nvidia device reservation here; the trainer picks up CUDA. - # - # docker compose -f docker-compose.collector.yml --profile train run --rm trainer inspect - # docker compose -f docker-compose.collector.yml --profile train run --rm trainer train --min-accuracy 0.85 - # docker compose -f docker-compose.collector.yml --profile train run --rm trainer evaluate --model /out//bodytype.onnx - # docker compose -f docker-compose.collector.yml --profile train run --rm trainer publish /out/ --url + # Phase B trainer — a small always-on job service beside the collector (CPU-only torch; + # idle it is a tiny Python HTTP server, torch loads only when a job runs). It reads the + # collector's SQLite + crops off the same volume (read-only) and keeps models, reports and + # job logs in its own volume. NOT published: only the collector reaches it, on this compose + # network, and the reviewer's login on the collector is the gate. The Training section of + # /review is its UI (readiness, Train / Evaluate / Publish, reports, logs). + # See wiki/decisions/bodytype-classifier-training.md. trainer: image: ${REGISTRY:-git.infra.msai.al/mca/parking_solution}/parking-trainer:${TAG:-dev} - profiles: ["train"] + restart: unless-stopped environment: - # Only `publish` needs it: a Gitea token with package:write for the model's generic package. + # Where `publish` PUTs a passing model (a Gitea generic package) and the token it uses + # (package:write). Only publishing needs the token; training runs without it. + TRAINER_PUBLISH_URL: ${TRAINER_PUBLISH_URL:-https://git.infra.msai.al/api/packages/mca/generic/parking-bodytype} TRAINER_PUBLISH_TOKEN: ${TRAINER_PUBLISH_TOKEN:-} volumes: - collector-data:/data:ro - - ${TRAINER_OUT:-./models}:/out + - trainer-out:/out volumes: collector-data: + trainer-out: diff --git a/komodo/resources.toml b/komodo/resources.toml index 0927f04..a45aa77 100644 --- a/komodo/resources.toml +++ b/komodo/resources.toml @@ -143,10 +143,9 @@ COLLECTOR_BIND=100.75.184.156 # to keep in sync, and rotating a booth touches one secret. The booth id is the booth's # pseudonymous CARWASH_REVIEW_BOOTH_ID, never a site name. Add a pair per booth. COLLECTOR_BOOTH_TOKENS=booth-2:[[wash_review_token_booth_2]] -# Phase-B trainer (profile "train", a one-off job on this host — never started by the deploy). -# Its output folder on the host, and the Gitea token `publish` uses to upload a passing model to -# the generic package registry (package:write). Uncomment when the first model is to be published. -#TRAINER_OUT=/opt/parking/models +# Phase-B trainer (the `trainer` service beside the collector; the Training section of /review +# is its UI). Only `publish` needs this: a Gitea token with package:write for the model's generic +# package. Uncomment when the first model is to be published. #TRAINER_PUBLISH_TOKEN=[[gitea_package_write_token]] COLLECTOR_REVIEWER_USER=reviewer COLLECTOR_REVIEWER_PASS=[[wash_collector_reviewer_pass]] diff --git a/wiki/concepts/vision-review-outbox.md b/wiki/concepts/vision-review-outbox.md index 00dd6ed..3dc15ef 100644 --- a/wiki/concepts/vision-review-outbox.md +++ b/wiki/concepts/vision-review-outbox.md @@ -130,9 +130,14 @@ Three surfaces, nothing else — it must not grow into a fleet console: label, the operator's category + classes, the camera's class + confidence, downgraded, at. Crops are not packaged: the phase-B trainer runs **on the same host** and reads the SQLite + crops straight off the volume, read-only ([[bodytype-classifier-training]]: CPU-only, the - Xeon is enough) — `docker-compose.collector.yml` carries it as the `trainer` service under - `profiles: ["train"]`, a one-off job never started by a deploy (built 2026-09-07; the CSV - export stays for a human with a spreadsheet). + Xeon is enough) — the `trainer` service beside the collector in + `docker-compose.collector.yml` (the CSV export stays for a human with a spreadsheet). +- **Training section on `/review`** (+ `/api/training/status|jobs|jobs/:id|versions/:v/report`) + — a thin proxy, behind the same reviewer login, to the trainer's job API on the compose + network (`COLLECTOR_TRAINER_URL`, unset = hidden): labels per class vs the minimum, Train + (mode / backbone / floor), the running job's log, the versions with Report / Evaluate / + Publish. The collector forwards only a fixed set of paths and knobs; the trainer validates + values and answers 409 while a job runs. **Where the data lives.** The collector writes to `/data` in its container: `collector.sqlite` and one JPEG per item at `crops//.jpg`. `/data` is the named Docker volume diff --git a/wiki/decisions/bodytype-classifier-training.md b/wiki/decisions/bodytype-classifier-training.md index 7c01c90..bb4d812 100644 --- a/wiki/decisions/bodytype-classifier-training.md +++ b/wiki/decisions/bodytype-classifier-training.md @@ -18,8 +18,13 @@ Xeon". This page is the loop as built; what is still outstanding is at the end. Each step is a place where a person decides. Nothing here runs on its own. 1. **Train** — `apps/trainer` (`parking-trainer`, Python/uv like the vision service; its own - image `parking-trainer`, a one-off job on the collector's host — never a booth service). - `train` reads the collector's `collector.sqlite` and `crops/` **straight off the volume** + image `parking-trainer`, the `trainer` service beside the collector on the reviewer's host — + never a booth service). **Started from the collector's UI:** the Training section of + `/review` (readiness, a Train button with mode / backbone / floor, the live log, the + versions with Report / Evaluate / Publish) drives a small job API the trainer serves on the + compose network (`serve`; stdlib HTTP, one job at a time, each job the CLI as a subprocess + with its log persisted under `/out/jobs/`). The collector proxies it behind the reviewer's + login; the trainer is never published. `train` reads the collector's `collector.sqlite` and `crops/` **straight off the volume** (read-only), takes only reviewed, usable rows (the operator's pick and the camera's class are never labels), **splits by TIME** (validation = the newest 20 % by *time seen*, so the number reflects tomorrow's traffic), drops classes with fewer than `--min-per-class` (20) labels from @@ -129,47 +134,44 @@ What the owner has: an **NVIDIA Quadro FX 3800** (in hand, not installed), and i host, so nothing moves. - **Consequences for the build (done):** the trainer image is **CPU-only PyTorch** (torch 2.14+cpu, ~200 MB of wheels, not the ~5 GB CUDA build); the `trainer` service in - `docker-compose.collector.yml` is real now — `profiles: ["train"]`, no device reservation - (one block to add if a modern card ever lands; the trainer would pick up CUDA), the collector - volume mounted read-only, output to `TRAINER_OUT` on the host (default `./models` beside the - compose file). + `docker-compose.collector.yml` is real — always on, serving the job API, no device + reservation (one block to add if a modern card ever lands; the trainer would pick up CUDA), + the collector volume mounted read-only, models/reports/logs in its own `trainer-out` volume. - **If faster is ever wanted:** a used mid-range card of the last few generations (~€200) turns the hour into a minute, given a slot and a PSU. **Renting a cloud GPU is rejected**: the crops would leave the premises, and even scrubbed of plates and site that runs against the whole privacy design of the outbox. -## Running it (on the collector host) +## Running it -``` -docker compose -f docker-compose.collector.yml --profile train run --rm trainer inspect -docker compose -f docker-compose.collector.yml --profile train run --rm trainer train --min-accuracy 0.85 -docker compose -f docker-compose.collector.yml --profile train run --rm trainer evaluate --model /out//bodytype.onnx -docker compose -f docker-compose.collector.yml --profile train run --rm trainer publish /out/ --url https://git.infra.msai.al/api/packages/mca/generic/parking-bodytype -``` +From the collector's `/review` page, Training section: **Train** (mode, backbone, floor) when +readiness says enough labels; watch the log; read the report under Versions; **Evaluate** a +written version against labels reviewed since; **Publish** it (needs `TRAINER_PUBLISH_TOKEN` +in the `wash-collector` stack — commented until the first publish). Then, in git: write the +version into `apps/vision/models/bodytype.version`, commit, let the build produce the image, +bump the booth's `TAG`. The pin stays a commit on purpose — it is the deploy control. -Then: write the version into `apps/vision/models/bodytype.version`, commit, let the build produce -the image, bump the booth's `TAG`. The trainer is never started by a deploy (a profile), and the -`wash-collector` stack's `TRAINER_OUT` / `TRAINER_PUBLISH_TOKEN` lines stay commented until the -first publish. +The CLI is still there for debugging, inside the running container: +`docker compose -f docker-compose.collector.yml exec trainer parking-trainer inspect`. -## Operating notes (first deploy, 2026-09-07) +## Operating notes (2026-09-07) -- **Only the collector shows as running — that is correct.** The trainer is not a service; it is - behind the `train` compose *profile*, so a deploy never starts it and `docker ps` on - `art-docker-station` lists one container. The trainer runs when invoked by hand, does its job, - exits, and leaves nothing behind (`--rm`). -- **A deploy does not pull profile services either.** The first `run` pulls the image itself, so - the host's Docker must be logged in to the registry (`docker compose -f - docker-compose.collector.yml --profile train pull trainer` is the check; if refused, `docker - login git.infra.msai.al` on the host first). -- **Why the wash-collector TAG bump mattered** although the collector code did not change: the - trainer service uses the stack's TAG, so a bump makes `run` resolve to an image that exists - (`stage-f7a262a` is the first tag that carries `parking-trainer`). -- **park-2 does not need the bump** until a model is pinned: the new vision image ships with an +- **First deploy (`stage-f7a262a`) shipped the trainer as a compose *profile*** — a one-off + job the owner had to start by hand with `docker compose … --profile train run …` from + wherever Komodo's periphery had cloned the repo (`/etc/komodo/stacks/wash-collector/`). + The user rightly called that "not so smart": the host runs a periphery, and the reviewer is + already in the collector's UI. **Superseded the same day:** the trainer is now a + **service** (`restart: unless-stopped`, the `serve` command) and the collector's + `/review` page carries the Training section. A deploy starts both containers; `docker ps` + shows two. +- **Why not a Docker socket in the collector** (the other way to a button): it would hand + root on the host to a service that accepts uploads from booths — the party the + [[threat-model]] distrusts. The job API keeps the trainer a normal container with a + read-only data mount and its own `trainer-out` volume. +- **park-2 does not need a bump** until a model is pinned: the vision image ships with an empty `bodytype.version`, phase B off, nothing for a booth to gain. -- **What to run, from the stack's directory on the host:** `inspect` first (label counts per - class, `ready: false` until enough are reviewed), then `train`, read `report.md` under - `TRAINER_OUT//`, then `publish`, pin, push, bump the booth. Commands under §Running it. +- **Reviewing is the bottleneck**: the Training section shows labels per class against the + minimum and keeps Train disabled until two classes clear it. ## Packaging rule (same as the vision service) diff --git a/wiki/decisions/fleet-deployment-komodo.md b/wiki/decisions/fleet-deployment-komodo.md index ee39241..b85c2b2 100644 --- a/wiki/decisions/fleet-deployment-komodo.md +++ b/wiki/decisions/fleet-deployment-komodo.md @@ -158,11 +158,10 @@ collector ([[vision-review-outbox]]) runs on the reviewer's GPU host as its own (`wash-collector`, `server = "art-docker-station"`, `file_paths = ["docker-compose.collector.yml"]`). Same repo, branch and pinned `TAG` promotion, its own secret references, and — because a stack names its compose files — nothing booth-side lands on that host and nothing of it on a booth. -The same stack carries the phase-B **trainer** as a compose *profile* (`train`, -[[bodytype-classifier-training]]): a deploy never starts it; the owner runs it by hand on the host -with `docker compose … --profile train run --rm trainer …`. So after a deploy of that stack -`docker ps` shows one container — expected; and a deploy pulls nothing for the profile, the -first `run` does (host Docker must be logged in to the registry). Its two env lines (`TRAINER_OUT`, the +The same stack carries the phase-B **trainer** as a second service ([[bodytype-classifier-training]]): +a deploy starts both, `docker ps` shows two containers, and the trainer is driven from the +collector's UI, never from the host's shell (a first cut as a compose *profile* run by hand was +replaced the same day — the host runs a periphery, nobody should be typing compose there). Its two env lines (`TRAINER_OUT`, the `TRAINER_PUBLISH_TOKEN` secret reference) stay commented in `resources.toml` until the first publish. diff --git a/wiki/log.md b/wiki/log.md index 9b0db85..11875d6 100644 --- a/wiki/log.md +++ b/wiki/log.md @@ -3144,6 +3144,20 @@ run; the Quadro FX 3800 is unusable (cc 1.3), the HD P530 irrelevant, the Xeon E compose seam drops the GPU reservation; cloud GPU rejected (crops stay on premises). Linked from [[opencv-anpr-service]], [[vision-review-outbox]], index. User: "No build just yet." +## [2026-09-07] build | Training from the collector UI — the trainer becomes a job service +User: the compose-profile trainer is "not so smart" (where is the compose file on a periphery +host? why not a button on the collector UI?). Built: `parking-trainer serve` — a stdlib job API +(`/health`, `/readiness`, `/versions`, `/versions//report`, `/jobs`), one job at a time, each +job the CLI as a subprocess with state + log persisted under `/out/jobs/`; the collector gained +`COLLECTOR_TRAINER_URL` + `/api/training/*` (reviewer-gated proxy, fixed paths, whitelisted +knobs, trainer status codes passed through, 503 unconfigured / 502 unreachable) and a Training +section on `/review` (readiness table, Train with mode/backbone/floor, live log, versions with +Report / Evaluate / Publish, the pin reminder). Compose: `trainer` is a service now +(`restart: unless-stopped`, `serve`, read-only data, own `trainer-out` volume, not published); +the Docker socket route was rejected (root on the host for a service booths upload to). Tests: +trainer 14, collector 7. Pages: [[bodytype-classifier-training]] (loop, running it, operating +notes superseded), [[vision-review-outbox]], [[fleet-deployment-komodo]]. + ## [2026-09-07] ingest | Trainer deployed as a profile; operating notes User pushed `stage-f7a262a`, bumped the `wash-collector` TAG, redeployed, and asked why only one service runs on art-docker-station. Expected: the trainer is a compose profile, never started or