⚡

流式IOSkill flowio

流式IO 是一个专为流式细胞术数据设计的Python库，用于解析和处理FCS（流式细胞术标准）文件，支持事件数据提取、元数据读取、文件创建和格式转换，适用于数据分析、预处理和生物医学研究。关键词：流式细胞术、FCS文件、Python、数据分析、数据预处理、生物医药、NumPy、Pandas。

数据分析 0 次安装 0 次浏览 更新于 3/16/2026

名称: flowio 描述: “解析 FCS（流式细胞术标准）文件 v2.0-3.1。将事件提取为 NumPy 数组，读取元数据/通道，转换为 CSV/DataFrame，用于流式细胞术数据预处理。”

FlowIO：流式细胞术标准文件处理器

概述

FlowIO 是一个轻量级的 Python 库，用于读写流式细胞术标准（FCS）文件。解析 FCS 元数据，提取事件数据，并以最少的依赖创建新的 FCS 文件。该库支持 FCS 版本 2.0、3.0 和 3.1，适用于后端服务、数据管道和基本的细胞术文件操作。

何时使用此技能

此技能应在以下情况使用：

  • 需要解析或提取元数据的 FCS 文件
  • 需要将流式细胞术数据转换为 NumPy 数组
  • 需要将事件数据导出为 FCS 格式
  • 需要分离多数据集 FCS 文件
  • 需要提取通道信息（散射、荧光、时间）
  • 需要验证或检查细胞术文件
  • 在高级分析前进行预处理工作流

相关工具： 对于高级流式细胞术分析，包括补偿、门控和 FlowJo/GatingML 支持，推荐使用 FlowKit 库作为 FlowIO 的配套工具。

安装

pip install flowio

要求 Python 3.9 或更高版本。

快速开始

基本文件读取

from flowio import FlowData

# 读取 FCS 文件
flow_data = FlowData('experiment.fcs')

# 访问基本信息
print(f"FCS 版本: {flow_data.version}")
print(f"事件: {flow_data.event_count}")
print(f"通道: {flow_data.pnn_labels}")

# 获取事件数据为 NumPy 数组
events = flow_data.as_array()  # 形状: (事件, 通道)

创建 FCS 文件

import numpy as np
from flowio import create_fcs

# 准备数据
data = np.array([[100, 200, 50], [150, 180, 60]])  # 2 事件, 3 通道
channels = ['FSC-A', 'SSC-A', 'FL1-A']

# 创建 FCS 文件
create_fcs('output.fcs', data, channels)

核心工作流

读取和解析 FCS 文件

FlowData 类提供了读取 FCS 文件的主要接口。

标准读取：

from flowio import FlowData

# 基本读取
flow = FlowData('sample.fcs')

# 访问属性
version = flow.version              # '3.0', '3.1' 等
event_count = flow.event_count      # 事件数量
channel_count = flow.channel_count  # 通道数量
pnn_labels = flow.pnn_labels        # 短通道名称
pns_labels = flow.pns_labels        # 描述性染色名称

# 获取事件数据
events = flow.as_array()            # 预处理（增益、对数缩放已应用）
raw_events = flow.as_array(preprocess=False)  # 原始数据

内存高效的元数据读取：

当仅需要元数据（无需事件数据）时：

# 仅解析 TEXT 段，跳过 DATA 和 ANALYSIS
flow = FlowData('sample.fcs', only_text=True)

# 访问元数据
metadata = flow.text  # TEXT 段关键词字典
print(metadata.get('$DATE'))  # 采集日期
print(metadata.get('$CYT'))   # 仪器名称

处理问题文件：

一些 FCS 文件有偏移差异或错误：

# 忽略 HEADER 和 TEXT 部分之间的偏移差异
flow = FlowData('problematic.fcs', ignore_offset_discrepancy=True)

# 使用 HEADER 偏移而不是 TEXT 偏移
flow = FlowData('problematic.fcs', use_header_offsets=True)

# 完全忽略偏移错误
flow = FlowData('problematic.fcs', ignore_offset_error=True)

排除空通道：

# 在解析期间排除特定通道
flow = FlowData('sample.fcs', null_channel_list=['Time', 'Null'])

提取元数据和通道信息

FCS 文件在 TEXT 段中包含丰富的元数据。

常见元数据关键词：

flow = FlowData('sample.fcs')

