设计理念与动机

引言：经典标准面临的现代挑战

DICOM（医学数字成像与通信标准）是医学影像领域的通用语言，统一了图像格式，支撑了跨厂商的互操作。然而，这一诞生于 20 世纪 80 年代的标准，其核心设计未预见到由 AI、大数据与高并发计算驱动的现代工作流。

在构建集成 AI 算法、三维可视化与海量数据管理的新一代影像工作站时，我们发现 DICOM 的若干设计已成为系统性能的结构性瓶颈。本文聚焦这些挑战，并介绍 DiCube——在保持对现有生态 100% 往返兼容的前提下，提供系统性重构的一种方案。

DICOM在现代工作流中的四大核心挑战

1. 文件碎片化：并发 I/O 的瓶颈

一个 CT 或 MR 序列通常由数百至数千个独立的 .dcm 文件构成，这是 DICOM 的核心性能制约。在现代工作站中，通信、三维渲染、数据库归档、AI 分析等模块需要并发访问同一序列。大量小文件在并发场景下触发频繁随机 I/O，形成显著瓶颈。尽管 SSD 能缓解，但对每天 TB 级增量的数据中心而言，成本更低的机械硬盘仍是长期存储主力；一旦数据转冷，访问延迟会快速升高。

2. 元数据冗余：存储与带宽的无效消耗

序列内绝大多数元数据（患者信息、检查参数、设备型号等）在切片间不变，但 DICOM 要求在每个 .dcm 文件重复保存。一个含 500 张切片的序列，等于把同一份信息复制 500 次，既占用磁盘，也浪费传输带宽。更糟的是，共享字段一旦在不同文件间不一致，会给后续处理引入数据质量风险。

3. 内置约束缺失：数据质量不确定

在实际应用中，DICOM 数据常见质量问题：序列内文件缺失、关键标签为空或错误、InstanceNumber 不连续或重复、像素间距不一致等。厂商往往需要自建复杂质控模块识别与兜底。由于不同 AI 算法对数据质量容忍度各异，想设计普适规则很难，进而增加集成难度与脆弱性。

4. 顺序解析：低效的元数据访问

DICOM 二进制缺乏全局索引，解析器需自文件头顺序扫描才能定位目标标签。即便只为获取 ImagePositionPatient、InstanceNumber 等少量字段，也要遍历解析所有文件的完整元数据块。主流库（如 PyDICOM）常默认完整加载元数据，这在批处理（如 AI 训练）或快速预览场景下效率低下。

DiCube 的系统性方案：重塑影像数据结构

DiCube 不是对 DICOM 的微调，而是面向现代工作流的存储与访问重构。核心设计：

1. 统一文件容器：将序列整合为单一文件，根除碎片化。
2. 智能元数据：分离共享/非共享元数据，建立高效索引。
3. 现代编解码：采用 HTJ2K 等技术，优化压缩率与编解码速度。
4. 100% 往返兼容：与标准 DICOM 生态无缝衔接。

从碎片到整体：单文件存储

DiCube 将整个 DICOM 序列合并为一个 .dcbs（DiCube Binary Sequence）文件，将对文件系统的随机读写转为高效顺序读写，从根本上缓解 I/O 瓶颈。

import dicube
import time
import os

# 示例DICOM目录和DiCube文件路径
dicom_dir = 'dicube-testdata/dicom/sample_200'
dcbs_file = 'dicube-testdata/test_sequence.dcbs'

# 转换DICOM序列为DiCube格式
dcb_image = dicube.load_from_dicom_folder(dicom_dir)
dicube.save(dcb_image, dcbs_file)

从冗余到索引：智能元数据管理

DiCube 将元数据分为两类：

• 共享元数据：如患者 ID、研究日期等，仅存一份；
• 非共享元数据：如 ImagePositionPatient 等逐切片变化的字段，以紧凑数组存储并建立索引。

该设计显著压缩元数据体积，并支持特定字段的毫秒级随机访问。

from dicube.dicom import CommonTags

# 仅加载元数据，速度极快
meta = dicube.load_meta(dcbs_file)

# O(1) 复杂度访问共享字段
patient_name = meta.get_shared_value(CommonTags.PatientName)
print(f"患者姓名: {patient_name}")

# 高效批量获取非共享字段
instance_numbers = meta.get_values(CommonTags.InstanceNumber)
print(f"一次性获取{len(instance_numbers)}个InstanceNumber, 前5个: {instance_numbers[:5]}")

高并发性能压测：贴近真实负载

为精确衡量DiCube在真实并发环境下的性能优势，我们设计了以下测试方案：

• 并发规模：10个进程并发执行。
• 工作负载：每个进程随机处理5个不同的影像序列。
• 测试数据：从一个包含1000个真实CT序列的大数据集中随机抽取50个序列，避免文件系统缓存干扰。
• 测试场景：
  1. 仅元数据读取：模拟PACS浏览器构建序列列表的场景。
  2. 元数据+像素读取：模拟三维重建或AI算法进行完整数据加载的场景。

