Robocop

scripts/speed_limit_vision/build_track_classifier_dataset.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import shutil
+
+from collections import Counter
+from pathlib import Path
+
+
+SPEED_VALUES = frozenset((15, 20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75))
+
+
+def parse_args() -> argparse.Namespace:
+  parser = argparse.ArgumentParser(description="Add trusted later-frame sign tracks to a classifier dataset.")
+  parser.add_argument("--base", type=Path, required=True)
+  parser.add_argument("--track-samples", type=Path, required=True)
+  parser.add_argument("--output", type=Path, required=True)
+  parser.add_argument("--train-ratio", type=float, default=0.85)
+  parser.add_argument("--min-growth", type=float, default=1.10)
+  parser.add_argument("--min-exact-confidence", type=float, default=0.80)
+  parser.add_argument("--min-detector-confidence", type=float, default=0.30)
+  parser.add_argument("--max-track-rank", type=int, default=3)
+  return parser.parse_args()
+
+
+def split_for_key(key: str, train_ratio: float) -> str:
+  fraction = int(hashlib.sha1(key.encode()).hexdigest()[:8], 16) / 0xFFFFFFFF
+  return "train" if fraction < train_ratio else "val"
+
+
+def link_or_copy(source: Path, destination: Path) -> None:
+  destination.parent.mkdir(parents=True, exist_ok=True)
+  if destination.exists():
+    return
+  try:
+    os.link(source, destination)
+  except OSError:
+    shutil.copy2(source, destination)
+
+
+def remove_appledouble_files(root: Path) -> int:
+  removed = 0
+  for path in root.rglob("._*"):
+    if path.is_file():
+      path.unlink()
+      removed += 1
+  return removed
+
+
+def trusted_track_row(row: dict[str, str], args: argparse.Namespace) -> bool:
+  try:
+    expected = int(row.get("expected_speed_limit_mph", ""))
+    predicted = int(row.get("predicted_speed_limit_mph", "") or 0)
+    read_confidence = float(row.get("read_confidence", "") or 0.0)
+    detector_confidence = float(row.get("detector_confidence", "") or 0.0)
+    growth = float(row.get("area_ratio_to_anchor", "") or 0.0)
+    rank = int(row.get("rank", "") or 999)
+  except ValueError:
+    return False
+  exact = predicted == expected and read_confidence >= args.min_exact_confidence
+  detector_snap = detector_confidence >= args.min_detector_confidence
+  return expected in SPEED_VALUES and growth >= args.min_growth and rank <= args.max_track_rank and (exact or detector_snap)
+
+
+def main() -> int:
+  args = parse_args()
+  base = args.base.expanduser().resolve()
+  output = args.output.expanduser().resolve()
+  counts: Counter[str] = Counter()
+  validation_routes: set[str] = set()
+  for split in ("train", "val"):
+    for class_dir in (base / split).iterdir():
+      if not class_dir.is_dir() or class_dir.name.startswith("._"):
+        continue
+      for source in class_dir.iterdir():
+        if not source.is_file() or source.name.startswith("._") or source.suffix.lower() not in (".jpg", ".jpeg", ".png"):
+          continue
+        link_or_copy(source, output / split / class_dir.name / f"base_{source.name}")
+        counts[f"base_{split}"] += 1
+
+  with args.track_samples.expanduser().resolve().open(encoding="utf-8", newline="") as input_file:
+    for row in csv.DictReader(input_file):
+      if not trusted_track_row(row, args):
+        counts["track_rejected"] += 1
+        continue
+      source = Path(row.get("crop_path", "")).expanduser().resolve()
+      if not source.is_file():
+        counts["track_rejected"] += 1
+        continue
+      speed = int(row["expected_speed_limit_mph"])
+      split = split_for_key(row.get("route") or row.get("track_key", ""), args.train_ratio)
+      if split == "val" and row.get("route"):
+        validation_routes.add(row["route"])
+      name = f"track_{row.get('track_key', '')}_{row.get('rank', '')}{source.suffix.lower()}"
+      link_or_copy(source, output / split / str(speed) / name)
+      counts[f"track_{split}"] += 1
+      counts[f"speed_{speed}"] += 1
+
+  counts["appledouble_removed"] = remove_appledouble_files(output)
+  (output / "track_validation_routes.txt").write_text("\n".join(sorted(validation_routes)) + "\n", encoding="ascii")
+  summary = {"base": str(base), "output": str(output), "counts": dict(sorted(counts.items()))}
+  (output / "track_dataset_summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="ascii")
+  print(json.dumps(summary, indent=2, sort_keys=True))
+  return 0
+
+
+if __name__ == "__main__":
+  raise SystemExit(main())

scripts/speed_limit_vision/build_track_direct_detector_dataset.py

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import shutil
+
+from collections import Counter
+from pathlib import Path
+
+import cv2
+
+if __package__ in (None, ""):
+  import sys
+  sys.path.insert(0, str(Path(__file__).resolve().parent))
+  from import_manual_review_queue import merged_review_rows, parse_speed  # type: ignore
+else:
+  from .import_manual_review_queue import merged_review_rows, parse_speed
+
+
+SPEED_VALUES = (15, 20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75)
+SPEED_TO_CLASS = {speed: index for index, speed in enumerate(SPEED_VALUES)}
+POSITIVE_STATUSES = frozenset(("accepted", "corrected"))
+
+
+def parse_args() -> argparse.Namespace:
+  parser = argparse.ArgumentParser(description="Add reviewed comma sign tracks to a direct speed-class detector dataset.")
+  parser.add_argument("--base", type=Path, required=True, help="Existing YOLO direct-value detector dataset.")
+  parser.add_argument("--queue", type=Path, required=True, help="Source reviewed manual_review_queue.csv.")
+  parser.add_argument("--labels", type=Path, help="Defaults to manual_review_labels.csv beside the queue.")
+  parser.add_argument("--track-samples", type=Path, required=True)
+  parser.add_argument("--output", type=Path, required=True)
+  parser.add_argument("--train-ratio", type=float, default=0.85)
+  parser.add_argument("--min-growth", type=float, default=1.10)
+  parser.add_argument("--min-exact-confidence", type=float, default=0.80)
+  parser.add_argument("--min-detector-confidence", type=float, default=0.30)
+  parser.add_argument("--max-track-rank", type=int, default=3)
+  return parser.parse_args()
+
+
+def parse_bbox(value: str) -> tuple[int, int, int, int] | None:
+  try:
+    values = tuple(int(round(float(part.strip()))) for part in value.split(","))
+  except ValueError:
+    return None
+  if len(values) != 4:
+    return None
+  x1, y1, x2, y2 = values
+  return values if x2 > x1 and y2 > y1 else None
+
+
+def split_for_key(key: str, train_ratio: float) -> str:
+  fraction = int(hashlib.sha1(key.encode()).hexdigest()[:8], 16) / 0xFFFFFFFF
+  return "train" if fraction < train_ratio else "val"
+
+
+def link_or_copy(source: Path, destination: Path) -> None:
+  destination.parent.mkdir(parents=True, exist_ok=True)
+  if destination.exists():
+    return
+  try:
+    os.link(source, destination)
+  except OSError:
+    shutil.copy2(source, destination)
+
+
+def yolo_label(speed: int, bbox: tuple[int, int, int, int], image_path: Path) -> str | None:
+  image = cv2.imread(str(image_path))
+  if image is None:
+    return None
+  height, width = image.shape[:2]
+  x1, y1, x2, y2 = bbox
+  x1 = max(min(x1, width - 1), 0)
+  y1 = max(min(y1, height - 1), 0)
+  x2 = max(min(x2, width), 0)
+  y2 = max(min(y2, height), 0)
+  if x2 <= x1 or y2 <= y1:
+    return None
+  values = (
+    SPEED_TO_CLASS[speed],
+    (x1 + x2) / (2 * width),
+    (y1 + y2) / (2 * height),
+    (x2 - x1) / width,
+    (y2 - y1) / height,
+  )
+  return f"{values[0]} {values[1]:.8f} {values[2]:.8f} {values[3]:.8f} {values[4]:.8f}\n"
+
+
+def copy_base_dataset(base: Path, output: Path) -> Counter[str]:
+  counts: Counter[str] = Counter()
+  for split in ("train", "val"):
+    image_dir = base / "images" / split
+    label_dir = base / "labels" / split
+    for source_image in image_dir.iterdir():
+      if not source_image.is_file() or source_image.name.startswith("._"):
+        continue
+      source_label = label_dir / f"{source_image.stem}.txt"
+      if not source_label.is_file():
+        continue
+      destination_stem = f"base_{source_image.stem}"
+      link_or_copy(source_image, output / "images" / split / f"{destination_stem}{source_image.suffix.lower()}")
+      link_or_copy(source_label, output / "labels" / split / f"{destination_stem}.txt")
+      counts[f"base_{split}"] += 1
+  return counts
+
+
+def add_reviewed_anchors(
+  queue_path: Path,
+  labels_path: Path,
+  output: Path,
+  train_ratio: float,
+  seen_images: set[Path],
+) -> Counter[str]:
+  counts: Counter[str] = Counter()
+  for row in merged_review_rows(queue_path, labels_path):
+    if row.get("review_status") not in POSITIVE_STATUSES:
+      continue
+    speed = parse_speed(row.get("review_speed_limit_mph", ""))
+    bbox = parse_bbox(row.get("review_bbox") or row.get("bbox", ""))
+    image_path = Path(row.get("frame_path", "")).expanduser()
+    if speed not in SPEED_TO_CLASS or bbox is None or not image_path.is_file():
+      continue
+    resolved_image = image_path.resolve()
+    if resolved_image in seen_images:
+      continue
+    label = yolo_label(speed, bbox, resolved_image)
+    if label is None:
+      continue
+    split = split_for_key(row.get("route") or row.get("record_key", ""), train_ratio)
+    stem = f"review_{row.get('record_key', hashlib.sha1(str(resolved_image).encode()).hexdigest()[:16])}"
+    destination_image = output / "images" / split / f"{stem}{resolved_image.suffix.lower()}"
+    link_or_copy(resolved_image, destination_image)
+    (output / "labels" / split / f"{stem}.txt").write_text(label, encoding="ascii")
+    seen_images.add(resolved_image)
+    counts[f"anchor_{split}"] += 1
+    counts[f"speed_{speed}"] += 1
+  return counts
+
+
+def trusted_track_row(row: dict[str, str], args: argparse.Namespace) -> bool:
+  try:
+    expected = int(row.get("expected_speed_limit_mph", ""))
+    predicted = int(row.get("predicted_speed_limit_mph", "") or 0)
+    read_confidence = float(row.get("read_confidence", "") or 0.0)
+    detector_confidence = float(row.get("detector_confidence", "") or 0.0)
+    growth = float(row.get("area_ratio_to_anchor", "") or 0.0)
+    rank = int(row.get("rank", "") or 999)
+  except ValueError:
+    return False
+  exact_read = predicted == expected and read_confidence >= args.min_exact_confidence
+  detector_snap = detector_confidence >= args.min_detector_confidence
+  return expected in SPEED_TO_CLASS and growth >= args.min_growth and rank <= args.max_track_rank and (exact_read or detector_snap)
+
+
+def add_track_samples(args: argparse.Namespace, output: Path, seen_images: set[Path]) -> Counter[str]:
+  counts: Counter[str] = Counter()
+  with args.track_samples.expanduser().resolve().open(encoding="utf-8", newline="") as input_file:
+    for row in csv.DictReader(input_file):
+      if not trusted_track_row(row, args):
+        counts["track_rejected"] += 1
+        continue
+      speed = int(row["expected_speed_limit_mph"])
+      bbox = parse_bbox(row.get("bbox", ""))
+      image_path = Path(row.get("frame_path", "")).expanduser().resolve()
+      if bbox is None or not image_path.is_file() or image_path in seen_images:
+        counts["track_rejected"] += 1
+        continue
+      label = yolo_label(speed, bbox, image_path)
+      if label is None:
+        counts["track_rejected"] += 1
+        continue
+      split = split_for_key(row.get("route") or row.get("track_key", ""), args.train_ratio)
+      stem = f"track_{row.get('track_key', '')}_{row.get('rank', '')}"
+      destination_image = output / "images" / split / f"{stem}{image_path.suffix.lower()}"
+      link_or_copy(image_path, destination_image)
+      (output / "labels" / split / f"{stem}.txt").write_text(label, encoding="ascii")
+      seen_images.add(image_path)
+      counts[f"track_{split}"] += 1
+      counts[f"speed_{speed}"] += 1
+  return counts
+
+
+def main() -> int:
+  args = parse_args()
+  base = args.base.expanduser().resolve()
+  queue_path = args.queue.expanduser().resolve()
+  labels_path = args.labels.expanduser().resolve() if args.labels else queue_path.with_name("manual_review_labels.csv")
+  output = args.output.expanduser().resolve()
+  for split in ("train", "val"):
+    (output / "images" / split).mkdir(parents=True, exist_ok=True)
+    (output / "labels" / split).mkdir(parents=True, exist_ok=True)
+
+  counts = copy_base_dataset(base, output)
+  seen_images: set[Path] = set()
+  counts.update(add_reviewed_anchors(queue_path, labels_path, output, args.train_ratio, seen_images))
+  counts.update(add_track_samples(args, output, seen_images))
+  yaml_lines = [
+    f"path: {output}",
+    "train: images/train",
+    "val: images/val",
+    "names:",
+    *(f"  {index}: speed_limit_{speed}" for index, speed in enumerate(SPEED_VALUES)),
+  ]
+  (output / "dataset.yaml").write_text("\n".join(yaml_lines) + "\n", encoding="ascii")
+  summary = {"base": str(base), "output": str(output), "counts": dict(sorted(counts.items()))}
+  (output / "track_dataset_summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="ascii")
+  print(json.dumps(summary, indent=2, sort_keys=True))
+  return 0
+
+
+if __name__ == "__main__":
+  raise SystemExit(main())

scripts/speed_limit_vision/evaluate_direct_value_detector.py

@@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import time
+
+from collections import Counter, defaultdict
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+import starpilot.system.speed_limit_vision as slv
+
+if __package__ in (None, ""):
+  import sys
+  sys.path.insert(0, str(Path(__file__).resolve().parent))
+  from evaluate_runtime_manifest import expected_value, first_present, is_negative, load_rows  # type: ignore
+  from evaluate_reviewed_route_events import load_cases  # type: ignore
+  from replay_route_runtime import RouteReplayDaemon  # type: ignore
+else:
+  from .evaluate_runtime_manifest import expected_value, first_present, is_negative, load_rows
+  from .evaluate_reviewed_route_events import load_cases
+  from .replay_route_runtime import RouteReplayDaemon
+
+
+SPEED_VALUES = (15, 20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75)
+
+
+def parse_args() -> argparse.Namespace:
+  parser = argparse.ArgumentParser(description="Evaluate a one-stage direct speed-value detector.")
+  parser.add_argument("--detector", type=Path, required=True)
+  parser.add_argument("--manifest", type=Path, help="Static reviewed-frame manifest.")
+  parser.add_argument("--queue", type=Path, help="Reviewed queue for cadence-aware route replay.")
+  parser.add_argument("--labels", type=Path)
+  parser.add_argument("--output-csv", type=Path)
+  parser.add_argument("--confidence", type=float, default=0.06)
+  parser.add_argument("--advisory-positive", action="store_true")
+  parser.add_argument("--include-uncertain", action="store_true")
+  parser.add_argument("--positive-only", action="store_true")
+  parser.add_argument("--max-cases", type=int, default=0)
+  parser.add_argument("--window-before", type=float, default=4.0)
+  parser.add_argument("--window-after", type=float, default=3.0)
+  parser.add_argument("--dedupe-seconds", type=float, default=3.0)
+  parser.add_argument("--measured-inference-seconds", type=float, default=0.44)
+  args = parser.parse_args()
+  if bool(args.manifest) == bool(args.queue):
+    parser.error("exactly one of --manifest or --queue is required")
+  return args
+
+
+class DirectValueDetector:
+  def __init__(self, model_path: Path, confidence: float):
+    self.model_path = model_path.expanduser().resolve()
+    self.confidence = confidence
+    self.input_size = slv.SpeedLimitVisionDaemon._read_onnx_square_input_size(self.model_path, 256)
+    self.net = cv2.dnn.readNetFromONNX(str(self.model_path))
+    self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV)
+    self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU)
+    self.last_forward_seconds = 0.0
+
+  def proposals(self, frame_bgr: np.ndarray) -> list[tuple[float, int, tuple[int, int, int, int]]]:
+    frame_height, frame_width = frame_bgr.shape[:2]
+    left_ratio, top_ratio, right_ratio, bottom_ratio = slv.ROI_WINDOWS[-1]["bounds"]
+    origin_x = int(frame_width * left_ratio)
+    origin_y = int(frame_height * top_ratio)
+    right = int(frame_width * right_ratio)
+    bottom = int(frame_height * bottom_ratio)
+    region = frame_bgr[origin_y:bottom, origin_x:right]
+    if region.size == 0:
+      return []
+    shape = (self.input_size, self.input_size)
+    letterboxed, ratio, pad_width, pad_height = slv.SpeedLimitVisionDaemon._letterbox(region, shape=shape)
+    blob = cv2.dnn.blobFromImage(letterboxed, scalefactor=1 / 255.0, size=shape, swapRB=True, crop=False)
+    self.net.setInput(blob)
+    started_at = time.monotonic()
+    predictions = np.squeeze(self.net.forward())
+    self.last_forward_seconds = time.monotonic() - started_at
+    if predictions.ndim != 2:
+      return []
+    if predictions.shape[0] < predictions.shape[1]:
+      predictions = predictions.T
+
+    region_height, region_width = region.shape[:2]
+    proposals: list[tuple[float, int, tuple[int, int, int, int]]] = []
+    for prediction in predictions:
+      class_scores = prediction[4:]
+      if class_scores.size not in (len(SPEED_VALUES) - 1, len(SPEED_VALUES)):
+        continue
+      class_id = int(np.argmax(class_scores))
+      confidence = float(class_scores[class_id])
+      if confidence < self.confidence:
+        continue
+      center_x, center_y, width, height = prediction[:4]
+      x1 = max(int((center_x - width / 2 - pad_width) / ratio), 0)
+      y1 = max(int((center_y - height / 2 - pad_height) / ratio), 0)
+      x2 = min(int((center_x + width / 2 - pad_width) / ratio), region_width)
+      y2 = min(int((center_y + height / 2 - pad_height) / ratio), region_height)
+      if x2 <= x1 or y2 <= y1:
+        continue
+      bbox = (x1 + origin_x, y1 + origin_y, x2 + origin_x, y2 + origin_y)
+      box_width = bbox[2] - bbox[0]
+      box_height = bbox[3] - bbox[1]
+      if box_width < slv.MODEL_PROPOSAL_MIN_WIDTH or box_height < slv.MODEL_PROPOSAL_MIN_HEIGHT:
+        continue
+      if box_width * box_height > frame_width * frame_height * slv.MODEL_PROPOSAL_MAX_AREA_RATIO:
+        continue
+      if (bbox[0] + bbox[2]) / 2 < frame_width * slv.MODEL_PROPOSAL_MIN_X_RATIO:
+        continue
+      if bbox[1] > frame_height * slv.MODEL_PROPOSAL_MAX_Y_RATIO:
+        continue
+      proposals.append((confidence, class_id, bbox))
+    return sorted(proposals, reverse=True)[:slv.MODEL_PROPOSAL_MAX_COUNT]
+
+  def detect(self, frame_bgr: np.ndarray) -> slv.Detection | None:
+    proposals = self.proposals(frame_bgr)
+    if not proposals:
+      return None
+    confidence, class_id, _bbox = proposals[0]
+    return slv.Detection(SPEED_VALUES[class_id], min(confidence, 0.95), confidence >= 0.80)
+
+
+class DirectRouteReplayDaemon(RouteReplayDaemon):
+  def __init__(self, detector: DirectValueDetector, measured_inference_seconds: float):
+    super().__init__(runtime_context=None, measured_inference_seconds=measured_inference_seconds)
+    self.direct_detector = detector
+
+  def _detect_sign(self, frame_bgr):
+    return self._publishable_detection(self.direct_detector.detect(frame_bgr))
+
+
+def evaluate_manifest(args: argparse.Namespace, detector: DirectValueDetector) -> dict[str, object]:
+  rows = load_rows(args.manifest.expanduser().resolve(), None)
+  if not args.include_uncertain:
+    rows = [row for row in rows if row.get("sample_type") != "uncertain_positive" and row.get("review_status") != "uncertain"]
+  counts: Counter[str] = Counter()
+  by_speed: dict[int, Counter[str]] = defaultdict(Counter)
+  output_rows: list[dict[str, object]] = []
+  for row in rows:
+    image_text = first_present(row, ("dataset_image", "frame_path", "source_frame", "image_path"))
+    frame_bgr = cv2.imread(str(Path(image_text).expanduser())) if image_text else None
+    if frame_bgr is None:
+      counts["unreadable"] += 1
+      continue
+    detection = detector.detect(frame_bgr)
+    predicted = detection.speed_limit_mph if detection else 0
+    expected = expected_value(row)
+    advisory_positive = args.advisory_positive and row.get("review_sign_type", "").strip().lower() == "advisory"
+    negative = False if advisory_positive else is_negative(row)
+    if negative:
+      counts.update(negative=1, negative_fp=int(bool(predicted)))
+    else:
+      counts.update(positive=1, positive_any=int(bool(predicted)), positive_exact=int(predicted == expected))
+      if expected is not None:
+        by_speed[expected].update(total=1, exact=int(predicted == expected), any=int(bool(predicted)))
+    output_rows.append({
+      "record_key": row.get("record_key", ""),
+      "expected_speed_limit_mph": expected or "",
+      "predicted_speed_limit_mph": predicted or "",
+      "confidence": detection.confidence if detection else "",
+      "negative": negative,
+      "image_path": image_text,
+    })
+  if args.output_csv:
+    args.output_csv.parent.mkdir(parents=True, exist_ok=True)
+    with args.output_csv.open("w", encoding="utf-8", newline="") as output_file:
+      writer = csv.DictWriter(output_file, fieldnames=output_rows[0].keys() if output_rows else ("record_key",))
+      writer.writeheader()
+      writer.writerows(output_rows)
+  return {"counts": dict(counts), "by_speed": {str(speed): dict(values) for speed, values in sorted(by_speed.items())}}
+
+
+def replay_video_cases(cases, detector: DirectValueDetector, args: argparse.Namespace):
+  daemons = {case.record_key: DirectRouteReplayDaemon(detector, args.measured_inference_seconds) for case in cases}
+  capture = cv2.VideoCapture(str(cases[0].source_video_path))
+  fps = capture.get(cv2.CAP_PROP_FPS) or 20.0
+  windows = {
+    case.record_key: (max(case.frame_time_s - args.window_before, 0.0), case.frame_time_s + args.window_after)
+    for case in cases
+  }
+  first_frame = max(int(min(window[0] for window in windows.values()) * fps), 0)
+  end_frame = max(int(max(window[1] for window in windows.values()) * fps), first_frame)
+  frame_index = 0
+  while frame_index < first_frame:
+    if not capture.grab():
+      break
+    frame_index += 1
+  while frame_index <= end_frame:
+    ok, frame_bgr = capture.read()
+    if not ok:
+      break
+    frame_time_s = frame_index / fps
+    for case in cases:
+      start_s, end_s = windows[case.record_key]
+      if start_s <= frame_time_s <= end_s:
+        daemons[case.record_key].process_frame(frame_time_s - start_s, frame_bgr)
+    frame_index += 1
+  capture.release()
+  return {
+    case.record_key: (
+      [int(event["candidateSpeedLimitMph"]) for event in daemons[case.record_key].events if event["event"] == "candidate"],
+      [int(event["speedLimitMph"]) for event in daemons[case.record_key].events if event["event"] == "publish"],
+    )
+    for case in cases
+  }
+
+
+def evaluate_queue(args: argparse.Namespace, detector: DirectValueDetector) -> dict[str, object]:
+  queue_path = args.queue.expanduser().resolve()
+  labels_path = args.labels.expanduser().resolve() if args.labels else queue_path.with_name("manual_review_labels.csv")
+  cases = load_cases(queue_path, labels_path, args.dedupe_seconds)
+  if args.positive_only:
+    cases = [case for case in cases if not case.negative]
+  if args.max_cases > 0:
+    cases = cases[:args.max_cases]
+  cases_by_video = defaultdict(list)
+  for case in cases:
+    cases_by_video[case.source_video_path].append(case)
+  results = {}
+  for index, video_cases in enumerate(cases_by_video.values(), start=1):
+    results.update(replay_video_cases(video_cases, detector, args))
+    if index % 10 == 0:
+      print(f"Replayed {index}/{len(cases_by_video)} video segments", flush=True)
+  counts: Counter[str] = Counter()
+  by_speed: dict[int, Counter[str]] = defaultdict(Counter)
+  for case in cases:
+    candidates, publishes = results.get(case.record_key, ([], []))
+    if case.negative:
+      counts.update(negative=1, candidate_fp=int(bool(candidates)), publish_fp=int(bool(publishes)))
+    else:
+      candidate_hit = case.expected_speed_limit_mph in candidates
+      publish_hit = case.expected_speed_limit_mph in publishes
+      counts.update(positive=1, candidate_hit=int(candidate_hit), publish_hit=int(publish_hit))
+      by_speed[case.expected_speed_limit_mph].update(total=1, candidate_hit=int(candidate_hit), publish_hit=int(publish_hit))
+  return {"counts": dict(counts), "by_speed": {str(speed): dict(values) for speed, values in sorted(by_speed.items())}}
+
+
+def main() -> int:
+  args = parse_args()
+  detector = DirectValueDetector(args.detector, args.confidence)
+  result = evaluate_manifest(args, detector) if args.manifest else evaluate_queue(args, detector)
+  print(json.dumps(result, indent=2, sort_keys=True))
+  return 0
+
+
+if __name__ == "__main__":
+  raise SystemExit(main())