# 文件级元数据
text_dict = flow.text
acquisition_date = text_dict.get('$DATE', '未知')
instrument = text_dict.get('$CYT', '未知')
data_type = flow.data_type  # 'I', 'F', 'D', 'A'

# 通道元数据
for i in range(flow.channel_count):
    pnn = flow.pnn_labels[i]      # 短名称（如 'FSC-A'）
    pns = flow.pns_labels[i]      # 描述性名称（如 'Forward Scatter'）
    pnr = flow.pnr_values[i]      # 范围/最大值
    print(f"通道 {i}: {pnn} ({pns}), 范围: {pnr}")

通道类型识别：

FlowIO 自动分类通道：

# 按通道类型获取索引
scatter_idx = flow.scatter_indices    # 对于 FSC, SSC 为 [0, 1]
fluoro_idx = flow.fluoro_indices      # 对于 FL 通道为 [2, 3, 4]
time_idx = flow.time_index            # 时间通道索引（或无）

# 访问特定通道类型
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluorescence_data = events[:, fluoro_idx]

ANALYSIS 段：

如果存在，访问处理结果：

if flow.analysis:
    analysis_keywords = flow.analysis  # ANALYSIS 关键词字典
    print(analysis_keywords)

创建新 FCS 文件

从 NumPy 数组或其他数据源生成 FCS 文件。

基本创建：

import numpy as np
from flowio import create_fcs

# 创建事件数据（行=事件，列=通道）
events = np.random.rand(10000, 5) * 1000

# 定义通道名称
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']

# 创建 FCS 文件
create_fcs('output.fcs', events, channel_names)

带描述性通道名称：

# 添加可选描述性名称（PnS）
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']
descriptive_names = ['Forward Scatter', 'Side Scatter', 'FITC', 'PE', 'Time']

create_fcs('output.fcs',
           events,
           channel_names,
           opt_channel_names=descriptive_names)

带自定义元数据：

# 添加 TEXT 段元数据
metadata = {
    '$SRC': 'Python 脚本',
    '$DATE': '19-OCT-2025',
    '$CYT': '合成仪器',
    '$INST': '实验室 A'
}

create_fcs('output.fcs',
           events,
           channel_names,
           opt_channel_names=descriptive_names,
           metadata=metadata)

注意： FlowIO 以 FCS 3.1 格式导出，使用单精度浮点数据。

导出修改后的数据

修改现有 FCS 文件并重新导出。

方法 1：使用 write_fcs() 方法：

from flowio import FlowData

# 读取原始文件
flow = FlowData('original.fcs')

# 写入更新后的元数据
flow.write_fcs('modified.fcs', metadata={'$SRC': '修改数据'})

方法 2：提取、修改和重建：

用于修改事件数据：

from flowio import FlowData, create_fcs

# 读取和提取数据
flow = FlowData('original.fcs')
events = flow.as_array(preprocess=False)

# 修改事件数据
events[:, 0] = events[:, 0] * 1.5  # 缩放第一通道

# 使用修改后的数据创建新 FCS 文件
create_fcs('modified.fcs',
           events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata=flow.text)

处理多数据集 FCS 文件

一些 FCS 文件在单个文件中包含多个数据集。

检测多数据集文件：

from flowio import FlowData, MultipleDataSetsError

try:
    flow = FlowData('sample.fcs')
except MultipleDataSetsError:
    print("文件包含多个数据集")
    # 使用 read_multiple_data_sets() 代替

读取所有数据集：

from flowio import read_multiple_data_sets

# 从文件读取所有数据集
datasets = read_multiple_data_sets('multi_dataset.fcs')

print(f"找到 {len(datasets)} 个数据集")

