flowio

Par mkurman · zorai

Analyse les fichiers FCS (Flow Cytometry Standard) v2.0-3.1. Extrait les événements sous forme de tableaux NumPy, lit les métadonnées/canaux, convertit en CSV/DataFrame, pour le prétraitement des données de cytométrie en flux.

npx skills add https://github.com/mkurman/zorai --skill flowio

FlowIO : Gestionnaire de fichiers Flow Cytometry Standard

Aperçu

FlowIO est une bibliothèque Python légère pour lire et écrire des fichiers Flow Cytometry Standard (FCS). Analysez les métadonnées FCS, extrayez les données d'événements et créez de nouveaux fichiers FCS avec des dépendances minimales. La bibliothèque supporte les versions FCS 2.0, 3.0 et 3.1, ce qui la rend idéale pour les services backend, les pipelines de données et les opérations basiques sur fichiers cytométriques.

Quand utiliser cette compétence

Cette compétence doit être utilisée quand :

  • Des fichiers FCS nécessitent un parsing ou une extraction de métadonnées
  • Des données de cytométrie de flux nécessitent une conversion en tableaux NumPy
  • Des données d'événements nécessitent une export au format FCS
  • Des fichiers FCS multi-dataset nécessitent une séparation
  • Une extraction d'information sur les canaux (scatter, fluorescence, time)
  • Une validation ou inspection de fichier cytométrique
  • Des workflows de pré-traitement avant une analyse avancée

Outils connexes : Pour l'analyse avancée de cytométrie de flux incluant la compensation, le gating et le support FlowJo/GatingML, recommandez la bibliothèque FlowKit en complément de FlowIO.

Installation

uv pip install flowio

Nécessite Python 3.9 ou ultérieur.

Démarrage rapide

Lecture basique de fichier

from flowio import FlowData

# Lire le fichier FCS
flow_data = FlowData('experiment.fcs')

# Accéder aux informations basiques
print(f"FCS Version: {flow_data.version}")
print(f"Events: {flow_data.event_count}")
print(f"Channels: {flow_data.pnn_labels}")

# Obtenir les données d'événements comme tableau NumPy
events = flow_data.as_array()  # Shape: (events, channels)

Création de fichiers FCS

import numpy as np
from flowio import create_fcs

# Préparer les données
data = np.array([[100, 200, 50], [150, 180, 60]])  # 2 events, 3 channels
channels = ['FSC-A', 'SSC-A', 'FL1-A']

# Créer un fichier FCS
create_fcs('output.fcs', data, channels)

Workflows principaux

Lecture et parsing de fichiers FCS

La classe FlowData fournit l'interface principale pour lire les fichiers FCS.

Lecture standard :

from flowio import FlowData

# Lecture basique
flow = FlowData('sample.fcs')

# Accéder aux attributs
version = flow.version              # '3.0', '3.1', etc.
event_count = flow.event_count      # Nombre d'événements
channel_count = flow.channel_count  # Nombre de canaux
pnn_labels = flow.pnn_labels        # Noms courts des canaux
pns_labels = flow.pns_labels        # Noms descriptifs des colorants

# Obtenir les données d'événements
events = flow.as_array()            # Pré-traitées (gain, log scaling appliqués)
raw_events = flow.as_array(preprocess=False)  # Données brutes

Lecture de métadonnées économe en mémoire :

