File size: 5,181 Bytes
23e73f9 | 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 | import os
import json
class NpEncoder(json.JSONEncoder):
def default(self, obj):
if hasattr(obj, 'item'):
return obj.item()
return super().default(obj)
import argparse
import torch
import numpy as np
from scipy.stats import entropy
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from core.registry import METHOD_REGISTRY
import methods
def compute_gini(array):
array = array.flatten()
array = array[array > 0]
if len(array) == 0: return 0.0
array = np.sort(array)
index = np.arange(1, array.shape[0] + 1)
n = array.shape[0]
return ((np.sum((2 * index - n - 1) * array)) / (n * np.sum(array)))
def generate_standard_grid(cameras, resolution=64):
"""基于相机分布生成标准查询网格 (默认 64^3 约 26 万点,兼顾精度与显存)"""
if not cameras:
min_bounds = torch.tensor([-5.0, -5.0, -5.0])
max_bounds = torch.tensor([5.0, 5.0, 5.0])
else:
cam_centers = torch.stack([c.camera_center for c in cameras])
min_bounds, _ = torch.min(cam_centers, dim=0)
max_bounds, _ = torch.max(cam_centers, dim=0)
extent = max_bounds - min_bounds
min_bounds -= extent * 0.1
max_bounds += extent * 0.1
X = torch.linspace(min_bounds[0], max_bounds[0], resolution)
Y = torch.linspace(min_bounds[1], max_bounds[1], resolution)
Z = torch.linspace(min_bounds[2], max_bounds[2], resolution)
grid_x, grid_y, grid_z = torch.meshgrid(X, Y, Z, indexing='ij')
query_points = torch.stack([grid_x.flatten(), grid_y.flatten(), grid_z.flatten()], dim=-1)
return query_points.cuda()
def analyze_physics(model, cameras):
stats = {}
# 1. 变体自述物理指标 (特异性指标,如畸变度、Billboard偏差等)
try:
specific_metrics = model.compute_physical_metrics(cameras=cameras)
stats.update(specific_metrics)
except NotImplementedError:
print(" ⚠️ [Physics] Wrapper 未实现 compute_physical_metrics,跳过特异性指标。")
except Exception as e:
print(f" ⚠️ [Physics] 特异性指标计算发生错误: {e}")
# 2. 宏观标量场评估 (Apple-to-Apple 的绝对公平空间测试)
try:
query_points = generate_standard_grid(cameras, resolution=64)
with torch.no_grad():
scalar_field = model.evaluate_spatial_field(query_points, cameras=cameras)
field_np = scalar_field.cpu().numpy()
stats['spatial_gini'] = compute_gini(field_np)
p = field_np / (field_np.sum() + 1e-7)
stats['density_entropy'] = entropy(p[p > 0])
stats["Field_Total_Volume"] = float(np.sum(field_np))
except NotImplementedError:
print(" ⚠️ [Physics] Wrapper 未实现 evaluate_spatial_field,跳过空间分布诊断。")
except Exception as e:
print(f" ⚠️ [Physics] 空间标量场评估发生异常: {e}")
return {k: round(v, 4) if isinstance(v, float) else v for k, v in stats.items()}
def load_cameras_from_json(cam_path):
"""为了兼容旧流水线,简单构建具有 camera_center 属性的 Dummy Cam"""
if not os.path.exists(cam_path): return []
with open(cam_path, 'r') as f: cams_data = json.load(f)
class DummyCam:
def __init__(self, center):
self.camera_center = torch.tensor(center, dtype=torch.float32)
return [DummyCam(c['center']) for c in cams_data if 'center' in c]
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--method", type=str, required=True, help="模型方法名称")
parser.add_argument("--source_path", type=str, default="", help="原始数据集路径")
parser.add_argument("--model_path", type=str, required=True, help="模型输出目录")
parser.add_argument("--iteration", type=int, default=30000)
args = parser.parse_args()
model_dir = os.path.abspath(args.model_path)
if not os.path.exists(model_dir):
print(f"⚠️ [Physics] 跳过: 模型目录不存在 {model_dir}")
return
folder_name = os.path.basename(model_dir)
print(f"🔬 [Physics] 正在精准解剖场景: {folder_name} ...")
# 实例化对应的 Wrapper 并加载资产
dataset_config = {"source_path": args.source_path, "model_path": args.model_path, "resolution": 1}
try:
model_class = methods.load_method(args.method)
model = model_class(dataset_config, hyperparams={})
model.load(args.model_path, args.iteration)
except Exception as e:
print(f"❌ [Physics] 实例化或加载模型失败: {e}")
return
cameras = model.scene.getTrainCameras() if hasattr(model, 'scene') else []
stats = analyze_physics(model, cameras)
if stats:
stats['experiment'] = folder_name
with open(os.path.join(model_dir, f"offline_physics_{args.iteration}.json"), 'w') as f:
json.dump(stats, f, indent=4, cls=NpEncoder)
print(f" ✨ [Physics] {folder_name} 分析完成!提取指标数: {len(stats)}")
if __name__ == "__main__":
main()
|