flowio

FlowIO: Flow Cytometry Standard ファイルハンドラー

概要

FlowIO は、Flow Cytometry Standard（FCS）ファイルの読み取りと書き込みを行う軽量な Python ライブラリです。最小限の依存関係で、FCS メタデータの解析、イベントデータの抽出、新しい FCS ファイルの作成を行えます。FCS バージョン 2.0、3.0、3.1 に対応しており、バックエンドサービス、データパイプライン、基本的なフローサイトメトリーファイル操作に最適です。

このスキルを使用するタイミング

このスキルは、次のような場合に使用します。

• 解析またはメタデータ抽出が必要な FCS ファイル
• NumPy 配列への変換が必要なフローサイトメトリーデータ
• FCS 形式へのエクスポートが必要なイベントデータ
• 分離が必要な複数データセットの FCS ファイル
• チャンネル情報の抽出（スキャター、蛍光、時間）
• フローサイトメトリーファイルの検証または確認
• 高度な解析の前処理ワークフロー

関連ツール: コンペンセーション、ゲーティング、FlowJo/GatingML への対応を含む高度なフローサイトメトリー解析には、FlowIO の補完ツールとして FlowKit ライブラリをおすすめします。

インストール

pip install flowio

Python 3.9 以降が必要です。

クイックスタート

基本的なファイル読み取り

from flowio import FlowData

# Read FCS file
flow_data = FlowData('experiment.fcs')

# Access basic information
print(f"FCS Version: {flow_data.version}")
print(f"Events: {flow_data.event_count}")
print(f"Channels: {flow_data.pnn_labels}")

# Get event data as NumPy array
events = flow_data.as_array()  # Shape: (events, channels)

FCS ファイルの作成

import numpy as np
from flowio import create_fcs

# Prepare data
data = np.array([[100, 200, 50], [150, 180, 60]])  # 2 events, 3 channels
channels = ['FSC-A', 'SSC-A', 'FL1-A']

# Create FCS file
create_fcs('output.fcs', data, channels)

主要なワークフロー

FCS ファイルの読み取りと解析

FlowData クラスは、FCS ファイルを読み取るための主要なインターフェースを提供します。

標準的な読み取り:

from flowio import FlowData

# Basic reading
flow = FlowData('sample.fcs')

# Access attributes
version = flow.version              # '3.0', '3.1', etc.
event_count = flow.event_count      # Number of events
channel_count = flow.channel_count  # Number of channels
pnn_labels = flow.pnn_labels        # Short channel names
pns_labels = flow.pns_labels        # Descriptive stain names

# Get event data
events = flow.as_array()            # Preprocessed (gain, log scaling applied)
raw_events = flow.as_array(preprocess=False)  # Raw data

メモリ効率の良いメタデータ読み取り:

イベントデータが不要で、メタデータのみが必要な場合:

# Only parse TEXT segment, skip DATA and ANALYSIS
flow = FlowData('sample.fcs', only_text=True)

# Access metadata
metadata = flow.text  # Dictionary of TEXT segment keywords
print(metadata.get('$DATE'))  # Acquisition date
print(metadata.get('$CYT'))   # Instrument name

問題のあるファイルの処理:

一部の FCS ファイルには、オフセットの不一致やエラーがあります。

# Ignore offset discrepancies between HEADER and TEXT sections
flow = FlowData('problematic.fcs', ignore_offset_discrepancy=True)

# Use HEADER offsets instead of TEXT offsets
flow = FlowData('problematic.fcs', use_header_offsets=True)

# Ignore offset errors entirely
flow = FlowData('problematic.fcs', ignore_offset_error=True)

Null チャンネルの除外:

# Exclude specific channels during parsing
flow = FlowData('sample.fcs', null_channel_list=['Time', 'Null'])

メタデータとチャンネル情報の抽出

FCS ファイルには、TEXT セグメントに豊富なメタデータが含まれています。

よく使われるメタデータキーワード:

flow = FlowData('sample.fcs')

# File-level metadata
text_dict = flow.text
acquisition_date = text_dict.get('$DATE', 'Unknown')
instrument = text_dict.get('$CYT', 'Unknown')
data_type = flow.data_type  # 'I', 'F', 'D', 'A'

# Channel metadata
for i in range(flow.channel_count):
    pnn = flow.pnn_labels[i]      # Short name (e.g., 'FSC-A')
    pns = flow.pns_labels[i]      # Descriptive name (e.g., 'Forward Scatter')
    pnr = flow.pnr_values[i]      # Range/max value
    print(f"Channel {i}: {pnn} ({pns}), Range: {pnr}")

チャンネルタイプの識別:

FlowIO はチャンネルを自動的に分類します。