测试环境配置（请按需修改路径）

import multiprocessing as mp
import random
import time
import os
import pydicom
import dicube
import numpy as np

# --- 配置您的数据路径 ---
dicom_base_dir = '/data/manifest-NLST_allCT/sample_1000'
dicube_base_dir = '/data/manifest-NLST_allCT/sample_1000_dcbs'
# -------------------------

# --- 准备测试数据 ---
num_processes = 10
series_per_process = 5
total_series_needed = num_processes * series_per_process

all_dicom_dirs = [d for d in os.listdir(dicom_base_dir) if os.path.isdir(os.path.join(dicom_base_dir, d))]
random.seed(time.time())
selected_series_names = random.sample(all_dicom_dirs, total_series_needed)
print('selected_series_names',selected_series_names)
selected_dicom_paths = [os.path.join(dicom_base_dir, name) for name in selected_series_names]
selected_dicube_paths = [os.path.join(dicube_base_dir, name + '.dcbs') for name in selected_series_names]

1. 存储空间占用对比

在性能测试前，我们首先计算这50个被选中序列分别以DICOM和DiCube格式存储时所占用的磁盘空间。

def get_dir_size(path):
    total = 0
    with os.scandir(path) as it:
        for entry in it:
            if entry.is_file():
                total += entry.stat().st_size
    return total

dicom_total_size = sum(get_dir_size(p) for p in selected_dicom_paths)
dicube_total_size = sum(os.path.getsize(p) for p in selected_dicube_paths)

print("--- 50个随机序列存储空间对比 ---")
print(f"DICOM 格式总大小: {dicom_total_size / (1024**2):.2f} MB")
print(f"DiCube 格式总大小: {dicube_total_size / (1024**2):.2f} MB")
print(f"空间节省率: {(1 - dicube_total_size / dicom_total_size) * 100:.1f}%")

2. 并发性能测试执行

我们定义了针对四种不同场景的读取函数，并使用多进程池来模拟并发访问。

# --- 读取函数定义 (Low-level, for a single item) ---
def read_dicom_series_meta_only(series_path):
    files = [os.path.join(series_path, f) for f in os.listdir(series_path) if f.endswith('.dcm')]
    for filepath in files:
        pydicom.dcmread(filepath, stop_before_pixels=True)

def read_dicom_series_full(series_path):
    dcm_series = [pydicom.dcmread(os.path.join(series_path, f)) for f in os.listdir(series_path) if f.endswith('.dcm')]
    # 模拟访问像素数据
    pixels = [ds.pixel_array for ds in dcm_series]
    return (len(pixels), pixels[0].shape)

def read_dicube_meta_only(dcbs_path):
    dicube.load_meta(dcbs_path)

def read_dicube_full(dcbs_path):
    dcb_image = dicube.load(dcbs_path)
    # 模拟访问像素数据
    pixels = dcb_image.raw_image
    return pixels.shape

# --- NEW: Top-level worker functions that are picklable ---
# These functions take a list of paths and process them.
def dicom_meta_worker(paths):
    for path in paths:
        read_dicom_series_meta_only(path)

def dicube_meta_worker(paths):
    for path in paths:
        read_dicube_meta_only(path)

def dicom_full_worker(paths):
    for path in paths:
        read_dicom_series_full(path)

def dicube_full_worker(paths):
    for path in paths:
        read_dicube_full(path)
        
# --- 测试执行框架 (Unchanged) ---
def run_performance_test(paths, worker_function, num_processes, series_per_process):
    tasks = [paths[i*series_per_process:(i+1)*series_per_process] for i in range(num_processes)]
    pool = mp.Pool(processes=num_processes)
    
    start_time = time.time()
    # Use the worker function directly on the list of task lists
    pool.map(worker_function, tasks)
    end_time = time.time()
    
    pool.close()
    pool.join()
    return end_time - start_time
        
# --- 执行所有测试 (UPDATED to use new worker functions) ---
print("\n--- 正在执行并发性能测试 ---")
# 元数据测试
dicom_meta_time = run_performance_test(selected_dicom_paths, dicom_meta_worker, num_processes, series_per_process)
dicube_meta_time = run_performance_test(selected_dicube_paths, dicube_meta_worker, num_processes, series_per_process)

# 完整数据测试
dicom_full_time = run_performance_test(selected_dicom_paths, dicom_full_worker, num_processes, series_per_process)
dicube_full_time = run_performance_test(selected_dicube_paths, dicube_full_worker, num_processes, series_per_process)

3. 测试结果与分析

print("\n--- 并发性能测试结果 (10进程 x 5序列) ---")

