flowio-flow-cytometry

FlowIO — フローサイトメトリーファイルハンドラー

概要

FlowIO は、フローサイトメトリースタンダード（FCS）ファイルの読み書きに対応した軽量な Python ライブラリです。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 ファイルは、次の 4 つのセグメントで構成されます。

セグメント	内容	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	TEXT より HEADER のオフセットを優先する
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	Dict	カスタムの TEXT セグメントのキーと値のペア

ベストプラクティス

1. メタデータのスキャンには only_text=True を使用する：多数のファイルを処理する場合は、DATA セグメントの解析をスキップすると、速度が 10～100 倍向上します。

2. データを変更する場合は preprocess=False を使用する：イベントのフィルタリングや変更を行うときは、常に生の値を使用してから再エクスポートしてください。前処理は元に戻せません。

3. アンチパターン — flow.events を直接変更する：FlowIO はイベントのインプレース変更に対応していません。as_array() で抽出して変更し、create_fcs() で保存してください。

4. 再エクスポート時にメタデータを保持する：元の取得情報を保持するには、create_fcs() のメタデータとして flow.text を渡してください。

5. 複数データセットを含むファイルを確認する：MultipleDataSetsError を捕捉し、read_multiple_data_sets() を使用してください。一部の機器は、複数の取得データを 1 つのファイルに書き込みます。

よく使うレシピ

レシピ：蛍光チャンネルのみを抽出する

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 チャンネルまたはパディングチャンネルが含まれている	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 et al.（2010）「Data File Standard for Flow Cytometry, Version FCS 3.1」— Cytometry Part A