Source code for cytoflow.operations.frame_stat

#!/usr/bin/env python3.8
# coding: latin-1

# (c) Massachusetts Institute of Technology 2015-2018
# (c) Brian Teague 2018-2022
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 2 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <>.


The `frame_stat` module contains one class:

`FrameStatisticOp` -- applies a function to subsets of a data set,
and adds the resulting statistic to the `Experiment`.  Unlike
`ChannelStatisticOp`, which operates on a single channel, this operation
operates on entire `pandas.DataFrame`.

from warnings import warn
import pandas as pd
import numpy as np

from traits.api import (HasStrictTraits, Str, List, Constant, provides, 
                        Callable, Any)

import cytoflow.utility as util

from .i_operation import IOperation

[docs]@provides(IOperation) class FrameStatisticOp(HasStrictTraits): """ Apply a function to subsets of a data set, and add it as a statistic to the experiment. The `apply` function groups the data by the variables in `by`, then applies the `function` callable to each `pandas.DataFrame` subset. The callable should take a `pandas.DataFrame` as its only parameter. The return type is arbitrary, but to be used with the rest of `cytoflow` it should probably be a numeric type or an iterable of numeric types. Attributes ---------- name : Str The operation name. Becomes the first element in the `Experiment.statistics` key tuple. function : Callable The function used to compute the statistic. Must take a `pandas.DataFrame` as its only argument. The return type is arbitrary, but to be used with the rest of `cytoflow` it should probably be a numeric type or an iterable of numeric types. If `statistic_name` is unset, the name of the function becomes the second in element in the `Experiment.statistics` key tuple. statistic_name : Str The name of the function; if present, becomes the second element in the `Experiment.statistics` key tuple. Particularly useful if `function` is a lambda. by : List(Str) A list of metadata attributes to aggregate the data before applying the function. For example, if the experiment has two pieces of metadata, ``Time`` and ``Dox``, setting ``by = ["Time", "Dox"]`` will apply `function` separately to each subset of the data with a unique combination of ``Time`` and ``Dox``. subset : Str A Python expression sent to Experiment.query() to subset the data before computing the statistic. fill : Any (default = 0) The value to use in the statistic if a slice of the data is empty. Examples -------- >>> stats_op = FrameStatisticOp(name = "ByDox", ... function = lambda x: np.mean(x["FITC-A"], ... statistic_name = "Mean", ... by = ["Dox"]) >>> ex2 = stats_op.apply(ex) """ id = Constant('') friendly_id = Constant("Statistics") name = Str function = Callable statistic_name = Str by = List(Str) subset = Str fill = Any(0)
[docs] def apply(self, experiment): if experiment is None: raise util.CytoflowOpError('experiment', "No experiment specified") if not raise util.CytoflowOpError('name', "Must specify a name") if != util.sanitize_identifier( raise util.CytoflowOpError('name', "Name can only contain letters, numbers and underscores." .format( if not self.function: raise util.CytoflowOpError('function', "Must specify a function") if not raise util.CytoflowOpError('by', "Must specify some grouping conditions " "in 'by'") stat_name = (, self.statistic_name) \ if self.statistic_name \ else (, self.function.__name__) if stat_name in experiment.statistics: raise util.CytoflowOpError('name', "{} is already in the experiment's statistics" .format(stat_name)) new_experiment = experiment.clone(deep = False) if self.subset: try: experiment = experiment.query(self.subset) except Exception as e: raise util.CytoflowOpError('subset', "Subset string '{0}' isn't valid" .format(self.subset)) from e if len(experiment) == 0: raise util.CytoflowOpError('subset', "Subset string '{0}' returned no events" .format(self.subset)) for b in if b not in experiment.conditions: raise util.CytoflowOpError('by', "Aggregation metadata {} not found, " " must be one of {}" .format(b, experiment.conditions)) unique =[b].unique() if len(unique) == 1: warn("Only one category for {}".format(b), util.CytoflowOpWarning) groupby = for group, data_subset in groupby: if len(data_subset) == 0: warn("Group {} had no data" .format(group), util.CytoflowOpWarning) # this shouldn't be necessary, but see pandas bug #38053 if len( == 1: idx = pd.Index(experiment[[0]].unique(), name =[0]) else: idx = pd.MultiIndex.from_product([experiment[x].unique() for x in], names = stat = pd.Series(data = self.fill, index = idx, name = "{} : {}".format(stat_name[0], stat_name[1]), dtype = np.dtype(object)).sort_index() for group, data_subset in groupby: if len(data_subset) == 0: continue try: v = self.function(data_subset)[group] = v except Exception as e: raise util.CytoflowOpError('function', "Your function threw an error in group {}" .format(group)) from e # check for, and warn about, NaNs. if pd.Series(stat.loc[group]).isna().any(): warn("Category {} returned {}".format(group, stat.loc[group]), util.CytoflowOpWarning) # try to convert to numeric, but if there are non-numeric bits ignore stat = pd.to_numeric(stat, errors = 'ignore') new_experiment.history.append(self.clone_traits(transient = lambda t: True)) new_experiment.statistics[stat_name] = stat return new_experiment