Source code for cytoflowgui.op_plugins.flowpeaks

#!/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 <>.

FlowPeaks Clustering

This module uses the **flowPeaks** algorithm to assign events to clusters in
an unsupervized manner.
.. object:: Name
    The operation name; determines the name of the new metadata
.. object:: X Channel, Y Channel
    The channels to apply the mixture model to.

.. object:: X Scale, Y Scale 

    Re-scale the data in **Channel** before fitting. 

.. object:: h, h0

    Scalar values that control the smoothness of the estimated distribution.
    Increasing **h** makes it "rougher," while increasing **h0** makes it
.. object:: tol

    How readily should clusters be merged?  Must be between 0 and 1.
.. object:: Merge Distance

    How far apart can clusters be before they are merged?
.. object:: By 

    A list of metadata attributes to aggregate the data before estimating
    the model.  For example, if the experiment has two pieces of metadata,
    ``Time`` and ``Dox``, setting **By** to ``["Time", "Dox"]`` will fit 
    the model separately to each subset of the data with a unique combination of
    ``Time`` and ``Dox``.

.. plot::
   :include-source: False

    import cytoflow as flow
    import_op = flow.ImportOp()
    import_op.tubes = [flow.Tube(file = "Plate01/RFP_Well_A3.fcs",
                                 conditions = {'Dox' : 10.0}),
                       flow.Tube(file = "Plate01/CFP_Well_A4.fcs",
                                 conditions = {'Dox' : 1.0})]
    import_op.conditions = {'Dox' : 'float'}
    ex = import_op.apply()

    fp_op = flow.FlowPeaksOp(name = 'Flow',
                             channels = ['V2-A', 'Y2-A'],
                             scale = {'V2-A' : 'log',
                                      'Y2-A' : 'log'})
    ex2 = fp_op.apply(ex)
    fp_op.default_view(density = True).plot(ex2)


from traits.api import provides, List
from traitsui.api import (View, Item, EnumEditor, VGroup, TextEditor, 
                          CheckListEditor, ButtonEditor)
from envisage.api import Plugin
from pyface.api import ImageResource

from ..view_plugins import ViewHandler
from ..view_plugins.scatterplot import ScatterplotParamsHandler
from ..view_plugins.density import DensityParamsHandler
from ..editors import SubsetListEditor, ColorTextEditor, InstanceHandlerEditor
from ..workflow.operations import FlowPeaksWorkflowOp, FlowPeaksWorkflowView
from ..subset_controllers import subset_handler_factory

from .i_op_plugin import IOperationPlugin, OP_PLUGIN_EXT
from .op_plugin_base import OpHandler, shared_op_traits_view, PluginHelpMixin

[docs]class FlowPeaksHandler(OpHandler): operation_traits_view = \ View(Item('name', editor = TextEditor(auto_set = False, placeholder = "None")), Item('xchannel', editor=EnumEditor(name='context_handler.previous_channels'), label = "X Channel"), Item('ychannel', editor=EnumEditor(name='context_handler.previous_channels'), label = "Y Channel"), Item('xscale', label = "X Scale"), Item('yscale', label = "Y Scale"), VGroup( Item('h', editor = TextEditor(auto_set = False, evaluate = float, format_func = lambda x: "" if x is None else str(x), placeholder = "None")), Item('h0', editor = TextEditor(auto_set = False, evaluate = float, format_func = lambda x: "" if x is None else str(x), placeholder = "None")), Item('tol', editor = TextEditor(auto_set = False, evaluate = float, format_func = lambda x: "" if x is None else str(x), placeholder = "None")), Item('merge_dist', editor = TextEditor(auto_set = False, evaluate = float, format_func = lambda x: "" if x is None else str(x), placeholder = "None"), label = "Merge\nDistance"), Item('by', editor = CheckListEditor(cols = 2, name = 'context_handler.previous_conditions_names'), label = 'Group\nEstimates\nBy', style = 'custom'), VGroup(Item('subset_list', show_label = False, editor = SubsetListEditor(conditions = "context_handler.previous_conditions", editor = InstanceHandlerEditor(view = 'subset_view', handler_factory = subset_handler_factory))), label = "Subset", show_border = False, show_labels = False), Item('do_estimate', editor = ButtonEditor(value = True, label = "Estimate!"), show_label = False), label = "Estimation parameters", show_border = False), shared_op_traits_view)
[docs]class FlowPeaksViewHandler(ViewHandler): view_traits_view = \ View(VGroup( VGroup(Item('xchannel', style = 'readonly'), Item('ychannel', style = 'readonly'), Item('show_density', label = "Show density plot?"), label = "Flow Peaks Default Plot", show_border = False)), Item('context.view_warning', resizable = True, visible_when = 'context.view_warning', editor = ColorTextEditor(foreground_color = "#000000", background_color = "#ffff99")), Item('context.view_error', resizable = True, visible_when = 'context.view_error', editor = ColorTextEditor(foreground_color = "#000000", background_color = "#ff9191"))) view_params_view = View(Item('scatterplot_plot_params', editor = InstanceHandlerEditor(view = 'view_params_view', handler_factory = ScatterplotParamsHandler), style = 'custom', show_label = False, visible_when = 'not object.show_density'), Item('density_plot_params', editor = InstanceHandlerEditor(view = 'view_params_view', handler_factory = DensityParamsHandler), style = 'custom', show_label = False, visible_when = 'object.show_density'))
[docs]@provides(IOperationPlugin) class FlowPeaksPlugin(Plugin, PluginHelpMixin): id = '' operation_id = '' view_id = '' short_name = "Flow Peaks" menu_group = "Gates"
[docs] def get_operation(self): return FlowPeaksWorkflowOp()
[docs] def get_handler(self, model, context): if isinstance(model, FlowPeaksWorkflowOp): return FlowPeaksHandler(model = model, context = context) elif isinstance(model, FlowPeaksWorkflowView): return FlowPeaksViewHandler(model = model, context = context)
[docs] def get_icon(self): return ImageResource('flowpeaks')
plugin = List(contributes_to = OP_PLUGIN_EXT) def _plugin_default(self): return [self]