Scannet V2数据集下载后，如何用Python快速检查与验证文件完整性？

当你花费数小时甚至数天时间下载完Scannet V2这个庞大的三维数据集后，最迫切的问题就是：这些文件真的完整吗？作为计算机视觉和三维重建领域的重要基准数据集，Scannet V2包含超过1500个扫描场景，每个场景包含点云、RGB-D序列、语义标注等多种数据类型。本文将分享一套完整的Python验证流程，帮助你快速确认数据完整性，避免因文件损坏导致的后续实验问题。

1. 准备工作与环境配置

在开始验证前，我们需要确保环境配置正确。推荐使用Python 3.7+环境，并安装以下依赖库：

pip install open3d numpy pandas tqdm

验证工作主要涉及以下几个核心模块：

• os ：用于文件系统操作和路径管理
• hashlib ：计算文件校验和
• json ：解析场景元数据文件
• open3d ：点云可视化验证

建议创建一个专门的验证目录结构，例如：

scannet_validation/
├── checklist/         # 存放官方提供的清单文件
├── scripts/           # 验证脚本
└── logs/              # 验证结果日志

2. 基于官方清单的基础验证

Scannet V2官方提供了场景ID的清单文件，这是验证的第一道关卡。通常你会从官方获取或从数据集根目录找到 scans.txt 和 scans_test.txt 两个文件。

import os
from tqdm import tqdm

def validate_scene_ids(data_root, checklist_path):
    """验证下载的场景ID是否与官方清单匹配"""
    with open(checklist_path) as f:
        official_ids = {line.strip() for line in f}
    
    downloaded_ids = set(os.listdir(os.path.join(data_root, 'scans')))
    missing_ids = official_ids - downloaded_ids
    
    print(f"应下载场景数: {len(official_ids)}")
    print(f"实际下载场景数: {len(downloaded_ids)}")
    print(f"缺失场景数: {len(missing_ids)}")
    
    if missing_ids:
        with open('missing_scenes.log', 'w') as f:
            f.write('\n'.join(sorted(missing_ids)))

对于测试集，需要使用 scans_test.txt 进行同样的验证。这个基础检查可以快速发现明显的下载缺失问题。

3. 文件完整性深度检查

3.1 关键文件存在性验证

每个Scannet场景应包含以下核心文件类型：

文件类型	描述	是否必需
_vh_clean_2.ply	清洁后的点云	是
.aggregation.json	实例聚合信息	是
_vh_clean_2.labels.ply	带标签的点云	可选
.sens	RGB-D传感器数据	可选

def check_required_files(scene_path):
    """检查单个场景的必要文件是否存在"""
    required_files = [
        '_vh_clean_2.ply',
        '.aggregation.json',
        '_vh_clean_2.0.010000.segs.json'
    ]
    
    scene_id = os.path.basename(scene_path)
    missing = []
    
    for f in required_files:
        if not os.path.exists(os.path.join(scene_path, f"{scene_id}{f}")):
            missing.append(f)
    
    return missing

3.2 文件大小验证

对于大型数据集，文件大小是快速判断完整性的有效指标。以下是典型文件大小的参考范围：

EXPECTED_SIZES = {
    '_vh_clean_2.ply': (50, 200),  # MB
    '.sens': (1000, 5000),         # MB
    '.aggregation.json': (0.1, 1)  # MB
}

def validate_file_sizes(scene_path):
    """验证文件大小是否在合理范围内"""
    anomalies = []
    scene_id = os.path.basename(scene_path)
    
    for suffix, (min_mb, max_mb) in EXPECTED_SIZES.items():
        file_path = os.path.join(scene_path, f"{scene_id}{suffix}")
        if os.path.exists(file_path):
            size_mb = os.path.getsize(file_path) / (1024 * 1024)
            if not (min_mb <= size_mb <= max_mb):
                anomalies.append((suffix, size_mb))
    
    return anomalies

3.3 校验和验证

对于关键文件，计算MD5校验和可以提供更可靠的验证：

import hashlib

def calculate_md5(file_path, chunk_size=8192):
    """计算大文件的MD5校验和"""
    md5 = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            md5.update(chunk)
    return md5.hexdigest()

