flowio

用途

FlowIO 是一个轻量级 Python 库，用于读写流式细胞术标准（FCS）文件，支持 FCS 版本 2.0、3.0 和 3.1。它适用于需要解析或提取 FCS 元数据、将流式细胞术数据转换为 NumPy 数组、导出事件数据到 FCS 格式、分离多数据集 FCS 文件、提取通道信息（散射、荧光、时间）、验证或检查细胞术文件，以及在进行高级分析之前的预处理工作流。

主要能力

读取与解析 FCS 文件：使用 FlowData 类读取文件，可访问版本、事件数、通道数、通道短名称（pnn_labels）、描述性染色名称（pns_labels）等属性。
提取事件数据：通过 as_array() 方法返回形状为（事件数, 通道数）的 NumPy 数组，可选择是否应用预处理（增益缩放、对数变换、时间缩放）。
内存高效读取：使用 only_text=True 仅解析 TEXT 段，跳过 DATA 和 ANALYSIS，适合仅需元数据的场景。
处理问题文件：支持 ignore_offset_discrepancy、use_header_offsets、ignore_offset_error 等参数应对偏移差异或错误。
排除空通道：通过 null_channel_list 参数在解析时排除特定通道。
通道分类：自动识别散射通道（scatter_indices）、荧光通道（fluoro_indices）和时间通道（time_index）。
ANALYSIS 段访问：如果存在，可获取处理后的结果关键字。
创建 FCS 文件：使用 create_fcs() 从 NumPy 数组生成 FCS 3.1 文件，支持自定义通道名称、描述性名称和 TEXT 段元数据。
导出修改后的数据：通过 write_fcs() 方法或提取数据后重建的方式修改并保存 FCS 文件。
处理多数据集文件：检测 MultipleDataSetsError，使用 read_multiple_data_sets() 读取所有数据集，或通过 nextdata_offset 参数读取特定数据集。
数据预处理：当 preprocess=True 时应用标准 FCS 预处理，包括增益缩放、对数变换（公式：value = a * 10^(b * raw_value)）和时间缩放。
错误处理：提供 FCSParsingError、DataOffsetDiscrepancyError、MultipleDataSetsError 等异常类。

使用方式

安装

uv pip install flowio

需要 Python 3.9 或更高版本。

基本读取

from flowio import FlowData
flow_data = FlowData('experiment.fcs')
print(f"FCS Version: {flow_data.version}")
print(f"Events: {flow_data.event_count}")
print(f"Channels: {flow_data.pnn_labels}")
events = flow_data.as_array()  # Shape: (events, channels)

创建 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)

可添加 opt_channel_names 描述性名称和 metadata 字典来自定义 TEXT 段。

处理多数据集文件

from flowio import read_multiple_data_sets
datasets = read_multiple_data_sets('multi_dataset.fcs')
for i, dataset in enumerate(datasets):
    events = dataset.as_array()

转换为 CSV

import pandas as pd
df = pd.DataFrame(flow.as_array(), columns=flow.pnn_labels)
df.attrs['fcs_version'] = flow.version
df.to_csv('output.csv', index=False)

过滤事件并重新导出

from flowio import FlowData, create_fcs
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)
mask = events[:, 0] > 500
filtered_events = events[mask]
create_fcs('filtered.fcs', filtered_events, flow.pnn_labels,
           opt_channel_names=flow.pns_labels, metadata={**flow.text, '$SRC': 'Filtered data'})

适用场景

检查 FCS 文件内容（文件版本、事件数、通道信息、关键元数据）。
批量处理多个 FCS 文件并生成摘要 DataFrame。
将 FCS 事件数据转换为 CSV 格式。
根据条件过滤事件并重新导出为 FCS。
提取特定通道（如荧光通道）并计算统计量。
为 Web 应用后端处理上传的 FCS 文件（依赖极少）。
作为高级流式分析（如补偿、门控）之前的预处理步骤。

限制

FlowIO 本身不提供高级分析功能，如补偿、门控或 FlowJo/GatingML 支持；此类需求建议搭配 FlowKit 库。
不支持直接修改事件数据；若需修改，必须提取数据、修改后再用 create_fcs() 保存。
创建 FCS 文件时固定导出为 FCS 3.1 格式，使用单精度浮点数据。
遇到偏移差异或错误时需手动调整参数处理，可能仍需额外的错误处理逻辑。
大文件可能导致内存不足，建议使用 only_text=True 或分块处理。
通道数量可能与预期不符，需检查空通道并通过 null_channel_list 排除。