scripts/speed_limit_vision/evaluate_reviewed_route_events.py

@@ -51,6 +51,8 @@ def parse_args() -> argparse.Namespace:
   parser.add_argument("--crop-ocr", action="store_true", help="Evaluate with crop OCR confirmation enabled.")
   parser.add_argument("--classifier-min-confidence", type=float, help="Override the value classifier confidence threshold.")
   parser.add_argument("--trusted-model-min-confidence", type=float, help="Override tiny-box trusted model confidence.")
+  parser.add_argument("--classifier-expansion-limit", type=int, help="Evaluate only the first N detector crop expansions.")
+  parser.add_argument("--classifier-expansion-indices", help="Comma-separated detector crop expansion indices to evaluate.")
   parser.add_argument("--strong-rescue-min-proposal-confidence", type=float, help="Override single-frame tiny-sign proposal confidence.")
   parser.add_argument("--strong-rescue-min-read-confidence", type=float, help="Override single-frame tiny-sign classifier confidence.")
   parser.add_argument("--low-speed-change-consistent-detections", type=int, help="Override reads required to change from 30+ mph to below 30 mph.")
@@ -66,6 +68,7 @@ def parse_args() -> argparse.Namespace:
   )
   parser.add_argument("--initial-speed-limit", type=int, default=0, help="Seed each replay window with a currently published speed limit.")
   parser.add_argument("--positive-only", action="store_true", help="Replay only reviewed speed signs, omitting ignored-crop windows.")