# 示例：预先记录已知正确的MD5值
KNOWN_MD5 = {
    'scene0001_00_vh_clean_2.ply': 'a1b2c3d4e5f6...',
    # 其他文件的MD5...
}

4. 数据可视化验证

4.1 点云可视化

使用Open3D进行点云可视化是最直观的验证方式：

import open3d as o3d

def visualize_random_pointcloud(data_root, num_samples=3):
    """随机抽样可视化几个点云"""
    scene_dirs = [d for d in os.listdir(os.path.join(data_root, 'scans')) 
                 if os.path.isdir(os.path.join(data_root, 'scans', d))]
    
    for scene_id in random.sample(scene_dirs, min(num_samples, len(scene_dirs))):
        ply_path = os.path.join(data_root, 'scans', scene_id, f"{scene_id}_vh_clean_2.ply")
        if os.path.exists(ply_path):
            pcd = o3d.io.read_point_cloud(ply_path)
            o3d.visualization.draw_geometries([pcd], window_name=scene_id)

4.2 语义标注验证

对于带有语义标注的场景，可以检查标注的完整性：

def validate_annotations(scene_path):
    """验证标注文件的完整性"""
    scene_id = os.path.basename(scene_path)
    json_path = os.path.join(scene_path, f"{scene_id}.aggregation.json")
    
    try:
        with open(json_path) as f:
            data = json.load(f)
            assert 'segGroups' in data
            assert len(data['segGroups']) > 0
        return True
    except:
        return False

5. 自动化验证脚本

将上述检查整合成一个自动化脚本：

import json
import random
from concurrent.futures import ThreadPoolExecutor
from collections import defaultdict

def full_validation(data_root, output_report='validation_report.json'):
    """执行完整的数据集验证流程"""
    report = defaultdict(list)
    scene_dirs = [os.path.join(data_root, 'scans', d) 
                 for d in os.listdir(os.path.join(data_root, 'scans'))
                 if os.path.isdir(os.path.join(data_root, 'scans', d))]
    
    with ThreadPoolExecutor() as executor:
        # 并行检查各场景
        results = list(executor.map(validate_scene, scene_dirs))
    
    for scene_id, checks in results:
        for check_type, status in checks.items():
            if not status['valid']:
                report[check_type].append(scene_id)
    
    with open(output_report, 'w') as f:
        json.dump(dict(report), f, indent=2)
    
    return report

def validate_scene(scene_path):
    """验证单个场景"""
    scene_id = os.path.basename(scene_path)
    checks = {
        'required_files': {
            'valid': not check_required_files(scene_path),
            'details': check_required_files(scene_path)
        },
        'file_sizes': {
            'valid': not validate_file_sizes(scene_path),
            'details': validate_file_sizes(scene_path)
        },
        'annotations': {
            'valid': validate_annotations(scene_path),
            'details': None
        }
    }
    return (scene_id, checks)

6. 增量下载与断点续传

对于下载中断的情况，可以基于验证结果生成增量下载列表：

def generate_redownload_list(report_path, output_file='redownload.txt'):
    """根据验证报告生成需要重新下载的场景列表"""
    with open(report_path) as f:
        report = json.load(f)
    
    affected_scenes = set()
    for scenes in report.values():
        affected_scenes.update(scenes)
    
    with open(output_file, 'w') as f:
        f.write('\n'.join(sorted(affected_scenes)))
    
    print(f"生成重新下载列表，共{len(affected_scenes)}个场景需要补全")

对于Python下载脚本，可以添加断点续传支持：

def download_with_resume(url, filename):
    """支持断点续传的下载函数"""
    if os.path.exists(filename):
        existing_size = os.path.getsize(filename)
        headers = {'Range': f'bytes={existing_size}-'}
    else:
        existing_size = 0
        headers = {}
    
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req) as response:
        with open(filename, 'ab' if existing_size else 'wb') as f:
            while chunk := response.read(8192):
                f.write(chunk)

这套验证流程在实际项目中帮助我发现了多个数据完整性问题，特别是在跨地区团队协作时，数据传输过程中容易出现文件损坏。建议在数据集下载完成后立即执行验证，避免在后续实验过程中才发现数据问题，造成时间浪费。