FlowIO 是一个轻量级 Python 库,用于读写流式细胞术标准(FCS)文件,支持 FCS 2.0、3.0 和 3.1 版本。其目标是为后端服务、数据管线和基础细胞术文件操作提供低依赖的 FCS 处理能力。
主要能力
读取与解析 FCS 文件:通过 FlowData 类读取 FCS 文件,可访问版本、事件数、通道数、通道短名(pnn_labels)和描述性染色名(pns_labels)等信息。支持将事件数据返回为 NumPy 数组(形状为事件 × 通道),可选择是否应用预处理(增益缩放、对数变换、时间缩放),也可获取原始数据。
元数据与通道信息提取:可读取 TEXT 段中的丰富的键值元数据,如采集日期($DATE)、仪器名称($CYT)、数据类型等。能自动识别通道类型,提供 scatter_indices、fluoro_indices 和 time_index,便于按类型提取数据。若存在 ANALYSIS 段,也可访问其处理结果。
创建新 FCS 文件:提供 create_fcs 函数,可从 NumPy 数组创建 FCS 3.1 单精度浮点格式文件,支持自定义通道名、描述性名称和元数据。
导出与转换:可将事件数据转换为 Pandas DataFrame 并导出为 CSV;可读取现有文件、修改事件数据后重新创建 FCS 文件;也可使用 write_fcs() 方法更新元数据后写出。
多数据集处理:对包含多个数据集的 FCS 文件,捕获 MultipleDataSetsError 后可使用 read_multiple_data_sets() 读取全部数据集,或通过 nextdata_offset 参数读取指定数据集。
内存高效读取:当只需元数据时,可使用 only_text=True 仅解析 TEXT 段,跳过 DATA 和 ANALYSIS,降低内存占用。
问题文件处理:提供 ignore_offset_discrepancy、use_header_offsets、ignore_offset_error 等参数应对偏移不一致或错误;可通过 null_channel_list 排除特定通道。
使用方式
安装:uv pip install flowio,需要 Python 3.9 或更高版本。
基本读取: from flowio import FlowData flow_data = FlowData('experiment.fcs') print(flow_data.version) print(flow_data.event_count) print(flow_data.pnn_labels) events = flow_data.as_array()
创建 FCS 文件: import numpy as np from flowio import create_fcs data = np.array([[100, 200, 50], [150, 180, 60]]) channels = ['FSC-A', 'SSC-A', 'FL1-A'] create_fcs('output.fcs', data, channels)
转换为 CSV: import pandas as pd df = pd.DataFrame(flow.as_array(), columns=flow.pnn_labels) df.to_csv('output.csv', index=False)
异常处理:可捕获 FCSParsingError、DataOffsetDiscrepancyError、MultipleDataSetsError 等,并配合相应参数或函数进行恢复。
适用场景
- 需要解析 FCS 文件或提取元数据的任务。
- 将流式细胞术数据转换为 NumPy 数组。
- 将事件数据导出为 FCS 格式。
- 需要分离多数据集 FCS 文件。
- 提取通道信息(散射、荧光、时间)。
- 细胞术文件验证或检查。
- 高级分析前的预处理工作流。
限制与注意事项
FlowIO 不支持直接原地修改事件数据;需提取数据、修改后使用 create_fcs() 保存。对于需要补偿、门控、FlowJo/GatingML 支持等高级流式细胞术分析,建议搭配 FlowKit 库使用。FlowIO 导出的 FCS 文件为 FCS 3.1 单精度浮点格式。详细 API 参考、FCS 关键字和扩展工作流可查阅 references/api_reference.md。