# 元数据结果
print("\n场景1: 仅读取元数据")
print(f"DICOM  总耗时: {dicom_meta_time:.2f} 秒")
print(f"DiCube 总耗时: {dicube_meta_time:.2f} 秒")
print(f"性能提升: {dicom_meta_time / dicube_meta_time:.1f} 倍")

# 完整数据结果
print("\n场景2: 读取元数据 + 像素数据")
print(f"DICOM  总耗时: {dicom_full_time:.2f} 秒")
print(f"DiCube 总耗时: {dicube_full_time:.2f} 秒")
print(f"性能提升: {dicom_full_time / dicube_full_time:.1f} 倍")

测试结果清晰地表明，DiCube在两种并发场景下均展现出压倒性的性能优势。尤其是在仅读取元数据的场景中，由于避免了对大量小文件的随机访问和重复解析，性能提升最为显著。在完整数据读取场景中，单文件容器带来的高效I/O同样使DiCube大幅领先。

值得注意的是，DiCube中的像素数据采用了高效压缩算法，即使考虑解压缩开销，其性能依然远超未压缩的DICOM文件。这种性能优势具有良好的可扩展性——随着并发度的提升，DiCube的领先幅度会进一步扩大。此外，本次测试环境为配备SSD缓存的混合硬盘，已经具备相当不错的读取性能。在纯机械硬盘或云对象存储等I/O受限的环境中，DiCube的性能优势将更加明显，差距可能会呈数量级增长。

无缝集成：100% DICOM 往返兼容

DiCube 的关键原则之一是与现有 DICOM 生态完全兼容，通过 100% 无损的往返转换保证：

1. 无损转换：从 DICOM 到 DiCube，再回到 DICOM，像素与元数据无信息损失；
2. 元数据完整：保留所有标准及私有 DICOM 标签；
3. 工作流兼容：可随时将 .dcbs 导出为标准 DICOM 文件集，用于 PACS 归档或与存量系统交互。

以下代码验证了多个关键字段在转换前后的一致性。

import shutil

# 使用之前加载的 dcb_image 对象
roundtrip_dicom_dir = 'dicube-testdata/roundtrip_dicom'
dicube.save_to_dicom_folder(dcb_image, roundtrip_dicom_dir)

# 读取原始DICOM文件和转换后的DICOM文件进行对比
original_dcm_path = os.path.join(dicom_dir, os.listdir(dicom_dir)[0])
original_dcm = pydicom.dcmread(original_dcm_path)

roundtrip_dcm_path = os.path.join(roundtrip_dicom_dir, 'slice_0000.dcm')
roundtrip_dcm = pydicom.dcmread(roundtrip_dcm_path)

print("\n--- 往返转换验证 ---")
fields_to_check = [
    'PatientName', 'StudyInstanceUID', 'SeriesDescription', 
    'ImageOrientationPatient', 'PixelSpacing'
]

for tag in fields_to_check:
    original_value = original_dcm.get(tag, 'N/A')
    roundtrip_value = roundtrip_dcm.get(tag, 'N/A')
    print(f"字段: {tag}")
    print(f"  - 原始值: {original_value}")
    print(f"  - 转换后: {roundtrip_value}")
    # 使用 numpy.allclose 进行浮点数组比较
    if isinstance(original_value, (pydicom.multival.MultiValue, list)):
        are_equal = np.allclose(np.array(original_value, dtype=float), np.array(roundtrip_value, dtype=float))
    else:
        are_equal = (original_value == roundtrip_value)
    print(f"  - 是否一致: {are_equal}")
    print("-" * 20)

print(f"元数据字段总数对比: {len(original_dcm)} (原始) vs {len(roundtrip_dcm)} (转换后)")

# 清理测试文件
os.remove(dcbs_file)
shutil.rmtree(roundtrip_dicom_dir)

总结：从性能瓶颈到技术赋能

DiCube通过系统性的结构优化，为解决DICOM在现代工作流中的核心挑战提供了有效方案。

问题领域	DICOM的局限性	DiCube的解决方案	性能提升
文件管理	大量碎片文件，并发 I/O 瓶颈	单文件容器	并发访问提升 3–10 倍
元数据处理	冗余存储，顺序解析	去重与索引化查询	元数据访问提升 10–50 倍
存储效率	缺乏高效标准压缩	集成 HTJ2K 编解码	空间节省 50–70%
工作流集成	解析转换逻辑复杂	现代化 API 与数据结构	集成效率显著提升

DiCube的核心价值：

1. 即时性能收益：无需改动上层业务，即可获得显著 I/O 与处理性能提升；
2. 赋能未来应用：为 AI 训练、实时分析、大规模并发等场景打好数据底座；
3. 零风险迁移：完整往返兼容，便于与现有 DICOM 生态无缝集成与回退。

通过在工作流中引入DiCube作为高性能的中间格式，开发团队可以更专注于业务逻辑创新，而不是耗费精力去规避底层数据格式的性能陷阱。这不仅能提升当前系统的用户体验，更能为未来的技术演进提供强大的支持。