"""Evaluation protocols (§4). * interpolation : render held-out frames, report PSNR / SSIM / LPIPS / D-RMSE (§4.2). * extrapolation : controlled lateral-shift sweep {1,2,3,4,6} m, report PSNR / SSIM / FID vs deviated GT (§4.3). Synthetic deviated GT is available because the world is known. * lane consistency: lane mIoU vs projected map + rendered-lane reprojection (§4.4). All tables are protocol templates: this fills the *measured* MapGS column; baseline columns remain TBD and must come from your own re-runs (integrity note). """ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, List, Optional import torch from mapgs.config import MapGSConfig from mapgs.data.synthetic import SyntheticDataset, render_deviated_gt from mapgs.hdmap.rasterize_map import project_polylines, render_lane_mask from mapgs.losses import perturb_pose, chamfer_2d from mapgs.eval.metrics import psnr, ssim, d_rmse, lane_iou, LPIPSMetric, FID from mapgs.model import MapGS from mapgs.model.dynamic import place_dynamic_gaussians from mapgs.render.rasterizer import GaussianRasterizer from mapgs.train.render import batched_plucker, render_scene_views, extract_scene_dynamic class Evaluator: def __init__(self, model: MapGS, cfg: MapGSConfig, device: str = "cuda", ras: Optional[GaussianRasterizer] = None): self.model = model.eval() self.cfg = cfg self.device = device self.ras = ras or GaussianRasterizer() self.H, self.W = cfg.data.height, cfg.data.width self.lpips = LPIPSMetric(device=device) self.fid = FID(device=device) @torch.no_grad() def _decode(self, sample): ckK = sample.ctx_K.to(self.device)[None] ckc = sample.ctx_c2w.to(self.device)[None] pl = batched_plucker(ckK, ckc, self.H, self.W) dyn = None if sample.box_centers.shape[0] > 0: dyn = { "box_centers": sample.box_centers[None], "box_rots": sample.box_rots[None], "box_size": sample.box_size[None], "canon_idx": sample.canon_idx[None], "valid": torch.ones(1, sample.box_centers.shape[0], dtype=torch.bool), } dyn = {k: v.to(self.device) for k, v in dyn.items()} decoded = self.model( sample.ctx_images.to(self.device)[None], pl, sample.ctx_tids.to(self.device)[None], sample.anchor_pos.to(self.device)[None], sample.anchor_type.to(self.device)[None], sample.anchor_normal.to(self.device)[None], s_t=self.cfg.model.tokens.s_max, dynamic=dyn, ) return decoded.scene(0), extract_scene_dynamic(dyn, 0) # ------------------------------------------------------------------ # @torch.no_grad() def interpolation(self, dataset, max_scenes: int = 50) -> Dict[str, float]: ps, ss, lp, dr = [], [], [], [] n = min(len(dataset), max_scenes) for i in range(n): s = dataset[i] g, dyn = self._decode(s) rend = render_scene_views(self.model, self.ras, g, dyn, s.sup_K.to(self.device), s.sup_c2w.to(self.device), s.sup_frame.to(self.device), self.H, self.W, self.model.uses_features) gt = s.sup_images.to(self.device) ps.append(float(psnr(rend["rgb"], gt))) ss.append(float(ssim(rend["rgb"], gt))) l = self.lpips(rend["rgb"], gt) if l is not None: lp.append(float(l)) dr.append(float(d_rmse(rend["depth"], s.sup_depth.to(self.device)))) return { "PSNR": _mean(ps), "SSIM": _mean(ss), "LPIPS": _mean(lp) if lp else float("nan"), "D-RMSE": _mean(dr), "n_scenes": n, } # ------------------------------------------------------------------ # @torch.no_grad() def extrapolation_sweep(self, dataset: SyntheticDataset, shifts=None, max_scenes: int = 30, frame: Optional[int] = None) -> Dict[float, Dict[str, float]]: shifts = shifts or list(self.cfg.eval.lateral_shifts) frame = frame if frame is not None else self.cfg.data.num_frames // 2 out: Dict[float, Dict[str, float]] = {} n = min(len(dataset), max_scenes) for sh in shifts: preds, gts, ps, ss = [], [], [], [] for i in range(n): s = dataset[i] scene = dataset.get_scene(i) g, dyn = self._decode(s) g_f = g if dyn is None else place_dynamic_gaussians( g, dyn["box_centers"], dyn["box_rots"], dyn["canon_idx"], frame) center_c2w = scene.cam2world[frame, 1] # center cam Kc = scene.K[1].to(self.device) dev = perturb_pose(center_c2w.to(self.device), lateral=float(sh)) out_r = self.ras.render(g_f, Kc[None], dev[None], self.H, self.W) pred = (self.model.feature_to_rgb(out_r.color) if self.model.uses_features else out_r.color[:, :3].clamp(0, 1))[0] gt, _ = render_deviated_gt(scene, frame, dev.cpu(), self.device, self.H, self.W, self.ras) gt = gt.to(self.device) preds.append(pred); gts.append(gt) ps.append(float(psnr(pred, gt))); ss.append(float(ssim(pred, gt))) fid = self.fid.compute(torch.stack(gts), torch.stack(preds)) if len(preds) >= 2 else None out[sh] = {"PSNR": _mean(ps), "SSIM": _mean(ss), "FID": fid if fid is not None else float("nan"), "n": n} return out # ------------------------------------------------------------------ # @torch.no_grad() def lane_consistency(self, dataset, max_scenes: int = 30, shift: float = 3.0, frame: Optional[int] = None) -> Dict[str, float]: frame = frame if frame is not None else self.cfg.data.num_frames // 2 ious, chamfers = [], [] n = min(len(dataset), max_scenes) for i in range(n): s = dataset[i] if len(s.lanes) == 0: continue g, dyn = self._decode(s) g_f = g if dyn is None else place_dynamic_gaussians( g, dyn["box_centers"], dyn["box_rots"], dyn["canon_idx"], frame) scene = dataset.get_scene(i) if isinstance(dataset, SyntheticDataset) else None base_c2w = (scene.cam2world[frame, 1] if scene is not None else s.ctx_c2w[1]).to(self.device) K = (scene.K[1] if scene is not None else s.ctx_K[1]).to(self.device) dev = perturb_pose(base_c2w, lateral=shift) out_r = self.ras.render(g_f, K[None], dev[None], self.H, self.W) lanes_dev = [l.to(self.device) for l in s.lanes] map_uv = project_polylines(lanes_dev, K[None], dev[None], self.H, self.W)[0] tgt = render_lane_mask(map_uv, self.H, self.W) pred_lane = out_r.aux[0, 0] if out_r.aux is not None else torch.zeros_like(tgt) ious.append(float(lane_iou(pred_lane, tgt))) # rendered-lane reprojection error (chamfer of supports) pv = (pred_lane > 0.5).nonzero(as_tuple=False).flip(-1).float() chamfers.append(float(chamfer_2d(pv, map_uv))) return {"lane_mIoU": _mean(ious), "lane_chamfer_px": _mean(chamfers), "n_scenes": len(ious)} def _mean(xs: List[float]) -> float: xs = [x for x in xs if x == x] # drop NaN return float(sum(xs) / len(xs)) if xs else float("nan")