FlowIO Workflows

These patterns target flowio==1.4.0. Examples below were exercised with synthetic files; placeholder filenames and source_events must be supplied for real data. Keep raw files immutable and write derived outputs separately.

1. Inventory a File Without Loading Events

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

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:

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

from flowio import FlowData

flow = FlowData("sample.fcs")

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

# Decoded DATA values without gain/log/time scaling (integer masks remain).
encoded = flow.as_array(preprocess=False)

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

Check shape against both declared counts: FlowIO infers the number of rows from DATA rather than validating $TOT. Integer DATA has already been masked to its PnR range even with preprocess=False. 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. The tested pandas 3.0.6 stack is for standalone FlowIO. FlowKit has separate dependency constraints, so retain the compatible pandas version in its environment.

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

import pandas as pd
from flowio import FlowData


def unique_labels(labels):
    bases = [value.strip() or f"channel_{i + 1}" for i, value in enumerate(labels)]
    reserved = set(bases)  # Do not steal a later detector's existing name.
    used = set()
    result = []
    for base in bases:
        name = base
        if name in used:
            suffix = 2
            while f"{base}__{suffix}" in reserved or f"{base}__{suffix}" in used:
                suffix += 1
            name = f"{base}__{suffix}"
        used.add(name)
        result.append(name)
    return result


flow = FlowData("sample.fcs")
columns = unique_labels(flow.pnn_labels)
assert len(columns) == len(set(columns))
values = flow.as_array(preprocess=True)
if values.shape != (flow.event_count, flow.channel_count):
    raise ValueError("DATA shape disagrees with declared event/channel counts")
frame = pd.DataFrame(
    values,
    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,
    }
)

For example, ["A", "A", "A__2"] becomes ["A", "A__3", "A__2"]. The position-to-source-label mapping is authoritative; never use renamed export labels to align a spillover matrix.

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:

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)
if values.shape != (flow.event_count, flow.channel_count):
    raise ValueError("DATA shape disagrees with declared event/channel counts")

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": "decoded DATA values; preprocess=False; integer range masks retained",
    "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:

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.

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:

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

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": "30-SEP-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:

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:

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:

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 decoded event container, temporary arrays, or metadata. It uses declared counts before loading; inconsistent loaded counts are rejected before the float64 allocation. A resource-limited process is needed for a hard peak-memory cap.

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.