---
name: flowio
description: "สกิลสำหรับ Flowio ใช้เมื่อต้องการแนวทาง ขั้นตอน Template หรือคำแนะนำเชิงปฏิบัติสำหรับงานวิเคราะห์ข้อมูล วิจัย AI ฐานข้อมูล วิทยาศาสตร์ข้อมูล และรายงานเชิงลึก ในหัวข้อนี้"
---

# Flowio

## Overview

FlowIO is a lightweight Python library for reading and writing Flow Cytometry Standard (FCS) files. Parse FCS metadata, extract event data, and create new FCS files with minimal dependencies. The library supports FCS versions 2.0, 3.0, and 3.1, making it ideal for backend services, data pipelines, and basic cytometry file operations.

## When to Use This Skill

This skill should be used when:

- FCS files requiring parsing or metadata extraction
- Flow cytometry data needing conversion to NumPy arrays
- Event data requiring export to FCS format
- Multi-dataset FCS files needing separation
- Channel information extraction (scatter, fluorescence, time)
- Cytometry file validation or inspection
- Pre-processing workflows before advanced analysis

**Related Tools:** For advanced flow cytometry analysis including compensation, gating, and FlowJo/GatingML support, recommend FlowKit library as a companion to FlowIO.

## Installation

```bash
uv pip install flowio
```

Requires Python 3.9 or later.

## Quick Start

### Basic File Reading

```python
from flowio import FlowData

# Flowio
flow_data = FlowData('experiment.fcs')

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

# Flowio
events = flow_data.as_array()  # Shape: (events, channels)
```

### Creating FCS Files

```python
import numpy as np
from flowio import create_fcs

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

# Flowio
create_fcs('output.fcs', data, channels)
```

## Core Workflows

### Reading and Parsing FCS Files

The FlowData class provides the primary interface for reading FCS files.

**Standard Reading:**

```python
from flowio import FlowData

# Flowio
flow = FlowData('sample.fcs')

# Flowio
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

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

**Memory-Efficient Metadata Reading:**

When only metadata is needed (no event data):

```python
# Flowio
flow = FlowData('sample.fcs', only_text=True)

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

**Handling Problematic Files:**

Some FCS files have offset discrepancies or errors:

```python
# Flowio
flow = FlowData('problematic.fcs', ignore_offset_discrepancy=True)

# Flowio
flow = FlowData('problematic.fcs', use_header_offsets=True)

# Flowio
flow = FlowData('problematic.fcs', ignore_offset_error=True)
```

**Excluding Null Channels:**

```python
# Flowio
flow = FlowData('sample.fcs', null_channel_list=['Time', 'Null'])
```

### Extracting Metadata and Channel Information

FCS files contain rich metadata in the TEXT segment.

**Common Metadata Keywords:**

```python
flow = FlowData('sample.fcs')

# Flowio
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'

# Flowio
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}")
```

**Channel Type Identification:**

FlowIO automatically categorizes channels:

```python
# Flowio
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)

# Flowio
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluorescence_data = events[:, fluoro_idx]
```

**ANALYSIS Segment:**

If present, access processed results:

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

### Creating New FCS Files

Generate FCS files from NumPy arrays or other data sources.

**Basic Creation:**

```python
import numpy as np
from flowio import create_fcs

# Flowio
events = np.random.rand(10000, 5) * 1000

# Flowio
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']

# Flowio
create_fcs('output.fcs', events, channel_names)
```

**With Descriptive Channel Names:**

```python
# Flowio
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)
```

**With Custom Metadata:**

```python
# Flowio
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)
```

**Note:** FlowIO exports as FCS 3.1 with single-precision floating-point data.

### Exporting Modified Data

Modify existing FCS files and re-export them.

**Approach 1: Using write_fcs() Method:**

```python
from flowio import FlowData

# Flowio
flow = FlowData('original.fcs')

# Flowio
flow.write_fcs('modified.fcs', metadata={'$SRC': 'Modified data'})
```

**Approach 2: Extract, Modify, and Recreate:**

For modifying event data:

```python
from flowio import FlowData, create_fcs

# Flowio
flow = FlowData('original.fcs')
events = flow.as_array(preprocess=False)

# Flowio
events[:, 0] = events[:, 0] * 1.5  # Scale first channel

# Flowio
create_fcs('modified.fcs',
           events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata=flow.text)
```

### Handling Multi-Dataset FCS Files

Some FCS files contain multiple datasets in a single file.

**Detecting Multi-Dataset Files:**

```python
from flowio import FlowData, MultipleDataSetsError

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

**Reading All Datasets:**

```python
from flowio import read_multiple_data_sets

# Flowio
datasets = read_multiple_data_sets('multi_dataset.fcs')

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

# Flowio
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)}")
```

**Reading Specific Dataset:**

```python
from flowio import FlowData

# Flowio
first_dataset = FlowData('multi.fcs', nextdata_offset=0)

# Flowio
next_offset = int(first_dataset.text['$NEXTDATA'])
if next_offset > 0:
    second_dataset = FlowData('multi.fcs', nextdata_offset=next_offset)