# 处理每个数据集
for i, dataset in enumerate(datasets):
    print(f"
数据集 {i}:")
    print(f"  事件: {dataset.event_count}")
    print(f"  通道: {dataset.pnn_labels}")

    # 获取此数据集的事件数据
    events = dataset.as_array()
    print(f"  形状: {events.shape}")
    print(f"  平均值: {events.mean(axis=0)}")

读取特定数据集：

from flowio import FlowData

# 读取第一个数据集（nextdata_offset=0）
first_dataset = FlowData('multi.fcs', nextdata_offset=0)

# 使用第一个的 NEXTDATA 偏移读取第二个数据集
next_offset = int(first_dataset.text['$NEXTDATA'])
if next_offset > 0:
    second_dataset = FlowData('multi.fcs', nextdata_offset=next_offset)

数据预处理

FlowIO 在 preprocess=True 时应用标准 FCS 预处理变换。

预处理步骤：

  1. 增益缩放： 乘以 PnG（增益）关键词值
  2. 对数变换： 如果存在，应用 PnE 指数变换
    • 公式: 值 = a * 10^(b * 原始值) 其中 PnE = “a,b”
  3. 时间缩放： 将时间值转换为适当单位

控制预处理：

# 预处理数据（默认）
preprocessed = flow.as_array(preprocess=True)

# 原始数据（无变换）
raw = flow.as_array(preprocess=False)

错误处理

适当处理常见的 FlowIO 异常。

from flowio import (
    FlowData,
    FCSParsingError,
    DataOffsetDiscrepancyError,
    MultipleDataSetsError
)

try:
    flow = FlowData('sample.fcs')
    events = flow.as_array()

except FCSParsingError as e:
    print(f"解析 FCS 文件失败: {e}")
    # 尝试宽松解析
    flow = FlowData('sample.fcs', ignore_offset_error=True)

except DataOffsetDiscrepancyError as e:
    print(f"检测到偏移差异: {e}")
    # 使用 ignore_offset_discrepancy 参数
    flow = FlowData('sample.fcs', ignore_offset_discrepancy=True)

except MultipleDataSetsError as e:
    print(f"检测到多个数据集: {e}")
    # 使用 read_multiple_data_sets 代替
    from flowio import read_multiple_data_sets
    datasets = read_multiple_data_sets('sample.fcs')

except Exception as e:
    print(f"意外错误: {e}")

常见用例

检查 FCS 文件内容

快速探索 FCS 文件结构：

from flowio import FlowData

flow = FlowData('unknown.fcs')

print("=" * 50)
print(f"文件: {flow.name}")
print(f"版本: {flow.version}")
print(f"大小: {flow.file_size:,} 字节")
print("=" * 50)

print(f"
事件: {flow.event_count:,}")
print(f"通道: {flow.channel_count}")

print("
通道信息:")
for i, (pnn, pns) in enumerate(zip(flow.pnn_labels, flow.pns_labels)):
    ch_type = "scatter" if i in flow.scatter_indices else \
              "fluoro" if i in flow.fluoro_indices else \
              "time" if i == flow.time_index else "other"
    print(f"  [{i}] {pnn:10s} | {pns:30s} | {ch_type}")

print("
关键元数据:")
for key in ['$DATE', '$BTIM', '$ETIM', '$CYT', '$INST', '$SRC']:
    value = flow.text.get(key, 'N/A')
    print(f"  {key:15s}: {value}")

批量处理多个文件

处理目录中的 FCS 文件：

from pathlib import Path
from flowio import FlowData
import pandas as pd

# 查找所有 FCS 文件
fcs_files = list(Path('data/').glob('*.fcs'))

# 提取摘要信息
summaries = []
for fcs_path in fcs_files:
    try:
        flow = FlowData(str(fcs_path), only_text=True)
        summaries.append({
            'filename': fcs_path.name,
            'version': flow.version,
            'events': flow.event_count,
            'channels': flow.channel_count,
            'date': flow.text.get('$DATE', 'N/A')
        })
    except Exception as e:
        print(f"处理 {fcs_path.name} 时出错: {e}")

# 创建摘要 DataFrame
df = pd.DataFrame(summaries)
print(df)

将 FCS 转换为 CSV

将事件数据导出为 CSV 格式：

from flowio import FlowData
import pandas as pd

# 读取 FCS 文件
flow = FlowData('sample.fcs')

# 转换为 DataFrame
df = pd.DataFrame(
    flow.as_array(),
    columns=flow.pnn_labels
)

# 添加元数据作为属性
df.attrs['fcs_version'] = flow.version
df.attrs['instrument'] = flow.text.get('$CYT', '未知')

# 导出到 CSV
df.to_csv('output.csv', index=False)
print(f"导出 {len(df)} 个事件到 CSV")

过滤事件并重新导出

应用过滤器并保存过滤后的数据：

from flowio import FlowData, create_fcs
import numpy as np

# 读取原始文件
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)

# 应用过滤（示例：第一通道阈值）
fsc_idx = 0
threshold = 500
mask = events[:, fsc_idx] > threshold
filtered_events = events[mask]

print(f"原始事件: {len(events)}")
print(f"过滤后事件: {len(filtered_events)}")

# 使用过滤后数据创建新 FCS 文件
create_fcs('filtered.fcs',
           filtered_events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata={**flow.text, '$SRC': '过滤数据'})

提取特定通道

提取和处理特定通道：

from flowio import FlowData
import numpy as np

flow = FlowData('sample.fcs')
events = flow.as_array()

# 仅提取荧光通道
fluoro_indices = flow.fluoro_indices
fluoro_data = events[:, fluoro_indices]
fluoro_names = [flow.pnn_labels[i] for i in fluoro_indices]

print(f"荧光通道: {fluoro_names}")
print(f"形状: {fluoro_data.shape}")

# 计算每通道统计信息
for i, name in enumerate(fluoro_names):
    channel_data = fluoro_data[:, i]
    print(f"
{name}:")
    print(f"  平均值: {channel_data.mean():.2f}")
    print(f"  中位数: {np.median(channel_data):.2f}")
    print(f"  标准差: {channel_data.std():.2f}")

最佳实践

  1. 内存效率： 当不需要事件数据时，使用 only_text=True
  2. 错误处理： 将文件操作包装在 try-except 块中以确保代码健壮
  3. 多数据集检测： 检查 MultipleDataSetsError 并使用适当函数
  4. 预处理控制： 根据分析需求明确设置 preprocess 参数
  5. 偏移问题： 如果解析失败，尝试 ignore_offset_discrepancy=True 参数
  6. 通道验证： 在处理前验证通道数量和名称是否符合预期
  7. 元数据保存： 修改文件时，保留原始 TEXT 段关键词

高级主题

理解 FCS 文件结构

FCS 文件由四个段组成：

  1. HEADER： FCS 版本和其他段的字节偏移
  2. TEXT： 键值对元数据（分隔符分隔）
  3. DATA： 原始事件数据（二进制/浮点/ASCII 格式）
  4. ANALYSIS（可选）： 数据处理结果

通过 FlowData 属性访问这些段：

  • flow.header - HEADER 段
  • flow.text - TEXT 段关键词
  • flow.events - DATA 段（作为字节）
  • flow.analysis - ANALYSIS 段关键词（如果存在）

详细 API 参考

有关完整 API 文档，包括所有参数、方法、异常和 FCS 关键词参考，请查阅详细参考文件：

阅读： references/api_reference.md

参考内容包括：

  • 完整的 FlowData 类文档
  • 所有实用函数（read_multiple_data_sets, create_fcs）
  • 异常类和处理
  • FCS 文件结构详情
  • 常见 TEXT 段关键词
  • 扩展示例工作流

处理复杂 FCS 操作或遇到不常见文件格式时，加载此参考以获取详细指导。

集成说明

NumPy 数组： 所有事件数据以 NumPy ndarrays 返回，形状为（事件, 通道）

Pandas DataFrames： 轻松转换为 DataFrame 进行分析：

import pandas as pd
df = pd.DataFrame(flow.as_array(), columns=flow.pnn_labels)

FlowKit 集成： 对于高级分析（补偿、门控、FlowJo 支持），使用 FlowKit 库，它基于 FlowIO 的解析能力

Web 应用程序： FlowIO 的最小依赖使其适用于处理 FCS 上传的 Web 后端服务

故障排除

问题： “偏移差异错误” 解决方案： 使用 ignore_offset_discrepancy=True 参数

问题： “多数据集错误” 解决方案： 使用 read_multiple_data_sets() 函数代替 FlowData 构造函数

问题： 大文件内存不足 解决方案： 对于仅元数据操作使用 only_text=True，或分块处理事件

问题： 意外通道数量 解决方案： 检查空通道；使用 null_channel_list 参数排除它们

问题： 无法就地修改事件数据 解决方案： FlowIO 不支持直接修改；提取数据，修改，然后使用 create_fcs() 保存

总结

FlowIO 为流式细胞术工作流提供必要的 FCS 文件处理能力。用于解析、元数据提取和文件创建。对于简单文件操作和数据提取，FlowIO 足够。对于复杂分析包括补偿和门控，与 FlowKit 或其他专门工具集成。