flowio

FlowIO

用途

将 FlowIO 用作轻量级底层读取器和写入器，以处理流式细胞术标准文件。 本技能中的示例针对 FlowIO 1.4.0，这是截至 2026-07-23 验证过的当前稳定版。

FlowIO 适用于：

• 读取 FCS 2.0、3.0 和 3.1 文件
• 检查 HEADER、TEXT、ANALYSIS 和通道元数据
• 将事件数据作为二维 NumPy 数组提取
• 读取包含多个数据集的旧版文件
• 写入列表模式、单精度 FCS 3.1 文件
• 为 pandas、机器学习或下游流式细胞术工具准备数据

FlowIO 不执行补偿、logicle/双指数变换、设门、聚类或 FlowJo 工作区处理。对于这些任务，请使用 FlowKit 或其他分析包。

安装

创建或激活 Python 环境，然后安装已验证的版本：

uv pip install "flowio==1.4.0"

确认运行时版本：

uv run python -c "import flowio; print(flowio.__version__)"

FlowIO 1.4.0 支持 Python 3.9 至 3.13，并依赖 NumPy。

操作工作流

1. 明确操作类型。 区分元数据清单、事件提取、文件修复、转换和下游生物学分析。
2. 加载事件前先检查。 仅处理元数据时（尤其是处理大型或不熟悉的文件时），使用 only_text=True。
3. 明确选择事件语义。 使用 as_array(preprocess=True) 应用 FCS 元数据中的增益／对数／时间缩放，或使用 preprocess=False 获取 DATA 段中编码的值。记录所选方式。
4. 默认严格解析。 不要自动抑制偏移量错误。仅当已知存在供应商格式缺陷时才放宽检查，并检查所得事件数据。
5. 将元数据视为可能包含敏感信息。 FCS TEXT 值可能包含样本、受试者、操作员和仪器标识符。仅导出任务所需的字段。
6. 通过重新打开文件验证写入结果。 每次导出 FCS 后，检查事件数／通道数、标签、元数据和代表性数值。

关键语义

TEXT 键会进行标准化

FlowData.text 中的键会转换为小写，并移除标准 FCS 关键字开头的 $：

from flowio import FlowData

flow = FlowData("sample.fcs", only_text=True)
acquisition_date = flow.text.get("date")
instrument = flow.text.get("cyt")
next_dataset = int(flow.text.get("nextdata", "0"))

不要查找 "$DATE"、"$CYT" 或其他以美元符号开头的大写键。 TEXT 值仍为字符串。FlowIO 1.4.0 还会从解码后的 TEXT 段中移除所有 $ 字符，包括值中的 $ 字符；如果需要精确保留元数据，请保留原始文件。

事件有两种表示形式

• flow.events 是未经处理的一维扁平事件数组。
• flow.as_array() 返回形状为 (event_count, channel_count) 的 NumPy float64 数组。
• flow.as_array(preprocess=True) 会应用 FCS 增益、对数和时间缩放。它不会应用补偿或 logicle／双指数显示变换。
• flow.as_array(preprocess=False) 会重塑编码后的事件值，但不执行这些缩放步骤。

as_array() 会在内存中创建另一个数组。FlowIO 不提供分块或内存映射的事件访问方式。

通道编号有两种约定

• NumPy 列以及 fluoro_indices、scatter_indices 和 time_index 使用从 0 开始的索引。
• flow.channels 使用从 1 开始的 FCS 参数编号。
• null_channels 包含通过 null_channel_list 提供的 PnN 标签字符串，包括未找到的已提供标签。
• pns_labels 的长度始终与 pnn_labels 相同；缺失的可选 PnS 标签会显示为空字符串。

写入功能有意保持有限

create_fcs() 需要：

• 已打开的二进制文件句柄
• 按事件／通道行优先顺序排列的扁平一维事件数据
• 每个通道对应一个 PnN 名称
• 可选的 PnS 名称，以及通过 metadata_dict 提供的字符串类型元数据

它会写入 FCS 3.1 列表模式（$MODE=L）单精度浮点数（$DATATYPE=F）数据。FlowIO 会生成必要的解释性关键字，且无法通过元数据覆盖这些关键字。

快速开始：读取 FCS 文件

from pathlib import Path

from flowio import FlowData

flow = FlowData(Path("sample.fcs"))
events = flow.as_array(preprocess=True)

print(
    {
        "version": flow.version,
        "events": flow.event_count,
        "channels": flow.channel_count,
        "shape": events.shape,
        "pnn": flow.pnn_labels,
        "pns": flow.pns_labels,
        "date": flow.text.get("date"),
        "instrument": flow.text.get("cyt"),
    }
)

仅读取元数据：

from flowio import FlowData

flow = FlowData("sample.fcs", only_text=True)
print(flow.version, flow.event_count, flow.pnn_labels)

对于仅包含元数据的实例，请勿调用 as_array()，因为其中的事件数据尚未加载。

优先使用路径或 Path，而不是调用方拥有的文件句柄。FlowData 解析后会关闭传入的句柄。在 FlowIO 1.4.0 中，read_multiple_data_sets(handle) 在读取第一个数据集后可能会失败，因为句柄已关闭；对于包含多个数据集的文件，请传入文件系统路径。

快速开始：读取多个数据集

请使用独立的辅助函数，而不要手动解析 $NEXTDATA 偏移量：

from flowio import read_multiple_data_sets

datasets = read_multiple_data_sets("legacy-multi-dataset.fcs")
for index, dataset in enumerate(datasets):
    values = dataset.as_array(preprocess=True)
    print(index, dataset.event_count, dataset.pnn_labels, values.shape)

