结肠息肉分割数据集 Kvasir-SEG 与 CVC-ClinicDB 实战：3步完成数据预处理与加载

结肠息肉分割数据集 Kvasir-SEG 与 CVC-ClinicDB 实战：3步完成数据预处理与加载

医学图像分析领域近年来发展迅猛，其中结肠息肉分割作为早期结直肠癌筛查的关键技术，正受到越来越多研究者的关注。对于刚进入该领域的研究者而言，如何快速获取和处理标准数据集往往是第一个需要跨越的障碍。本文将聚焦两个广泛使用的息肉分割数据集——Kvasir-SEG和CVC-ClinicDB，通过三个核心步骤演示完整的数据预处理流程，并提供可直接运行的PyTorch代码实现。

1. 数据集概览与获取

在开始任何息肉分割实验前，充分理解所用数据集的特性和获取方式是至关重要的。Kvasir-SEG和CVC-ClinicDB是目前息肉分割领域最具代表性的两个公开数据集，它们为算法开发提供了标准化的评估基准。

Kvasir-SEG数据集由挪威奥斯陆大学医院内窥镜专家采集并标注，包含1000张从胃肠道视频中提取的息肉图像。每张图像都配有精确的像素级分割标注，由专业医师手工绘制。数据集的主要特点包括：

• 图像分辨率从332x487到1920x1072不等
• 包含各种大小、形态和纹理的息肉样本
• 标注涵盖息肉、器械和气泡等多种对象类别

CVC-ClinicDB数据集则来自医学图像计算机与计算机辅助干预国际会议，包含612张结肠镜检查图像，特点为：

• 统一分辨率384x288
• 主要关注小型扁平息肉
• 标注仅包含息肉区域

获取这两个数据集的推荐方式如下：

# 数据集下载链接（需遵守数据使用协议） KVASIR_SEG_URL = "https://datasets.simula.no/kvasir-seg/" CVC_CLINICDB_URL = "https://polyp.grand-challenge.org/CVCClinicDB/"

提示：下载前请仔细阅读数据集的使用条款和引用要求，部分数据集需要注册并签署数据使用协议。

下表对比了两个数据集的关键参数：

2. 数据预处理流程

获得原始数据后，需要进行系统化的预处理才能用于模型训练。我们将这一过程分解为三个关键步骤：统一尺寸、数据增强和数据集划分。

2.1 图像尺寸标准化

不同分辨率的输入会给模型训练带来困难，因此需要将所有图像调整到统一尺寸。基于常见论文实践，我们选择352x352作为标准尺寸，这一大小在计算效率和细节保留间取得了良好平衡。

import cv2 import numpy as np def resize_image_mask(image, mask, target_size=(352, 352)): """ 统一调整图像和对应掩码尺寸 参数: image: 原始图像 (H,W,C) mask: 原始掩码 (H,W) target_size: 目标尺寸 (宽,高) 返回: resized_image: 调整后的图像 resized_mask: 调整后的掩码 """ # 使用双三次插值调整图像 resized_image = cv2.resize(image, target_size, interpolation=cv2.INTER_CUBIC) # 使用最近邻插值调整掩码，避免引入虚假边缘 resized_mask = cv2.resize(mask, target_size, interpolation=cv2.INTER_NEAREST) # 将掩码二值化 resized_mask = (resized_mask > 0).astype(np.uint8) return resized_image, resized_mask

2.2 数据增强策略

医学图像数据通常有限，适当的数据增强可以显著提高模型泛化能力。我们设计了一套针对息肉图像的增强方案：

import albumentations as A def get_augmentations(): """返回训练集的数据增强管道""" return A.Compose([ A.HorizontalFlip(p=0.5), A.VerticalFlip(p=0.5), A.Rotate(limit=45, p=0.7), A.GaussianBlur(blur_limit=(3,7), p=0.2), A.RandomBrightnessContrast( brightness_limit=(-0.1, 0.1), contrast_limit=(-0.1, 0.1), p=0.5 ), A.ElasticTransform( alpha=50, sigma=7, alpha_affine=10, p=0.3 ) ], additional_targets={'mask': 'mask'})

注意：增强只应用于训练集，验证集和测试集应保持原始数据以评估真实性能。

2.3 数据集划分与加载

按照标准实践，我们将数据划分为训练集、验证集和测试集。对于Kvasir-SEG，使用900张训练、100张测试；CVC-ClinicDB则用550张训练、62张测试。

from torch.utils.data import Dataset, DataLoader import os import glob class PolypDataset(Dataset): def __init__(self, image_dir, mask_dir, transform=None): self.image_paths = sorted(glob.glob(os.path.join(image_dir, "*.jpg"))) self.mask_paths = sorted(glob.glob(os.path.join(mask_dir, "*.png"))) self.transform = transform def __len__(self): return len(self.image_paths) def __getitem__(self, idx): image = cv2.imread(self.image_paths[idx]) image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) mask = cv2.imread(self.mask_paths[idx], cv2.IMREAD_GRAYSCALE) # 统一尺寸 image, mask = resize_image_mask(image, mask) if self.transform: augmented = self.transform(image=image, mask=mask) image = augmented['image'] mask = augmented['mask'] # 归一化并转换维度 image = image.astype(np.float32) / 255.0 image = np.transpose(image, (2, 0, 1)) # HWC to CHW mask = mask.astype(np.float32) mask = np.expand_dims(mask, axis=0) # 添加通道维度 return image, mask

