用Python自动化统计生物识别数据集关键信息的完整指南

面对动辄数千个视频文件的生物识别数据集，手动统计subject数量、训练测试划分、真假样本比例等信息不仅耗时耗力，还容易出错。本文将带你用Python脚本快速解析CASIA、MSU、Replay和OULU四大主流数据集的目录结构，自动生成结构化统计报告。

1. 数据集解析的核心思路

生物识别数据集通常以压缩包形式分发，解压后往往呈现复杂的嵌套目录结构。以CASIA数据集为例，其目录可能如下：

CASIA/
├── train/
│   ├── subject_001/
│   │   ├── real_1.avi
│   │   ├── attack_1.avi
│   │   └── ...
│   └── subject_002/
└── test/
    ├── subject_031/
    └── subject_032/

自动化统计的关键在于 ：

• 递归遍历目录树识别subject文件夹
• 通过文件名模式判断样本类型（real/attack）
• 根据路径中的train/test等关键词确定数据划分
• 汇总统计结果并可视化

2. 基础环境准备与工具链配置

2.1 安装必要依赖

推荐使用conda创建专用环境：

conda create -n dataset_stats python=3.8
conda activate dataset_stats
pip install pandas matplotlib seaborn tqdm pathlib2

2.2 项目目录结构

建议按如下方式组织代码：

dataset_analyzer/
├── configs/
│   ├── casia.yaml
│   └── msu.yaml
├── utils/
│   └── file_utils.py
├── analyzer.py
└── requirements.txt

3. 核心代码实现解析

3.1 通用数据集分析类框架

from pathlib import Path
import pandas as pd
from tqdm import tqdm

class DatasetAnalyzer:
    def __init__(self, dataset_path):
        self.dataset_path = Path(dataset_path)
        self.results = {
            'subjects': set(),
            'splits': {'train': 0, 'test': 0, 'devel': 0},
            'types': {'real': 0, 'attack': 0}
        }
    
    def _is_real_sample(self, filename):
        """子类需实现的抽象方法"""
        raise NotImplementedError
        
    def analyze(self):
        """主分析流程"""
        for split_dir in self.dataset_path.iterdir():
            if split_dir.is_dir() and split_dir.name in self.results['splits']:
                self._process_split(split_dir)
        
        self._generate_report()
    
    def _process_split(self, split_dir):
        for subject_dir in tqdm(list(split_dir.iterdir())):
            if not subject_dir.is_dir():
                continue
                
            self.results['subjects'].add(subject_dir.name)
            
            for video_file in subject_dir.glob('*.avi'):
                if self._is_real_sample(video_file.name):
                    self.results['types']['real'] += 1
                else:
                    self.results['types']['attack'] += 1
                
                self.results['splits'][split_dir.name] += 1

3.2 CASIA数据集专用解析器

class CASIAAnalyzer(DatasetAnalyzer):
    def _is_real_sample(self, filename):
        return filename.startswith(('1_', '2_', 'HR_1'))
    
    def _generate_report(self):
        df = pd.DataFrame({
            'Metric': ['Subjects', 'Total Videos', 'Real Videos', 'Attack Videos'],
            'Train': [
                len([s for s in self.results['subjects'] if s.startswith('train')]),
                self.results['splits']['train'],
                self.results['types']['real'],
                self.results['types']['attack']
            ],
            'Test': [
                len([s for s in self.results['subjects'] if s.startswith('test')]),
                self.results['splits']['test'],
                self.results['types']['real'],
                self.results['types']['attack']
            ]
        })
        return df

4. 高级统计与可视化技巧

4.1 生成交互式统计报告

import plotly.express as px

def generate_interactive_report(analyzer):
    df = analyzer._generate_report()
    
    fig = px.bar(df, x='Metric', y=['Train', 'Test'], 
                 barmode='group', title='Dataset Distribution')
    fig.show()
    
    pie_fig = px.pie(values=[analyzer.results['types']['real'], 
                           analyzer.results['types']['attack']],
                    names=['Real', 'Attack'],
                    title='Real vs Attack Distribution')
    pie_fig.show()