+  parser.add_argument("--route-file", type=Path, help="Only replay routes listed one per line in this file.")
   parser.add_argument("--max-cases", type=int, default=0, help="Optional evaluation cap after deduplication.")
   return parser.parse_args()
 
@@ -134,8 +137,13 @@ def replay_video_cases(cases: list[ReviewedCase], args: argparse.Namespace) -> d
   }
   first_frame = max(int(min(window[0] for window in windows.values()) * fps), 0)
   end_frame = max(int(max(window[1] for window in windows.values()) * fps), first_frame)
-  capture.set(cv2.CAP_PROP_POS_FRAMES, first_frame)
-  frame_index = first_frame
+  # Raw comma HEVC streams do not contain a usable random-access index.
+  # OpenCV accepts CAP_PROP_POS_FRAMES but still returns frame zero.
+  frame_index = 0
+  while frame_index < first_frame:
+    if not capture.grab():
+      break
+    frame_index += 1
 
   while frame_index <= end_frame:
     ok, frame_bgr = capture.read()
@@ -168,6 +176,15 @@ def main() -> int:
     slv.US_CLASSIFIER_MIN_CONFIDENCE = args.classifier_min_confidence
   if args.trusted_model_min_confidence is not None:
     slv.DETECTOR_CLASSIFIER_TRUSTED_MODEL_MIN_READ_CONFIDENCE = args.trusted_model_min_confidence
