Speed Limit Review

#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import json from datetime import UTC, datetime from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse QUEUE_REVIEW_FIELDS = [ "review_status", "review_speed_limit_mph", "review_sign_type", "review_bbox", "review_ignore_reason", "review_notes", ] LABEL_FIELDNAMES = [ "record_key", "review_status", "review_speed_limit_mph", "review_sign_type", "review_bbox", "review_ignore_reason", "review_notes", "reviewed_at_unix", ] HTML = r""" Speed Limit Review

Keys: Space/p accept model, type speed to correct, u uncertain, i/x ignore, Enter save correction, j/k next/prev, s school, r regulatory, a advisory

Crop

Frame

""" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Serve a small browser UI for manually reviewing speed-limit queue rows.") parser.add_argument("--manifest", type=Path, required=True, help="manual_review_queue.csv from build_manual_review_queue.py") parser.add_argument("--labels-out", type=Path, help="Defaults to /manual_review_labels.csv") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) return parser.parse_args() def load_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8", newline="") as handle: return list(csv.DictReader(handle)) def load_labels(path: Path) -> dict[str, dict[str, str]]: if not path.is_file(): return {} rows = load_csv(path) return {row["record_key"]: row for row in rows if row.get("record_key")} def write_labels(path: Path, labels: dict[str, dict[str, str]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=LABEL_FIELDNAMES, extrasaction="ignore") writer.writeheader() for key in sorted(labels): writer.writerow(labels[key]) def merged_rows(rows: list[dict[str, str]], labels: dict[str, dict[str, str]]) -> list[dict[str, str]]: merged = [] for row in rows: item = dict(row) label = labels.get(row.get("record_key", "")) if label: item.update({field: label.get(field, "") for field in QUEUE_REVIEW_FIELDS}) merged.append(item) return merged def filter_rows(rows: list[dict[str, str]], filter_name: str) -> list[dict[str, str]]: if filter_name == "all": return rows if filter_name == "school": return [row for row in rows if "school_zone" in row.get("detector_class", "") or "school_zone_candidate" in row.get("review_reasons", "")] if filter_name == "priority": return [row for row in rows if "priority_30_65" in row.get("review_reasons", "")] if filter_name == "disagreement": return [row for row in rows if "disagreement" in row.get("review_reasons", "") or "multi_value_votes" in row.get("review_reasons", "")] if filter_name == "negative": return [row for row in rows if row.get("detector_class") == "negative_empty"] return [row for row in rows if not row.get("review_status")] class ReviewServer(ThreadingHTTPServer): def __init__(self, server_address, handler_class, manifest_path: Path, labels_path: Path): super().__init__(server_address, handler_class) self.manifest_path = manifest_path self.labels_path = labels_path self.rows = load_csv(manifest_path) self.row_by_key = {row["record_key"]: row for row in self.rows} self.labels = load_labels(labels_path) class Handler(BaseHTTPRequestHandler): server: ReviewServer def log_message(self, _format, *args): return def send_json(self, data, status=HTTPStatus.OK): body = json.dumps(data).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def send_text(self, text: str, status=HTTPStatus.OK, content_type="text/html; charset=utf-8"): body = text.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): parsed = urlparse(self.path) if parsed.path == "/": self.send_text(HTML) return if parsed.path == "/api/queue": params = parse_qs(parsed.query) filter_name = params.get("filter", ["unreviewed"])[0] rows = filter_rows(merged_rows(self.server.rows, self.server.labels), filter_name) self.send_json({"rows": rows, "count": len(rows), "reviewed": len(self.server.labels)}) return if parsed.path.startswith("/media/"): parts = parsed.path.strip("/").split("/") if len(parts) != 3: self.send_error(HTTPStatus.NOT_FOUND) return _, record_key, kind = parts row = self.server.row_by_key.get(record_key) if row is None: self.send_error(HTTPStatus.NOT_FOUND) return image_path = Path(row.get("crop_path" if kind == "crop" else "frame_path", "")) if not image_path.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return body = image_path.read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "image/jpeg") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return self.send_error(HTTPStatus.NOT_FOUND) def do_POST(self): if urlparse(self.path).path != "/api/review": self.send_error(HTTPStatus.NOT_FOUND) return length = int(self.headers.get("Content-Length", "0")) try: payload = json.loads(self.rfile.read(length).decode("utf-8")) except Exception: self.send_error(HTTPStatus.BAD_REQUEST, "Invalid JSON") return record_key = str(payload.get("record_key") or "") if record_key not in self.server.row_by_key: self.send_error(HTTPStatus.BAD_REQUEST, "Unknown record_key") return label = {"record_key": record_key, "reviewed_at_unix": f"{datetime.now(UTC).timestamp():.3f}"} for field in QUEUE_REVIEW_FIELDS: label[field] = str(payload.get(field) or "") self.server.labels[record_key] = label write_labels(self.server.labels_path, self.server.labels) self.send_json({"ok": True, "reviewed": len(self.server.labels)}) def main() -> int: args = parse_args() manifest_path = args.manifest.expanduser().resolve() labels_path = args.labels_out.expanduser().resolve() if args.labels_out else manifest_path.with_name("manual_review_labels.csv") server = ReviewServer((args.host, args.port), Handler, manifest_path, labels_path) print(f"Review UI: http://{args.host}:{args.port}") print(f"Manifest: {manifest_path}") print(f"Labels: {labels_path}") try: server.serve_forever() except KeyboardInterrupt: pass return 0 if __name__ == "__main__": raise SystemExit(main())