flowio-flow-cytometry

FlowIO — 流式细胞术文件处理器

概述

FlowIO 是一个轻量级 Python 库，用于读取和写入流式细胞术标准（FCS）文件。它可以解析 FCS 元数据，将事件数据提取为 NumPy 数组，并创建新的 FCS 文件。支持 FCS 2.0、3.0 和 3.1 版本。依赖项极少，非常适合用于数据管道和高级分析前的预处理。

适用场景

• 解析 FCS 文件，将事件数据提取为 NumPy 数组
• 从 FCS 文件读取通道元数据（名称、范围、类型）
• 将流式细胞术数据转换为 pandas DataFrame 或 CSV
• 从 NumPy 数组或处理后的数据创建新的 FCS 文件
• 处理包含多个数据集的 FCS 文件（分离合并的数据集）
• 批量处理目录中的 FCS 文件
• 在下游分析前预处理流式细胞术数据
• 补偿、门控和 FlowJo 工作区支持，请改用 FlowKit
• 高级细胞术可视化（密度图、门控图），请使用 matplotlib 或 plotly

前置条件

pip install flowio numpy pandas

需要 Python 3.9+。没有编译依赖项，可在任何平台上安装。

快速入门

from flowio import FlowData

flow = FlowData("experiment.fcs")
print(f"Events: {flow.event_count}, Channels: {flow.channel_count}")
print(f"Channels: {flow.pnn_labels}")

events = flow.as_array()  # Shape: (n_events, n_channels)
print(f"Data shape: {events.shape}")

核心 API

1. 读取 FCS 文件

FlowData 类是读取 FCS 文件的主要接口。

from flowio import FlowData

# Standard reading
flow = FlowData("sample.fcs")
print(f"Version: {flow.version}")          # '3.0', '3.1', etc.
print(f"Events: {flow.event_count}")
print(f"Channels: {flow.channel_count}")

# Event data
events = flow.as_array()                   # Preprocessed (gain, log scaling)
raw = flow.as_array(preprocess=False)      # Raw values
print(f"Shape: {events.shape}")            # (n_events, n_channels)

# Memory-efficient: metadata only (skip DATA segment)
flow_meta = FlowData("sample.fcs", only_text=True)
print(f"Instrument: {flow_meta.text.get('$CYT', 'Unknown')}")

# Handle problematic files
flow = FlowData("bad.fcs", ignore_offset_discrepancy=True)
flow = FlowData("bad.fcs", use_header_offsets=True)

# Exclude null channels
flow = FlowData("sample.fcs", null_channel_list=["Time", "Null"])

2. 通道元数据

从 FCS 文件中提取通道名称、类型和范围。

flow = FlowData("sample.fcs")

# Channel names
pnn = flow.pnn_labels   # Short names: ['FSC-A', 'SSC-A', 'FL1-A', ...]
pns = flow.pns_labels   # Descriptive: ['Forward Scatter', 'Side Scatter', 'FITC', ...]
pnr = flow.pnr_values   # Range/max values per channel

# Channel type indices
scatter_idx = flow.scatter_indices   # [0, 1] — FSC, SSC
fluoro_idx = flow.fluoro_indices     # [2, 3, 4] — fluorescence channels
time_idx = flow.time_index           # Time channel index (or None)

# Access by type
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluoro_data = events[:, fluoro_idx]

# Full metadata (TEXT segment dictionary)
text = flow.text
print(f"Date: {text.get('$DATE', 'N/A')}")
print(f"Instrument: {text.get('$CYT', 'N/A')}")

3. 创建 FCS 文件

从 NumPy 数组生成新的 FCS 文件。

import numpy as np
from flowio import create_fcs

# Basic creation
events = np.random.rand(10000, 5) * 1000
channels = ["FSC-A", "SSC-A", "FL1-A", "FL2-A", "Time"]
create_fcs("output.fcs", events, channels)