+  if args.classifier_expansion_indices:
+    indices = tuple(int(value) for value in args.classifier_expansion_indices.split(","))
+    if not indices or min(indices) < 0 or max(indices) >= len(slv.DETECTOR_CLASSIFIER_EXPANSIONS):
+      raise ValueError("--classifier-expansion-indices contains an invalid index")
+    slv.DETECTOR_CLASSIFIER_EXPANSIONS = tuple(slv.DETECTOR_CLASSIFIER_EXPANSIONS[index] for index in indices)
+  elif args.classifier_expansion_limit is not None:
+    if args.classifier_expansion_limit < 1:
+      raise ValueError("--classifier-expansion-limit must be at least 1")
+    slv.DETECTOR_CLASSIFIER_EXPANSIONS = slv.DETECTOR_CLASSIFIER_EXPANSIONS[:args.classifier_expansion_limit]
   if args.strong_rescue_min_proposal_confidence is not None:
     slv.DETECTOR_CLASSIFIER_STRONG_RESCUE_MIN_PROPOSAL_CONFIDENCE = args.strong_rescue_min_proposal_confidence
   if args.strong_rescue_min_read_confidence is not None:
@@ -179,6 +196,11 @@ def main() -> int:
   if args.enable_strong_model_consensus:
     slv.DETECTOR_CLASSIFIER_STRONG_MODEL_CONSENSUS_ENABLED = True
   cases = load_cases(queue_path, labels_path, args.dedupe_seconds)