FCS 3.1 规范已弃用在一个文件中包含多个数据集的做法，但 FlowIO 可以读取采用这种格式的旧文件。

快速开始：创建 FCS 3.1 文件

from pathlib import Path

import numpy as np
from flowio import FlowData, create_fcs

values = np.asarray(
    [[100.0, 200.0, 50.0], [150.0, 180.0, 60.0]],
    dtype=np.float32,
)
pnn_labels = ["FSC-A", "SSC-A", "FITC-A"]
pns_labels = ["Forward scatter", "Side scatter", "CD3"]

output = Path("output.fcs")
with output.open("xb") as handle:
    create_fcs(
        handle,
        values.ravel(order="C"),
        pnn_labels,
        opt_channel_names=pns_labels,
        metadata_dict={
            "date": "23-JUL-2026",
            "cyt": "Example instrument",
            "src": "Validated NumPy array",
        },
    )

roundtrip = FlowData(output)
assert roundtrip.event_count == values.shape[0]
assert roundtrip.pnn_labels == pnn_labels
np.testing.assert_allclose(
    roundtrip.as_array(preprocess=False),
    values,
    rtol=1e-6,
    atol=1e-6,
)

元数据键可以使用混合大小写或带有 $ 的形式，但使用不带 $ 的小写键与 FlowIO 规范化后的表示形式一致，也不易出错。元数据值必须是字符串。

复制或重写现有文件

如果事件数据无需更改，请使用 write_fcs()：

from flowio import FlowData

flow = FlowData("source.fcs")

# Preserve selected source metadata (cyt, date, and spill/spillover when present).
flow.write_fcs("copy.fcs")

# Write only required metadata plus the custom fields supplied here.
flow.write_fcs("deidentified.fcs", metadata={"src": "Deidentified export"})

传入 metadata=None 会保留 FlowIO 选定的默认值。传入任何字典（包括 {}）都会替换这些默认值，而不是与其合并。write_fcs() 始终生成 FCS 3.1 浮点输出；写入前会对非浮点源事件进行预处理。它会以覆盖模式打开目标文件，因此除非确实要替换文件，否则调用前应拒绝已存在的输出路径。对于浮点源数据，它可以保留编码后的事件，但会丢弃 PnG 或 timestep，从而改变后续 as_array(preprocess=True) 的结果。请验证原始数据和预处理数据的往返结果。

如果事件值、事件数量或通道布局发生变化，请改用 create_fcs()。

内置检查器

scripts/inspect_fcs.py 无需网络访问即可清点一个或多个数据集。默认情况下，它仅读取元数据，输出结构字段和通道标签，但不输出完整的 TEXT/ANALYSIS 值；对于超过可配置大小限制的文件，它会拒绝处理。

请将 FLOWIO_SKILL_DIR 设置为已安装技能的目录。从此仓库的根目录出发，请使用 skills/flowio：

FLOWIO_SKILL_DIR="skills/flowio"

# Metadata and channel inventory
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs

# Include all normalized TEXT metadata; review output for identifiers
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs --include-text

# Load events and compute finite-value statistics using FlowIO preprocessing
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs --stats

# Compute statistics from encoded values instead
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs --stats --raw

请使用 --help 查看输出文件、输入/数组内存限制、空通道标签以及受控的偏移量恢复选项。

参考资料

请仅阅读当前任务所需的参考资料：

• references/api_reference.md — FlowIO 1.4.0 的确切公共 API 和签名
• references/workflows.md — 清点、DataFrame/CSV、批处理、写入和往返验证模式
• references/fcs_semantics.md — FCS 结构、元数据规范化、预处理公式、索引和写入器行为
• references/troubleshooting.md — 偏移量错误、多数据集文件、内存限制、验证、安全和隐私
• references/sources.md — 本次更新所依据的权威上游文档、发行说明、源代码和 FCS 3.1 出版物

不可妥协的检查项

• 绝不可声称 FlowIO 会应用补偿或门控。
• 绝不可将 as_array(preprocess=True) 视为原始采集值。
• 绝不可将二维数组或路径直接传给 create_fcs()。
• 绝不可假设 TEXT 键会保留 $ 或大写拼写。
• 绝不可在未说明原因并验证数据的情况下忽略偏移量错误。
• 绝不可将 FlowIO 的事件加载描述为流式或分块加载。

Agent 操作流程

1. 检查环境。 确认工具版本、参考基因组／注释版本以及输入格式（FASTQ、BAM、VCF、h5ad）。
2. 明确输入。 根据数据或用户提供的信息确认格式、标识符和参数。任何会改变结果的值都应询问，不要猜测。
3. 先运行小规模版本。 先在小型子集（一个样本、一条染色体、几千个细胞）上运行流水线。
4. 使用以上说明和参考资料执行完整任务。
5. 验证结果。 在解读结果前，检查 QC 指标、样本身份、基因组版本一致性和批次效应。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及哪些事项仍不确定。

如果发生这种情况	请这样做
输入之间的基因组版本或标识符不匹配	停止并统一（liftover、ID 映射）后再继续。
这些说明中的函数、标志或端点在已安装版本中不存在	查阅已安装版本自带的文档（help()、--help、官方文档），进行调整并告知用户。切勿编造 API。
必需的输入、标识符或参数不明确	询问用户，或在运行前明确说明所作的假设。

诚信规则

• 切勿捏造结果、参数、标识符、引文或统计数据。如果无法运行或验证某项内容，请如实说明。
• 不要解读超出统计结果所支持范围的生物学意义；报告多重检验校正情况。
• 请将此处与版本相关的细节视为可能已过时：根据已安装版本的官方文档进行确认。
• 对于会产生费用、占用共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，执行前必须先询问。