Quand seules les métadonnées sont nécessaires (pas de données d'événements) :

# Parser uniquement le segment TEXT, ignorer DATA et ANALYSIS
flow = FlowData('sample.fcs', only_text=True)

# Accéder aux métadonnées
metadata = flow.text  # Dictionnaire des mots-clés du segment TEXT
print(metadata.get('$DATE'))  # Date d'acquisition
print(metadata.get('$CYT'))   # Nom de l'instrument

Gestion de fichiers problématiques :

Certains fichiers FCS ont des discordances d'offset ou des erreurs :

# Ignorer les discordances d'offset entre les sections HEADER et TEXT
flow = FlowData('problematic.fcs', ignore_offset_discrepancy=True)

# Utiliser les offsets du HEADER au lieu des offsets TEXT
flow = FlowData('problematic.fcs', use_header_offsets=True)

# Ignorer les erreurs d'offset entièrement
flow = FlowData('problematic.fcs', ignore_offset_error=True)

Exclusion de canaux nuls :

# Exclure des canaux spécifiques lors du parsing
flow = FlowData('sample.fcs', null_channel_list=['Time', 'Null'])

Extraction de métadonnées et d'informations sur les canaux

Les fichiers FCS contiennent des métadonnées riches dans le segment TEXT.

Mots-clés de métadonnées courants :

flow = FlowData('sample.fcs')

# Métadonnées au niveau du fichier
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'

# Métadonnées de canal
for i in range(flow.channel_count):
    pnn = flow.pnn_labels[i]      # Nom court (ex : 'FSC-A')
    pns = flow.pns_labels[i]      # Nom descriptif (ex : 'Forward Scatter')
    pnr = flow.pnr_values[i]      # Plage/valeur max
    print(f"Channel {i}: {pnn} ({pns}), Range: {pnr}")

Identification du type de canal :

FlowIO catégorise automatiquement les canaux :

# Obtenir les indices par type de canal
scatter_idx = flow.scatter_indices    # [0, 1] pour FSC, SSC
fluoro_idx = flow.fluoro_indices      # [2, 3, 4] pour canaux FL
time_idx = flow.time_index            # Index du canal time (ou None)

# Accéder à des types de canaux spécifiques
events = flow.as_array()
scatter_data = events[:, scatter_idx]
fluorescence_data = events[:, fluoro_idx]

Segment ANALYSIS :

Si présent, accédez aux résultats traités :

if flow.analysis:
    analysis_keywords = flow.analysis  # Dictionnaire des mots-clés ANALYSIS
    print(analysis_keywords)

Création de nouveaux fichiers FCS

Générez des fichiers FCS à partir de tableaux NumPy ou d'autres sources de données.

Création basique :

import numpy as np
from flowio import create_fcs

# Créer les données d'événements (lignes=events, colonnes=channels)
events = np.random.rand(10000, 5) * 1000

# Définir les noms des canaux
channel_names = ['FSC-A', 'SSC-A', 'FL1-A', 'FL2-A', 'Time']

# Créer un fichier FCS
create_fcs('output.fcs', events, channel_names)

Avec noms de canaux descriptifs :

# Ajouter des noms descriptifs optionnels (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)

Avec métadonnées personnalisées :

# Ajouter des métadonnées de segment TEXT
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 exporte en FCS 3.1 avec données en virgule flottante simple précision.

Export de données modifiées

Modifiez les fichiers FCS existants et réexportez-les.

Approche 1 : Utiliser la méthode write_fcs() :

from flowio import FlowData

# Lire le fichier original
flow = FlowData('original.fcs')

# Écrire avec métadonnées mises à jour
flow.write_fcs('modified.fcs', metadata={'$SRC': 'Modified data'})

Approche 2 : Extraire, modifier et recréer :

Pour modifier les données d'événements :

from flowio import FlowData, create_fcs

# Lire et extraire les données
flow = FlowData('original.fcs')
events = flow.as_array(preprocess=False)

# Modifier les données d'événements
events[:, 0] = events[:, 0] * 1.5  # Mettre à l'échelle le premier canal

# Créer un nouveau fichier FCS avec les données modifiées
create_fcs('modified.fcs',
           events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata=flow.text)

Gestion de fichiers FCS multi-dataset

Certains fichiers FCS contiennent plusieurs datasets dans un seul fichier.

Détection de fichiers multi-dataset :

from flowio import FlowData, MultipleDataSetsError

try:
    flow = FlowData('sample.fcs')
except MultipleDataSetsError:
    print("Le fichier contient plusieurs datasets")
    # Utiliser read_multiple_data_sets() à la place

Lecture de tous les datasets :

from flowio import read_multiple_data_sets

# Lire tous les datasets du fichier
datasets = read_multiple_data_sets('multi_dataset.fcs')

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

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

    # Obtenir les données d'événements pour ce dataset
    events = dataset.as_array()
    print(f"  Shape: {events.shape}")
    print(f"  Mean values: {events.mean(axis=0)}")

Lecture d'un dataset spécifique :

from flowio import FlowData

# Lire le premier dataset (nextdata_offset=0)
first_dataset = FlowData('multi.fcs', nextdata_offset=0)

# Lire le second dataset en utilisant l'offset NEXTDATA du premier
next_offset = int(first_dataset.text['$NEXTDATA'])
if next_offset > 0:
    second_dataset = FlowData('multi.fcs', nextdata_offset=next_offset)

Pré-traitement des données

FlowIO applique les transformations de pré-traitement FCS standard quand preprocess=True.

Étapes de pré-traitement :

  1. Mise à l'échelle du gain : Multiplier les valeurs par le mot-clé PnG (gain)
  2. Transformation logarithmique : Appliquer la transformation exponentielle PnE si présente
    • Formule : value = a * 10^(b * raw_value) où PnE = "a,b"
  3. Mise à l'échelle du temps : Convertir les valeurs de temps en unités appropriées

Contrôle du pré-traitement :

# Données pré-traitées (par défaut)
preprocessed = flow.as_array(preprocess=True)

# Données brutes (pas de transformations)
raw = flow.as_array(preprocess=False)

Gestion des erreurs

Gérez les exceptions FlowIO courantes de manière appropriée.

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}")
    # Essayer avec un parsing plus permissif
    flow = FlowData('sample.fcs', ignore_offset_error=True)