# With descriptive names and metadata
create_fcs(
    "output.fcs",
    events,
    channels,
    opt_channel_names=["Forward Scatter", "Side Scatter", "FITC", "PE", "Time"],
    metadata={"$SRC": "Python pipeline", "$DATE": "17-FEB-2026", "$CYT": "Synthetic"},
)
# Output: FCS 3.1, single-precision float

4. 多数据集 FCS 文件

处理包含多个数据集的 FCS 文件。

from flowio import FlowData, read_multiple_data_sets, MultipleDataSetsError

# Detect multi-dataset files
try:
    flow = FlowData("sample.fcs")
except MultipleDataSetsError:
    datasets = read_multiple_data_sets("sample.fcs")
    print(f"Found {len(datasets)} datasets")
    for i, ds in enumerate(datasets):
        print(f"Dataset {i}: {ds.event_count} events, {ds.channel_count} channels")
        events = ds.as_array()

# Read specific dataset by offset
first = FlowData("multi.fcs", nextdata_offset=0)
next_offset = int(first.text.get("$NEXTDATA", "0"))
if next_offset > 0:
    second = FlowData("multi.fcs", nextdata_offset=next_offset)

5. 修改并重新导出

读取、修改并保存 FCS 数据。

from flowio import FlowData, create_fcs

# Read original
flow = FlowData("original.fcs")
events = flow.as_array(preprocess=False)  # Use raw for modification

# Filter events (e.g., threshold on FSC)
mask = events[:, 0] > 500
filtered = events[mask]
print(f"Before: {len(events)}, After: {len(filtered)}")

# Save filtered data as new FCS
create_fcs(
    "filtered.fcs",
    filtered,
    flow.pnn_labels,
    opt_channel_names=flow.pns_labels,
    metadata={**flow.text, "$SRC": "Filtered"},
)

# Or write with updated metadata (no event modification)
flow.write_fcs("updated.fcs", metadata={"$SRC": "Updated"})

核心概念

FCS 文件结构

FCS 文件由四个部分组成：

部分	内容	FlowData 属性
HEADER	版本、字节偏移量	flow.header
TEXT	键值元数据（$DATE、$CYT、通道名称）	flow.text
DATA	事件数据（二进制/浮点数）	flow.events（字节）、flow.as_array()
ANALYSIS	可选的处理结果	flow.analysis

预处理（as_array）

当 preprocess=True（默认值）时，FlowIO 会执行以下操作：

1. 增益缩放：乘以 PnG 增益值
2. 对数变换：如果存在 PnE 指数变换，则应用该变换（value = a × 10^(b × raw)）
3. 时间缩放：将时间通道转换为正确的单位

如果需要原始值进行修改或自定义变换，请使用 preprocess=False。

常见工作流

工作流：批量汇总 FCS 文件

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