3. PyTorch数据加载器实现

完成预处理后，我们需要构建高效的数据加载管道。PyTorch的DataLoader提供了多线程加载和批处理功能，能显著加速训练过程。

3.1 创建DataLoader

def create_dataloaders(config): """创建训练、验证和测试数据加载器""" # 初始化数据集 train_dataset = PolypDataset( image_dir=config['train_image_dir'], mask_dir=config['train_mask_dir'], transform=get_augmentations() ) val_dataset = PolypDataset( image_dir=config['val_image_dir'], mask_dir=config['val_mask_dir'], transform=None ) test_dataset = PolypDataset( image_dir=config['test_image_dir'], mask_dir=config['test_mask_dir'], transform=None ) # 创建数据加载器 train_loader = DataLoader( train_dataset, batch_size=config['batch_size'], shuffle=True, num_workers=config['num_workers'], pin_memory=True ) val_loader = DataLoader( val_dataset, batch_size=config['batch_size'], shuffle=False, num_workers=config['num_workers'], pin_memory=True ) test_loader = DataLoader( test_dataset, batch_size=config['batch_size'], shuffle=False, num_workers=config['num_workers'], pin_memory=True ) return train_loader, val_loader, test_loader

3.2 配置参数示例

# 典型配置参数 config = { 'train_image_dir': 'path/to/kvasir_seg/train/images', 'train_mask_dir': 'path/to/kvasir_seg/train/masks', 'val_image_dir': 'path/to/kvasir_seg/val/images', 'val_mask_dir': 'path/to/kvasir_seg/val/masks', 'test_image_dir': 'path/to/kvasir_seg/test/images', 'test_mask_dir': 'path/to/kvasir_seg/test/masks', 'batch_size': 8, 'num_workers': 4, 'image_size': (352, 352) }

3.3 数据可视化检查

在正式训练前，建议检查数据加载是否正确：

import matplotlib.pyplot as plt def visualize_sample(images, masks, num_samples=4): """可视化批处理中的样本""" plt.figure(figsize=(15, 5*num_samples)) for i in range(num_samples): # 显示图像 plt.subplot(num_samples, 2, 2*i+1) img = images[i].permute(1, 2, 0).numpy() plt.imshow(img) plt.title(f"Image {i+1}") plt.axis('off') # 显示掩码 plt.subplot(num_samples, 2, 2*i+2) mask = masks[i].squeeze().numpy() plt.imshow(mask, cmap='gray') plt.title(f"Mask {i+1}") plt.axis('off') plt.tight_layout() plt.show() # 使用示例 train_loader, _, _ = create_dataloaders(config) images, masks = next(iter(train_loader)) visualize_sample(images, masks)

4. 高级技巧与常见问题

掌握了基本流程后，下面介绍一些提升数据预处理效果的实用技巧。

4.1 类别不平衡处理

息肉分割任务中，前景（息肉）像素通常远少于背景像素。为解决这种不平衡，可采用以下策略：

• 加权损失函数：为前景像素分配更高权重
• 过采样：复制包含大息肉样本
• 难例挖掘：聚焦难以分割的区域

def calculate_class_weights(mask_dir): """计算类别权重以处理不平衡""" mask_paths = glob.glob(os.path.join(mask_dir, "*.png")) pixel_counts = {'background': 0, 'polyp': 0} for path in mask_paths: mask = cv2.imread(path, cv2.IMREAD_GRAYSCALE) mask = (mask > 0).astype(np.uint8) pixel_counts['background'] += np.sum(mask == 0) pixel_counts['polyp'] += np.sum(mask == 1) total = pixel_counts['background'] + pixel_counts['polyp'] weight_background = total / (2 * pixel_counts['background']) weight_polyp = total / (2 * pixel_counts['polyp']) return torch.tensor([weight_background, weight_polyp], dtype=torch.float32)

4.2 多数据集联合训练

结合多个数据集可以提升模型泛化能力。处理要点包括：

1. 统一不同数据集的标注标准
2. 平衡各数据集的样本比例
3. 处理分辨率差异

class CombinedDataset(Dataset): """合并多个息肉数据集""" def __init__(self, datasets): self.datasets = datasets self.lengths = [len(d) for d in datasets] self.total_length = sum(self.lengths) def __len__(self): return self.total_length def __getitem__(self, idx): dataset_idx = 0 while idx >= self.lengths[dataset_idx]: idx -= self.lengths[dataset_idx] dataset_idx += 1 return self.datasets[dataset_idx][idx]

4.3 性能优化技巧

• 预加载数据：对于小数据集，可全部加载到内存
• 混合精度：减少内存占用，加速训练
• 缓存机制：存储预处理结果避免重复计算

class CachedDataset(Dataset): """带缓存的数据集，加速后续epoch""" def __init__(self, base_dataset, cache_size=1000): self.base_dataset = base_dataset self.cache = [None] * len(base_dataset) self.cache_size = cache_size def __len__(self): return len(self.base_dataset) def __getitem__(self, idx): if self.cache[idx] is None: self.cache[idx] = self.base_dataset[idx] # 简单缓存淘汰策略 if len([x for x in self.cache if x is not None]) > self.cache_size: oldest = next(i for i, x in enumerate(self.cache) if x is not None) self.cache[oldest] = None return self.cache[idx]