4.2 支持多数据集的工厂模式

def create_analyzer(dataset_name, dataset_path):
    if dataset_name.lower() == 'casia':
        return CASIAAnalyzer(dataset_path)
    elif dataset_name.lower() == 'msu':
        return MSUAnalyzer(dataset_path)
    elif dataset_name.lower() == 'replay':
        return ReplayAnalyzer(dataset_path)
    elif dataset_name.lower() == 'oulu':
        return OULUAnalyzer(dataset_path)
    else:
        raise ValueError(f"Unsupported dataset: {dataset_name}")

5. 实战：处理非连续subject编号问题

MSU数据集存在subject编号不连续的情况，需要特殊处理：

class MSUAnalyzer(DatasetAnalyzer):
    def __init__(self, dataset_path):
        super().__init__(dataset_path)
        self.valid_subjects = {
            'test': [1, 13, 14, 23, 24, 26, 28, 29, 30, 32, 33, 35, 
                    36, 37, 39, 42, 48, 49, 50, 51],
            'train': [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 15, 16, 
                     17, 18, 19, 20, 21, 22, 25, 27, 31, 34, 38, 
                     40, 41, 43, 44, 45, 46, 47]
        }
    
    def _process_split(self, split_dir):
        for subject_dir in tqdm(list(split_dir.iterdir())):
            try:
                subject_id = int(subject_dir.name)
                if subject_id in self.valid_subjects[split_dir.name]:
                    self.results['subjects'].add(subject_dir.name)
                    
                    for video_file in subject_dir.glob('*.avi'):
                        if self._is_real_sample(video_file.name):
                            self.results['types']['real'] += 1
                        else:
                            self.results['types']['attack'] += 1
                            
                        self.results['splits'][split_dir.name] += 1
            except ValueError:
                continue

6. 性能优化与批量处理

对于超大规模数据集（如OULU的4950个视频），建议使用多进程处理：

from multiprocessing import Pool

def process_subject(subject_dir):
    # 实现单个subject的处理逻辑
    pass

class ParallelAnalyzer(DatasetAnalyzer):
    def _process_split(self, split_dir):
        with Pool(processes=4) as pool:
            results = list(tqdm(
                pool.imap(process_subject, split_dir.iterdir()),
                total=len(list(split_dir.iterdir()))
            ))
        
        # 汇总结果
        for res in results:
            self.results['subjects'].update(res['subjects'])
            for k in self.results['types']:
                self.results['types'][k] += res['types'][k]
            for k in self.results['splits']:
                self.results['splits'][k] += res['splits'][k]

7. 生成Markdown格式报告

最后将统计结果输出为可读性强的报告：

def generate_markdown_report(analyzer, output_path):
    with open(output_path, 'w') as f:
        f.write(f"# Dataset Analysis Report\n\n")
        f.write(f"## Basic Statistics\n")
        f.write(f"- Total Subjects: {len(analyzer.results['subjects'])}\n")
        f.write(f"- Total Videos: {sum(analyzer.results['types'].values())}\n")
        
        f.write("\n## Split Distribution\n")
        for split, count in analyzer.results['splits'].items():
            f.write(f"- {split.capitalize()}: {count} videos\n")
            
        f.write("\n## Real vs Attack Distribution\n")
        f.write(f"- Real: {analyzer.results['types']['real']} "
               f"({analyzer.results['types']['real']/sum(analyzer.results['types'].values()):.1%})\n")
        f.write(f"- Attack: {analyzer.results['types']['attack']} "
               f"({analyzer.results['types']['attack']/sum(analyzer.results['types'].values()):.1%})\n")

在实际项目中，这套脚本帮助我将原本需要数小时的手动统计工作缩短到几分钟内完成，同时保证了结果的准确性。特别是在处理Replay数据集复杂的目录结构时，自动化脚本的优势尤为明显。