FlowIO Workflows

These patterns target flowio==1.4.0. Keep raw files immutable and write derived outputs to a separate location.

1. Inventory a File Without Loading Events

Use only_text=True when the goal is file structure, labels, or acquisition metadata:

python

from pathlib import Path

from flowio import FlowData

path = Path("sample.fcs")
flow = FlowData(path, only_text=True)

summary = {
    "file": path.name,
    "size_bytes": path.stat().st_size,
    "fcs_version": flow.version,
    "data_type": flow.data_type,
    "event_count": flow.event_count,
    "channel_count": flow.channel_count,
    "pnn_labels": flow.pnn_labels,
    "pns_labels": flow.pns_labels,
    "date": flow.text.get("date"),
    "instrument": flow.text.get("cyt"),
}
print(summary)

Do not call as_array() on this object. only_text=True skips DATA loading but still parses ANALYSIS.

For an inventory that also handles multi-dataset files and emits JSON, use:

bash

FLOWIO_SKILL_DIR="skills/flowio"  # Set to the installed skill directory.
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs

2. Load Encoded or Metadata-Scaled Events

python

from flowio import FlowData

flow = FlowData("sample.fcs")

# Gain/log/time scaling from FCS metadata.
scaled = flow.as_array(preprocess=True)

# Values reshaped as encoded in DATA.
encoded = flow.as_array(preprocess=False)

assert scaled.shape == (flow.event_count, flow.channel_count)
assert encoded.shape == scaled.shape

Neither array is compensated or gated. Name output variables to preserve this distinction; avoid generic names such as processed.

3. Build a DataFrame

This example assumes pandas is already present in the project environment. Add and lock an environment-compatible pandas release only when tabular integration is required; FlowIO itself does not depend on pandas.

FCS files can contain duplicate or empty labels. Make DataFrame names unique instead of silently assuming PnN values are valid columns:

python

from collections import Counter

import pandas as pd
from flowio import FlowData


def unique_labels(labels):
    counts = Counter()
    result = []
    for index, value in enumerate(labels):
        base = value.strip() or f"channel_{index + 1}"
        counts[base] += 1
        suffix = "" if counts[base] == 1 else f"__{counts[base]}"
        result.append(f"{base}{suffix}")
    return result


flow = FlowData("sample.fcs")
columns = unique_labels(flow.pnn_labels)
frame = pd.DataFrame(
    flow.as_array(preprocess=True),
    columns=columns,
)

frame.attrs.update(
    {
        "flowio_version": "1.4.0",
        "fcs_version": flow.version,
        "event_semantics": "gain/log/time scaled; uncompensated; ungated",
        "source_pnn_labels": flow.pnn_labels,
        "source_pns_labels": flow.pns_labels,
    }
)

DataFrame attributes are not preserved by every export format. Store a sidecar JSON provenance record for durable pipelines.

4. Export CSV Safely

CSV loses most FCS metadata and type information. Export a sidecar record:

python

import json
from pathlib import Path

import pandas as pd
from flowio import FlowData

source = Path("sample.fcs")
flow = FlowData(source)
values = flow.as_array(preprocess=False)

frame = pd.DataFrame(
    values,
    columns=unique_labels(flow.pnn_labels),  # Helper from the previous workflow.
)
csv_path = Path("sample.events.csv")
with csv_path.open("x", encoding="utf-8", newline="") as handle:
    frame.to_csv(handle, index=False)

provenance = {
    "source_file": source.name,
    "flowio_version": "1.4.0",
    "fcs_version": flow.version,
    "event_semantics": "encoded DATA values reshaped; preprocess=False",
    "pnn_labels": flow.pnn_labels,
    "pns_labels": flow.pns_labels,
    "event_count": flow.event_count,
    "channel_count": flow.channel_count,
}
with Path("sample.events.json").open("x", encoding="utf-8") as handle:
    handle.write(json.dumps(provenance, indent=2) + "\n")

Before sharing either output, review labels and metadata for subject or sample identifiers.

5. Batch Metadata Inventory

This pattern reads no event arrays:

python

from pathlib import Path

from flowio import read_multiple_data_sets
from flowio.exceptions import FlowIOException

records = []
failures = []

for path in sorted(Path("fcs").rglob("*.fcs")):
    try:
        datasets = read_multiple_data_sets(path, only_text=True)
        for dataset_index, flow in enumerate(datasets):
            records.append(
                {
                    "file": str(path),
                    "dataset_index": dataset_index,
                    "fcs_version": flow.version,
                    "event_count": flow.event_count,
                    "channel_count": flow.channel_count,
                    "pnn_labels": flow.pnn_labels,
                }
            )
    except (FlowIOException, OSError, ValueError) as error:
        failures.append({"file": str(path), "error": str(error)})

