worldmodel/plans/iphone/data_format_specification_part2.md

title, date, draft, tags, categories

title	date	draft	tags	categories
酒店场景建模项目 - 数据格式规范（续）	2026-05-20	false	iPhone 3D 重建 RoomPlan 物理 酒店场景	RoomPlan

酒店场景建模项目 - 数据格式规范（续）

本文档是 data_format_specification.md 的续篇

---

五、物理数据格式（续）

5.3 材质物理属性（续）

material_props.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

{
  "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（完整示例）

{
  "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 跨平台交换格式优先级

点云格式:
  首选: 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 数据压缩策略

点云压缩

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)

数据集打包

# 场景级打包
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大规模数据格式

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 完整性检查清单

必需文件检查:
  - 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 自动化验证脚本

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加载器

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