From dbbb051ebd927c2f1c021962a487489ee65b3e08 Mon Sep 17 00:00:00 2001 From: Julian Cuni Date: Mon, 7 Sep 2026 09:38:55 +0200 Subject: [PATCH] feat(carwash): entry-stream sampling for the review outbox; park-2 wired to the collector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wash stream is small; the entry camera photographs every car in exactly the view the classifier is trained on. The booth can now queue entry vehicle reads as pure training material — crop + the camera's class, no order, no operator, no category. - Core announces every vehicle read (deviceEvents.emitVehicleRead from snapshot.ts); the Car Wash module listens, samples entry reads in-process (sampleEntry: exactly one in N) and queues them (enqueueEntry). CARWASH_REVIEW_ENTRY_SAMPLE=N; 1 = every entry (storage and bandwidth are not the limit — user); 0/unset = off. Forwarded by compose. - Packages carry kind: "wash" | "entry". Collector: kind column, entry meta validated without the operator fields, review screen shows an entry sample as such, export has a kind column, operator agreement computed from wash items only. Setup line shows "1 in N entries sampled"; status carries entrySample. - komodo: park-2's four review lines enabled (collector URL by Netbird DNS name, booth-2, the shared per-booth secret, every entry sampled) — the collector is up on the overlay. - Tests on both sides. Wiki: vision-review-outbox (entry stream + the internet-feed assessment), log. Claude-Session: https://claude.ai/code/session_01FWncR69HgGPuei1dLrW3cU --- apps/collector/src/app.test.ts | 20 ++++- apps/collector/src/app.ts | 53 +++++++----- apps/collector/src/db.ts | 25 +++--- apps/collector/src/review-page.ts | 9 ++- apps/server/.env.example | 5 ++ apps/server/src/device-events.ts | 19 +++++ apps/server/src/modules/carwash/index.ts | 8 ++ .../src/modules/carwash/review-outbox.test.ts | 30 ++++++- .../src/modules/carwash/review-outbox.ts | 80 +++++++++++++++---- apps/server/src/modules/carwash/routes.ts | 2 +- apps/server/src/snapshot.ts | 11 ++- apps/web/src/lib/i18n/en.ts | 1 + apps/web/src/lib/i18n/sq.ts | 1 + apps/web/src/modules/carwash/api.ts | 2 + docker-compose.yml | 1 + komodo/resources.toml | 9 ++- wiki/concepts/vision-review-outbox.md | 18 +++++ wiki/log.md | 12 +++ 18 files changed, 242 insertions(+), 64 deletions(-) diff --git a/apps/collector/src/app.test.ts b/apps/collector/src/app.test.ts index c8d6755..3635e98 100644 --- a/apps/collector/src/app.test.ts +++ b/apps/collector/src/app.test.ts @@ -109,8 +109,8 @@ describe("review + export", () => { const stats = (await app.inject({ method: "GET", url: "/api/stats", headers: { authorization: basic } })).json(); expect(stats.booths).toEqual([ - { booth: "booth-7", received: 2, pending: 0, reviewed: 2 }, - { booth: "booth-9", received: 1, pending: 0, reviewed: 1 }, + { booth: "booth-7", received: 2, pending: 0, reviewed: 2, entries: 0 }, + { booth: "booth-9", received: 1, pending: 0, reviewed: 1, entries: 0 }, ]); expect(stats.operators).toEqual([ { booth: "booth-7", operatorRef: "ab12cd34ef56ab12", reviewed: 2, agree: 1, disagree: 1, unusable: 0 }, @@ -120,9 +120,21 @@ describe("review + export", () => { const csv = await app.inject({ method: "GET", url: "/export/labels.csv", headers: { authorization: basic } }); expect(csv.statusCode).toBe(200); const lines = csv.body.trim().split("\n"); - expect(lines[0]).toBe("item,booth,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"); + expect(lines[0]).toBe("item,booth,kind,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"); expect(lines).toHaveLength(3); // header + 2 usable labels; the unusable one is left out - expect(lines[1]).toContain('"item-1","booth-7","crops/booth-7/item-1.jpg","suv","Vetura","car|sedan|hatchback","suv"'); + expect(lines[1]).toContain('"item-1","booth-7","wash","crops/booth-7/item-1.jpg","suv","Vetura","car|sedan|hatchback","suv"'); + + // An ENTRY sample: no order, no operator — accepted, reviewable, in the export, and + // never counted in any operator's agreement. + const entry = await ingest({ v: 1, kind: "entry", booth: "booth-7", item: "entry-1", at: "2026-09-06T11:00:00.000Z", vision: { class: "car", confidence: 0.7 }, image: { width: 300, height: 180, plateBlurred: true } }); + expect(entry.statusCode).toBe(201); + expect((await ingest({ v: 1, kind: "entry", booth: "booth-7", item: "entry-2", at: "x", vision: { class: "car", confidence: 0.7 }, image: { width: 1, height: 1, plateBlurred: true } })).statusCode).toBe(422); + expect((await post("entry-1", "suv")).statusCode).toBe(200); + const stats2 = (await app.inject({ method: "GET", url: "/api/stats", headers: { authorization: basic } })).json(); + expect(stats2.booths[0]).toEqual({ booth: "booth-7", received: 3, pending: 0, reviewed: 3, entries: 1 }); + expect(stats2.operators.find((o: { booth: string }) => o.booth === "booth-7")).toMatchObject({ reviewed: 2, agree: 1, disagree: 1 }); + const csv3 = (await app.inject({ method: "GET", url: "/export/labels.csv", headers: { authorization: basic } })).body; + expect(csv3).toContain('"entry-1","booth-7","entry","crops/booth-7/entry-1.jpg","suv","","","car"'); // A booth-supplied name that looks like a spreadsheet formula is neutralised in the export. await ingest(meta({ item: "item-4", operatorCategory: { id: "x", name: "=HYPERLINK(\"http://evil\")", classes: ["car"] } })); diff --git a/apps/collector/src/app.ts b/apps/collector/src/app.ts index cbaa298..739e489 100644 --- a/apps/collector/src/app.ts +++ b/apps/collector/src/app.ts @@ -20,15 +20,18 @@ import { reviewPage } from "./review-page.js"; /** The package's `meta` part, as the booth sends it (review-outbox.ts). */ interface IngestMeta { v: number; + /** "wash" (default when absent) = a desk decision; "entry" = a sampled entry read with + * no order and no operator — crop + the camera's class only. */ + kind?: "wash" | "entry"; booth: string; item: string; - order: string; + order?: string; at: string; - operator: string; - operatorCategory: { id: string; name: string; classes?: string[] }; - service: string; - vision: { class: string; confidence: number; categoryId: string | null }; - downgraded: boolean; + operator?: string; + operatorCategory?: { id: string; name: string; classes?: string[] }; + service?: string; + vision: { class: string; confidence: number; categoryId?: string | null }; + downgraded?: boolean; image: { width: number; height: number; plateBlurred: boolean }; } @@ -46,17 +49,21 @@ function checkMeta(m: unknown, booth: string): { ok: true; meta: IngestMeta } | if (x.v !== 1) return { ok: false, why: "unsupported meta version" }; if (x.booth !== booth) return { ok: false, why: "meta.booth does not match the token's booth" }; if (!str(x.item, 64) || !ID_RE.test(x.item as string)) return { ok: false, why: "bad item id" }; - if (!str(x.order, 64)) return { ok: false, why: "bad order ref" }; if (!str(x.at, 40) || Number.isNaN(Date.parse(x.at as string))) return { ok: false, why: "bad timestamp" }; - if (!str(x.operator, 64)) return { ok: false, why: "bad operator ref" }; - const oc = x.operatorCategory as Record | undefined; - if (!oc || !str(oc.id, 64) || !str(oc.name, 120)) return { ok: false, why: "bad operatorCategory" }; - if (oc.classes !== undefined && (!Array.isArray(oc.classes) || !oc.classes.every(isVehicleClass))) return { ok: false, why: "bad operatorCategory.classes" }; - if (!str(x.service, 120)) return { ok: false, why: "bad service" }; + const kind = x.kind === undefined ? "wash" : x.kind; + if (kind !== "wash" && kind !== "entry") return { ok: false, why: "bad kind" }; const v = x.vision as Record | undefined; if (!v || !isVehicleClass(v.class) || typeof v.confidence !== "number" || v.confidence < 0 || v.confidence > 1) return { ok: false, why: "bad vision read" }; if (v.categoryId != null && !str(v.categoryId, 64)) return { ok: false, why: "bad vision.categoryId" }; - if (typeof x.downgraded !== "boolean") return { ok: false, why: "bad downgraded" }; + if (kind === "wash") { + if (!str(x.order, 64)) return { ok: false, why: "bad order ref" }; + if (!str(x.operator, 64)) return { ok: false, why: "bad operator ref" }; + const oc = x.operatorCategory as Record | undefined; + if (!oc || !str(oc.id, 64) || !str(oc.name, 120)) return { ok: false, why: "bad operatorCategory" }; + if (oc.classes !== undefined && (!Array.isArray(oc.classes) || !oc.classes.every(isVehicleClass))) return { ok: false, why: "bad operatorCategory.classes" }; + if (!str(x.service, 120)) return { ok: false, why: "bad service" }; + if (typeof x.downgraded !== "boolean") return { ok: false, why: "bad downgraded" }; + } const im = x.image as Record | undefined; if (!im || typeof im.width !== "number" || typeof im.height !== "number" || typeof im.plateBlurred !== "boolean") return { ok: false, why: "bad image meta" }; return { ok: true, meta: x as unknown as IngestMeta }; @@ -148,16 +155,18 @@ export async function buildCollector(cfg: CollectorConfig, opts: { dbFile?: stri const rel = path.posix.join("crops", booth, `${meta.item}.jpg`); await mkdir(path.join(cfg.dataDir, "crops", booth), { recursive: true }); await writeFile(path.join(cfg.dataDir, rel), image); + const kind = meta.kind ?? "wash"; db.insert({ id: meta.item, booth, - orderRef: meta.order, + kind, + orderRef: meta.order ?? "", at: meta.at, - operatorRef: meta.operator, - operatorCategoryId: meta.operatorCategory.id, - operatorCategoryName: meta.operatorCategory.name, - operatorClasses: JSON.stringify(meta.operatorCategory.classes ?? []), - service: meta.service, + operatorRef: meta.operator ?? "", + operatorCategoryId: meta.operatorCategory?.id ?? "", + operatorCategoryName: meta.operatorCategory?.name ?? "", + operatorClasses: JSON.stringify(meta.operatorCategory?.classes ?? []), + service: meta.service ?? "", visionClass: meta.vision.class, visionConfidence: meta.vision.confidence, visionCategoryId: meta.vision.categoryId ?? null, @@ -168,7 +177,7 @@ export async function buildCollector(cfg: CollectorConfig, opts: { dbFile?: stri imagePath: rel, receivedAt: new Date().toISOString(), }); - req.log.info(`ingest: ${booth} item ${meta.item} (${meta.vision.class} → ${meta.operatorCategory.name})`); + req.log.info(`ingest: ${booth} ${kind} ${meta.item} (${meta.vision.class}${kind === "wash" ? ` → ${meta.operatorCategory!.name}` : ""})`); return reply.code(201).send({ ok: true }); }); @@ -214,9 +223,9 @@ export async function buildCollector(cfg: CollectorConfig, opts: { dbFile?: stri if (/^[=+\-@\t\r]/.test(v)) v = `'${v}`; return `"${v.replace(/"/g, '""')}"`; }; - const head = "item,booth,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"; + const head = "item,booth,kind,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"; const lines = rows.map((r) => - [r.id, r.booth, r.imagePath, r.reviewLabel, r.operatorCategoryName, JSON.parse(r.operatorClasses).join("|"), r.visionClass, r.visionConfidence, r.downgraded, r.at, r.reviewedAt].map(q).join(","), + [r.id, r.booth, r.kind, r.imagePath, r.reviewLabel, r.operatorCategoryName, JSON.parse(r.operatorClasses).join("|"), r.visionClass, r.visionConfidence, r.downgraded, r.at, r.reviewedAt].map(q).join(","), ); return reply.type("text/csv; charset=utf-8").header("content-disposition", 'attachment; filename="labels.csv"').send([head, ...lines].join("\n") + "\n"); }); diff --git a/apps/collector/src/db.ts b/apps/collector/src/db.ts index 5273482..abaf8db 100644 --- a/apps/collector/src/db.ts +++ b/apps/collector/src/db.ts @@ -9,6 +9,9 @@ import type { VehicleClass } from "@parking/shared"; export interface ItemRow { id: string; booth: string; + /** "wash" = a desk decision (operator fields set); "entry" = a sampled entry read (pure + * training material: crop + the camera's class, operator fields empty). */ + kind: "wash" | "entry"; orderRef: string; at: string; operatorRef: string; @@ -44,11 +47,12 @@ export class CollectorDb { CREATE TABLE IF NOT EXISTS items ( id TEXT PRIMARY KEY, booth TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT 'wash', order_ref TEXT NOT NULL, at TEXT NOT NULL, - operator_ref TEXT NOT NULL, - operator_category_id TEXT NOT NULL, - operator_category_name TEXT NOT NULL, + operator_ref TEXT NOT NULL DEFAULT '', + operator_category_id TEXT NOT NULL DEFAULT '', + operator_category_name TEXT NOT NULL DEFAULT '', operator_classes TEXT NOT NULL DEFAULT '[]', service TEXT NOT NULL, vision_class TEXT NOT NULL, @@ -77,6 +81,7 @@ export class CollectorDb { return { id: r.id as string, booth: r.booth as string, + kind: r.kind === "entry" ? "entry" : "wash", orderRef: r.order_ref as string, at: r.at as string, operatorRef: r.operator_ref as string, @@ -107,10 +112,10 @@ export class CollectorDb { insert(row: Omit): void { this.#db .prepare( - `INSERT INTO items (id, booth, order_ref, at, operator_ref, operator_category_id, operator_category_name, + `INSERT INTO items (id, booth, kind, order_ref, at, operator_ref, operator_category_id, operator_category_name, operator_classes, service, vision_class, vision_confidence, vision_category_id, downgraded, image_width, image_height, plate_blurred, image_path, received_at) - VALUES (@id, @booth, @orderRef, @at, @operatorRef, @operatorCategoryId, @operatorCategoryName, + VALUES (@id, @booth, @kind, @orderRef, @at, @operatorRef, @operatorCategoryId, @operatorCategoryName, @operatorClasses, @service, @visionClass, @visionConfidence, @visionCategoryId, @downgraded, @imageWidth, @imageHeight, @plateBlurred, @imagePath, @receivedAt)`, ) @@ -142,19 +147,21 @@ export class CollectorDb { * reviewer's class fell inside the operator's chosen category (agree) or outside * (disagree) — the honest-mistake / fraud rate the outbox exists for. */ stats(): { - booths: { booth: string; received: number; pending: number; reviewed: number }[]; + booths: { booth: string; received: number; pending: number; reviewed: number; entries: number }[]; operators: { booth: string; operatorRef: string; reviewed: number; agree: number; disagree: number; unusable: number }[]; } { const booths = this.#db .prepare( `SELECT booth, COUNT(*) AS received, SUM(CASE WHEN reviewed_at IS NULL THEN 1 ELSE 0 END) AS pending, - SUM(CASE WHEN reviewed_at IS NOT NULL THEN 1 ELSE 0 END) AS reviewed + SUM(CASE WHEN reviewed_at IS NOT NULL THEN 1 ELSE 0 END) AS reviewed, + SUM(CASE WHEN kind = 'entry' THEN 1 ELSE 0 END) AS entries FROM items GROUP BY booth ORDER BY booth`, ) - .all() as { booth: string; received: number; pending: number; reviewed: number }[]; + .all() as { booth: string; received: number; pending: number; reviewed: number; entries: number }[]; + // Operator agreement is a WASH thing — an entry sample has no operator decision. const reviewed = this.#db - .prepare("SELECT booth, operator_ref, operator_classes, review_label FROM items WHERE reviewed_at IS NOT NULL") + .prepare("SELECT booth, operator_ref, operator_classes, review_label FROM items WHERE reviewed_at IS NOT NULL AND kind = 'wash'") .all() as { booth: string; operator_ref: string; operator_classes: string; review_label: string }[]; const ops = new Map(); for (const r of reviewed) { diff --git a/apps/collector/src/review-page.ts b/apps/collector/src/review-page.ts index 0e42479..9f9fd11 100644 --- a/apps/collector/src/review-page.ts +++ b/apps/collector/src/review-page.ts @@ -80,10 +80,13 @@ async function next() { el.innerHTML = '' + '
' + - '
operator chose
' + esc(it.operatorCategoryName) + ' (' + esc(opClasses.join(', ') || 'no classes mapped') + ')
' + + (it.kind === 'entry' + ? '
sample
entry stream — no wash, no operator decision; label the vehicle
' + : '
operator chose
' + esc(it.operatorCategoryName) + ' (' + esc(opClasses.join(', ') || 'no classes mapped') + ')
') + '
camera saw
' + esc(it.visionClass) + ' ' + Math.round(it.visionConfidence * 100) + '%' + (it.downgraded ? ' flagged downgrade at the booth' : '') + '
' + - '
service
' + esc(it.service) + '
' + - '
booth · operator
' + esc(it.booth) + ' · ' + esc(it.operatorRef) + '
' + + (it.kind === 'entry' ? '
booth
' + esc(it.booth) + '
' : + '
service
' + esc(it.service) + '
' + + '
booth · operator
' + esc(it.booth) + ' · ' + esc(it.operatorRef) + '
') + '
at
' + esc(it.at) + '
' + '
' + '
' + diff --git a/apps/server/.env.example b/apps/server/.env.example index c0c724d..f1e6c1e 100644 --- a/apps/server/.env.example +++ b/apps/server/.env.example @@ -99,3 +99,8 @@ WS_ALLOWED_ORIGINS=http://localhost:5173,tauri://localhost,http://tauri.localhos # CARWASH_REVIEW_TOKEN= # CARWASH_REVIEW_BOOTH_ID= # CARWASH_REVIEW_INTERVAL_SEC=60 +# Entry-stream sampling: also queue one in N ENTRY vehicle reads (no wash, no operator) as +# pure training material in the gate view — many times the wash stream, zero domain shift. +# 1 = every entry (the reviewer labels what they have time for; the rest waits and stays +# useful), N = one in N, 0/unset = off. Needs the three settings above. +# CARWASH_REVIEW_ENTRY_SAMPLE=1 diff --git a/apps/server/src/device-events.ts b/apps/server/src/device-events.ts index b7821fc..508751c 100644 --- a/apps/server/src/device-events.ts +++ b/apps/server/src/device-events.ts @@ -1,6 +1,7 @@ import { EventEmitter } from "node:events"; import type { PrinterStatus } from "@parking/devices"; import type { LedgerEventRow } from "@parking/db"; +import type { VehicleRead } from "@parking/shared"; // Internal event bus for device-originated events (button presses, etc.). // Hardware drivers / inbound device pushes emit here; business logic (entry @@ -105,6 +106,17 @@ export interface PlateRecognizedEvent { readonly direction: "entry" | "exit"; } +/** Emitted when vision classified the vehicle in an entry/exit frame (advisory; stored on + * the read row like the plate). A module may sample these — the Car Wash review outbox + * queues one in N ENTRY reads for the remote reviewer, in the gate view the classifier + * will be trained on (wiki/concepts/vision-review-outbox.md). The core emits; it never + * knows who listens. */ +export interface VehicleReadEvent { + readonly identity: string; + readonly direction: "entry" | "exit"; + readonly read: VehicleRead; +} + /** Per-lane RADAR presence — a vehicle-presence INPUT (loop/radar) is shorted at the * entry/exit barrier, i.e. "something is in the lane vicinity" BEFORE the camera has * confirmed a vehicle. Same signal that makes the physical button lamp (relay 3) blink: @@ -197,6 +209,13 @@ class DeviceEventBus extends EventEmitter { this.on("plate-recognized", cb); return () => this.off("plate-recognized", cb); } + emitVehicleRead(event: VehicleReadEvent): void { + this.emit("vehicle-read", event); + } + onVehicleRead(cb: (event: VehicleReadEvent) => void): () => void { + this.on("vehicle-read", cb); + return () => this.off("vehicle-read", cb); + } } /** Process-wide device event bus. */ diff --git a/apps/server/src/modules/carwash/index.ts b/apps/server/src/modules/carwash/index.ts index f266f31..2555f60 100644 --- a/apps/server/src/modules/carwash/index.ts +++ b/apps/server/src/modules/carwash/index.ts @@ -1,3 +1,4 @@ +import { deviceEvents } from "../../device-events.js"; import type { ServerModule } from "../index.js"; import { ReviewOutbox, reviewUploadConfigFromEnv } from "./review-outbox.js"; import { carwashRoutes } from "./routes.js"; @@ -18,6 +19,13 @@ export const carwashModule: ServerModule = { const outbox = new ReviewOutbox(deps.db, app.log, cfg); app.log.info(cfg ? `carwash review upload: on → ${new URL(cfg.url).host} as ${cfg.boothId}` : "carwash review upload: off"); outbox.start(); + // Entry-stream sampling: one in N entry vehicle reads goes to the reviewer as pure + // training material (the gate view, no order attached). The core announces the read; + // the module decides. Off unless CARWASH_REVIEW_ENTRY_SAMPLE is set. + const offVehicleRead = deviceEvents.onVehicleRead((e) => { + if (e.direction === "entry" && outbox.sampleEntry()) void outbox.enqueueEntry(e.read); + }); + app.addHook("onClose", async () => offVehicleRead()); app.addHook("onClose", async () => outbox.stop()); const service = new CarwashService(deps, app.log, outbox); // A wash ordered with payAt = "booth" is a charge line on the parking settlement; diff --git a/apps/server/src/modules/carwash/review-outbox.test.ts b/apps/server/src/modules/carwash/review-outbox.test.ts index cf349a5..d83dfe8 100644 --- a/apps/server/src/modules/carwash/review-outbox.test.ts +++ b/apps/server/src/modules/carwash/review-outbox.test.ts @@ -3,6 +3,7 @@ import sharp from "sharp"; import { createTestDb } from "@parking/db/testing"; import { carwashOrders, carwashReviewOutbox, deviceEvents, snapshots, type Db } from "@parking/db"; import type { FastifyInstance } from "fastify"; +import { deviceEvents as deviceEventBus } from "../../device-events.js"; import { buildServer } from "../../server.js"; import { login, makeLog, minutesAgo, seedTariff, seedUser, silentLogger } from "../../test-helpers.js"; import { EXPIRE_DAYS, ReviewOutbox, makeReviewCrop, operatorRef, reviewUploadConfigFromEnv } from "./review-outbox.js"; @@ -81,7 +82,7 @@ describe("queue + drain", () => { }); afterEach(() => close()); - const cfg = { url: "https://collector.overlay/ingest", token: "secret-1", boothId: "booth-7", intervalSec: 60 }; + const cfg = { url: "https://collector.overlay/ingest", token: "secret-1", boothId: "booth-7", intervalSec: 60, entrySample: 0 }; const read = { bodyType: "car" as const, confidence: 0.86, snapshotId: "snap-1", box: CAR, plateBox: PLATE }; const item = { orderId: "o-1", createdAt: "2026-09-06T10:00:00.000Z", createdBy: "lavazhier", categoryId: "car", categoryName: "Vetura", categoryClasses: ["car", "sedan"], serviceName: "Standard", visionCategoryId: "car", downgraded: false }; @@ -105,7 +106,7 @@ describe("queue + drain", () => { const row = db.select().from(carwashReviewOutbox).all()[0]!; expect(row.status).toBe("queued"); expect(row.image!.length).toBeGreaterThan(500); - expect(row.payload).toMatchObject({ v: 1, booth: "booth-7", order: "o-1", operatorCategory: { id: "car", name: "Vetura", classes: ["car", "sedan"] }, vision: { class: "car", confidence: 0.86 }, downgraded: false, image: { plateBlurred: true } }); + expect(row.payload).toMatchObject({ v: 1, kind: "wash", booth: "booth-7", order: "o-1", operatorCategory: { id: "car", name: "Vetura", classes: ["car", "sedan"] }, vision: { class: "car", confidence: 0.86 }, downgraded: false, image: { plateBlurred: true } }); expect(JSON.stringify(row.payload)).not.toContain("lavazhier"); expect(await ob.drain()).toEqual({ sent: 1, failed: 0, deferred: 0 }); @@ -159,6 +160,18 @@ describe("queue + drain", () => { expect(fetchFn.mock.calls.length).toBe(before); expect(ob.status().failed).toBe(3); + // Entry sampling: one in N entry reads becomes a package with the crop and the + // camera's class only — no order, no operator, no category. + const sampler = new ReviewOutbox(db, silentLogger(), { ...cfg, entrySample: 3 }, fetchFn); + expect([sampler.sampleEntry(), sampler.sampleEntry(), sampler.sampleEntry(), sampler.sampleEntry()]).toEqual([false, false, true, false]); + expect(ob.sampleEntry()).toBe(false); // entrySample 0 = off + expect(await sampler.enqueueEntry(read)).toBe(true); + const entryRow = db.select().from(carwashReviewOutbox).where(eq(carwashReviewOutbox.orderId, "entry:snap-1")).get()!; + expect(entryRow.payload).toMatchObject({ v: 1, kind: "entry", booth: "booth-7", vision: { class: "car", confidence: 0.86 }, image: { plateBlurred: true } }); + expect(entryRow.payload).not.toHaveProperty("operator"); + expect(entryRow.payload).not.toHaveProperty("operatorCategory"); + expect(entryRow.image!.length).toBeGreaterThan(500); + // No vehicle box, no snapshot, or upload off → nothing queued. expect(await ob.enqueue(item, { ...read, box: null })).toBe(false); expect(await ob.enqueue(item, { ...read, snapshotId: "gone" })).toBe(false); @@ -178,6 +191,7 @@ describe("through the app", () => { process.env.CARWASH_REVIEW_URL = "https://collector.overlay/ingest"; process.env.CARWASH_REVIEW_TOKEN = "tok"; process.env.CARWASH_REVIEW_BOOTH_ID = "booth-9"; + process.env.CARWASH_REVIEW_ENTRY_SAMPLE = "1"; const t = createTestDb(); db = t.db; close = t.close; @@ -187,7 +201,7 @@ describe("through the app", () => { afterEach(async () => { await app.close(); close(); - for (const k of ["CARWASH_REVIEW_URL", "CARWASH_REVIEW_TOKEN", "CARWASH_REVIEW_BOOTH_ID"]) { + for (const k of ["CARWASH_REVIEW_URL", "CARWASH_REVIEW_TOKEN", "CARWASH_REVIEW_BOOTH_ID", "CARWASH_REVIEW_ENTRY_SAMPLE"]) { if (saved[k] === undefined) delete process.env[k]; else process.env[k] = saved[k]; } @@ -214,6 +228,14 @@ describe("through the app", () => { // Enqueue is fire-and-forget: give the crop a moment. await vi.waitFor(() => expect(db.select().from(carwashReviewOutbox).all()).toHaveLength(1)); const status = (await app.inject({ method: "GET", url: "/api/carwash/review/status", headers: { cookie: a.cookie } })).json(); - expect(status).toMatchObject({ enabled: true, boothId: "booth-9", queued: 1, sent: 0 }); + expect(status).toMatchObject({ enabled: true, boothId: "booth-9", queued: 1, sent: 0, entrySample: 1 }); + + // An ENTRY vehicle read announced by the core (snapshot.ts) is sampled by the module + // (1 in 1 here) into an entry package; an exit read is not. + deviceEventBus.emitVehicleRead({ identity: "T-X", direction: "exit", read: { bodyType: "car", confidence: 0.8, snapshotId: "snap-r", box: CAR, plateBox: PLATE } }); + deviceEventBus.emitVehicleRead({ identity: "T-R", direction: "entry", read: { bodyType: "car", confidence: 0.8, snapshotId: "snap-r", box: CAR, plateBox: PLATE } }); + await vi.waitFor(() => expect(db.select().from(carwashReviewOutbox).all()).toHaveLength(2)); + const rows = db.select().from(carwashReviewOutbox).all(); + expect(rows.map((r) => (r.payload as { kind: string }).kind).sort()).toEqual(["entry", "wash"]); }); }); diff --git a/apps/server/src/modules/carwash/review-outbox.ts b/apps/server/src/modules/carwash/review-outbox.ts index e4ce895..751cc2b 100644 --- a/apps/server/src/modules/carwash/review-outbox.ts +++ b/apps/server/src/modules/carwash/review-outbox.ts @@ -34,6 +34,10 @@ export interface ReviewUploadConfig { /** Pseudonymous booth id — a label the reviewer maps to a site; never the site name. */ readonly boothId: string; readonly intervalSec: number; + /** Queue one in N ENTRY vehicle reads (no order attached) for the reviewer — the gate + * view is exactly what the classifier is trained on, and the entry stream is many times + * the wash stream. 0 = off. */ + readonly entrySample: number; } /** From the server env (Komodo stack env). All three of URL, token and booth id, or off. */ @@ -43,7 +47,12 @@ export function reviewUploadConfigFromEnv(env: NodeJS.ProcessEnv = process.env): const boothId = (env.CARWASH_REVIEW_BOOTH_ID ?? "").trim(); if (!url || !token || !boothId) return null; const raw = Number(env.CARWASH_REVIEW_INTERVAL_SEC ?? 60); - return { url, token, boothId, intervalSec: Number.isFinite(raw) && raw >= 10 ? raw : 60 }; + const sample = Number(env.CARWASH_REVIEW_ENTRY_SAMPLE ?? 0); + return { + url, token, boothId, + intervalSec: Number.isFinite(raw) && raw >= 10 ? raw : 60, + entrySample: Number.isInteger(sample) && sample > 0 ? sample : 0, + }; } /** The crop's longest edge, in pixels — enough for a reviewer and a classifier, small @@ -145,6 +154,8 @@ export interface OutboxStatus { readonly failed: number; readonly lastSentAt: string | null; readonly lastError: string | null; + /** 0 = entry sampling off; N = one in N entry reads is queued. */ + readonly entrySample: number; } export class ReviewOutbox { @@ -154,6 +165,7 @@ export class ReviewOutbox { readonly #fetch: FetchLike; #timer: NodeJS.Timeout | null = null; #draining = false; + #entrySeen = 0; constructor(db: Db, logger: FastifyBaseLogger, cfg: ReviewUploadConfig | null, fetchFn?: FetchLike) { this.#db = db; @@ -172,35 +184,68 @@ export class ReviewOutbox { * sample) or when upload is not configured (an unbounded queue nobody drains). */ async enqueue(item: ReviewItemInput, read: VehicleRead): Promise { if (!this.#cfg) return false; + return this.#queue(item.orderId, read, (id, crop) => ({ + v: 1, + kind: "wash", + booth: this.#cfg!.boothId, + item: id, + order: item.orderId, + at: item.createdAt, + operator: operatorRef(this.#cfg!.boothId, item.createdBy), + operatorCategory: { id: item.categoryId, name: item.categoryName, classes: [...item.categoryClasses] }, + service: item.serviceName, + vision: { class: read.bodyType, confidence: read.confidence, categoryId: item.visionCategoryId }, + downgraded: item.downgraded, + image: crop, + })); + } + + /** Every Nth entry read is a sample (N = entrySample); the caller queues it. Counted + * in-process, so "1 in 5" is exactly that across a booth's day. */ + sampleEntry(): boolean { + const n = this.#cfg?.entrySample ?? 0; + if (n <= 0) return false; + this.#entrySeen += 1; + return this.#entrySeen % n === 0; + } + + /** Queue an ENTRY sample: the crop and the camera's class only — no order, no operator, + * no category. Pure training material in the gate view; the reviewer labels it. */ + async enqueueEntry(read: VehicleRead): Promise { + if (!this.#cfg) return false; + return this.#queue(`entry:${read.snapshotId ?? "?"}`, read, (id, crop) => ({ + v: 1, + kind: "entry", + booth: this.#cfg!.boothId, + item: id, + at: new Date().toISOString(), + vision: { class: read.bodyType, confidence: read.confidence }, + image: crop, + })); + } + + async #queue( + ref: string, + read: VehicleRead, + build: (id: string, image: { width: number; height: number; plateBlurred: boolean }) => Record, + ): Promise { if (!read.box || !read.snapshotId) return false; try { const snap = this.#db.select().from(snapshots).where(eq(snapshots.id, read.snapshotId)).get(); if (!snap) { - this.#logger.info(`carwash review: snapshot ${read.snapshotId} gone (pruned) — order ${item.orderId} not queued`); + this.#logger.info(`carwash review: snapshot ${read.snapshotId} gone (pruned) — ${ref} not queued`); return false; } const crop = await makeReviewCrop(snap.bytes, read.box, read.plateBox); const id = randomUUID(); - const payload = { - v: 1, - booth: this.#cfg.boothId, - item: id, - order: item.orderId, - at: item.createdAt, - operator: operatorRef(this.#cfg.boothId, item.createdBy), - operatorCategory: { id: item.categoryId, name: item.categoryName, classes: [...item.categoryClasses] }, - service: item.serviceName, - vision: { class: read.bodyType, confidence: read.confidence, categoryId: item.visionCategoryId }, - downgraded: item.downgraded, - image: { width: crop.width, height: crop.height, plateBlurred: crop.plateBlurred }, - }; + const payload = build(id, { width: crop.width, height: crop.height, plateBlurred: crop.plateBlurred }); this.#db .insert(carwashReviewOutbox) - .values({ id, orderId: item.orderId, createdAt: new Date().toISOString(), status: "queued", attempts: 0, nextAttemptAt: null, image: crop.bytes, payload }) + .values({ id, orderId: ref, createdAt: new Date().toISOString(), status: "queued", attempts: 0, nextAttemptAt: null, image: crop.bytes, payload }) .run(); return true; } catch (err) { - this.#logger.warn(`carwash review: could not queue order ${item.orderId}: ${(err as Error).message}`); + this.#logger.warn(`carwash review: could not queue ${ref}: ${(err as Error).message}`); return false; } } @@ -320,6 +365,7 @@ export class ReviewOutbox { return { enabled: this.enabled, boothId: this.#cfg?.boothId ?? null, + entrySample: this.#cfg?.entrySample ?? 0, queued: count("queued"), sent: count("sent"), failed: count("failed"), diff --git a/apps/server/src/modules/carwash/routes.ts b/apps/server/src/modules/carwash/routes.ts index fcecda6..6badf94 100644 --- a/apps/server/src/modules/carwash/routes.ts +++ b/apps/server/src/modules/carwash/routes.ts @@ -42,7 +42,7 @@ export async function carwashRoutes(app: FastifyInstance, deps: ServerModuleDeps // The review outbox's health (Setup → Car wash): how many decisions wait for the // reviewer, how many went, the last error. Site admin's read. app.get("/api/carwash/review/status", { preHandler: settingsRead }, async () => - outbox?.status() ?? { enabled: false, boothId: null, queued: 0, sent: 0, failed: 0, lastSentAt: null, lastError: null }, + outbox?.status() ?? { enabled: false, boothId: null, queued: 0, sent: 0, failed: 0, lastSentAt: null, lastError: null, entrySample: 0 }, ); app.put<{ Body: SettingsBody }>("/api/carwash/settings", { preHandler: settingsWrite }, async (req, reply) => { diff --git a/apps/server/src/snapshot.ts b/apps/server/src/snapshot.ts index 65c072f..6dc03c7 100644 --- a/apps/server/src/snapshot.ts +++ b/apps/server/src/snapshot.ts @@ -210,7 +210,16 @@ async function recognizePlate( occurredAt: new Date().toISOString(), }) .run(); - if (result.vehicle) logger.info(`vision vehicle '${result.vehicle.bodyType}' (${result.vehicle.confidence.toFixed(3)}) for ${identity}`); + if (result.vehicle) { + logger.info(`vision vehicle '${result.vehicle.bodyType}' (${result.vehicle.confidence.toFixed(3)}) for ${identity}`); + if (vehicleBox) { + deviceEvents.emitVehicleRead({ + identity, + direction, + read: { bodyType: result.vehicle.bodyType, confidence: result.vehicle.confidence, snapshotId, box: vehicleBox, plateBox }, + }); + } + } if (!plate) return; logger.info(`anpr plate '${plate}' (${result.plate!.confidence.toFixed(3)}) for ${identity}`); // The session's entry/exit event already shipped without this (async) plate — tell the diff --git a/apps/web/src/lib/i18n/en.ts b/apps/web/src/lib/i18n/en.ts index 6743aeb..faa2271 100644 --- a/apps/web/src/lib/i18n/en.ts +++ b/apps/web/src/lib/i18n/en.ts @@ -143,6 +143,7 @@ export const en: Catalog = { reviewTitle: "Remote review", reviewOff: "off — no collector configured for this booth", reviewCounts: "{{queued}} waiting · {{sent}} delivered · {{failed}} abandoned", + reviewEntrySample: "1 in {{n}} entries sampled", reviewHint: "Each wash order sends the vehicle crop (plate blurred) and the chosen category to a trusted reviewer over the private network. One-way; nothing that names this site leaves.", visionClassesHint: "The camera's fixed vocabulary (set in code, not here). Tick the classes this category covers.", visionThreshold: "Camera confidence to flag a downgrade", diff --git a/apps/web/src/lib/i18n/sq.ts b/apps/web/src/lib/i18n/sq.ts index 2343a70..9149cce 100644 --- a/apps/web/src/lib/i18n/sq.ts +++ b/apps/web/src/lib/i18n/sq.ts @@ -145,6 +145,7 @@ export const sq = { reviewTitle: "Shqyrtim në distancë", reviewOff: "joaktiv — asnjë mbledhës i konfiguruar për këtë kabinë", reviewCounts: "{{queued}} në pritje · {{sent}} të dërguara · {{failed}} të braktisura", + reviewEntrySample: "1 në {{n}} hyrje merret mostër", reviewHint: "Çdo porosi lavazhi dërgon prerjen e mjetit (targa e turbulluar) dhe kategorinë e zgjedhur te një shqyrtues i besuar përmes rrjetit privat. Njëkahësh; asgjë që emërton këtë vend nuk del.", visionClassesHint: "Fjalori i fiksuar i kamerës (vendoset në kod, jo këtu). Shëno klasat që mbulon kjo kategori.", visionThreshold: "Siguria e kamerës për të shënuar një ulje kategorie", diff --git a/apps/web/src/modules/carwash/api.ts b/apps/web/src/modules/carwash/api.ts index 6d4a68c..b163c7d 100644 --- a/apps/web/src/modules/carwash/api.ts +++ b/apps/web/src/modules/carwash/api.ts @@ -40,6 +40,8 @@ export interface CarwashReviewStatus { failed: number; lastSentAt: string | null; lastError: string | null; + /** 0 = entry sampling off; N = one in N entry reads is queued as training material. */ + entrySample: number; } export function fetchCarwashReviewStatus(): Promise { return apiFetch("/api/carwash/review/status"); diff --git a/docker-compose.yml b/docker-compose.yml index 5fcc985..7fc1380 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,7 @@ services: CARWASH_REVIEW_URL: ${CARWASH_REVIEW_URL:-} CARWASH_REVIEW_TOKEN: ${CARWASH_REVIEW_TOKEN:-} CARWASH_REVIEW_BOOTH_ID: ${CARWASH_REVIEW_BOOTH_ID:-} + CARWASH_REVIEW_ENTRY_SAMPLE: ${CARWASH_REVIEW_ENTRY_SAMPLE:-0} # CRITICAL on the plain-HTTP booth LAN: cookies are Secure (HTTPS-only) by DEFAULT, # so without COOKIE_SECURE=0 the auth cookie is never sent over http and operators # CANNOT LOG IN. Leave unset only behind TLS. See disk-os-hardening "deploy-time runbook". diff --git a/komodo/resources.toml b/komodo/resources.toml index 843384c..92b763f 100644 --- a/komodo/resources.toml +++ b/komodo/resources.toml @@ -94,9 +94,12 @@ MODULES_ENTITLED=parking,carwash # Car Wash review outbox (wiki/concepts/vision-review-outbox.md): the collector's ingest URL # on the Netbird overlay, this booth's pseudonymous id, and its token — the SAME secret the # wash-collector stack lists under that id. Leave all three unset to keep the outbox off. -#CARWASH_REVIEW_URL=http://docker-station.nb.infra:8090/ingest -#CARWASH_REVIEW_BOOTH_ID=booth-2 -#CARWASH_REVIEW_TOKEN=[[wash_review_token_booth_2]] +CARWASH_REVIEW_URL=http://docker-station.nb.infra:8090/ingest +CARWASH_REVIEW_BOOTH_ID=booth-2 +CARWASH_REVIEW_TOKEN=[[wash_review_token_booth_2]] +# Also send ENTRY reads as training material (gate view, no wash): 1 = every entry (storage +# and bandwidth are not the limit; review what you have time for). N = one in N. 0 = off. +CARWASH_REVIEW_ENTRY_SAMPLE=1 VISION_ENABLED=1 # Desktop app WS handshake: Origin is tauri://localhost (set explicitly by # platform-ws.ts, since the native WS plugin has no page context to auto-attach diff --git a/wiki/concepts/vision-review-outbox.md b/wiki/concepts/vision-review-outbox.md index 6a02e61..8d1959b 100644 --- a/wiki/concepts/vision-review-outbox.md +++ b/wiki/concepts/vision-review-outbox.md @@ -39,6 +39,24 @@ locked-down collector without exposing anything to the open internet. image is dropped from the row once delivered; a voided order is abandoned unsent; anything older than 14 days is abandoned ("expired") rather than resurfacing a fortnight in a burst. +## The entry stream — the real accelerator (built 2026-09-07) + +The wash stream is small; the **entry camera photographs every car**, in exactly the view the +classifier is trained on, with zero domain shift. So the booth can also queue **one in N entry +vehicle reads** as pure training material: the crop and the camera's class, *no* order, *no* +operator, *no* category — same crop-and-blur pipeline, same one-way path, same privacy +properties. `CARWASH_REVIEW_ENTRY_SAMPLE=N` (0/unset = off; needs the three upload settings). +Seam: the core announces every vehicle read (`deviceEvents.emitVehicleRead`, snapshot.ts, entry +and exit) and the Car Wash module decides — it samples entry reads in-process (`sampleEntry()`, +exactly one in N) and calls `enqueueEntry()`; the core never imports the module. Packages carry +`kind: "wash" | "entry"`; the collector stores the kind, the review screen shows an entry sample +as "entry stream — label the vehicle", the export carries a `kind` column, and **operator +agreement is computed from wash items only** (an entry sample has no operator decision). + +An internet feed was considered the same day and kept OUT of the collector's ingest: licensed +sets only, in a separate folder with provenance, used as warm-up and weighted down, and never +the judge of accuracy — the evaluation set is gate crops only. + ## The package `multipart/form-data`: `meta` (JSON) + `image` (JPEG). Meta = `{ v, booth, item, order, at, diff --git a/wiki/log.md b/wiki/log.md index 422fed8..deba920 100644 --- a/wiki/log.md +++ b/wiki/log.md @@ -3121,3 +3121,15 @@ the overlay address; commented `trainer` profile seam for the GPU), a third buil build-images.yml, and a `wash-collector` stack on `art-docker-station` in komodo/resources.toml (secret refs to fill). Booth payload now carries `operatorCategory.classes`. Tests: app.test.ts. Updated [[vision-review-outbox]], [[fleet-deployment-komodo]]. + +## [2026-09-07] ingest | Entry-stream sampling for the review outbox +User asked about feeding internet pictures through the collector; assessment: licensed only, +separate folder, warm-up weight, never the evaluation set — and the stronger accelerator is the +ENTRY stream (every car, the gate view, zero domain shift). Built: `deviceEvents.emitVehicleRead` +from snapshot.ts (core announces; the module listens), `ReviewOutbox.sampleEntry()` (one in N, +in-process) + `enqueueEntry()` (crop + camera class, no order/operator/category), env +`CARWASH_REVIEW_ENTRY_SAMPLE` (compose + resources template + .env.example), packages carry +`kind`; the collector stores kind, the review screen shows entry samples as such, export has a +kind column, operator agreement is wash-only. Setup line shows "1 in N entries sampled". Also: +Setup → Car wash is a two-column grid (the master-data card was squeezed at max-w-2xl). Tests +on both sides. Updated [[vision-review-outbox]].