Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 29594f8bad | |||
| 4ff31557a8 | |||
| 3e77a4ad7c |
@@ -16,7 +16,7 @@ const basic = "Basic " + Buffer.from(`${REVIEWER.user}:${REVIEWER.pass}`).toStri
|
|||||||
|
|
||||||
beforeEach(async () => {
|
beforeEach(async () => {
|
||||||
dir = await mkdtemp(path.join(tmpdir(), "collector-"));
|
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();
|
await app.ready();
|
||||||
});
|
});
|
||||||
afterEach(async () => {
|
afterEach(async () => {
|
||||||
|
|||||||
@@ -15,6 +15,8 @@ import { reviewPage } from "./review-page.js";
|
|||||||
// /review + /api/* the reviewer's screen (HTTP Basic, one login)
|
// /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
|
// 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)
|
// 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.
|
// 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). */
|
/** 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");
|
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<unknown> {
|
||||||
|
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<Record<string, unknown>>;
|
||||||
|
};
|
||||||
|
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<string, unknown> }>("/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<string, unknown> = {};
|
||||||
|
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;
|
return app;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,9 @@ export interface CollectorConfig {
|
|||||||
readonly boothTokens: ReadonlyMap<string, string>;
|
readonly boothTokens: ReadonlyMap<string, string>;
|
||||||
/** The single reviewer login; null = review screen and export refuse (503). */
|
/** The single reviewer login; null = review screen and export refuse (503). */
|
||||||
readonly reviewer: { readonly user: string; readonly pass: string } | null;
|
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). */
|
/** "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",
|
dataDir: env.COLLECTOR_DATA_DIR ?? "/data",
|
||||||
boothTokens: parseBoothTokens(env.COLLECTOR_BOOTH_TOKENS ?? ""),
|
boothTokens: parseBoothTokens(env.COLLECTOR_BOOTH_TOKENS ?? ""),
|
||||||
reviewer: user && pass.length >= 8 ? { user, pass } : null,
|
reviewer: user && pass.length >= 8 ? { user, pass } : null,
|
||||||
|
trainerUrl: (env.COLLECTOR_TRAINER_URL ?? "").trim().replace(/\/+$/, "") || null,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ const cfg = configFromEnv();
|
|||||||
const app = await buildCollector(cfg);
|
const app = await buildCollector(cfg);
|
||||||
if (cfg.boothTokens.size === 0) app.log.warn("COLLECTOR_BOOTH_TOKENS is empty — no booth can ingest");
|
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");
|
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 });
|
await app.listen({ host: cfg.host, port: cfg.port });
|
||||||
|
|
||||||
const stop = async () => {
|
const stop = async () => {
|
||||||
|
|||||||
@@ -35,6 +35,16 @@ export function reviewPage(): string {
|
|||||||
td, th { text-align:left; padding:.2rem .5rem; border-bottom:1px solid #2a2a2a; }
|
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; }
|
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; }
|
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; }
|
||||||
</style>
|
</style>
|
||||||
</head>
|
</head>
|
||||||
<body>
|
<body>
|
||||||
@@ -46,6 +56,10 @@ export function reviewPage(): string {
|
|||||||
<section class="card">
|
<section class="card">
|
||||||
<table id="stats"><thead><tr><th>booth</th><th>operator</th><th>reviewed</th><th>agree</th><th>disagree</th><th>unusable</th></tr></thead><tbody></tbody></table>
|
<table id="stats"><thead><tr><th>booth</th><th>operator</th><th>reviewed</th><th>agree</th><th>disagree</th><th>unusable</th></tr></thead><tbody></tbody></table>
|
||||||
</section>
|
</section>
|
||||||
|
<section class="card" id="training" hidden>
|
||||||
|
<h2>Training</h2>
|
||||||
|
<div id="tr-body"></div>
|
||||||
|
</section>
|
||||||
<p class="muted">Keys: <kbd>1</kbd>–<kbd>9</kbd>, <kbd>0</kbd> pick a class in order · <kbd>u</kbd> unusable · <kbd>s</kbd> skip. Skipped items come back after a reload. Your verdict is the training label; the operator's pick is only compared against it.</p>
|
<p class="muted">Keys: <kbd>1</kbd>–<kbd>9</kbd>, <kbd>0</kbd> pick a class in order · <kbd>u</kbd> unusable · <kbd>s</kbd> skip. Skipped items come back after a reload. Your verdict is the training label; the operator's pick is only compared against it.</p>
|
||||||
</main>
|
</main>
|
||||||
<script>
|
<script>
|
||||||
@@ -114,6 +128,106 @@ document.addEventListener('keydown', e => {
|
|||||||
|
|
||||||
next().catch(e => { document.getElementById('item').innerHTML = '<p class="warn">' + esc(e.message) + '</p>'; });
|
next().catch(e => { document.getElementById('item').innerHTML = '<p class="warn">' + esc(e.message) + '</p>'; });
|
||||||
loadStats().catch(() => {});
|
loadStats().catch(() => {});
|
||||||
|
|
||||||
|
// ---- Training: the trainer's job API, proxied by the collector -------------------------
|
||||||
|
// Readiness (labels per class vs the minimum), one job at a time with a live log, the
|
||||||
|
// versions a run produced (written or refused) with Report / Evaluate / Publish. Pinning a
|
||||||
|
// published version into the vision image stays a git commit — that is the deploy control.
|
||||||
|
let trPoll = null;
|
||||||
|
let trShownReport = null;
|
||||||
|
const trDefaults = { mode: 'features', backbone: 'resnet18', minAccuracy: 0.85 };
|
||||||
|
|
||||||
|
function pct(x) { return x == null ? '—' : Math.round(x * 100) + ' %'; }
|
||||||
|
|
||||||
|
async function training() {
|
||||||
|
const box = document.getElementById('training');
|
||||||
|
const el = document.getElementById('tr-body');
|
||||||
|
let s;
|
||||||
|
try { s = await api('/api/training/status'); } catch (e) { box.hidden = false; el.innerHTML = '<p class="warn">' + esc(e.message) + '</p>'; return; }
|
||||||
|
if (!s.configured) { box.hidden = true; return; }
|
||||||
|
box.hidden = false;
|
||||||
|
if (!s.reachable) { el.innerHTML = '<p class="warn">trainer not reachable: ' + esc(s.error || '') + '</p>'; schedule(true); return; }
|
||||||
|
const r = s.readiness, run = r.run || {}, minPer = run.minPerClass || 20;
|
||||||
|
const byClass = (r.labelled && r.labelled.byClass) || {};
|
||||||
|
const classes = Object.keys(byClass);
|
||||||
|
const cur = s.current;
|
||||||
|
const readyLine = r.ready
|
||||||
|
? '<span class="ok">enough labels to train</span> — classes this run: ' + esc((run.classes || []).join(', '))
|
||||||
|
: '<span class="warn">not enough labels yet</span> — a class needs ' + minPer + ' reviewed crops; two classes must clear it';
|
||||||
|
let html = '<p>' + readyLine + ' <span class="muted">(' + (r.labelled ? r.labelled.total : 0) + ' labelled, ' + (r.missingCrops || 0) + ' missing crop files)</span></p>';
|
||||||
|
html += '<table><thead><tr><th>class</th><th>reviewed</th><th>train</th><th>val</th><th></th></tr></thead><tbody>' +
|
||||||
|
(classes.map(c => '<tr><td class="mono">' + esc(c) + '</td><td>' + byClass[c] + '</td><td>' + ((run.train || {})[c] ?? '—') + '</td><td>' + ((run.val || {})[c] ?? '—') + '</td><td class="muted">' + (byClass[c] < minPer ? 'below ' + minPer + ' — dropped' : '') + '</td></tr>').join('') || '<tr><td colspan="5" class="muted">no labels yet — review crops above</td></tr>') +
|
||||||
|
'</tbody></table>';
|
||||||
|
const d = Object.assign({}, trDefaults, r.defaults || {});
|
||||||
|
html += '<div class="row" style="margin-top:.8rem">' +
|
||||||
|
'<label>mode <select id="tr-mode">' + (r.modes || ['features', 'finetune']).map(m => '<option' + (m === d.mode ? ' selected' : '') + '>' + m + '</option>').join('') + '</select></label>' +
|
||||||
|
'<label>backbone <select id="tr-backbone">' + (r.backbones || ['resnet18']).map(b => '<option' + (b === d.backbone ? ' selected' : '') + '>' + b + '</option>').join('') + '</select></label>' +
|
||||||
|
'<label>floor <input id="tr-floor" type="number" min="0" max="1" step="0.01" value="' + d.minAccuracy + '"></label>' +
|
||||||
|
'<button id="tr-train"' + (r.ready && !cur ? '' : ' disabled') + '>Train</button>' +
|
||||||
|
(cur ? '<span class="warn">running: ' + esc(cur.kind) + ' ' + esc(cur.id) + '</span>' : '') +
|
||||||
|
'</div>';
|
||||||
|
const last = cur || (s.jobs && s.jobs[0]);
|
||||||
|
if (last) {
|
||||||
|
const cls = last.status === 'done' ? 'ok' : last.status === 'running' ? 'warn' : 'bad';
|
||||||
|
html += '<p style="margin:.8rem 0 0"><span class="' + cls + '">' + esc(last.status) + '</span> <span class="mono">' + esc(last.kind) + ' ' + esc(last.id) + '</span> <span class="muted">' + esc(last.startedAt || '') + (last.exitCode != null ? ' · exit ' + last.exitCode : '') + '</span> <button class="small" data-job="' + esc(last.id) + '">log</button></p>' +
|
||||||
|
'<pre id="tr-log" hidden></pre>';
|
||||||
|
}
|
||||||
|
const vs = s.versions || [];
|
||||||
|
html += '<h2 style="margin-top:1rem">Versions</h2>';
|
||||||
|
html += vs.length
|
||||||
|
? '<table><thead><tr><th>version</th><th>model</th><th>accuracy</th><th>classes</th><th>mode</th><th></th></tr></thead><tbody>' +
|
||||||
|
vs.map(v => '<tr><td class="mono">' + esc(v.version) + '</td><td>' + (v.written ? '<span class="ok">written</span>' : '<span class="bad">refused</span>') + '</td><td>' + pct(v.accuracy) + (v.floor != null ? ' <span class="muted">/ floor ' + pct(v.floor) + '</span>' : '') + '</td><td class="muted">' + esc((v.classes || []).join(', ')) + '</td><td class="muted">' + esc(v.mode || '') + '</td><td>' +
|
||||||
|
'<button class="small" data-report="' + esc(v.version) + '">report</button> ' +
|
||||||
|
(v.written ? '<button class="small" data-eval="' + esc(v.version) + '"' + (cur ? ' disabled' : '') + '>evaluate</button> <button class="small" data-publish="' + esc(v.version) + '"' + (cur ? ' disabled' : '') + '>publish</button>' : '') +
|
||||||
|
'</td></tr>').join('') + '</tbody></table>'
|
||||||
|
: '<p class="muted">no runs yet</p>';
|
||||||
|
html += '<pre id="tr-report" hidden></pre>';
|
||||||
|
html += '<p class="muted" style="margin:.8rem 0 0">A written model is only a file here. To put it on a booth: publish, then pin the version in <span class="mono">apps/vision/models/bodytype.version</span>, commit, and bump the TAG of the booth.</p>';
|
||||||
|
el.innerHTML = html;
|
||||||
|
|
||||||
|
const trainBtn = document.getElementById('tr-train');
|
||||||
|
if (trainBtn) trainBtn.addEventListener('click', () => startJob({ kind: 'train', mode: document.getElementById('tr-mode').value, backbone: document.getElementById('tr-backbone').value, minAccuracy: Number(document.getElementById('tr-floor').value) }));
|
||||||
|
el.querySelectorAll('button[data-eval]').forEach(b => b.addEventListener('click', () => startJob({ kind: 'evaluate', version: b.dataset.eval })));
|
||||||
|
el.querySelectorAll('button[data-publish]').forEach(b => b.addEventListener('click', () => { if (confirm('Publish ' + b.dataset.publish + ' to the package registry?')) startJob({ kind: 'publish', version: b.dataset.publish }); }));
|
||||||
|
el.querySelectorAll('button[data-job]').forEach(b => b.addEventListener('click', () => showLog(b.dataset.job)));
|
||||||
|
el.querySelectorAll('button[data-report]').forEach(b => b.addEventListener('click', () => showReport(b.dataset.report)));
|
||||||
|
if (cur) showLog(cur.id).catch(() => {});
|
||||||
|
if (trShownReport) showReport(trShownReport).catch(() => {});
|
||||||
|
schedule(!!cur);
|
||||||
|
}
|
||||||
|
|
||||||
|
function schedule(soon) {
|
||||||
|
if (trPoll) clearTimeout(trPoll);
|
||||||
|
trPoll = setTimeout(() => training().catch(() => {}), soon ? 4000 : 60000);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function startJob(body) {
|
||||||
|
try {
|
||||||
|
const r = await fetch('/api/training/jobs', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) });
|
||||||
|
if (!r.ok) { const e = await r.json().catch(() => ({})); alert('trainer: ' + (e.error || ('HTTP ' + r.status))); }
|
||||||
|
} catch (e) { alert(e.message); }
|
||||||
|
training().catch(() => {});
|
||||||
|
}
|
||||||
|
|
||||||
|
async function showLog(id) {
|
||||||
|
const j = await api('/api/training/jobs/' + encodeURIComponent(id));
|
||||||
|
const pre = document.getElementById('tr-log');
|
||||||
|
if (!pre) return;
|
||||||
|
pre.hidden = false;
|
||||||
|
pre.textContent = j.log || '(no output yet)';
|
||||||
|
pre.scrollTop = pre.scrollHeight;
|
||||||
|
}
|
||||||
|
|
||||||
|
async function showReport(v) {
|
||||||
|
const r = await fetch('/api/training/versions/' + encodeURIComponent(v) + '/report');
|
||||||
|
const pre = document.getElementById('tr-report');
|
||||||
|
if (!pre) return;
|
||||||
|
trShownReport = v;
|
||||||
|
pre.hidden = false;
|
||||||
|
pre.textContent = r.ok ? await r.text() : 'no report for ' + v + ' (HTTP ' + r.status + ')';
|
||||||
|
}
|
||||||
|
|
||||||
|
training().catch(() => {});
|
||||||
</script>
|
</script>
|
||||||
</body>
|
</body>
|
||||||
</html>`;
|
</html>`;
|
||||||
|
|||||||
@@ -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<void> {
|
||||||
|
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<void>((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<void>((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("<script>") + 8, html.lastIndexOf("</script>"));
|
||||||
|
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);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
{
|
{
|
||||||
"$schema": "https://schema.tauri.app/config/2",
|
"$schema": "https://schema.tauri.app/config/2",
|
||||||
"productName": "Parking System",
|
"productName": "Parking System",
|
||||||
"version": "0.1.0",
|
"version": "0.2.0",
|
||||||
"identifier": "com.parking.desktop",
|
"identifier": "com.parking.desktop",
|
||||||
"build": {
|
"build": {
|
||||||
"devUrl": "http://localhost:5173",
|
"devUrl": "http://localhost:5173",
|
||||||
|
|||||||
+13
-7
@@ -1,9 +1,11 @@
|
|||||||
# syntax=docker/dockerfile:1.7
|
# syntax=docker/dockerfile:1.7
|
||||||
# Parking TRAINER image: the phase-B body-type classifier job. Build CONTEXT is apps/trainer
|
# Parking TRAINER image: the phase-B body-type classifier. Build CONTEXT is apps/trainer
|
||||||
# (self-contained Python package). A ONE-OFF JOB on the reviewer's host (art-docker-station),
|
# (self-contained Python package). Runs on the reviewer's host (art-docker-station) beside
|
||||||
# never a booth service: it reads the wash collector's volume (collector.sqlite + crops/)
|
# the collector, never on a booth: by default it SERVES the job API the collector's Training
|
||||||
# and writes a versioned model folder. CPU-only PyTorch — the host has no usable GPU and a
|
# section drives (`serve`); the same image runs the CLI one-off (`train`, `inspect`, …). It
|
||||||
# few thousand crops train in minutes/an hour on four Xeon cores.
|
# 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.
|
# See wiki/decisions/bodytype-classifier-training.md.
|
||||||
|
|
||||||
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS base
|
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
|
USER trainer
|
||||||
|
|
||||||
ENV TRAINER_DATA_DIR=/data \
|
ENV TRAINER_DATA_DIR=/data \
|
||||||
TRAINER_OUT_DIR=/out
|
TRAINER_OUT_DIR=/out \
|
||||||
|
TRAINER_PORT=8091
|
||||||
VOLUME ["/out"]
|
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"]
|
ENTRYPOINT ["uv", "run", "--no-sync", "parking-trainer"]
|
||||||
CMD ["inspect"]
|
CMD ["serve"]
|
||||||
|
|||||||
@@ -25,12 +25,12 @@ A passing run writes `<out>/<version>/`:
|
|||||||
| `report.md` | the human report: accuracy, per-class recall/precision, confusion matrix, dropped classes, loss weights |
|
| `report.md` | the human report: accuracy, per-class recall/precision, confusion matrix, dropped classes, loss weights |
|
||||||
| `metrics.json` | the same numbers, machine-readable |
|
| `metrics.json` | the same numbers, machine-readable |
|
||||||
|
|
||||||
On the reviewer's host (the `wash-collector` stack):
|
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 /
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer inspect
|
Evaluate / Publish, reports and logs. Jobs run as subprocesses of the CLI, one at a time; state
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer train --min-accuracy 0.85
|
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
|
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.
|
runs without the extra (torch tests skip), matching CI.
|
||||||
|
|||||||
@@ -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"
|
||||||
@@ -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)")
|
_log(f"{f} missing — nothing to publish (a run below the floor writes no model)")
|
||||||
return 1
|
return 1
|
||||||
version = Sidecar.read(d / SIDECAR_FILE).version
|
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", "")
|
token = a.token or os.environ.get("TRAINER_PUBLISH_TOKEN", "")
|
||||||
if not token:
|
if not token:
|
||||||
_log("no token: pass --token or set TRAINER_PUBLISH_TOKEN")
|
_log("no token: pass --token or set TRAINER_PUBLISH_TOKEN")
|
||||||
@@ -289,6 +292,20 @@ def cmd_publish(a: argparse.Namespace) -> int:
|
|||||||
return 0
|
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 = sub.add_parser("publish", help="PUT a version folder to a Gitea generic package")
|
||||||
u.add_argument("dir", type=Path, help="the <version>/ folder a passing run wrote")
|
u.add_argument("dir", type=Path, help="the <version>/ folder a passing run wrote")
|
||||||
u.add_argument(
|
u.add_argument(
|
||||||
"--url", required=True, help="https://<gitea>/api/packages/<owner>/generic/parking-bodytype"
|
"--url",
|
||||||
|
default=os.environ.get("TRAINER_PUBLISH_URL", ""),
|
||||||
|
help="https://<gitea>/api/packages/<owner>/generic/parking-bodytype (or TRAINER_PUBLISH_URL)",
|
||||||
)
|
)
|
||||||
u.add_argument("--token", default="", help="Gitea token with package:write (or TRAINER_PUBLISH_TOKEN)")
|
u.add_argument("--token", default="", help="Gitea token with package:write (or TRAINER_PUBLISH_TOKEN)")
|
||||||
u.set_defaults(fn=cmd_publish)
|
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
|
return p
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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 `<out>/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 <out>/<version>/ folder: written?, metrics, sidecar
|
||||||
|
GET /versions/<v>/report report.md (text/markdown)
|
||||||
|
GET /jobs recent jobs, newest first
|
||||||
|
GET /jobs/<id> 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)
|
||||||
@@ -22,29 +22,30 @@ services:
|
|||||||
COLLECTOR_REVIEWER_USER: ${COLLECTOR_REVIEWER_USER:-reviewer}
|
COLLECTOR_REVIEWER_USER: ${COLLECTOR_REVIEWER_USER:-reviewer}
|
||||||
COLLECTOR_REVIEWER_PASS: ${COLLECTOR_REVIEWER_PASS:?set COLLECTOR_REVIEWER_PASS in the stack env}
|
COLLECTOR_REVIEWER_PASS: ${COLLECTOR_REVIEWER_PASS:?set COLLECTOR_REVIEWER_PASS in the stack env}
|
||||||
LOG_LEVEL: ${LOG_LEVEL:-info}
|
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:
|
volumes:
|
||||||
- collector-data:/data
|
- collector-data:/data
|
||||||
|
|
||||||
# Phase B trainer — a ONE-OFF JOB on this host's CPU, not a service (profile "train": it
|
# Phase B trainer — a small always-on job service beside the collector (CPU-only torch;
|
||||||
# only runs when asked). Reads the collector's SQLite + crops straight off the same volume
|
# idle it is a tiny Python HTTP server, torch loads only when a job runs). It reads the
|
||||||
# (read-only), writes a versioned model folder under TRAINER_OUT on the host. CPU-only
|
# collector's SQLite + crops off the same volume (read-only) and keeps models, reports and
|
||||||
# PyTorch: the Xeon E3-1225 v5 trains a few thousand crops in minutes (features mode) to an
|
# job logs in its own volume. NOT published: only the collector reaches it, on this compose
|
||||||
# hour (full fine-tune) — see wiki/decisions/bodytype-classifier-training.md. If a modern GPU
|
# network, and the reviewer's login on the collector is the gate. The Training section of
|
||||||
# ever lands in the host, add an nvidia device reservation here; the trainer picks up CUDA.
|
# /review is its UI (readiness, Train / Evaluate / Publish, reports, logs).
|
||||||
#
|
# See wiki/decisions/bodytype-classifier-training.md.
|
||||||
# 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/<version>/bodytype.onnx
|
|
||||||
# docker compose -f docker-compose.collector.yml --profile train run --rm trainer publish /out/<version> --url <gitea generic package url>
|
|
||||||
trainer:
|
trainer:
|
||||||
image: ${REGISTRY:-git.infra.msai.al/mca/parking_solution}/parking-trainer:${TAG:-dev}
|
image: ${REGISTRY:-git.infra.msai.al/mca/parking_solution}/parking-trainer:${TAG:-dev}
|
||||||
profiles: ["train"]
|
restart: unless-stopped
|
||||||
environment:
|
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:-}
|
TRAINER_PUBLISH_TOKEN: ${TRAINER_PUBLISH_TOKEN:-}
|
||||||
volumes:
|
volumes:
|
||||||
- collector-data:/data:ro
|
- collector-data:/data:ro
|
||||||
- ${TRAINER_OUT:-./models}:/out
|
- trainer-out:/out
|
||||||
|
|
||||||
volumes:
|
volumes:
|
||||||
collector-data:
|
collector-data:
|
||||||
|
trainer-out:
|
||||||
|
|||||||
@@ -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
|
# 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.
|
# pseudonymous CARWASH_REVIEW_BOOTH_ID, never a site name. Add a pair per booth.
|
||||||
COLLECTOR_BOOTH_TOKENS=booth-2:[[wash_review_token_booth_2]]
|
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).
|
# Phase-B trainer (the `trainer` service beside the collector; the Training section of /review
|
||||||
# Its output folder on the host, and the Gitea token `publish` uses to upload a passing model to
|
# is its UI). Only `publish` needs this: a Gitea token with package:write for the model's generic
|
||||||
# the generic package registry (package:write). Uncomment when the first model is to be published.
|
# package. Uncomment when the first model is to be published.
|
||||||
#TRAINER_OUT=/opt/parking/models
|
|
||||||
#TRAINER_PUBLISH_TOKEN=[[gitea_package_write_token]]
|
#TRAINER_PUBLISH_TOKEN=[[gitea_package_write_token]]
|
||||||
COLLECTOR_REVIEWER_USER=reviewer
|
COLLECTOR_REVIEWER_USER=reviewer
|
||||||
COLLECTOR_REVIEWER_PASS=[[wash_collector_reviewer_pass]]
|
COLLECTOR_REVIEWER_PASS=[[wash_collector_reviewer_pass]]
|
||||||
|
|||||||
@@ -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.
|
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 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
|
+ 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
|
Xeon is enough) — the `trainer` service beside the collector in
|
||||||
`profiles: ["train"]`, a one-off job never started by a deploy (built 2026-09-07; the CSV
|
`docker-compose.collector.yml` (the CSV export stays for a human with a spreadsheet).
|
||||||
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`
|
**Where the data lives.** The collector writes to `/data` in its container: `collector.sqlite`
|
||||||
and one JPEG per item at `crops/<booth-id>/<item-id>.jpg`. `/data` is the named Docker volume
|
and one JPEG per item at `crops/<booth-id>/<item-id>.jpg`. `/data` is the named Docker volume
|
||||||
|
|||||||
@@ -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.
|
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
|
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).
|
image `parking-trainer`, the `trainer` service beside the collector on the reviewer's host —
|
||||||
`train` reads the collector's `collector.sqlite` and `crops/` **straight off the volume**
|
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
|
(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
|
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
|
reflects tomorrow's traffic), drops classes with fewer than `--min-per-class` (20) labels from
|
||||||
@@ -129,28 +134,44 @@ What the owner has: an **NVIDIA Quadro FX 3800** (in hand, not installed), and i
|
|||||||
host, so nothing moves.
|
host, so nothing moves.
|
||||||
- **Consequences for the build (done):** the trainer image is **CPU-only PyTorch** (torch
|
- **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
|
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
|
`docker-compose.collector.yml` is real — always on, serving the job API, no device
|
||||||
(one block to add if a modern card ever lands; the trainer would pick up CUDA), the collector
|
reservation (one block to add if a modern card ever lands; the trainer would pick up CUDA),
|
||||||
volume mounted read-only, output to `TRAINER_OUT` on the host (default `./models` beside the
|
the collector volume mounted read-only, models/reports/logs in its own `trainer-out` volume.
|
||||||
compose file).
|
|
||||||
- **If faster is ever wanted:** a used mid-range card of the last few generations (~€200) turns
|
- **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
|
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
|
would leave the premises, and even scrubbed of plates and site that runs against the whole
|
||||||
privacy design of the outbox.
|
privacy design of the outbox.
|
||||||
|
|
||||||
## Running it (on the collector host)
|
## Running it
|
||||||
|
|
||||||
```
|
From the collector's `/review` page, Training section: **Train** (mode, backbone, floor) when
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer inspect
|
readiness says enough labels; watch the log; read the report under Versions; **Evaluate** a
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer train --min-accuracy 0.85
|
written version against labels reviewed since; **Publish** it (needs `TRAINER_PUBLISH_TOKEN`
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer evaluate --model /out/<version>/bodytype.onnx
|
in the `wash-collector` stack — commented until the first publish). Then, in git: write the
|
||||||
docker compose -f docker-compose.collector.yml --profile train run --rm trainer publish /out/<version> --url https://git.infra.msai.al/api/packages/mca/generic/parking-bodytype
|
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 CLI is still there for debugging, inside the running container:
|
||||||
the image, bump the booth's `TAG`. The trainer is never started by a deploy (a profile), and the
|
`docker compose -f docker-compose.collector.yml exec trainer parking-trainer inspect`.
|
||||||
`wash-collector` stack's `TRAINER_OUT` / `TRAINER_PUBLISH_TOKEN` lines stay commented until the
|
|
||||||
first publish.
|
## Operating notes (2026-09-07)
|
||||||
|
|
||||||
|
- **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.
|
||||||
|
- **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)
|
## Packaging rule (same as the vision service)
|
||||||
|
|
||||||
|
|||||||
@@ -484,3 +484,16 @@ booth, `pkexec dpkg -i`, polkit dialog, relaunch, badge shows 0.1.7). The prompt
|
|||||||
- **Operator-facing consequence:** the in-app prompt now says the install needs the
|
- **Operator-facing consequence:** the in-app prompt now says the install needs the
|
||||||
administrator password (i18n `update.prompt`, en + sq). An operator who accepts and can't
|
administrator password (i18n `update.prompt`, en + sq). An operator who accepts and can't
|
||||||
authenticate simply stays on the current version; nothing breaks, and the failure is logged.
|
authenticate simply stays on the current version; nothing breaks, and the failure is logged.
|
||||||
|
|
||||||
|
### v0.2.0 — the first feature release of the desktop bundle (2026-09-07)
|
||||||
|
|
||||||
|
Every tag from v0.1.0 to v0.1.7 was a desktop-shell fix (origins, cookies, WS tickets, the
|
||||||
|
updater manifest). Since v0.1.7 the SPA the bundle carries (`frontendDist: ../../web/dist`)
|
||||||
|
gained the venue-module registry, the Car Wash module with per-till shifts and the wash-desk
|
||||||
|
printer role, roles that remember their jobs with signed edits, the advisory vehicle category
|
||||||
|
from the entry camera, the review outbox status in Setup, and the two-column Car Wash setup —
|
||||||
|
31 commits, none of them shell fixes. Under 0.x that is a **minor** bump, not a patch: **v0.2.0**.
|
||||||
|
`tauri.conf.json` now says 0.2.0 too (the release workflow still rewrites it from the tag, so
|
||||||
|
the file only matters for local bundles). The README's release gate — run the real bundle, LIVE,
|
||||||
|
one mutation, a frontend log row — is still the step between the tag and the push of the tag.
|
||||||
|
|
||||||
|
|||||||
@@ -158,9 +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"]`).
|
(`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
|
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.
|
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`,
|
The same stack carries the phase-B **trainer** as a second service ([[bodytype-classifier-training]]):
|
||||||
[[bodytype-classifier-training]]): a deploy never starts it; the owner runs it by hand on the host
|
a deploy starts both, `docker ps` shows two containers, and the trainer is driven from the
|
||||||
with `docker compose … --profile train run --rm trainer …`. Its two env lines (`TRAINER_OUT`, 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
|
`TRAINER_PUBLISH_TOKEN` secret reference) stay commented in `resources.toml` until the first
|
||||||
publish.
|
publish.
|
||||||
|
|
||||||
|
|||||||
+29
@@ -3144,6 +3144,35 @@ 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
|
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."
|
[[opencv-anpr-service]], [[vision-review-outbox]], index. User: "No build just yet."
|
||||||
|
|
||||||
|
## [2026-09-07] decision | Desktop v0.2.0 — a minor bump, not a patch
|
||||||
|
User: "Do you think we are ready for version 0.2.0? The actual version is 0.1.7." Yes: v0.1.x
|
||||||
|
were all shell fixes; the bundled SPA now carries the module registry, Car Wash + per-till
|
||||||
|
shifts, roles jobs, the vision category and the review outbox (31 commits since v0.1.7).
|
||||||
|
`tauri.conf.json` set to 0.2.0, annotated tag `v0.2.0` created locally; the release gate in
|
||||||
|
apps/desktop/README.md (real bundle, LIVE, a mutation, a frontend log row) stands between the
|
||||||
|
tag and its push. Recorded on [[desktop-shell-tauri]].
|
||||||
|
|
||||||
|
## [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/<v>/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
|
||||||
|
pulled by a deploy; run by hand, exits. Recorded on [[bodytype-classifier-training]] §Operating
|
||||||
|
notes (why the TAG bump still mattered, park-2 needs no bump until a pin, the registry login for
|
||||||
|
the first pull, the order of commands) and [[fleet-deployment-komodo]].
|
||||||
|
|
||||||
## [2026-09-07] build | Phase B trainer + the classifier stage on the booth
|
## [2026-09-07] build | Phase B trainer + the classifier stage on the booth
|
||||||
User: "Shall we go and build the trainer for the Xeon?" Built `apps/trainer` (`parking-trainer`:
|
User: "Shall we go and build the trainer for the Xeon?" Built `apps/trainer` (`parking-trainer`:
|
||||||
`inspect` / `train` / `evaluate` / `publish`; reads the collector volume read-only, time split,
|
`inspect` / `train` / `evaluate` / `publish`; reads the collector volume read-only, time split,
|
||||||
|
|||||||
Reference in New Issue
Block a user