File size: 7,005 Bytes
eafbe80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | #!/usr/bin/env python3
"""Aggregate per-view revisit closure + minimal cross-view 2D metrics (open-domain multiview)."""
from __future__ import annotations
import argparse
import json
import os
import sys
from typing import Any, Dict, List, Optional, Tuple
import cv2
import numpy as np
_metrics_dir = os.path.dirname(os.path.abspath(__file__))
if _metrics_dir not in sys.path:
sys.path.insert(0, _metrics_dir)
import psnr_lpips as _pl # noqa: E402
try:
from skimage.metrics import structural_similarity as _skimage_ssim
_HAS_SKIMAGE = True
except Exception:
_skimage_ssim = None # type: ignore
_HAS_SKIMAGE = False
def _read_json(path: str) -> Dict[str, Any]:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def _read_last_frame_rgb(video_path: str) -> Optional[np.ndarray]:
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
return None
n = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
if n <= 0:
cap.release()
return None
cap.set(cv2.CAP_PROP_POS_FRAMES, max(0, n - 1))
ok, bgr = cap.read()
cap.release()
if not ok:
return None
return cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
def _mse(a: np.ndarray, b: np.ndarray) -> float:
d = a.astype(np.float64) - b.astype(np.float64)
return float(np.mean(d ** 2))
def _psnr(mse_v: float) -> float:
if mse_v <= 0:
return 100.0
return float(10.0 * np.log10((255.0 ** 2) / mse_v))
def _ssim(a: np.ndarray, b: np.ndarray) -> Optional[float]:
if not _HAS_SKIMAGE or _skimage_ssim is None:
return None
try:
if a.shape != b.shape:
b = cv2.resize(b, (a.shape[1], a.shape[0]), interpolation=cv2.INTER_AREA)
try:
return float(_skimage_ssim(a, b, channel_axis=2, data_range=255))
except TypeError:
return float(_skimage_ssim(a, b, multichannel=True, data_range=255))
except Exception:
return None
def main() -> int:
ap = argparse.ArgumentParser(description="Aggregate multiview closure + cross-view vs ref first frame")
ap.add_argument(
"--multiview_root",
required=True,
help="e.g. .../static_consistency/open_domain/multiview_revisit",
)
ap.add_argument("--ref_view", type=str, default="0", help="view_id used as reference first frame")
ap.add_argument("--output_json", required=True)
ap.add_argument("--device", default="cuda", help="LPIPS device")
args = ap.parse_args()
root = os.path.abspath(args.multiview_root)
summary_path = os.path.join(root, "multiview_revisit_summary.json")
if not os.path.isfile(summary_path):
out = {
"error": f"missing {summary_path}",
"multiview_root": root,
"per_view_closure": [],
"cross_view_vs_ref": [],
}
os.makedirs(os.path.dirname(os.path.abspath(args.output_json)) or ".", exist_ok=True)
with open(args.output_json, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2)
return 0
summary_data = _read_json(summary_path)
rows: List[Dict[str, Any]] = list(summary_data.get("summary") or [])
ref_row: Optional[Dict[str, Any]] = None
for r in rows:
if str(r.get("view_id")) == str(args.ref_view):
ref_row = r
break
ref_first_rgb: Optional[np.ndarray] = None
if ref_row and ref_row.get("first_frame_image"):
p = str(ref_row["first_frame_image"])
if os.path.isfile(p):
bgr = cv2.imread(p, cv2.IMREAD_COLOR)
if bgr is not None:
ref_first_rgb = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
per_view_closure: List[Dict[str, Any]] = []
cross_view: List[Dict[str, Any]] = []
lpips_model = _pl._lpips_model(device=args.device)
for r in rows:
vid = str(r.get("view_id", ""))
out_dir = str(r.get("output_dir") or "")
if not out_dir or not os.path.isdir(out_dir):
continue
closure_p = os.path.join(out_dir, "revisit_closure_metrics.json")
entry: Dict[str, Any] = {"view_id": vid, "output_dir": out_dir}
if os.path.isfile(closure_p):
try:
c = _read_json(closure_p)
entry["closure_first_vs_last_mse"] = c.get("closure_first_vs_last_mse")
entry["closure_first_vs_last_psnr"] = c.get("closure_first_vs_last_psnr")
entry["num_chunks"] = c.get("num_chunks")
except Exception as e:
entry["closure_error"] = str(e)
else:
entry["closure_error"] = "missing revisit_closure_metrics.json"
per_view_closure.append(entry)
if ref_first_rgb is None or str(vid) == str(args.ref_view):
continue
mp4 = os.path.join(out_dir, "combo_revisit_4chunk_gen_only.mp4")
if not os.path.isfile(mp4):
cross_view.append({"view_id": vid, "error": "missing combo_revisit_4chunk_gen_only.mp4"})
continue
last_rgb = _read_last_frame_rgb(mp4)
if last_rgb is None:
cross_view.append({"view_id": vid, "error": "cannot read last frame"})
continue
if last_rgb.shape[:2] != ref_first_rgb.shape[:2]:
ref_r = cv2.resize(ref_first_rgb, (last_rgb.shape[1], last_rgb.shape[0]), interpolation=cv2.INTER_AREA)
else:
ref_r = ref_first_rgb
mse_v = _mse(last_rgb, ref_r)
row = {
"view_id": vid,
"ref_view": str(args.ref_view),
"last_vs_ref_first_mse": mse_v,
"last_vs_ref_first_psnr": _psnr(mse_v),
"last_vs_ref_first_ssim": _ssim(last_rgb, ref_r),
"last_vs_ref_first_lpips": _pl.lpips_distance(last_rgb, ref_r, lpips_model, device=args.device),
}
cross_view.append(row)
metric_definitions = {
"per_view_closure": "From run_combo_revisit_fixed_first revisit_closure_metrics.json (same-view first vs last).",
"cross_view_vs_ref": (
"Heuristic: last frame of view k generated video vs reference view input first frame. "
"Not multi-view geometry; use when opendomain images depict the same object/scene."
),
}
out = {
"multiview_root": root,
"ref_view": str(args.ref_view),
"metric_definitions": metric_definitions,
"per_view_closure": per_view_closure,
"cross_view_vs_ref": cross_view,
"notes": [
"Key-object ROI not applied (full frame).",
"3D consistency requires depth+pose; open_domain uses 2D proxies only.",
],
}
outp = os.path.abspath(args.output_json)
os.makedirs(os.path.dirname(outp) or ".", exist_ok=True)
with open(outp, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2)
print(f"[aggregate_multiview_open_domain_metrics] views={len(per_view_closure)} -> {outp}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|