except DataOffsetDiscrepancyError as e:
    print(f"Offset discrepancy detected: {e}")
    # Utiliser le paramètre ignore_offset_discrepancy
    flow = FlowData('sample.fcs', ignore_offset_discrepancy=True)

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

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

Cas d'usage courants

Inspection du contenu d'un fichier FCS

Exploration rapide de la structure d'un fichier 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}")

Traitement par lot de fichiers multiples

Traitez un répertoire de fichiers FCS :

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

# Trouver tous les fichiers FCS
fcs_files = list(Path('data/').glob('*.fcs'))

# Extraire les informations de synthèse
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}")

# Créer un DataFrame de synthèse
df = pd.DataFrame(summaries)
print(df)

Conversion FCS vers CSV

Exportez les données d'événements au format CSV :

from flowio import FlowData
import pandas as pd

# Lire le fichier FCS
flow = FlowData('sample.fcs')

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

# Ajouter les métadonnées comme attributs
df.attrs['fcs_version'] = flow.version
df.attrs['instrument'] = flow.text.get('$CYT', 'Unknown')

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

Filtrage d'événements et réexport

Appliquez des filtres et enregistrez les données filtrées :

from flowio import FlowData, create_fcs
import numpy as np

# Lire le fichier original
flow = FlowData('sample.fcs')
events = flow.as_array(preprocess=False)

# Appliquer un filtre (exemple : seuil sur le premier canal)
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)}")

# Créer un nouveau fichier FCS avec les données filtrées
create_fcs('filtered.fcs',
           filtered_events,
           flow.pnn_labels,
           opt_channel_names=flow.pns_labels,
           metadata={**flow.text, '$SRC': 'Filtered data'})

Extraction de canaux spécifiques

Extrayez et traitez des canaux spécifiques :

from flowio import FlowData
import numpy as np

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

# Extraire uniquement les canaux de fluorescence
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}")

# Calculer les statistiques par canal
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}")

Bonnes pratiques

  1. Efficacité mémoire : Utilisez only_text=True quand les données d'événements ne sont pas nécessaires
  2. Gestion des erreurs : Enveloppez les opérations sur fichiers dans des blocs try-except pour un code robuste
  3. Détection multi-dataset : Vérifiez MultipleDataSetsError et utilisez la fonction appropriée
  4. Contrôle du pré-traitement : Définissez explicitement le paramètre preprocess selon les besoins d'analyse
  5. Problèmes d'offset : Si le parsing échoue, essayez le paramètre ignore_offset_discrepancy=True
  6. Validation des canaux : Vérifiez que les nombres et noms de canaux correspondent aux attentes avant le traitement
  7. Préservation des métadonnées : Quand vous modifiez des fichiers, préservez les mots-clés du segment TEXT original