```

## Data Preprocessing

FlowIO applies standard FCS preprocessing transformations when `preprocess=True`.

**Preprocessing Steps:**

1. **Gain Scaling:** Multiply values by PnG (gain) keyword
2. **Logarithmic Transformation:** Apply PnE exponential transformation if present
   - Formula: `value = a * 10^(b * raw_value)` where PnE = "a,b"
3. **Time Scaling:** Convert time values to appropriate units

**Controlling Preprocessing:**

```python
# Flowio
preprocessed = flow.as_array(preprocess=True)

# Flowio
raw = flow.as_array(preprocess=False)
```

## Error Handling

Handle common FlowIO exceptions appropriately.

```python
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}")
```

## Common Use Cases

### Inspecting FCS File Contents

Quick exploration of FCS file structure:

```python
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}")
```

### Batch Processing Multiple Files

Process a directory of FCS files:

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

# Flowio
fcs_files = list(Path('data/').glob('*.fcs'))

# Flowio
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}")

# Flowio
df = pd.DataFrame(summaries)
print(df)
```

### Converting FCS to CSV

Export event data to CSV format:

```python
from flowio import FlowData
import pandas as pd

# Flowio
flow = FlowData('sample.fcs')

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

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

# Flowio
df.to_CSV('output.CSV', index=False)
print(f"Exported {len(df)} events to CSV")
```

### Filtering Events and Re-exporting

Apply filters and save filtered data:

```python
from flowio import FlowData, create_fcs
import numpy as np

# Flowio
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)

# Flowio
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)}")

# Flowio
create_fcs('filtered.fcs',
           filtered_events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata={**flow.text, '$SRC': 'Filtered data'})
```

### Extracting Specific Channels

Extract and process specific channels:

```python
from flowio import FlowData
import numpy as np

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

# Flowio
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}")

# Flowio
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}")
```

## Best Practices

1. **Memory Efficiency:** Use `only_text=True` when event data is not needed
2. **Error Handling:** Wrap file operations in try-except blocks for robust code
3. **Multi-Dataset Detection:** Check for MultipleDataSetsError and use appropriate function
4. **Preprocessing Control:** Explicitly set `preprocess` parameter based on analysis needs
5. **Offset Issues:** If parsing fails, try `ignore_offset_discrepancy=True` parameter
6. **Channel Validation:** Verify channel counts and names match expectations before processing
7. **Metadata Preservation:** When modifying files, preserve original TEXT segment keywords

## Advanced Topics

### Understanding FCS File Structure

FCS files consist of four segments:

1. **HEADER:** FCS version and byte offsets for other segments
2. **TEXT:** Key-value metadata pairs (delimiter-separated)
3. **DATA:** Raw event data (binary/float/ASCII format)
4. **ANALYSIS** (optional): Results from data processing

Access these segments via FlowData attributes:
- `flow.header` - HEADER segment
- `flow.text` - TEXT segment keywords
- `flow.events` - DATA segment (as bytes)
- `flow.analysis` - ANALYSIS segment keywords (if present)

### Detailed API Reference

For comprehensive API documentation including all parameters, methods, exceptions, and FCS keyword reference, consult the detailed reference file:

**Read:** `references/api_reference.md`

The reference includes:
- Complete FlowData class documentation
- All utility functions (read_multiple_data_sets, create_fcs)
- Exception classes and handling
- FCS file structure details
- Common TEXT segment keywords
- Extended example workflows

When working with complex FCS operations or encountering unusual file formats, load this reference for detailed guidance.

## Integration Notes

**NumPy Arrays:** All event data is returned as NumPy ndarrays with shape (events, channels)

**Pandas DataFrames:** Easily convert to DataFrames for analysis:
```python
import pandas as pd
df = pd.DataFrame(flow.as_array(), columns=flow.pnn_labels)
```

**FlowKit Integration:** For advanced analysis (compensation, gating, FlowJo support), use FlowKit library which builds on FlowIO's parsing capabilities

**Web Applications:** FlowIO's minimal dependencies make it ideal for web backend services processing FCS uploads

## Troubleshooting

**Problem:** "Offset discrepancy error"
**Solution:** Use `ignore_offset_discrepancy=True` parameter

**Problem:** "Multiple datasets error"
**Solution:** Use `read_multiple_data_sets()` function instead of FlowData constructor

**Problem:** Out of memory with large files
**Solution:** Use `only_text=True` for metadata-only operations, or process events in chunks

**Problem:** Unexpected channel counts
**Solution:** Check for null channels; use `null_channel_list` parameter to exclude them

**Problem:** Cannot modify event data in place
**Solution:** FlowIO doesn't support direct modification; extract data, modify, then use `create_fcs()` to save

## Summary

FlowIO provides essential FCS file handling capabilities for flow cytometry workflows. Use it for parsing, metadata extraction, and file creation. For simple file operations and data extraction, FlowIO is sufficient. For complex analysis including compensation and gating, integrate with FlowKit or other specialized tools.