+  if args.route_file:
+    selected_routes = {
+      line.strip() for line in args.route_file.expanduser().resolve().read_text(encoding="utf-8").splitlines() if line.strip()
+    }
+    cases = [case for case in cases if case.route in selected_routes]
   if args.positive_only:
     cases = [case for case in cases if not case.negative]
   if args.max_cases > 0:

scripts/speed_limit_vision/evaluate_runtime_manifest.py

@@ -43,6 +43,8 @@ def parse_args() -> argparse.Namespace:
   crop_ocr_group.add_argument("--crop-ocr", action="store_true", dest="crop_ocr", default=None, help="Enable crop OCR confirmation.")
   crop_ocr_group.add_argument("--no-crop-ocr", action="store_false", dest="crop_ocr", help="Evaluate the model-only detector/classifier path.")
   parser.add_argument("--separate-reject-classifier", action="store_true", help="Enable the optional second-stage reject classifier during eval.")
+  parser.add_argument("--classifier-expansion-limit", type=int, help="Evaluate only the first N detector crop expansions.")
+  parser.add_argument("--classifier-expansion-indices", help="Comma-separated detector crop expansion indices to evaluate.")
   parser.add_argument("--include-uncertain", action="store_true", help="Include uncertain_positive review rows in positive metrics.")
   parser.add_argument("--advisory-positive", action="store_true", help="Score reviewed advisory rows as readable speed positives.")
   parser.add_argument("--strict-positive-recall", type=float, help="Exit non-zero if positive exact recall is below this value.")
@@ -165,6 +167,15 @@ def main() -> int:
     slv.US_REJECT_CLASSIFIER_MIN_CONFIDENCE = args.classifier_reject_min_confidence
   if args.trusted_model_min_confidence is not None:
     slv.DETECTOR_CLASSIFIER_TRUSTED_MODEL_MIN_READ_CONFIDENCE = args.trusted_model_min_confidence
+  if args.classifier_expansion_indices:
+    indices = tuple(int(value) for value in args.classifier_expansion_indices.split(","))
+    if not indices or min(indices) < 0 or max(indices) >= len(slv.DETECTOR_CLASSIFIER_EXPANSIONS):
+      raise ValueError("--classifier-expansion-indices contains an invalid index")
+    slv.DETECTOR_CLASSIFIER_EXPANSIONS = tuple(slv.DETECTOR_CLASSIFIER_EXPANSIONS[index] for index in indices)
+  elif args.classifier_expansion_limit is not None:
+    if args.classifier_expansion_limit < 1:
+      raise ValueError("--classifier-expansion-limit must be at least 1")
+    slv.DETECTOR_CLASSIFIER_EXPANSIONS = slv.DETECTOR_CLASSIFIER_EXPANSIONS[:args.classifier_expansion_limit]
   if args.strong_rescue_min_proposal_confidence is not None:
     slv.DETECTOR_CLASSIFIER_STRONG_RESCUE_MIN_PROPOSAL_CONFIDENCE = args.strong_rescue_min_proposal_confidence
   if args.strong_rescue_min_read_confidence is not None:

scripts/speed_limit_vision/mine_reviewed_sign_tracks.py