# Get indices by channel type
scatter_idx = flow.scatter_indices    # [0, 1] for FSC, SSC
fluoro_idx = flow.fluoro_indices      # [2, 3, 4] for FL channels
time_idx = flow.time_index            # Index of time channel (or None)

# Access specific channel types
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluorescence_data = events[:, fluoro_idx]

ANALYSIS セグメント:

存在する場合は、処理済みの結果にアクセスします。

if flow.analysis:
    analysis_keywords = flow.analysis  # Dictionary of ANALYSIS keywords
    print(analysis_keywords)

新しい FCS ファイルの作成

NumPy 配列やその他のデータソースから FCS ファイルを生成します。

基本的な作成:

import numpy as np
from flowio import create_fcs

# Create event data (rows=events, columns=channels)
events = np.random.rand(10000, 5) * 1000

# Define channel names
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']

# Create FCS file
create_fcs('output.fcs', events, channel_names)

わかりやすいチャンネル名を付ける場合:

# Add optional descriptive 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)

カスタムメタデータを付ける場合:

# Add TEXT segment metadata
metadata = {
    '$SRC': 'Python script',
    '$DATE': '19-OCT-2025',
    '$CYT': 'Synthetic Instrument',
    '$INST': 'Laboratory 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

# Read original file
flow = FlowData('original.fcs')

# Write with updated metadata
flow.write_fcs('modified.fcs', metadata={'$SRC': 'Modified data'})

方法 2: 抽出、変更、再作成:

イベントデータを変更する場合:

from flowio import FlowData, create_fcs

# Read and extract data
flow = FlowData('original.fcs')
events = flow.as_array(preprocess=False)

# Modify event data
events[:, 0] = events[:, 0] * 1.5  # Scale first channel

# Create new FCS file with modified data
create_fcs('modified.fcs',
           events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata=flow.text)

複数データセットの FCS ファイルの処理

一部の FCS ファイルには、1 つのファイル内に複数のデータセットが含まれています。

複数データセットのファイルの検出:

from flowio import FlowData, MultipleDataSetsError

try:
    flow = FlowData('sample.fcs')
except MultipleDataSetsError:
    print("File contains multiple datasets")
    # Use read_multiple_data_sets() instead

すべてのデータセットの読み取り:

from flowio import read_multiple_data_sets

# Read all datasets from file
datasets = read_multiple_data_sets('multi_dataset.fcs')

print(f"Found {len(datasets)} datasets")

# Process each dataset
for i, dataset in enumerate(datasets):
    print(f"\nDataset {i}:")
    print(f"  Events: {dataset.event_count}")
    print(f"  Channels: {dataset.pnn_labels}")

    # Get event data for this dataset
    events = dataset.as_array()
    print(f"  Shape: {events.shape}")
    print(f"  Mean values: {events.mean(axis=0)}")

特定のデータセットの読み取り:

from flowio import FlowData

# Read first dataset (nextdata_offset=0)
first_dataset = FlowData('multi.fcs', nextdata_offset=0)

# Read second dataset using NEXTDATA offset from first
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 が存在する場合、指数変換を適用します
   • 計算式: value = a * 10^(b * raw_value)（PnE = "a,b" の場合）
3. 時間スケーリング: 時間の値を適切な単位に変換します

前処理の制御:

# Preprocessed data (default)
preprocessed = flow.as_array(preprocess=True)

# Raw data (no transformations)
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"Failed to parse FCS file: {e}")
    # Try with relaxed parsing
    flow = FlowData('sample.fcs', ignore_offset_error=True)

except DataOffsetDiscrepancyError as e:
    print(f"Offset discrepancy detected: {e}")
    # Use ignore_offset_discrepancy parameter
    flow = FlowData('sample.fcs', ignore_offset_discrepancy=True)

except MultipleDataSetsError as e:
    print(f"Multiple datasets detected: {e}")
    # Use read_multiple_data_sets instead
    from flowio import read_multiple_data_sets
    datasets = read_multiple_data_sets('sample.fcs')

except Exception as e:
    print(f"Unexpected error: {e}")

よくある利用例

FCS ファイルの内容の確認

FCS ファイルの構造を手早く確認します。

from flowio import FlowData

flow = FlowData('unknown.fcs')

print("=" * 50)
print(f"File: {flow.name}")
print(f"Version: {flow.version}")
print(f"Size: {flow.file_size:,} bytes")
print("=" * 50)

print(f"\nEvents: {flow.event_count:,}")
print(f"Channels: {flow.channel_count}")

print("\nChannel Information:")
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("\nKey Metadata:")
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

# Find all FCS files
fcs_files = list(Path('data/').glob('*.fcs'))

# Extract summary information
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"Error processing {fcs_path.name}: {e}")

# Create summary DataFrame
df = pd.DataFrame(summaries)
print(df)

FCS から CSV への変換

イベントデータを CSV 形式でエクスポートします。