fcs_files = list(Path("data/").glob("*.fcs"))
summaries = []
for f in fcs_files:
    try:
        flow = FlowData(str(f), only_text=True)
        summaries.append({
            "file": f.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"Error: {f.name}: {e}")

df = pd.DataFrame(summaries)
print(df)

工作流：将 FCS 转换为包含通道统计信息的 DataFrame

from flowio import FlowData
import pandas as pd
import numpy as np

flow = FlowData("sample.fcs")
df = pd.DataFrame(flow.as_array(), columns=flow.pnn_labels)

# Per-channel statistics
for col in df.columns:
    print(f"{col}: mean={df[col].mean():.1f}, median={df[col].median():.1f}, std={df[col].std():.1f}")

# Export
df.to_csv("output.csv", index=False)
print(f"Exported {len(df)} events, {len(df.columns)} channels")

关键参数

参数	功能	默认值	选项	效果
preprocess	as_array()	True	True/False	应用增益/对数缩放
only_text	FlowData()	False	True/False	跳过 DATA 部分（仅读取元数据）
ignore_offset_discrepancy	FlowData()	False	True/False	容许 HEADER/TEXT 偏移量不一致
use_header_offsets	FlowData()	False	True/False	优先使用 HEADER 偏移量而非 TEXT 偏移量
ignore_offset_error	FlowData()	False	True/False	跳过所有偏移量验证
null_channel_list	FlowData()	None	名称列表	解析时排除指定通道
nextdata_offset	FlowData()	None	字节偏移量	读取多数据集文件中的指定数据集
opt_channel_names	create_fcs()	None	名称列表	描述性通道名称（PnS）
metadata	create_fcs()	None	字典	自定义 TEXT 部分的键值对

最佳实践

1. 使用 only_text=True 扫描元数据：处理大量文件时，跳过 DATA 段解析，可提速 10–100 倍。

2. 使用 preprocess=False 修改数据：筛选或修改事件时，始终使用原始值，然后重新导出。预处理无法撤销。

3. 反模式——直接修改 flow.events：FlowIO 不支持原地修改事件。使用 as_array() 提取数据，修改后再通过 create_fcs() 保存。

4. 重新导出时保留元数据：将 flow.text 作为元数据传给 create_fcs()，以保留原始采集信息。

5. 检查文件是否包含多个数据集：捕获 MultipleDataSetsError 并使用 read_multiple_data_sets()——某些仪器会将多次采集写入同一个文件。

常见用法

用法：仅提取荧光通道

from flowio import FlowData
import numpy as np

flow = FlowData("sample.fcs")
events = flow.as_array()
fluoro = events[:, flow.fluoro_indices]
names = [flow.pnn_labels[i] for i in flow.fluoro_indices]
print(f"Fluorescence channels: {names}, shape: {fluoro.shape}")

用法：文件检查报告

from flowio import FlowData

flow = FlowData("unknown.fcs")
print(f"Version: {flow.version} | Events: {flow.event_count:,} | Channels: {flow.channel_count}")
for i, (pnn, pns) in enumerate(zip(flow.pnn_labels, flow.pns_labels)):
    ctype = "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} | {ctype}")
for key in ["$DATE", "$CYT", "$INST", "$SRC"]:
    print(f"  {key}: {flow.text.get(key, 'N/A')}")

用法：将事件归一化到 [0, 1] 范围

使用场景：为机器学习或跨样本比较准备荧光通道。

from flowio import FlowData
import numpy as np

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

# Normalize each fluorescence channel to [0, 1]
fluoro_idx = flow.fluoro_indices
fluoro = events[:, fluoro_idx]
pnr = np.array(flow.pnr_values)[fluoro_idx]  # Per-channel max range
normalized = fluoro / pnr
print(f"Normalized shape: {normalized.shape}, range: [{normalized.min():.3f}, {normalized.max():.3f}]")

故障排除

问题	原因	解决方案
DataOffsetDiscrepancyError	HEADER/TEXT 偏移不匹配	使用 ignore_offset_discrepancy=True
MultipleDataSetsError	文件包含多个数据集	改用 read_multiple_data_sets()
FCSParsingError	FCS 文件损坏或不符合标准	尝试 ignore_offset_error=True；确认文件是有效的 FCS 文件
大型文件导致内存不足	一次性加载了数百万个事件	使用 only_text=True 读取元数据；按通道分块处理
通道数量异常	文件中存在空值／填充通道	使用 null_channel_list=["Time", "Null"] 排除这些通道
修改后的数据值错误	修改前应用了预处理	修改事件时，对原始数据使用 preprocess=False
缺少通道名称（PnS 为空）	仪器未设置描述性名称	改用 pnn_labels（短名称）；PnS 在 FCS 规范中为可选项

相关技能

• matplotlib-scientific-plotting — 根据提取的细胞术数据创建散点图、密度图和直方图
• scikit-learn-machine-learning — 对细胞术事件数据进行聚类和降维

参考资料

• FlowIO 文档 — 官方 GitHub 仓库和 API
• FCS 文件格式规范 — ISAC 流式细胞术数据标准
• Spidlen 等人（2010）《流式细胞术数据文件标准，FCS 3.1 版》— Cytometry Part A