from __future__ import annotations import json import math from pathlib import Path from typing import Any, Sequence import numpy as np from PIL import Image, ImageDraw, ImageFont from utils import load_record, resolve_path, sha256_file CLASS_NAMES = ( "car", "truck", "construction_vehicle", "bus", "trailer", "barrier", "motorcycle", "bicycle", "pedestrian", "traffic_cone", ) MODEL_CAMERA_ORDER = ( "CAM_FRONT", "CAM_FRONT_RIGHT", "CAM_FRONT_LEFT", "CAM_BACK", "CAM_BACK_LEFT", "CAM_BACK_RIGHT", ) GRID_CAMERA_ORDER = ( "CAM_FRONT_LEFT", "CAM_FRONT", "CAM_FRONT_RIGHT", "CAM_BACK_LEFT", "CAM_BACK", "CAM_BACK_RIGHT", ) EDGES = ( (0, 1), (1, 2), (2, 3), (3, 0), (4, 5), (5, 6), (6, 7), (7, 4), (0, 4), (1, 5), (2, 6), (3, 7), ) def load_font(size: int = 18): for path in ( "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf", ): if Path(path).is_file(): return ImageFont.truetype(path, size) return ImageFont.load_default() def corners_3d(box: np.ndarray) -> np.ndarray: x, y, z, w, length, h, yaw = [float(v) for v in box[:7]] x_c = np.array([length / 2, length / 2, -length / 2, -length / 2, length / 2, length / 2, -length / 2, -length / 2]) y_c = np.array([w / 2, -w / 2, -w / 2, w / 2, w / 2, -w / 2, -w / 2, w / 2]) z_c = np.array([h / 2, h / 2, h / 2, h / 2, -h / 2, -h / 2, -h / 2, -h / 2]) rot = np.array( [ [math.cos(yaw), -math.sin(yaw), 0.0], [math.sin(yaw), math.cos(yaw), 0.0], [0.0, 0.0, 1.0], ], dtype=np.float64, ) return (rot @ np.stack([x_c, y_c, z_c], axis=0)).T + np.array([x, y, z], dtype=np.float64) def project_box(box: np.ndarray, lidar2img: np.ndarray, image_size: tuple[int, int]) -> np.ndarray | None: corners = corners_3d(box) homo = np.concatenate([corners, np.ones((8, 1), dtype=np.float64)], axis=1) proj = (lidar2img.astype(np.float64) @ homo.T).T depth = proj[:, 2] if np.count_nonzero(depth > 1e-3) < 8: return None points = proj[:, :2] / depth[:, None] width, height = image_size if not ( (points[:, 0] >= -100).any() and (points[:, 0] <= width + 100).any() and (points[:, 1] >= -100).any() and (points[:, 1] <= height + 100).any() ): return None return points def draw_projected_box(draw: ImageDraw.ImageDraw, points: np.ndarray, color: tuple[int, int, int], width: int = 3) -> None: xy = [(float(x), float(y)) for x, y in points] for i, j in EDGES: draw.line([xy[i], xy[j]], fill=color, width=width) def draw_label(draw: ImageDraw.ImageDraw, xy: tuple[float, float], text: str, color: tuple[int, int, int], font) -> None: x = max(0.0, min(float(xy[0]), 790.0)) y = max(0.0, min(float(xy[1]), 440.0)) bbox = draw.textbbox((x, y), text, font=font) draw.rectangle(bbox, fill=(0, 0, 0)) draw.text((x, y), text, fill=color, font=font) def select_predictions( boxes: np.ndarray, scores: np.ndarray, labels: np.ndarray, score_thr: float, max_boxes: int, ) -> list[tuple[np.ndarray, float, str]]: boxes = np.asarray(boxes, dtype=np.float64) scores = np.asarray(scores, dtype=np.float64).reshape(-1) labels = np.asarray(labels, dtype=np.int64).reshape(-1) order = np.argsort(-scores, kind="stable") out: list[tuple[np.ndarray, float, str]] = [] for idx in order: if float(scores[idx]) < float(score_thr): continue label = int(labels[idx]) name = CLASS_NAMES[label] if 0 <= label < len(CLASS_NAMES) else str(label) out.append((boxes[idx], float(scores[idx]), name)) if len(out) >= int(max_boxes): break return out def camera_records(frame_assets: dict[str, Any], repo_root: str | Path) -> dict[str, Path]: record = frame_assets["assets"]["camera_images"] return {item["name"]: resolve_path(item["path"], repo_root) for item in record["images"]} def load_lidar2img_by_camera(frame_assets: dict[str, Any], repo_root: str | Path) -> dict[str, np.ndarray]: value = load_record(frame_assets["assets"]["lidar2img"], repo_root).astype(np.float64).reshape(1, 6, 4, 4)[0] return {name: value[index] for index, name in enumerate(MODEL_CAMERA_ORDER)} def wrap_text(text: str, width: int = 42) -> list[str]: words = str(text).split() lines: list[str] = [] current = "" for word in words: trial = word if not current else current + " " + word if len(trial) <= width: current = trial else: if current: lines.append(current) current = word if current: lines.append(current) return lines def add_sidebar(sheet: Image.Image, frame_index: int, branch: str, score_thr: float, max_boxes: int, result_dir: str | Path | None) -> Image.Image: sidebar_w = 520 canvas = Image.new("RGB", (sheet.width + sidebar_w, sheet.height), (250, 250, 250)) canvas.paste(sheet, (0, 0)) draw = ImageDraw.Draw(canvas) title_font = load_font(24) body_font = load_font(18) small_font = load_font(15) x0 = sheet.width + 24 y = 24 draw.rectangle((sheet.width, 0, canvas.width - 1, canvas.height - 1), fill=(245, 247, 250), outline=(180, 180, 180), width=2) draw.text((x0, y), "Final Detection Visualization", fill=(0, 0, 0), font=title_font) y += 46 lines = [ f"frame{frame_index:03d} / {branch}", "Final Model: W8A8 per_channel", "Role: AidLite/QNN board demo", "Input: six raw camera JPGs", "Preprocess: BGR->RGB normalize resize CHW pad", "Temporal: scene-start reset, then prev_bev recursion", f"Score threshold: {score_thr}", f"Max boxes: {max_boxes}", "Red: W8A8 final detections", ] for raw in lines: color = (0, 0, 0) if raw.startswith("Final"): color = (180, 0, 0) elif raw.startswith("Red"): color = (220, 20, 20) for line in wrap_text(raw, width=42): draw.text((x0, y), line, fill=color, font=body_font) y += 27 y += 4 if result_dir is not None: y += 14 draw.text((x0, y), "Input result directory:", fill=(0, 0, 0), font=body_font) y += 28 for line in wrap_text(str(result_dir), width=48): draw.text((x0, y), line, fill=(80, 80, 80), font=small_font) y += 22 return canvas def save_camera_grid_visualization( output_dir: str | Path, frame_index: int, frame_assets: dict[str, Any], repo_root: str | Path, boxes: np.ndarray, scores: np.ndarray, labels: np.ndarray, score_thr: float = 0.25, max_boxes: int = 80, panel_width: int = 900, draw_labels: bool = False, metrics_panel: bool = True, result_dir: str | Path | None = None, ) -> dict[str, Any]: output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) font = load_font(18) label_font = load_font(15) cameras = camera_records(frame_assets, repo_root) lidar2img = load_lidar2img_by_camera(frame_assets, repo_root) preds = select_predictions(boxes, scores, labels, score_thr, max_boxes) panels = [] per_camera = {} for camera_name in GRID_CAMERA_ORDER: image = Image.open(cameras[camera_name]).convert("RGB").resize((800, 450)) draw = ImageDraw.Draw(image) count = 0 for box, score, class_name in preds: points = project_box(box, lidar2img[camera_name], image.size) if points is None: continue draw_projected_box(draw, points, (255, 40, 40), width=3) if draw_labels: draw_label(draw, tuple(points[:, :2].min(axis=0)), f"W8A8 {class_name} {score:.2f}", (255, 180, 180), label_font) count += 1 header = Image.new("RGB", (image.width, 44), (20, 20, 20)) hdraw = ImageDraw.Draw(header) hdraw.text((10, 10), f"{camera_name} | W8A8 {count}", fill=(255, 255, 255), font=font) panel = Image.new("RGB", (image.width, image.height + header.height), (0, 0, 0)) panel.paste(header, (0, 0)) panel.paste(image, (0, header.height)) panel = panel.resize((panel_width, int(panel.height * panel_width / panel.width))) panels.append(panel) per_camera[camera_name] = {"w8a8": int(count)} cols, rows = 3, 2 panel_w, panel_h = panels[0].size title_h = 70 sheet = Image.new("RGB", (cols * panel_w, rows * panel_h + title_h), (245, 245, 245)) sheet_draw = ImageDraw.Draw(sheet) branch = "scene_start" if bool(frame_assets.get("is_scene_start", False)) else "temporal" token = str(frame_assets.get("sample_token", ""))[:16] title = f"Sample4 frame{frame_index:03d} ({branch}) token={token} | red=W8A8 final | score_thr={score_thr}" sheet_draw.text((16, 18), title, fill=(0, 0, 0), font=font) for index, panel in enumerate(panels): x = (index % cols) * panel_w y = title_h + (index // cols) * panel_h sheet.paste(panel, (x, y)) if metrics_panel: sheet = add_sidebar(sheet, frame_index, branch, score_thr, max_boxes, result_dir) out_file = output_path / f"frame{frame_index:03d}_camera_grid.png" sheet.save(out_file) return { "path": str(out_file), "sha256": sha256_file(out_file), "mode": "camera_grid_w8a8_only", "score_thr": float(score_thr), "max_boxes": int(max_boxes), "candidate_count": int(len(preds)), "per_camera_projected_counts": per_camera, } def save_camera_grid_gif(records: Sequence[dict[str, Any]], output_dir: str | Path, name: str = "sample4_camera_grid.gif", width: int = 1800, duration_ms: int = 700) -> dict[str, Any] | None: paths = [Path(record["path"]) for record in records if record.get("path")] if not paths: return None frames = [] for path in paths: image = Image.open(path).convert("RGB") height = int(image.height * width / image.width) frames.append(image.resize((width, height))) out_file = Path(output_dir) / name frames[0].save(out_file, save_all=True, append_images=frames[1:], duration=duration_ms, loop=0) return {"path": str(out_file), "sha256": sha256_file(out_file), "frame_count": len(frames)} def save_camera_grid_summary(output_dir: str | Path, records: Sequence[dict[str, Any]], gif_record: dict[str, Any] | None) -> dict[str, Any]: output_path = Path(output_dir) summary = { "status": "PASS", "mode": "sample4_camera_grid_w8a8_only", "visualizations": list(records), "gif": gif_record, } path = output_path / "sample4_camera_grid_summary.json" path.write_text(json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") summary["path"] = str(path) summary["sha256"] = sha256_file(path) return summary