| name | flowio |
| description | Parse FCS (Flow Cytometry Standard) files v2.0-3.1. Extract events as NumPy arrays, read metadata/channels, convert to CSV/DataFrame, for flow cytometry data preprocessing. |
| license | BSD-3-Clause license |
| metadata | version: "1.0" skill-author: K-Dense Inc. |
FlowIO: Flow Cytometry Standard File Handler
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
uv pip install flowio
Requires Python 3.9 or later.
Quick Start
Basic File Reading
from flowio import FlowData
flow_data = FlowData('experiment.fcs')
print(f"FCS Version: {flow_data.version}")
print(f"Events: {flow_data.event_count}")
print(f"Channels: {flow_data.pnn_labels}")
events = flow_data.as_array()
Creating FCS Files
import numpy as np
from flowio import create_fcs
data = np.array([[100, 200, 50], [150, 180, 60]])
channels = ['FSC-A', 'SSC-A', 'FL1-A']
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:
from flowio import FlowData
flow = FlowData('sample.fcs')
version = flow.version
event_count = flow.event_count
channel_count = flow.channel_count
pnn_labels = flow.pnn_labels
pns_labels = flow.pns_labels
events = flow.as_array()
raw_events = flow.as_array(preprocess=False)
Memory-Efficient Metadata Reading:
When only metadata is needed (no event data):
flow = FlowData('sample.fcs', only_text=True)
metadata = flow.text
print(metadata.get('$DATE'))
print(metadata.get('$CYT'))
Handling Problematic Files:
Some FCS files have offset discrepancies or errors:
flow = FlowData('problematic.fcs', ignore_offset_discrepancy=True)
flow = FlowData('problematic.fcs', use_header_offsets=True)
flow = FlowData('problematic.fcs', ignore_offset_error=True)
Excluding Null Channels:
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:
flow = FlowData('sample.fcs')
text_dict = flow.text
acquisition_date = text_dict.get('$DATE', 'Unknown')
instrument = text_dict.get('$CYT', 'Unknown')
data_type = flow.data_type
for i in range(flow.channel_count):
pnn = flow.pnn_labels[i]
pns = flow.pns_labels[i]
pnr = flow.pnr_values[i]
print(f"Channel {i}: {pnn} ({pns}), Range: {pnr}")
Channel Type Identification:
FlowIO automatically categorizes channels:
scatter_idx = flow.scatter_indices
fluoro_idx = flow.fluoro_indices
time_idx = flow.time_index
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluorescence_data = events[:, fluoro_idx]
ANALYSIS Segment:
If present, access processed results:
if flow.analysis:
analysis_keywords = flow.analysis
print(analysis_keywords)
Creating New FCS Files
Generate FCS files from NumPy arrays or other data sources.
Basic Creation:
import numpy as np
from flowio import create_fcs
events = np.random.rand(10000, 5) * 1000
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']
create_fcs('output.fcs', events, channel_names)
With Descriptive Channel Names:
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:
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:
from flowio import FlowData
flow = FlowData('original.fcs')
flow.write_fcs('modified.fcs', metadata={'$SRC': 'Modified data'})
Approach 2: Extract, Modify, and Recreate:
For modifying event data:
from flowio import FlowData, create_fcs
flow = FlowData('original.fcs')
events = flow.as_array(preprocess=False)
events[:, 0] = events[:, 0] * 1.5
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:
from flowio import FlowData, MultipleDataSetsError
try:
flow = FlowData('sample.fcs')
except MultipleDataSetsError:
print("File contains multiple datasets")
Reading All Datasets:
from flowio import read_multiple_data_sets
datasets = read_multiple_data_sets('multi_dataset.fcs')
print(f"Found {len(datasets)} datasets")
for i, dataset in enumerate(datasets):
print(f"\nDataset {i}:")
print(f" Events: {dataset.event_count}")
print(f" Channels: {dataset.pnn_labels}")
events = dataset.as_array()
print(f" Shape: {events.shape}")
print(f" Mean values: {events.mean(axis=0)}")
Reading Specific Dataset:
from flowio import FlowData
first_dataset = FlowData('multi.fcs', nextdata_offset=0)
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:
- Gain Scaling: Multiply values by PnG (gain) keyword
- Logarithmic Transformation: Apply PnE exponential transformation if present
- Formula:
value = a * 10^(b * raw_value) where PnE = "a,b"
- Time Scaling: Convert time values to appropriate units
Controlling Preprocessing:
preprocessed = flow.as_array(preprocess=True)
raw = flow.as_array(preprocess=False)
Error Handling
Handle common FlowIO exceptions appropriately.
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}")
flow = FlowData('sample.fcs', ignore_offset_error=True)
except DataOffsetDiscrepancyError as e:
print(f"Offset discrepancy detected: {e}")
flow = FlowData('sample.fcs', ignore_offset_discrepancy=True)
except MultipleDataSetsError as e:
print(f"Multiple datasets detected: {e}")
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:
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:
from pathlib import Path
from flowio import FlowData
import pandas as pd
fcs_files = list(Path('data/').glob('*.fcs'))
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}")
df = pd.DataFrame(summaries)
print(df)
Converting FCS to CSV
Export event data to CSV format:
from flowio import FlowData
import pandas as pd
flow = FlowData('sample.fcs')
df = pd.DataFrame(
flow.as_array(),
columns=flow.pnn_labels
)
df.attrs['fcs_version'] = flow.version
df.attrs['instrument'] = flow.text.get('$CYT', 'Unknown')
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:
from flowio import FlowData, create_fcs
import numpy as np
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)
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_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:
from flowio import FlowData
import numpy as np
flow = FlowData('sample.fcs')
events = flow.as_array()
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}")
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
- Memory Efficiency: Use
only_text=True when event data is not needed
- Error Handling: Wrap file operations in try-except blocks for robust code
- Multi-Dataset Detection: Check for MultipleDataSetsError and use appropriate function
- Preprocessing Control: Explicitly set
preprocess parameter based on analysis needs
- Offset Issues: If parsing fails, try
ignore_offset_discrepancy=True parameter
- Channel Validation: Verify channel counts and names match expectations before processing
- Metadata Preservation: When modifying files, preserve original TEXT segment keywords
Advanced Topics
Understanding FCS File Structure
FCS files consist of four segments:
- HEADER: FCS version and byte offsets for other segments
- TEXT: Key-value metadata pairs (delimiter-separated)
- DATA: Raw event data (binary/float/ASCII format)
- ANALYSIS (optional): Results from data processing
Access these segments via FlowData attributes:
flow.header - HEADER segmentflow.text - TEXT segment keywordsflow.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:
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.