from flowio import FlowData
import pandas as pd

# Read FCS file
flow = FlowData('sample.fcs')

# Convert to DataFrame
df = pd.DataFrame(
    flow.as_array(),
    columns=flow.pnn_labels
)

# Add metadata as attributes
df.attrs['fcs_version'] = flow.version
df.attrs['instrument'] = flow.text.get('$CYT', 'Unknown')

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

イベントのフィルタリングと再エクスポート

フィルターを適用し、フィルター済みのデータを保存します。

from flowio import FlowData, create_fcs
import numpy as np

# Read original file
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)

# Apply filtering (example: threshold on first channel)
fsc_idx = 0
threshold = 500
mask = events[:, fsc_idx] > threshold
filtered_events = events[mask]

print(f"Original events: {len(events)}")
print(f"Filtered events: {len(filtered_events)}")

# Create new FCS file with filtered data
create_fcs('filtered.fcs',
           filtered_events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata={**flow.text, '$SRC': 'Filtered data'})

特定のチャンネルの抽出

特定のチャンネルを抽出して処理します。

from flowio import FlowData
import numpy as np

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

# Extract fluorescence channels only
fluoro_indices = flow.fluoro_indices
fluoro_data = events[:, fluoro_indices]
fluoro_names = [flow.pnn_labels[i] for i in fluoro_indices]

print(f"Fluorescence channels: {fluoro_names}")
print(f"Shape: {fluoro_data.shape}")

# Calculate statistics per channel
for i, name in enumerate(fluoro_names):
    channel_data = fluoro_data[:, i]
    print(f"\n{name}:")
    print(f"  Mean: {channel_data.mean():.2f}")
    print(f"  Median: {np.median(channel_data):.2f}")
    print(f"  Std Dev: {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 ファイルは 4 つのセグメントで構成されています。

1. HEADER: FCS のバージョンと、他のセグメントのバイトオフセット
2. TEXT: キーと値のメタデータのペア（区切り文字で区切られています）
3. DATA: 生のイベントデータ（バイナリ/浮動小数点/ASCII 形式）
4. ANALYSIS（任意）: データ処理の結果

これらのセグメントには FlowData の属性を介してアクセスできます。

• flow.header - HEADER セグメント
• flow.text - TEXT セグメントのキーワード
• flow.events - DATA セグメント（バイト列として）
• flow.analysis - ANALYSIS セグメントのキーワード（存在する場合）

詳細な API リファレンス

すべてのパラメーター、メソッド、例外、FCS キーワードのリファレンスを含む包括的な API ドキュメントについては、詳細なリファレンスファイルを参照してください。

読み込み: references/api_reference.md

リファレンスには次の内容が含まれます。

• FlowData クラスの完全なドキュメント
• すべてのユーティリティ関数（read_multiple_data_sets、create_fcs）
• 例外クラスと例外処理
• FCS ファイル構造の詳細
• 一般的な TEXT セグメントのキーワード
• 拡張されたサンプルワークフロー

複雑な FCS の操作を行う場合や、通常とは異なるファイル形式に遭遇した場合は、このリファレンスを読み込んで詳しいガイダンスを確認してください。

統合に関する注意

NumPy 配列: すべてのイベントデータは、形状 (events, channels) の NumPy ndarray として返されます。

Pandas DataFrame: 解析のために DataFrame へ簡単に変換できます。

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

FlowKit との統合: 高度な解析（補正、ゲーティング、FlowJo 対応）には、FlowIO の解析機能を基盤とする FlowKit ライブラリを使用してください。

Web アプリケーション: FlowIO は依存関係が最小限のため、FCS のアップロードを処理する Web バックエンドサービスに適しています。

トラブルシューティング

問題: 「Offset discrepancy error」（オフセット不一致エラー） 解決策: ignore_offset_discrepancy=True パラメーターを使用する

問題: 「Multiple datasets error」（複数データセットエラー） 解決策: FlowData コンストラクターではなく read_multiple_data_sets() 関数を使用する

問題: 大きなファイルでメモリ不足になる 解決策: メタデータのみを扱う操作では only_text=True を使用するか、イベントをチャンク単位で処理する

問題: チャンネル数が想定と異なる 解決策: null チャンネルがないか確認し、null_channel_list パラメーターを使用して除外する

問題: イベントデータをその場で変更できない 解決策: FlowIO は直接の変更をサポートしていません。データを抽出して変更した後、create_fcs() を使用して保存してください。

まとめ

FlowIO は、フローサイトメトリーのワークフローに必要な FCS ファイル処理機能を提供します。解析、メタデータの抽出、ファイルの作成に使用してください。単純なファイル操作やデータ抽出であれば FlowIO で十分です。補正やゲーティングを含む複雑な解析には、FlowKit や他の専門ツールと統合してください。