Source code for mesmerize.pyqtgraphCore.flowchart.library.Display

# -*- coding: utf-8 -*-
from ... import PlotDataItem, ComboBox
from ....analysis import simple_plot_window
from .common import *
import numpy as np
import pandas as pd
from ....plotting.widgets import HeatmapTracerWidget, ScatterPlotWidget, \
    BeeswarmPlotWindow, ProportionsWidget, CrossCorrelationWidget, SpaceMapWidget
from ....plotting.variants.timeseries import TimeseriesPlot
from graphviz import Digraph
from ....common.utils import make_workdir
import os
import subprocess
from random import randrange


[docs]class Plot(CtrlNode): """Plot curves and/or scatter points""" nodeName = 'Plot' uiTemplate = [('Show', 'check', {'checked': True}), ('Apply', 'check', {'checked': False, 'applyBox': True}), ('data_column', 'combo', {}) ] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}})#, 'Out': {'io': 'out'}}) self.trans_ids = [] self.plot_widget = simple_plot_window.PlotWindow() self.ctrls['Apply'].clicked.connect(self.update) self.ctrls['Show'].clicked.connect(self.plot_widget.setVisible) def process(self, **kwargs): transmissions = kwargs['In'] if not len(transmissions) > 0: raise Exception('No incoming transmissions') transmissions = merge_transmissions(transmissions) columns = pd.concat([t.df for t in transmissions]).columns dcols = organize_dataframe_columns(columns)[0] self.ctrls['data_column'].setItems(dcols) if self.ctrls['Apply'].isChecked() is False: return data_column = self.ctrls['data_column'].currentText() self.plot_widget.graphicsView.clear() srcs = [] plots = [] colors = ['m', 'r', 'y', 'g', 'c', 'b'] ci = 0 for t in transmissions: if t is None: srcs.append(str(type(None))) continue if id(t) in self.trans_ids: continue for ix, r in t.df.iterrows(): if data_column in r.index: data = r[data_column] # elif 'peak_curve' in r.index: # data = r['peak_curve'] else: srcs.append('No data found in chosen column!') continue if data is None: srcs.append('No data found in chosen column!') continue plot = PlotDataItem() try: plot.setData(data) # plot.setData(data / np.maximum(np.min(data), 0.0000000001)) plot.setPen(colors[ci]) except Exception as e: srcs.append('Plotting error: ' + str(e)) continue self.plot_widget.graphicsView.addItem(plot) ci += 1 self.plot_widget.set_history_widget(srcs)
[docs]class FrequencyDomainMagnitude(CtrlNode): """Plot Frequency vs. Frequency Domain Magnitude""" nodeName = 'FDM Plot' uiTemplate = [('Show', 'check', {'checked': True}), ('Apply', 'check', {'checked': True, 'applyBox': True}) ]
[docs]class Timeseries(CtrlNode): """Seaborn Timeseries plot""" nodeName = 'Timeseries' uiTemplate = [('data_column', 'combo', {}), ('err_style', 'combo', {'items': ['band', 'bars']}), ('Show', 'button', {'text': 'Show'}), ('Apply', 'check', {'checked': False, 'applyBox': True}), ] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.ctrls['Apply'].clicked.connect(self.update) self.plot_widget = TimeseriesPlot() self.ctrls['Show'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): transmissions = kwargs['In'] self.transmissions_list = merge_transmissions(transmissions) self.t = Transmission.merge(self.transmissions_list) columns = self.t.df.columns.to_list() self.ctrls['data_column'].setItems(columns) if not self.ctrls['Apply'].isChecked(): return data_column = self.ctrls['data_column'].currentText() data = np.vstack(self.t.df[data_column].values) es = self.ctrls['err_style'].currentText() self.plot_widget.set(data=data, err_style=es)
[docs]class Heatmap(CtrlNode): """Stack 1-D arrays and plot visually like a heatmap""" nodeName = "Heatmap" uiTemplate = [('Show', 'button', {'text': 'Show GUI'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}})#, 'Out': {'io': 'out'}}) self.plot_widget = HeatmapTracerWidget() self.ctrls['Show'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): self.transmissions = kwargs['In'] self.transmissions_list = merge_transmissions(self.transmissions) if len(self.transmissions_list) == 1: self.t = self.transmissions_list[0] else: self.t = Transmission.merge(self.transmissions_list) self.plot_widget.set_input(self.t)
[docs]class ScatterPlot(CtrlNode): """Scatter Plot, useful for visualizing transformed data and clusters""" nodeName = "ScatterPlot" uiTemplate = [('Show', 'button', {'text': 'Show'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.plot_widget = ScatterPlotWidget() self.ctrls['Show'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): transmissions = kwargs['In'] tran_list = merge_transmissions(transmissions) merged = Transmission.merge(tran_list) self.plot_widget.set_input(merged)
[docs]class BeeswarmPlots(CtrlNode): """Beeswarm and Violin plots""" nodeName = 'BeeswarmPlots' uiTemplate = [('Apply', 'check', {'checked': False, 'applyBox': True}), ('ShowGUI', 'button', {'text': 'OpenGUI'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.plot_widget = None self.ctrls['ShowGUI'].clicked.connect(self._open_plot_gui) def process(self, **kwargs): if (self.ctrls['Apply'].isChecked() is False) or self.plot_widget is None: return transmissions = kwargs['In'] transmissions_list = merge_transmissions(transmissions) self.plot_widget.update_input_transmissions(transmissions_list) def _open_plot_gui(self): if self.plot_widget is None: self.plot_widget = BeeswarmPlotWindow(parent=None) self.plot_widget.show()
[docs]class Proportions(CtrlNode): """Plot proportions of one categorical column vs another""" nodeName = 'Proportions' uiTemplate = [('show gui', 'button', {'text': 'Show Gui'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.plot_widget = ProportionsWidget() self.ctrls['show gui'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): self.transmissions = kwargs['In'] self.transmissions_list = merge_transmissions(self.transmissions) self.t = Transmission.merge(self.transmissions_list) self.plot_widget.set_input(self.t)
[docs]class CrossCorr(CtrlNode): """Cross Correlation""" nodeName = 'CrossCorr' uiTemplate = [('ShowGUI', 'button', {'text': 'Show Gui'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.plot_widget = CrossCorrelationWidget() self.ctrls['ShowGUI'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): self.transmissions = kwargs['In'] self.transmissions_list = merge_transmissions(self.transmissions) self.t = Transmission.merge(self.transmissions_list) self.plot_widget.set_input(self.t)
[docs]class SpaceMap(CtrlNode): """Visualize spatial maps of a categorical variable""" nodeName = 'SpaceMap' uiTemplate = [('ShowGUI', 'button', {'text': 'Show GUI'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.plot_widget =SpaceMapWidget() self.ctrls['ShowGUI'].clicked.connect(self.plot_widget.show) def process(self, **kwargs): self.transmissions = kwargs['In'] self.transmissions = merge_transmissions(self.transmissions) self.t = Transmission.merge(self.transmissions) self.plot_widget.set_input(self.t)
[docs]class AnalysisGraph(CtrlNode): """Graph of the analysis log""" nodeName = 'AnalysisGraph' uiTemplate = [('data_blocks', 'list_widget', {'toolTip': 'Double click a data block to open the pdf'})] def __init__(self, name): CtrlNode.__init__(self, name, terminals={'In': {'io': 'in', 'multi': True}}) self.ctrls['data_blocks'].itemDoubleClicked.connect(self.open_file) def process(self, **kwargs): self.transmissions = kwargs['In'] self.transmissions_list = merge_transmissions(self.transmissions) self.t = Transmission.merge(self.transmissions_list) workdir = make_workdir(f'analysis_graphs{randrange(1000, 9999)}') self.fnames = [] for db in self.t.history_trace.data_blocks: fname = os.path.join(workdir, f"{db}") g = Digraph('G', filename=fname, format='pdf', node_attr={'shape': 'record'}) db_history = self.t.history_trace.get_data_block_history(db) o_encounters = [] o_n_l = [] for op in db_history: o = list(op.keys())[0] params = op[o] params = cleanup_params(o, params) s = [] for k in params.keys(): p = str(params[k]).replace("{", "\\n --\> ").replace("}", "").replace("|", "\|") # .replace("[", " ").replace("(", " ").replace(")", " ") s.append(f"{k}: {p}") o_n = f"{o}.{o_encounters.count(o)}" o_n_l.append(o_n) s = '{ ' + o_n + ' | ' + '\\n'.join(s) + ' }' g.node(o_n, s) o_encounters.append(o) es = [] for i in range(1, len(o_n_l)): es.append((o_n_l[i - 1], o_n_l[i])) g.edges(es) g.render(view=False) self.fnames.append(fname + '.pdf') self.ctrls['data_blocks'].setItems(list(map(str, range(len(self.fnames))))) def open_file(self, item): ix = self.ctrls['data_blocks'].currentRow() subprocess.call(["xdg-open", self.fnames[ix]])
def cleanup_params(operation, params): log = params.copy() if operation == 'rfft': log.pop('frequencies') elif operation =='fcluster': log.pop('linkage_matrix') return log