--- title: "酒店场景建模项目 - 数据格式规范（续）" date: 2026-05-20 draft: false tags: ["iPhone", "3D 重建", "RoomPlan", "物理", "酒店场景"] categories: ["worldmodel"] --- # 酒店场景建模项目 - 数据格式规范（续） > 本文档是 [`data_format_specification.md`](data_format_specification.md) 的续篇 --- ## 五、物理数据格式（续） ### 5.3 材质物理属性（续） #### material_props.json（完整示例） ```json { "version": "1.0", "materials": [ { "material_id": "wood_oak", "name": "Oak Wood", "density": 600.0, "static_friction": 0.5, "dynamic_friction": 0.4, "restitution": 0.3, "young_modulus": 11000000000.0, "poisson_ratio": 0.3, "damping": 0.1 }, { "material_id": "fabric_cotton", "name": "Cotton Fabric", "density": 200.0, "static_friction": 0.7, "dynamic_friction": 0.6, "restitution": 0.1, "young_modulus": 1000000.0, "poisson_ratio": 0.4, "damping": 0.5 }, { "material_id": "glass", "name": "Glass", "density": 2500.0, "static_friction": 0.4, "dynamic_friction": 0.3, "restitution": 0.8, "young_modulus": 70000000000.0, "poisson_ratio": 0.24, "damping": 0.01, "transparency": 0.9, "ior": 1.52 }, { "material_id": "metal_steel", "name": "Steel", "density": 7800.0, "static_friction": 0.6, "dynamic_friction": 0.5, "restitution": 0.5, "young_modulus": 200000000000.0, "poisson_ratio": 0.3, "damping": 0.05 } ] } ``` --- ## 六、元数据格式 ### 6.1 全局元数据 #### metadata.json ```json { "dataset_name": "HotelScene-Dataset", "version": "1.0.0", "creation_date": "2026-05-16", "description": "High-fidelity digital twin dataset for hotel scenes", "license": "CC BY-NC 4.0", "citation": "@article{hotel2026, title={Hotel Scene Reconstruction}, ...}", "authors": [ { "name": "Zhang San", "affiliation": "University", "email": "zhangsan@university.edu", "orcid": "0000-0000-0000-0000" } ], "statistics": { "num_scenes": 10, "num_frames": 5000, "num_instances": 523, "total_size_gb": 750.5, "scene_types": { "lobby": 2, "corridor": 2, "room": 5, "bathroom": 5 } }, "coordinate_system": { "type": "right_hand", "up_axis": "Z", "forward_axis": "X", "unit": "meter" }, "sensors": { "lidar": { "model": "Livox Mid-360", "frequency_hz": 10, "range_m": 70, "accuracy_m": 0.02, "fov_deg": 360 }, "camera_rgb": { "model": "Azure Kinect DK", "resolution": [1920, 1080], "fps": 30, "fov_deg": 90, "sensor_size_mm": [6.4, 4.8] }, "camera_depth": { "model": "Azure Kinect DK ToF", "resolution": [640, 576], "fps": 30, "range_m": [0.25, 5.46], "accuracy_m": 0.01 }, "imu": { "model": "Xsens MTi-630", "frequency_hz": 400, "accel_range_g": 16, "gyro_range_dps": 2000 } }, "processing_info": { "slam_method": "FAST-LIO2", "reconstruction_method": "3D Gaussian Splatting", "semantic_method": "YOLO-World + SAM + CLIP", "software_versions": { "python": "3.10.12", "pytorch": "2.1.0", "open3d": "0.18.0", "colmap": "3.8" } } } ``` ### 6.2 场景元数据 #### scene_info.json（完整示例） ```json { "scene_id": "room_301", "scene_type": "hotel_room", "hotel_info": { "name": "Grand Hotel", "address": "123 Main St, Shanghai", "star_rating": 5, "floor": 3, "room_number": "301", "room_type": "deluxe" }, "capture_info": { "date": "2026-05-16", "time_start": "10:30:00", "time_end": "10:50:00", "duration_minutes": 20, "operator": "Zhang San", "device": "iPhone 15 Pro Max", "weather": "sunny", "lighting_condition": "natural + artificial", "temperature_celsius": 22, "humidity_percent": 45 }, "geometry": { "floor_area_sqm": 25.5, "ceiling_height_m": 2.8, "bbox_min": [0.0, 0.0, 0.0], "bbox_max": [5.0, 5.1, 2.8], "volume_m3": 71.4, "wall_thickness_m": 0.2 }, "statistics": { "num_frames": 500, "num_rgb_images": 500, "num_depth_images": 500, "num_lidar_scans": 200, "num_points_raw": 12345678, "num_points_processed": 5432109, "num_instances": 23, "trajectory_length_m": 15.2, "scan_coverage_sqm": 24.3 }, "quality_metrics": { "slam": { "loop_closure_error_m": 0.003, "trajectory_rmse_m": 0.015, "num_loop_closures": 5 }, "point_cloud": { "density_points_per_sqm": 5000, "coverage_percentage": 95.2, "noise_std_m": 0.018 }, "reconstruction": { "psnr_db": 28.5, "ssim": 0.87, "lpips": 0.14 }, "semantic": { "detection_map_50": 0.72, "instance_iou": 0.68, "scene_graph_completeness": 0.85 } }, "processing_status": { "data_collection": { "status": "completed", "timestamp": "2026-05-16T10:50:00Z" }, "slam": { "status": "completed", "timestamp": "2026-05-16T12:30:00Z", "duration_minutes": 45 }, "3dgs": { "status": "completed", "timestamp": "2026-05-16T18:00:00Z", "duration_minutes": 180, "iterations": 30000 }, "semantic": { "status": "completed", "timestamp": "2026-05-16T19:30:00Z", "duration_minutes": 60 }, "physics": { "status": "in_progress", "timestamp": null, "progress_percent": 45 } }, "notes": "High-quality scan with good lighting. Minor occlusion behind wardrobe.", "tags": ["deluxe_room", "high_quality", "complete_coverage"] } ``` --- ## 七、数据交换与压缩 ### 7.1 跨平台交换格式优先级 ```yaml 点云格式: 首选: PLY (binary) - 优势: 通用性最好，所有工具支持 - 劣势: 文件较大 备选: PCD (binary) - 优势: Open3D原生格式 - 劣势: 部分工具不支持 大规模: LAS/LAZ - 优势: 压缩率高（50-80%） - 劣势: 需要专门库 网格格式: 首选: OBJ + MTL - 优势: 最通用，人类可读 - 劣势: 文件较大，不支持动画 Web: GLTF/GLB - 优势: Web友好，支持PBR - 劣势: 桌面工具支持有限 游戏: FBX - 优势: Unity/Unreal原生 - 劣势: 专有格式 物理: USD/USDZ - 优势: 支持物理属性 - 劣势: 学习曲线陡峭 图像格式: 无损: PNG 有损: JPEG (quality=95) HDR: EXR 深度格式: 标准: 16-bit PNG 高精度: 32-bit EXR 数组: NPY (NumPy) 语义格式: 人类可读: JSON 大规模: HDF5 高效传输: Protocol Buffers ``` ### 7.2 数据压缩策略 #### 点云压缩 ```python import open3d as o3d import numpy as np def compress_pointcloud(input_ply, output_ply, voxel_size=0.01): """体素下采样压缩点云""" pcd = o3d.io.read_point_cloud(input_ply) # 体素下采样 pcd_down = pcd.voxel_down_sample(voxel_size) # 保存 o3d.io.write_point_cloud(output_ply, pcd_down, write_ascii=False) # 统计 original_size = len(pcd.points) compressed_size = len(pcd_down.points) ratio = compressed_size / original_size print(f"Compression ratio: {ratio:.2%}") print(f"Points: {original_size} → {compressed_size}") # LAZ压缩（需要laspy库） import laspy def compress_to_laz(points, colors, output_laz): """压缩为LAZ格式""" header = laspy.LasHeader(point_format=3, version="1.4") header.offsets = np.min(points, axis=0) header.scales = np.array([0.001, 0.001, 0.001]) las = laspy.LasData(header) las.x = points[:, 0] las.y = points[:, 1] las.z = points[:, 2] las.red = (colors[:, 0] * 65535).astype(np.uint16) las.green = (colors[:, 1] * 65535).astype(np.uint16) las.blue = (colors[:, 2] * 65535).astype(np.uint16) las.write(output_laz) ``` #### 数据集打包 ```bash # 场景级打包 tar -czf room_301.tar.gz room_301/ # 分卷压缩（大文件） tar -czf - room_301/ | split -b 1G - room_301.tar.gz.part # 解压分卷 cat room_301.tar.gz.part* | tar -xzf - # 7z高压缩率 7z a -t7z -m0=lzma2 -mx=9 room_301.7z room_301/ ``` ### 7.3 HDF5大规模数据格式 ```python import h5py import numpy as np def save_scene_hdf5(filepath, scene_data): """保存场景到HDF5（完整示例）""" with h5py.File(filepath, 'w') as f: # 元数据（属性） f.attrs['scene_id'] = scene_data['scene_id'] f.attrs['version'] = '1.0' f.attrs['creation_date'] = '2026-05-16' # 点云组 pc_group = f.create_group('point_cloud') pc_group.create_dataset( 'points', data=scene_data['points'], compression='gzip', compression_opts=9 ) pc_group.create_dataset('colors', data=scene_data['colors']) pc_group.create_dataset('normals', data=scene_data['normals']) pc_group.attrs['num_points'] = len(scene_data['points']) # 轨迹组 traj_group = f.create_group('trajectory') traj_group.create_dataset('timestamps', data=scene_data['timestamps']) traj_group.create_dataset('poses', data=scene_data['poses']) traj_group.attrs['num_poses'] = len(scene_data['poses']) # 图像组（可选，大数据） if 'images' in scene_data: img_group = f.create_group('images') for i, img in enumerate(scene_data['images']): img_group.create_dataset( f'frame_{i:07d}', data=img, compression='gzip' ) # 语义组 sem_group = f.create_group('semantic') sem_group.create_dataset('instance_ids', data=scene_data['instance_ids']) sem_group.create_dataset('features', data=scene_data['features']) # 场景图（JSON字符串） if 'scene_graph' in scene_data: import json sg_json = json.dumps(scene_data['scene_graph']) f.create_dataset('scene_graph_json', data=sg_json) def load_scene_hdf5(filepath): """从HDF5加载场景""" scene_data = {} with h5py.File(filepath, 'r') as f: # 元数据 scene_data['scene_id'] = f.attrs['scene_id'] scene_data['version'] = f.attrs['version'] # 点云 scene_data['points'] = f['point_cloud/points'][:] scene_data['colors'] = f['point_cloud/colors'][:] scene_data['normals'] = f['point_cloud/normals'][:] # 轨迹 scene_data['timestamps'] = f['trajectory/timestamps'][:] scene_data['poses'] = f['trajectory/poses'][:] # 语义 scene_data['instance_ids'] = f['semantic/instance_ids'][:] scene_data['features'] = f['semantic/features'][:] # 场景图 if 'scene_graph_json' in f: import json sg_json = f['scene_graph_json'][()] if isinstance(sg_json, bytes): sg_json = sg_json.decode('utf-8') scene_data['scene_graph'] = json.loads(sg_json) return scene_data # 流式读取（大数据） def stream_images_hdf5(filepath): """流式读取图像""" with h5py.File(filepath, 'r') as f: img_group = f['images'] for key in sorted(img_group.keys()): yield img_group[key][:] ``` --- ## 八、数据验证与质量控制 ### 8.1 完整性检查清单 ```yaml 必需文件检查: - scene_info.json - raw/timestamps.txt - processed/slam/trajectory.txt - processed/slam/dense_map.ply 可选文件检查: - processed/3dgs/input.ply - processed/mesh/scene.obj - processed/semantic/scene_graph.json - processed/physics/scene.usd 数据一致性检查: - RGB帧数 == 深度帧数 - 时间戳数量 == 帧数 - 轨迹长度 == 帧数 - 实例ID连续性 元数据完整性: - 所有必需字段存在 - 数值范围合理 - 时间戳格式正确 - 坐标系定义明确 质量指标检查: - SLAM误差 < 0.5% - 点云密度 > 1000 points/m² - 覆盖率 > 90% - PSNR > 25 dB ``` ### 8.2 自动化验证脚本 ```python import os import json from pathlib import Path from typing import List, Dict class DatasetValidator: """数据集完整性验证器""" def __init__(self, dataset_root: str): self.root = Path(dataset_root) self.errors = [] self.warnings = [] self.info = [] def validate_dataset(self) -> Dict: """验证整个数据集""" # 1. 检查全局元数据 self._check_global_metadata() # 2. 检查所有场景 scenes_dir = self.root / 'scenes' if scenes_dir.exists(): for scene_dir in scenes_dir.iterdir(): if scene_dir.is_dir(): self.validate_scene(scene_dir) else: self.errors.append("Missing scenes directory") # 3. 生成报告 return self.generate_report() def validate_scene(self, scene_path: Path): """验证单个场景""" scene_id = scene_path.name self.info.append(f"Validating scene: {scene_id}") # 必需文件检查 required_files = { 'scene_info.json': 'Scene metadata', 'raw/timestamps.txt': 'Timestamp alignment', 'processed/slam/trajectory.txt': 'SLAM trajectory' } for file, desc in required_files.items(): if not (scene_path / file).exists(): self.errors.append(f"{scene_id}: Missing {desc} ({file})") # 数据一致性检查 self._check_frame_consistency(scene_path, scene_id) # 元数据检查 self._check_scene_metadata(scene_path, scene_id) # 质量指标检查 self._check_quality_metrics(scene_path, scene_id) def _check_global_metadata(self): """检查全局元数据""" metadata_file = self.root / 'metadata.json' if not metadata_file.exists(): self.errors.append("Missing global metadata.json") return with open(metadata_file) as f: metadata = json.load(f) required_fields = ['dataset_name', 'version', 'license'] for field in required_fields: if field not in metadata: self.errors.append(f"Missing metadata field: {field}") def _check_frame_consistency(self, scene_path: Path, scene_id: str): """检查帧数一致性""" rgb_dir = scene_path / 'raw/rgb' depth_dir = scene_path / 'raw/depth' if rgb_dir.exists() and depth_dir.exists(): rgb_files = sorted(rgb_dir.glob('*.jpg')) depth_files = sorted(depth_dir.glob('*.png')) if len(rgb_files) != len(depth_files): self.warnings.append( f"{scene_id}: Frame count mismatch - " f"RGB={len(rgb_files)}, Depth={len(depth_files)}" ) # 检查文件名连续性 for i, (rgb_file, depth_file) in enumerate(zip(rgb_files, depth_files)): expected_name = f"{i:07d}" if rgb_file.stem != expected_name: self.warnings.append( f"{scene_id}: RGB frame naming issue at {i}" ) if depth_file.stem != expected_name: self.warnings.append( f"{scene_id}: Depth frame naming issue at {i}" ) def _check_scene_metadata(self, scene_path: Path, scene_id: str): """检查场景元数据""" info_file = scene_path / 'scene_info.json' if not info_file.exists(): return with open(info_file) as f: info = json.load(f) # 必需字段 required_fields = ['scene_id', 'scene_type', 'capture_info', 'geometry'] for field in required_fields: if field not in info: self.errors.append(f"{scene_id}: Missing metadata field '{field}'") # 检查scene_id一致性 if info.get('scene_id') != scene_id: self.warnings.append( f"{scene_id}: scene_id mismatch in metadata " f"(expected: {scene_id}, got: {info.get('scene_id')})" ) def _check_quality_metrics(self, scene_path: Path, scene_id: str): """检查质量指标""" info_file = scene_path / 'scene_info.json' if not info_file.exists(): return with open(info_file) as f: info = json.load(f) if 'quality_metrics' not in info: self.warnings.append(f"{scene_id}: Missing quality metrics") return metrics = info['quality_metrics'] # SLAM质量 if 'slam' in metrics: slam = metrics['slam'] if slam.get('loop_closure_error_m', 1.0) > 0.01: self.warnings.append( f"{scene_id}: High SLAM loop closure error " f"({slam.get('loop_closure_error_m'):.4f}m)" ) # 点云质量 if 'point_cloud' in metrics: pc = metrics['point_cloud'] if pc.get('coverage_percentage', 0) < 90: self.warnings.append( f"{scene_id}: Low coverage " f"({pc.get('coverage_percentage'):.1f}%)" ) if pc.get('density_points_per_sqm', 0) < 1000: self.warnings.append( f"{scene_id}: Low point cloud density " f"({pc.get('density_points_per_sqm')} points/m²)" ) # 渲染质量 if 'reconstruction' in metrics: recon = metrics['reconstruction'] if recon.get('psnr_db', 0) < 25: self.warnings.append( f"{scene_id}: Low PSNR ({recon.get('psnr_db'):.1f} dB)" ) def generate_report(self) -> Dict: """生成验证报告""" report = { 'status': 'PASS' if len(self.errors) == 0 else 'FAIL', 'summary': { 'total_errors': len(self.errors), 'total_warnings': len(self.warnings), 'total_info': len(self.info) }, 'errors': self.errors, 'warnings': self.warnings, 'info': self.info } return report def print_report(self): """打印报告""" report = self.generate_report() print("=" * 60) print("DATASET VALIDATION REPORT") print("=" * 60) print(f"Status: {report['status']}") print(f"Errors: {report['summary']['total_errors']}") print(f"Warnings: {report['summary']['total_warnings']}") print() if report['errors']: print("ERRORS:") for error in report['errors']: print(f" ❌ {error}") print() if report['warnings']: print("WARNINGS:") for warning in report['warnings']: print(f" ⚠️ {warning}") print() print("=" * 60) # 使用示例 validator = DatasetValidator('hotel_dataset') validator.validate_dataset() validator.print_report() ``` --- ## 九、数据加载工具库 ### 9.1 完整的Python加载器 ```python import json import numpy as np import open3d as o3d import cv2 from pathlib import Path from typing import Dict, List, Optional, Tuple from dataclasses import dataclass @dataclass class SceneMetadata: """场景元数据""" scene_id: str scene_type: str floor_area: float ceiling_height: float num_frames: int num_instances: int class HotelSceneLoader: """酒店场景数据加载器""" def __init__(self, dataset_root: str): self.root = Path(dataset_root) self._load_global_metadata() def _load_global_metadata(self): """加载全局元数据""" metadata_file = self.root / 'metadata.json' if metadata_file.exists(): with open(metadata_file) as f: self.global_metadata = json.load(f) else: self.global_metadata = {} def list_scenes(self) -> List[str]: """列出所有场景""" scenes_dir = self.root / 'scenes' return [d.name for d in scenes_dir.iterdir() if d.is_dir()] def load_scene_metadata(self, scene_id: str) -> SceneMetadata: """加载场景元数据""" info_file = self.root / 'scenes' / scene_id / 'scene_info.json' with open(info_file) as f: info = json.load(f) return SceneMetadata( scene_id=info['scene_id'], scene_type=info['scene_type'], floor_area=info['geometry']['floor_area_sqm'], ceiling_height=info['geometry']['ceiling_height_m'], num_frames=info['statistics']['num_frames'], num_instances=info['statistics']['num_instances'] ) def load_point_cloud(self, scene_id: str, cloud_type: str = 'dense') -> o3d.geometry.PointCloud: """ 加载点云 Args: scene_id: 场景ID cloud_type: 'sparse' 或 'dense' """ scene_path = self.root / 'scenes' / scene_id if cloud_type == 'dense': pc_file = scene_path / 'processed/slam/dense_map.ply' else: pc_file = scene_path / 'processed/slam/sparse_map.ply' if pc_file.exists(): return o3d.io.read_point_cloud(str(pc_file)) return None def load_trajectory(self, scene_id: str) -> Tuple[np.ndarray, np.ndarray]: """ 加载轨迹 Returns: timestamps: (N,) 时间戳 poses: (N, 4, 4) 位姿矩阵 """ traj_file = self.root / 'scenes' / scene_id / 'processed/slam/trajectory.txt' if not traj_file.exists(): return None, None data = np.loadtxt(traj_file) timestamps = data[:, 0] # 转换为4x4矩阵 from scipy.spatial.transform import Rotation positions = data[:, 1:4] quaternions = data[:, 4:8] rotations = Rotation.from_quat(quaternions).as_matrix() poses = np.zeros((len(data), 4, 4)) poses[:, :3, :3] = rotations poses[:, :3, 3] = positions poses[:, 3, 3] = 1.0 return timestamps, poses def load_mesh(self, scene_id: str) -> o3d.geometry.TriangleMesh: """加载网格模型""" mesh_file = self.root / 'scenes' / scene_id / 'processed/mesh/scene.obj' if mesh_file.exists(): return o3d.io