feat(trainer): training from the collector UI — the trainer becomes a job service, the review page gains a Training section
Trainer: `parking-trainer serve` — a stdlib HTTP job API on the compose network (never
published): /health, /readiness, /versions, /versions/<v>/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
This commit is contained in:
+13
-7
@@ -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"]
|
||||
|
||||
@@ -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 |
|
||||
| `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.
|
||||
|
||||
@@ -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)")
|
||||
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 <version>/ folder a passing run wrote")
|
||||
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.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
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user