Source code for cytoflow.operations.xform_stat

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

# (c) Massachusetts Institute of Technology 2015-2018
# (c) Brian Teague 2018-2019
# 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 <>.


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

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

import cytoflow.utility as util

from .i_operation import IOperation

[docs]@provides(IOperation) class TransformStatisticOp(HasStrictTraits): """ Apply a function to a statistic, creating a new statistic. The function can be applied to the entire statistic, or it can be applied individually to groups of the statistic. The function should take a :class:`pandas.Series` as its only argument. Return type is arbitrary, but a to be used with the rest of :class:`cytoflow` it should probably be a numeric type or an iterable of numeric types. As a special case, if the function returns a :class:`pandas.Series` *with the same index that it was passed*, it is interpreted as a transformation. The resulting statistic will have the same length, index names and index levels as the original statistic. Attributes ---------- name : Str The operation name. Becomes the first element in the :attr:`~Experiment.statistics` key tuple. statistic : Tuple(Str, Str) The statistic to apply the function to. function : Callable The function used to transform the statistic. :attr:`function` must take a :class:`pandas.Series` as its only parameter. The return type is arbitrary, but to work with the rest of :class:`cytoflow` it should probably be a numeric type or an iterable of numeric types.. If :attr:`statistic_name` is unset, the name of the function becomes the second in element in the :attr:`~Experiment.statistics` key tuple. statistic_name : Str The name of the function; if present, becomes the second element in the :attr:`~Experiment.statistics` key tuple. by : List(Str) A list of metadata attributes to aggregate the input statistic before applying the function. For example, if the statistic has two indices ``Time`` and ``Dox``, setting ``by = ["Time", "Dox"]`` will apply :attr:`function` separately to each subset of the data with a unique combination of ``Time`` and ``Dox``. fill : Any (default = 0) Value to use in the statistic if a slice of the data is empty. Examples -------- >>> stats_op = ChannelStatisticOp(name = "Mean", ... channel = "Y2-A", ... function = np.mean, ... by = ["Dox"]) >>> ex2 = stats_op.apply(ex) >>> log_op = TransformStatisticOp(name = "LogMean", ... statistic = ("Mean", "mean"), ... function = np.log) >>> ex3 = log_op.apply(ex2) """ id = Constant('') friendly_id = Constant("Transform Statistic") name = CStr statistic = Tuple(Str, Str) function = Callable statistic_name = Str by = List(Str) fill = Any(0)
[docs] def apply(self, experiment): """ Applies :attr:`function` to a statistic. Parameters ---------- experiment : Experiment The experiment to apply the operation to Returns ------- Experiment The same as the old experiment, but with a new statistic that results from applying :attr:`function` to the statistic specified in :attr:`statistic`. """ if experiment is None: raise util.CytoflowOpError('experiment', "Must specify an experiment") 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.statistic: raise util.CytoflowViewError('statistic', "Statistic not set") if self.statistic not in experiment.statistics: raise util.CytoflowViewError('statistic', "Can't find the statistic {} in the experiment" .format(self.statistic)) else: stat = experiment.statistics[self.statistic] if not self.function: raise util.CytoflowOpError('function', "Must specify a function") 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)) for b in if b not in stat.index.names: raise util.CytoflowOpError('by', "{} is not a statistic index; " " must be one of {}" .format(b, stat.index.names)) data = stat.reset_index() if idx = pd.MultiIndex.from_product([data[x].unique() for x in], names = else: idx = stat.index.copy() new_stat = pd.Series(data = self.fill, index = idx, dtype = np.dtype(object)).sort_index() if for group in data[].itertuples(index = False, name = None): if isinstance(stat.index, pd.MultiIndex): s = stat.xs(group, level =, drop_level = False) else: s = stat.loc[list(group)] if len(s) == 0: continue try: new_stat[group] = self.function(s) 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 np.any(np.isnan(new_stat.loc[group])): warn("Category {} returned {}".format(group, new_stat.loc[group]), util.CytoflowOpWarning) else: new_stat = self.function(stat) if not isinstance(new_stat, pd.Series): raise util.CytoflowOpError('by', "Transform function {} does not return a Series; " "in this case, you must set 'by'" .format(self.function)) = "{} : {}".format(stat_name[0], stat_name[1]) matched_series = True for group in data[].itertuples(index = False, name = None): if isinstance(stat.index, pd.MultiIndex): s = stat.xs(group, level =, drop_level = False) else: s = stat.loc[list(group)] if isinstance(new_stat.loc[group], pd.Series) and \ s.index.equals(new_stat.loc[group].index): pass else: matched_series = False break if matched_series and len( > 0: new_stat = pd.concat(new_stat.values) # try to convert to numeric, but if there are non-numeric bits ignore new_stat = pd.to_numeric(new_stat, errors = 'ignore') # sort the index, for performance new_stat = new_stat.sort_index() new_experiment = experiment.clone() new_experiment.history.append(self.clone_traits(transient = lambda t: True)) if self.statistic_name: new_experiment.statistics[(, self.statistic_name)] = new_stat else: new_experiment.statistics[(, self.function.__name__)] = new_stat return new_experiment