Do not automatically retry failed files with every relaxed offset option. Triage each error type and file source.

6. Read Legacy Multi-Dataset Files

Pass a path, not an open handle. FlowIO 1.4.0 closes caller-provided handles, so the utility cannot reuse one for the second dataset.

python

from flowio import read_multiple_data_sets

datasets = read_multiple_data_sets(
    "legacy.lmd",
    only_text=False,
)

for dataset_index, flow in enumerate(datasets):
    encoded = flow.as_array(preprocess=False)
    print(dataset_index, encoded.shape, flow.text.get("nextdata"))

If a known off-by-one vendor file requires recovery:

python

datasets = read_multiple_data_sets(
    "known-vendor-file.lmd",
    ignore_offset_error=True,
)

Capture the FlowIO warning and validate event values before using them.

7. Write a New FCS File From a 2-D Array

python

from pathlib import Path

import numpy as np
from flowio import create_fcs

events = np.asarray(source_events, dtype=np.float32)
channel_names = ["FSC-A", "SSC-A", "FITC-A"]
stain_names = ["Forward scatter", "Side scatter", "CD3"]

if events.ndim != 2:
    raise ValueError("source_events must be a 2-D events x channels array")
if events.shape[1] != len(channel_names):
    raise ValueError("array columns and channel_names differ")
if len(stain_names) != len(channel_names):
    raise ValueError("PnS and PnN label counts differ")
if not np.isfinite(events).all():
    raise ValueError("decide how to handle NaN/inf before FCS export")

destination = Path("derived.fcs")
if destination.exists():
    raise FileExistsError(destination)

with destination.open("xb") as handle:
    create_fcs(
        handle,
        events.ravel(order="C"),
        channel_names,
        opt_channel_names=stain_names,
        metadata_dict={
            "date": "23-JUL-2026",
            "src": "Derived event matrix",
        },
    )

Mode "xb" prevents accidental overwrite. Use "wb" only when replacement is intentional and controlled.

8. Rewrite Metadata Without Changing Event Count

write_fcs() is useful for selected metadata replacement:

python

from pathlib import Path

from flowio import FlowData

source = Path("source.fcs")
destination = Path("deidentified.fcs")
if destination.exists():
    raise FileExistsError(destination)

flow = FlowData(source)
flow.write_fcs(
    destination,
    metadata={
        "src": "Deidentified derivative",
        "cyt": flow.text.get("cyt", "Unknown instrument"),
    },
)

Passing a metadata dictionary replaces FlowIO's default selected metadata. It does not merge with all source TEXT fields. This is useful for minimizing metadata, but inspect the reopened file rather than assuming which fields remain.

write_fcs() converts the representation to FCS 3.1 single-precision float. It is not a lossless metadata patcher. For floating-point sources it can retain encoded values while dropping PnG or timestep, so compare both preprocess=False and preprocess=True arrays after reopening.

9. Round-Trip Validation

Validate every file you create:

python

import numpy as np
from flowio import FlowData

check = FlowData("derived.fcs")

assert check.version == "3.1"
assert check.data_type == "F"
assert check.event_count == events.shape[0]
assert check.channel_count == events.shape[1]
assert check.pnn_labels == channel_names

roundtrip = check.as_array(preprocess=False)
np.testing.assert_allclose(
    roundtrip,
    events,
    rtol=1e-6,
    atol=1e-6,
)

Also verify:

Expected PnS labels
Expected custom metadata
No unintended subject identifiers
Representative channel distributions
Spillover label order, if stored
Readability in the intended downstream tool

10. Compute Inspection Statistics

The bundled inspector computes finite-value statistics only when explicitly requested:

bash

FLOWIO_SKILL_DIR="skills/flowio"  # Set to the installed skill directory.

# Metadata-scaled values
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs --stats

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

# Save JSON without overwriting an existing file
uv run --no-project --with "flowio==1.4.0" \
  python "$FLOWIO_SKILL_DIR/scripts/inspect_fcs.py" sample.fcs \
  --output sample.inspect.json

Statistics require full event loading and another float64 array. Estimate memory before using --stats on a large file. The inspector's --max-array-bytes guard limits the estimated float64 array; this estimate does not include the encoded event container.

11. Hand Off to Higher-Level Analysis

Use FlowIO to preserve file semantics, then hand the data to a cytometry analysis library when the task includes compensation or gating.

A defensible handoff records:

Whether FlowIO preprocessing was used
Whether the compensation matrix came from spill/spillover or an external control
Channel/PnN order used for matrix alignment
Transform names and parameters
Gate definitions and software version

Do not apply FlowIO preprocessing blindly before a higher-level library. Check whether that library expects encoded, compensated, or already transformed values to avoid double scaling.