| """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] |
| 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))) |
| |
| 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] |
| return float(sum(xs) / len(xs)) if xs else float("nan") |
|
|