Sujets avancés

Compréhension de la structure d'un fichier FCS

Les fichiers FCS se composent de quatre segments :

  1. HEADER : Version FCS et offsets d'octets pour les autres segments
  2. TEXT : Paires clé-valeur de métadonnées (séparées par délimiteur)
  3. DATA : Données brutes d'événements (format binaire/float/ASCII)
  4. ANALYSIS (optionnel) : Résultats du traitement des données

Accédez à ces segments via les attributs FlowData :

  • flow.header - segment HEADER
  • flow.text - mots-clés du segment TEXT
  • flow.events - segment DATA (comme octets)
  • flow.analysis - mots-clés du segment ANALYSIS (si présent)

Référence API détaillée

Pour une documentation complète incluant tous les paramètres, méthodes, exceptions et référence des mots-clés FCS, consultez le fichier de référence détaillée :

Lire : references/api_reference.md

La référence inclut :

  • Documentation complète de la classe FlowData
  • Toutes les fonctions utilitaires (read_multiple_data_sets, create_fcs)
  • Classes d'exception et gestion
  • Détails de la structure des fichiers FCS
  • Mots-clés courants du segment TEXT
  • Workflows d'exemples étendus

Quand vous travaillez avec des opérations FCS complexes ou rencontrez des formats de fichier inusités, chargez cette référence pour des conseils détaillés.

Notes d'intégration

Tableaux NumPy : Toutes les données d'événements sont renvoyées comme ndarrays NumPy avec la forme (events, channels)

DataFrames Pandas : Convertissez facilement en DataFrames pour l'analyse :

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

Intégration FlowKit : Pour une analyse avancée (compensation, gating, support FlowJo), utilisez la bibliothèque FlowKit qui s'appuie sur les capacités de parsing de FlowIO

Applications Web : Les dépendances minimales de FlowIO le rendent idéal pour les services backend web traitant les uploads FCS

Dépannage

Problème : « Offset discrepancy error » Solution : Utilisez le paramètre ignore_offset_discrepancy=True

Problème : « Multiple datasets error » Solution : Utilisez la fonction read_multiple_data_sets() au lieu du constructeur FlowData

Problème : Manque de mémoire avec de gros fichiers Solution : Utilisez only_text=True pour les opérations portant uniquement sur les métadonnées, ou traitez les événements par chunks

Problème : Nombres de canaux inattendus Solution : Vérifiez les canaux nuls ; utilisez le paramètre null_channel_list pour les exclure

Problème : Impossible de modifier les données d'événements sur place Solution : FlowIO ne supporte pas la modification directe ; extrayez les données, modifiez-les, puis utilisez create_fcs() pour sauvegarder

Résumé

FlowIO fournit les capacités essentielles de gestion des fichiers FCS pour les workflows de cytométrie de flux. Utilisez-le pour le parsing, l'extraction de métadonnées et la création de fichiers. Pour les opérations simples sur fichiers et l'extraction de données, FlowIO est suffisant. Pour une analyse complexe incluant la compensation et le gating, intégrez FlowKit ou d'autres outils spécialisés.

Skills similaires

imaging-data-commons
mkurman / zorai

Interroger et télécharger des données d'imagerie médicale publiques depuis NCI IDC.

309
pyhealth
mkurman / zorai

Développer des modèles d'IA clinique sur données de santé avec PyHealth.

309
zarr-python
mkurman / zorai

Stocker et manipuler efficacement de grands tableaux N-dimensionnels avec chunking et compression.

309
dask
mkurman / zorai

Traiter des données massives en parallèle avec Dask sur CPU unique ou cluster distribué.

309
scikit-learn
mkurman / zorai

Construire et évaluer des pipelines de machine learning classique avec scikit-learn.

309