| import os |
| import sys |
| import subprocess |
| import json |
| import csv |
| import glob |
| from tqdm import tqdm |
| import numpy as np |
|
|
| BASE_DIR = "/root/autodl-tmp/SplatAtlas" |
| OUT_BASE = os.path.join(BASE_DIR, "outputs") |
| DATASET_ROOT = "/root/autodl-tmp/dataset" |
| CSV_PATH = os.path.join(BASE_DIR, "scripts/phase1_validation/validation_results_final.csv") |
| ERR_LOG = os.path.join(BASE_DIR, "scripts/phase1_validation/final_batch_errors.log") |
|
|
| |
| ALL_METHODS = [ |
| "3dgsmcmc", "analyticsplatting", "atomgs", "coadaptgs", "conegs", |
| "erankgs", "ges", "ghap", "gof", "gslpm", "lightgaussian", "opti3dgs", |
| "pgsr", "reactgs", "steepgs", "vanilla_3dgs", "absgs", "gaussianpro", |
| "minisplatting", "pixelgs" |
| ] |
|
|
| def build_scene_database(): |
| """绝对白名单机制，屏蔽衍生目录""" |
| db = {} |
| allowed_roots = ["360", "tnt", "deepblending_clean"] |
| |
| for root in allowed_roots: |
| search_path = os.path.join(DATASET_ROOT, root, "*") |
| for path in glob.glob(search_path): |
| if not os.path.isdir(path): continue |
| |
| scene_name = os.path.basename(path).lower() |
| |
| if root == "tnt": |
| res, dataset_type = 2, "tnt" |
| elif root == "deepblending_clean": |
| res, dataset_type = 1, "db" |
| elif root == "360": |
| outdoor = ['garden', 'stump', 'treehill', 'bicycle', 'flowers'] |
| res = 4 if any(o in scene_name for o in outdoor) else 2 |
| dataset_type = "mip360_outdoor" if res == 4 else "mip360_indoor" |
| |
| db[scene_name] = {"source": path, "res": res, "dataset": dataset_type, "bg": "0,0,0"} |
| return db |
|
|
| def log_error(cell_name, phase, stdout, stderr): |
| with open(ERR_LOG, "a") as f: |
| f.write(f"\n{'='*50}\n[{phase} ERROR] Cell: {cell_name}\n") |
| if stdout: f.write(f"--- STDOUT ---\n" + "\n".join(stdout.strip().split("\n")[-20:]) + "\n") |
| if stderr: f.write(f"--- STDERR ---\n" + "\n".join(stderr.strip().split("\n")[-20:]) + "\n") |
|
|
| def get_actual_cell_dir(method, scene): |
| target = f"{method}_{scene}".lower() |
| if os.path.exists(OUT_BASE): |
| for d in os.listdir(OUT_BASE): |
| if d.lower() == target: |
| return os.path.join(OUT_BASE, d) |
| return None |
|
|
| def phase_a_native(method, scene_key, scene_cfg, cell_dir): |
| json_path = os.path.join(cell_dir, "metrics_test_iter30000.json") |
| if os.path.exists(json_path): |
| try: |
| with open(json_path, 'r') as f: |
| data = json.load(f) |
| if "PSNR" in data or ("photometric" in data and "PSNR" in data["photometric"]): |
| return True |
| except: pass |
|
|
| cmd_render = [ |
| "python", "scripts/main_render.py", "--method", method, |
| "--source_path", scene_cfg["source"], "--model_path", cell_dir, |
| "--iteration", "30000", "--resolution", str(scene_cfg["res"]) |
| ] |
| res = subprocess.run(cmd_render, cwd=BASE_DIR, capture_output=True, text=True) |
| if res.returncode != 0: |
| log_error(f"{method}_{scene_key}", "PHASE_A_RENDER", res.stdout, res.stderr) |
| return False |
|
|
| py_metric = f""" |
| import sys, json, os |
| sys.path.append("ufd_evalkit") |
| from photometric import compute_photometric_metrics |
| r_dir = "{cell_dir}/renders_test_30000/renders" |
| g_dir = "{cell_dir}/renders_test_30000/gt" |
| if not os.path.exists(r_dir): |
| r_dir = "{cell_dir}/renders_test_30000" |
| g_dir = "{cell_dir}/gt_test_30000" |
| metrics = compute_photometric_metrics(r_dir, g_dir) |
| with open("{json_path}", "w") as f: |
| json.dump({{"PSNR": metrics.get("PSNR", 0)}}, f, indent=4) |
| """ |
| res2 = subprocess.run(["python", "-c", py_metric], cwd=BASE_DIR, capture_output=True, text=True) |
| if res2.returncode != 0: |
| log_error(f"{method}_{scene_key}", "PHASE_A_METRIC", res2.stdout, res2.stderr) |
| return False |
| return True |
|
|
| def phase_b_gsplat(ply_path, scene_cfg, cell_dir, cell_name): |
| out_dir = os.path.join(OUT_BASE, "phase1_validation", cell_name) |
| os.makedirs(out_dir, exist_ok=True) |
| cmd = [ |
| "python", "scripts/phase1_validation/render_single.py", |
| "--ply_path", ply_path, "--source_path", scene_cfg["source"], |
| "--model_path", cell_dir, "--output_dir", out_dir, |
| "--resolution", str(scene_cfg["res"]), "--bg_color", scene_cfg["bg"] |
| ] |
| res = subprocess.run(cmd, cwd=BASE_DIR, capture_output=True, text=True) |
| |
| if res.returncode != 0: |
| log_error(cell_name, "PHASE_B_GSPLAT", res.stdout, res.stderr) |
| return None, None, None, "ERROR_B" |
|
|
| native_psnr, gsplat_psnr, delta = None, None, None |
| for line in res.stdout.split('\n'): |
| if "Mean PSNR (ours/gsplat)" in line: |
| try: gsplat_psnr = float(line.split(":")[1].strip().split()[0]) |
| except: pass |
| if "Mean PSNR (native baseline)" in line: |
| try: native_psnr = float(line.split(":")[1].strip().split()[0]) |
| except: pass |
| if "Delta" in line and "dB" in line and "STATUS" not in line: |
| try: delta = float(line.split(":")[1].strip().split()[0]) |
| except: pass |
| |
| if gsplat_psnr is None: |
| try: |
| with open(os.path.join(out_dir, "psnr_results.json"), "r") as f: |
| d = json.load(f) |
| gsplat_psnr = float(np.mean(list(d.values()))) |
| except: pass |
|
|
| notes = "" |
| if gsplat_psnr is not None and gsplat_psnr < 15.0: |
| notes = "low PSNR" |
|
|
| return native_psnr, gsplat_psnr, delta, notes |
|
|
| def main(): |
| print("=== SplatAtlas Phase 1 Final Full Batch ===") |
| SCENES = build_scene_database() |
| print(f"[Info] Loaded {len(SCENES)} verified scenes.") |
| |
| completed = set() |
| write_header = not os.path.exists(CSV_PATH) |
| if not write_header: |
| with open(CSV_PATH, "r") as f: |
| reader = csv.DictReader(f) |
| for row in reader: |
| completed.add((row['method'], row['scene'])) |
|
|
| valid_cells = [] |
| for method in ALL_METHODS: |
| for scene, cfg in SCENES.items(): |
| cell_dir = get_actual_cell_dir(method, scene) |
| if cell_dir: |
| valid_cells.append((method, scene, cfg, cell_dir)) |
|
|
| print(f"[Info] Found {len(valid_cells)} pre-trained cells across {len(ALL_METHODS)} methods.\n") |
|
|
| pbar = tqdm(total=len(valid_cells), desc="Validating Cells", unit="cell") |
|
|
| with open(CSV_PATH, "a", newline='') as f: |
| writer = csv.writer(f) |
| if write_header: |
| writer.writerow(["method", "scene", "dataset", "resolution", "has_ply", "native_psnr", "gsplat_psnr", "delta_db", "status", "notes"]) |
|
|
| for method, scene, cfg, cell_dir in valid_cells: |
| if (method, scene) in completed: |
| pbar.update(1) |
| continue |
|
|
| ply_path = os.path.join(cell_dir, "point_cloud/iteration_30000/point_cloud.ply") |
| if not os.path.exists(ply_path): |
| writer.writerow([method, scene, cfg["dataset"], cfg["res"], "False", "", "", "", "SKIP", "PLY missing"]) |
| f.flush() |
| pbar.set_postfix_str(f"[{method}/{scene}] SKIP") |
| pbar.update(1) |
| continue |
|
|
| cell_name = os.path.basename(cell_dir) |
| |
| if not phase_a_native(method, scene, cfg, cell_dir): |
| writer.writerow([method, scene, cfg["dataset"], cfg["res"], "True", "", "", "", "ERROR_A", "Native render failed"]) |
| f.flush() |
| pbar.set_postfix_str(f"[{method}/{scene}] ERROR_A") |
| pbar.update(1) |
| continue |
|
|
| native, gsplat, delta, notes = phase_b_gsplat(ply_path, cfg, cell_dir, cell_name) |
| |
| if native is None and gsplat is None: |
| writer.writerow([method, scene, cfg["dataset"], cfg["res"], "True", "", "", "", "ERROR_B", notes]) |
| f.flush() |
| pbar.set_postfix_str(f"[{method}/{scene}] ERROR_B") |
| else: |
| if delta is None and native is not None and gsplat is not None: |
| delta = gsplat - native |
| |
| status = "PASS" |
| if delta is not None and abs(delta) >= 1.0: |
| status = "FAIL" |
| notes = f"Systematic Bias | {notes}".strip(" | ") |
| |
| writer.writerow([method, scene, cfg["dataset"], cfg["res"], "True", |
| f"{native:.4f}" if native else "", |
| f"{gsplat:.4f}" if gsplat else "", |
| f"{delta:.4f}" if delta else "", |
| status, notes]) |
| f.flush() |
| pbar.set_postfix_str(f"[{method}/{scene}] {status}") |
| |
| pbar.update(1) |
| |
| pbar.close() |
| print("\n========================================") |
| print(f"FINAL BATCH COMPLETED. Results saved to: {CSV_PATH}") |
| print("========================================") |
|
|
| if __name__ == "__main__": |
| main() |
|
|