@@ -0,0 +1,432 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import math
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+import starpilot.system.speed_limit_vision as slv
+
+if __package__ in (None, ""):
+  import sys
+  sys.path.insert(0, str(Path(__file__).resolve().parent))
+  from import_manual_review_queue import merged_review_rows, parse_speed  # type: ignore
+  from replay_route_runtime import configure_models  # type: ignore
+else:
+  from .import_manual_review_queue import merged_review_rows, parse_speed
+  from .replay_route_runtime import configure_models
+
+
+POSITIVE_STATUSES = frozenset(("accepted", "corrected"))
+
+
+@dataclass(frozen=True)
+class TrackCase:
+  source_row: dict[str, str]
+  track_key: str
+  video_path: Path
+  frame_time_s: float
+  expected_speed_mph: int
+  anchor_bbox: tuple[int, int, int, int]
+
+
+@dataclass(frozen=True)
+class TrackSample:
+  time_s: float
+  bbox: tuple[int, int, int, int]
+  crop_bbox: tuple[int, int, int, int]
+  detector_confidence: float
+  predicted_speed_mph: int
+  read_confidence: float
+  sharpness: float
+  brightness: float
+  area_ratio_to_anchor: float
+  score: float
+  frame_jpeg: bytes
+  crop_jpeg: bytes
+
+
+def parse_args() -> argparse.Namespace:
+  parser = argparse.ArgumentParser(description="Track human-reviewed signs into later, larger route frames.")
+  parser.add_argument("--queue", type=Path, required=True, help="Reviewed manual_review_queue.csv.")
+  parser.add_argument("--labels", type=Path, help="Defaults to manual_review_labels.csv beside the queue.")
+  parser.add_argument("--models-dir", type=Path, default=Path("starpilot/assets/vision_models"))
+  parser.add_argument("--output-dir", type=Path, required=True)
+  parser.add_argument("--window-after", type=float, default=2.5, help="Seconds to track after the reviewed anchor frame.")
+  parser.add_argument("--sample-interval", type=float, default=0.10, help="Minimum spacing between ranked samples.")
+  parser.add_argument("--detector-interval", type=float, default=0.20, help="How often to snap optical flow to detector proposals.")
+  parser.add_argument("--max-samples-per-track", type=int, default=4)
+  parser.add_argument("--dedupe-seconds", type=float, default=3.0)
+  parser.add_argument("--min-area-growth", type=float, default=0.85)
+  parser.add_argument("--limit", type=int, default=0)
+  return parser.parse_args()
+
+
+def parse_bbox(value: str) -> tuple[int, int, int, int] | None:
+  try:
+    values = tuple(int(round(float(part.strip()))) for part in value.split(","))
+  except ValueError:
+    return None
+  if len(values) != 4:
+    return None
+  x1, y1, x2, y2 = values
+  return values if x2 > x1 and y2 > y1 else None
+
+
+def bbox_area(bbox: tuple[int, int, int, int]) -> int:
+  x1, y1, x2, y2 = bbox
+  return max(x2 - x1, 0) * max(y2 - y1, 0)
+
+
+def bbox_iou(first: tuple[int, int, int, int], second: tuple[int, int, int, int]) -> float:
+  ax1, ay1, ax2, ay2 = first
+  bx1, by1, bx2, by2 = second
+  intersection = max(min(ax2, bx2) - max(ax1, bx1), 0) * max(min(ay2, by2) - max(ay1, by1), 0)
+  union = bbox_area(first) + bbox_area(second) - intersection
+  return intersection / union if union > 0 else 0.0
+
+
+def clamp_bbox(bbox: tuple[float, float, float, float], width: int, height: int) -> tuple[int, int, int, int] | None:
+  x1, y1, x2, y2 = bbox
+  result = (
+    max(min(int(round(x1)), width - 1), 0),
+    max(min(int(round(y1)), height - 1), 0),
+    max(min(int(round(x2)), width), 0),
+    max(min(int(round(y2)), height), 0),
+  )
+  return result if bbox_area(result) > 0 else None
+
+
+def expanded_bbox(bbox: tuple[int, int, int, int], width: int, height: int, padding: float = 0.12) -> tuple[int, int, int, int]:
+  x1, y1, x2, y2 = bbox
+  box_width = x2 - x1
+  box_height = y2 - y1
+  return (
+    max(int(x1 - box_width * padding), 0),
+    max(int(y1 - box_height * padding), 0),
+    min(int(x2 + box_width * padding), width),
+    min(int(y2 + box_height * padding), height),
+  )
+
+
+def feature_points(gray: np.ndarray, bbox: tuple[int, int, int, int]) -> np.ndarray | None:
+  mask = np.zeros_like(gray)
+  x1, y1, x2, y2 = bbox
+  inset_x = max((x2 - x1) // 12, 1)
+  inset_y = max((y2 - y1) // 12, 1)
+  mask[y1 + inset_y:y2 - inset_y, x1 + inset_x:x2 - inset_x] = 255
+  return cv2.goodFeaturesToTrack(gray, mask=mask, maxCorners=60, qualityLevel=0.01, minDistance=3, blockSize=5)
+
+
+def flow_bbox(
+  previous_gray: np.ndarray,
+  current_gray: np.ndarray,
+  bbox: tuple[int, int, int, int],
+  points: np.ndarray | None,
+) -> tuple[tuple[int, int, int, int] | None, np.ndarray | None]:
+  if points is None or len(points) < 6:
+    points = feature_points(previous_gray, bbox)
+  if points is None or len(points) < 4:
+    return None, None
+  next_points, status, errors = cv2.calcOpticalFlowPyrLK(
+    previous_gray,
+    current_gray,
+    points,
+    None,
+    winSize=(25, 25),
+    maxLevel=3,
+    criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 20, 0.03),
+  )
+  if next_points is None or status is None:
+    return None, None
+  good = status.reshape(-1).astype(bool)
+  if errors is not None:
+    good &= errors.reshape(-1) < 35.0
+  old = points.reshape(-1, 2)[good]
+  new = next_points.reshape(-1, 2)[good]
+  if len(old) < 4:
+    return None, None
+  transform, inliers = cv2.estimateAffinePartial2D(old, new, method=cv2.RANSAC, ransacReprojThreshold=3.0)
+  if transform is None or inliers is None or int(inliers.sum()) < 4:
+    return None, None
+  scale = math.hypot(float(transform[0, 0]), float(transform[0, 1]))
+  if not 0.88 <= scale <= 1.18:
+    return None, None
+  x1, y1, x2, y2 = bbox
+  corners = np.float32(((x1, y1), (x2, y1), (x2, y2), (x1, y2))).reshape(-1, 1, 2)
+  moved = cv2.transform(corners, transform).reshape(-1, 2)
+  tracked = clamp_bbox(
+    (float(moved[:, 0].min()), float(moved[:, 1].min()), float(moved[:, 0].max()), float(moved[:, 1].max())),
+    current_gray.shape[1],
+    current_gray.shape[0],
+  )
+  if tracked is None:
+    return None, None
+  inlier_points = new[inliers.reshape(-1).astype(bool)].reshape(-1, 1, 2)
+  return tracked, inlier_points
+
+
+def matching_proposal(daemon: slv.SpeedLimitVisionDaemon, frame: np.ndarray, tracked_bbox: tuple[int, int, int, int]):
+  tx1, ty1, tx2, ty2 = tracked_bbox
+  track_center = np.array(((tx1 + tx2) / 2, (ty1 + ty2) / 2))
+  track_diagonal = max(math.hypot(tx2 - tx1, ty2 - ty1), 1.0)
+  best = None
+  best_score = 0.0
+  for confidence, _class_id, bbox in daemon._collect_detector_classifier_proposals(frame):
+    x1, y1, x2, y2 = bbox
+    center = np.array(((x1 + x2) / 2, (y1 + y2) / 2))
+    distance_ratio = float(np.linalg.norm(center - track_center)) / track_diagonal
+    overlap = bbox_iou(tracked_bbox, bbox)
+    if overlap < 0.06 and distance_ratio > 0.85:
+      continue
+    score = overlap * 2.0 + max(1.0 - distance_ratio, 0.0) + float(confidence) * 0.25
+    if score > best_score:
+      best_score = score
+      best = float(confidence), bbox
+  return best
+
+
+def encode_jpeg(image: np.ndarray, quality: int = 90) -> bytes:
+  ok, encoded = cv2.imencode(".jpg", image, (cv2.IMWRITE_JPEG_QUALITY, quality))
+  if not ok:
+    raise RuntimeError("Could not encode tracked frame")
+  return encoded.tobytes()
+
+
+def load_cases(queue_path: Path, labels_path: Path, dedupe_seconds: float) -> tuple[list[TrackCase], list[str]]:
+  with queue_path.open(encoding="utf-8", newline="") as queue_file:
+    fieldnames = list(csv.DictReader(queue_file).fieldnames or ())
+  rows = merged_review_rows(queue_path, labels_path)
+  seen: set[tuple[str, int, int, int]] = set()
+  cases: list[TrackCase] = []
+  for row in rows:
+    if row.get("review_status") not in POSITIVE_STATUSES:
+      continue
+    speed = parse_speed(row.get("review_speed_limit_mph", ""))
+    bbox = parse_bbox(row.get("review_bbox") or row.get("bbox", ""))
+    try:
+      frame_time_s = float(row.get("frame_time_s", ""))
+      segment = int(row.get("segment", ""))
+    except ValueError:
+      continue
+    video_path = Path(row.get("source_video_path", "")).expanduser()
+    if not speed or bbox is None or not video_path.is_file():
+      continue
+    bucket = int(frame_time_s / max(dedupe_seconds, 0.1))
+    dedupe_key = (row.get("route", ""), segment, bucket, speed)
+    if dedupe_key in seen:
+      continue
+    seen.add(dedupe_key)
+    digest = hashlib.sha1(f"{row.get('record_key')}:{speed}".encode()).hexdigest()[:16]
+    cases.append(TrackCase(row, f"sign_track_{digest}", video_path.resolve(), frame_time_s, speed, bbox))
+  return cases, fieldnames
+
+
+def mine_case(case: TrackCase, daemon: slv.SpeedLimitVisionDaemon, args: argparse.Namespace) -> list[TrackSample]:
+  capture = cv2.VideoCapture(str(case.video_path))
+  fps = capture.get(cv2.CAP_PROP_FPS) or 20.0
+  anchor_frame_index = max(int(round(case.frame_time_s * fps)), 0)
+  # Raw comma HEVC streams have no seek index. CAP_PROP_POS_FRAMES silently
+  # returns frame zero, so advance sequentially to preserve timestamp alignment.
+  for _frame_index in range(anchor_frame_index):
+    if not capture.grab():
+      capture.release()
+      return []
+  ok, anchor_frame = capture.read()
+  if not ok or anchor_frame is None:
+    capture.release()
+    return []
+
+  height, width = anchor_frame.shape[:2]
+  bbox = clamp_bbox(case.anchor_bbox, width, height)
+  if bbox is None:
+    capture.release()
+    return []
+  anchor_area = max(bbox_area(bbox), 1)
+  previous_gray = cv2.cvtColor(anchor_frame, cv2.COLOR_BGR2GRAY)
+  points = feature_points(previous_gray, bbox)
+  next_sample_at = case.frame_time_s
+  next_detector_at = case.frame_time_s
+  end_frame_index = anchor_frame_index + int(round(args.window_after * fps))
+  candidates: list[TrackSample] = []
+
+  current_frame = anchor_frame
+  current_index = anchor_frame_index
+  while current_index <= end_frame_index:
+    time_s = current_index / fps
+    if current_index > anchor_frame_index:
+      current_gray = cv2.cvtColor(current_frame, cv2.COLOR_BGR2GRAY)
+      bbox, points = flow_bbox(previous_gray, current_gray, bbox, points)
+      previous_gray = current_gray
+      if bbox is None:
+        break
+      x1, y1, x2, y2 = bbox
+      if bbox_area(bbox) > anchor_area * 12.0 or x1 <= 0 or y1 <= 0 or x2 >= width or y2 >= height:
+        break
+    detector_confidence = 0.0
+    if time_s + 1e-6 >= next_detector_at:
+      proposal = matching_proposal(daemon, current_frame, bbox)
+      next_detector_at = time_s + args.detector_interval
+      if proposal is not None:
+        detector_confidence, proposal_bbox = proposal
+        bbox = proposal_bbox
+        points = feature_points(previous_gray, bbox)
+    if time_s + 1e-6 >= next_sample_at:
+      next_sample_at = time_s + args.sample_interval
+      crop_box = expanded_bbox(bbox, width, height)
+      x1, y1, x2, y2 = crop_box
+      crop = current_frame[y1:y2, x1:x2]
+      if crop.size:
+        read = daemon._classify_speed_limit_from_model(crop)
+        predicted_speed = int(read[0]) if read is not None else 0
+        read_confidence = float(read[1]) if read is not None else 0.0
+        gray_crop = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY)
+        sharpness = float(cv2.Laplacian(gray_crop, cv2.CV_64F).var())
+        brightness = float(gray_crop.mean())
+        growth = bbox_area(bbox) / anchor_area
+        exact_bonus = read_confidence * 2.0 if predicted_speed == case.expected_speed_mph else 0.0
+        wrong_penalty = read_confidence * 1.5 if predicted_speed and predicted_speed != case.expected_speed_mph else 0.0
+        score = (
+          math.log2(max(growth, 0.25)) * 0.55 +
+          min(sharpness / 180.0, 1.0) * 0.35 +
+          detector_confidence * 0.45 +
+          exact_bonus - wrong_penalty +
+          min(max(time_s - case.frame_time_s, 0.0), 1.5) * 0.08
+        )
+        if growth >= args.min_area_growth:
+          candidates.append(TrackSample(
+            time_s=time_s,
+            bbox=bbox,
+            crop_bbox=crop_box,
+            detector_confidence=detector_confidence,
+            predicted_speed_mph=predicted_speed,
+            read_confidence=read_confidence,
+            sharpness=sharpness,
+            brightness=brightness,
+            area_ratio_to_anchor=growth,
+            score=score,
+            frame_jpeg=encode_jpeg(current_frame),
+            crop_jpeg=encode_jpeg(crop, 95),
+          ))
+    if current_index >= end_frame_index:
+      break
+    ok, current_frame = capture.read()
+    if not ok:
+      break
+    current_index += 1
+  capture.release()
+
+  selected: list[TrackSample] = []
+  for candidate in sorted(candidates, key=lambda item: item.score, reverse=True):
+    if any(abs(candidate.time_s - kept.time_s) < max(args.sample_interval * 1.5, 0.12) for kept in selected):
+      continue
+    selected.append(candidate)
+    if len(selected) >= args.max_samples_per_track:
+      break
+  return selected
+
+
+def main() -> int:
+  args = parse_args()
+  queue_path = args.queue.expanduser().resolve()
+  labels_path = args.labels.expanduser().resolve() if args.labels else queue_path.with_name("manual_review_labels.csv")
+  output_dir = args.output_dir.expanduser().resolve()
+  frame_dir = output_dir / "frames"
+  crop_dir = output_dir / "crops"
+  frame_dir.mkdir(parents=True, exist_ok=True)
+  crop_dir.mkdir(parents=True, exist_ok=True)
+  configure_models(args.models_dir.expanduser().resolve())
+  daemon = slv.SpeedLimitVisionDaemon(use_runtime=False)
+  cases, source_fieldnames = load_cases(queue_path, labels_path, args.dedupe_seconds)
+  if args.limit > 0:
+    cases = cases[:args.limit]
+
+  queue_rows: list[dict[str, str]] = []
+  sample_rows: list[dict[str, str]] = []
+  for index, case in enumerate(cases, start=1):
+    samples = mine_case(case, daemon, args)
+    if not samples:
+      continue
+    for rank, sample in enumerate(samples, start=1):
+      stem = f"{case.track_key}_r{rank:02d}_t{sample.time_s:07.3f}".replace(".", "p")
+      frame_path = frame_dir / f"{stem}.jpg"
+      crop_path = crop_dir / f"{stem}_crop.jpg"
+      frame_path.write_bytes(sample.frame_jpeg)
+      crop_path.write_bytes(sample.crop_jpeg)
+      sample_rows.append({
+        "track_key": case.track_key,
+        "source_record_key": case.source_row.get("record_key", ""),
+        "route": case.source_row.get("route", ""),
+        "segment": case.source_row.get("segment", ""),
+        "frame_time_s": f"{sample.time_s:.3f}",
+        "expected_speed_limit_mph": str(case.expected_speed_mph),
+        "review_sign_type": case.source_row.get("review_sign_type", ""),
+        "frame_path": str(frame_path),
+        "crop_path": str(crop_path),
+        "source_video_path": str(case.video_path),
+        "bbox": ",".join(str(value) for value in sample.bbox),
+        "crop_bbox": ",".join(str(value) for value in sample.crop_bbox),
+        "detector_confidence": f"{sample.detector_confidence:.6f}",
+        "predicted_speed_limit_mph": str(sample.predicted_speed_mph or ""),
+        "read_confidence": f"{sample.read_confidence:.6f}",
+        "sharpness": f"{sample.sharpness:.3f}",
+        "brightness": f"{sample.brightness:.3f}",
+        "area_ratio_to_anchor": f"{sample.area_ratio_to_anchor:.4f}",
+        "track_score": f"{sample.score:.6f}",
+        "rank": str(rank),
+      })
+      if rank == 1:
+        review_row = dict(case.source_row)
+        review_row.update({
+          "record_key": case.track_key,
+          "frame_time_s": f"{sample.time_s:.3f}",
+          "frame_path": str(frame_path),
+          "crop_path": str(crop_path),
+          "source_video_path": str(case.video_path),
+          "bbox": ",".join(str(value) for value in sample.bbox),
+          "crop_bbox": ",".join(str(value) for value in sample.crop_bbox),
+          "candidate_speed_limit_mph": str(case.expected_speed_mph),
+          "candidate_confidence": f"{sample.read_confidence:.6f}",
+          "model_read": str(sample.predicted_speed_mph or ""),
+          "review_status": "",
+          "review_speed_limit_mph": "",
+          "review_sign_type": case.source_row.get("review_sign_type", "regulatory"),
+          "review_bbox": "",
+          "review_ignore_reason": "",
+          "review_notes": f"tracked from {case.source_row.get('record_key', '')}",
+          "source_record_key": case.source_row.get("record_key", ""),
+          "source_review_status": case.source_row.get("review_status", ""),
+          "source_review_speed_limit_mph": str(case.expected_speed_mph),
+        })
+        queue_rows.append(review_row)
+    if index % 10 == 0:
+      print(f"Tracked {index}/{len(cases)} reviewed signs; review rows={len(queue_rows)} samples={len(sample_rows)}", flush=True)
+
+  queue_fieldnames = list(source_fieldnames)
+  for extra in ("source_record_key",):
+    if extra not in queue_fieldnames:
+      queue_fieldnames.append(extra)
+  with (output_dir / "manual_review_queue.csv").open("w", encoding="utf-8", newline="") as output_file:
+    writer = csv.DictWriter(output_file, fieldnames=queue_fieldnames, extrasaction="ignore")
+    writer.writeheader()
+    writer.writerows(queue_rows)
+  sample_fieldnames = tuple(sample_rows[0]) if sample_rows else (
+    "track_key", "source_record_key", "route", "segment", "frame_time_s", "expected_speed_limit_mph",
+  )
+  with (output_dir / "track_samples.csv").open("w", encoding="utf-8", newline="") as output_file:
+    writer = csv.DictWriter(output_file, fieldnames=sample_fieldnames)
+    writer.writeheader()
+    writer.writerows(sample_rows)
+  print(f"Track mining complete: cases={len(cases)} review_rows={len(queue_rows)} samples={len(sample_rows)} output={output_dir}")
+  return 0
+
+
+if __name__ == "__main__":
+  raise SystemExit(main())

starpilot/system/speed_limit_vision.py

@@ -162,7 +162,6 @@ US_REJECT_CLASSIFIER_MIN_CONFIDENCE = 0.85
 DETECTOR_CLASSIFIER_EXPANSIONS = (
   (0.00, 0.00, 0.00, 0.00, 1.10),
   (0.10, 0.06, 0.10, 0.12, 1.00),
-  (0.06, 0.00, 0.10, 0.10, 0.75),
   (0.00, 0.00, 0.18, 0.18, 0.55),
 )
 SCHOOL_ZONE_DIRECT_EXPANSIONS = (