Source code for sfepy.base.log_plotter

"""
Plotting class to be used by Log.
"""
import time
import numpy as nm

from sfepy.base.base import Output, Struct

[docs] def draw_data(ax, xdata, ydata, label, plot_kwargs, swap_axes=False): """ Draw log data to a given axes, obeying `swap_axes`. """ def _update_plot_kwargs(lines): plot_kwargs['color'] = lines[0].get_color() alpha = lines[0].get_alpha() plot_kwargs['alpha'] = 0.5 if alpha is None else 0.5 * alpha plot_kwargs = plot_kwargs.copy() if not swap_axes: if nm.isrealobj(ydata): ax.plot(xdata, ydata, label=label, **plot_kwargs) else: lines = ax.plot(xdata, ydata.real, label='Re ' + label, **plot_kwargs) _update_plot_kwargs(lines) ax.plot(xdata, ydata.imag, label='Im ' + label, **plot_kwargs) else: if nm.isrealobj(ydata): ax.plot(ydata, xdata, label=label, **plot_kwargs) else: lines = ax.plot(ydata.real, xdata, label='Re ' + label, **plot_kwargs) _update_plot_kwargs(lines) ax.plot(ydata.imag, xdata, label='Im ' + label, **plot_kwargs)
[docs] class LogPlotter(Struct): """ LogPlotter to be used by :class:`sfepy.base.log.Log`. """ output = Output('plotter:') output = staticmethod(output) def __init__(self, aggregate=100, sleep=1.0): Struct.__init__(self, aggregate=aggregate, sleep=sleep, xdata={}, ydata={}, plot_kwargs={}, clear_axes={}, show_legends=False)
[docs] def process_command(self, command): self.output(command[0]) if command[0] == 'plot': ig, ip, xd, yd = command[1:] xdata = self.xdata.setdefault((ig, ip), []) ydata = self.ydata.setdefault((ig, ip), []) xdata.append(xd) ydata.append(yd) elif command[0] == 'vline': ig, x, kwargs = command[1:] self.vlines[ig].append((x, kwargs)) elif command[0] == 'clear': ig = command[1] self.clear_axes[ig] = True elif command[0] == 'legends': self.show_legends = True elif command[0] == 'add_axis': ig, names, yscale, xlabel, ylabel, plot_kwargs = command[1:] self.data_names[ig] = names self.yscales[ig] = yscale self.xlabels[ig] = xlabel self.ylabels[ig] = ylabel self.plot_kwargs[ig] = plot_kwargs self.n_gr = len(self.data_names) self.make_axes() elif command[0] == 'save': self.fig.savefig(command[1]) self.pipe.send(True) # Acknowledge save.
[docs] def apply_commands(self): from matplotlib.ticker import LogLocator, AutoLocator for key in sorted(self.ydata.keys()): ig, ip = key xdata = nm.array(self.xdata[(ig, ip)]) ydata = nm.array(self.ydata[(ig, ip)]) ax = self.ax[ig] if self.clear_axes[ig]: ax.cla() self.clear_axes[ig] = False ax.set_yscale(self.yscales[ig]) ax.yaxis.grid(True) draw_data(ax, nm.array(xdata), nm.array(ydata), self.data_names[ig][ip], self.plot_kwargs[ig][ip]) if self.yscales[ig] == 'log': ymajor_formatter = ax.yaxis.get_major_formatter() yminor_locator = LogLocator() else: yminor_locator = AutoLocator() self.ax[ig].yaxis.set_minor_locator(yminor_locator) for ig, ax in enumerate(self.ax): if self.show_legends: try: ax.legend() except: pass if self.xlabels[ig]: ax.set_xlabel(self.xlabels[ig]) if self.ylabels[ig]: ax.set_ylabel(self.ylabels[ig]) for x, kwargs in self.vlines[ig]: ax.axvline(x, **kwargs) try: self.plt.tight_layout(pad=0.5) except: pass
[docs] def terminate(self): if self.ii: self.output('processed %d commands' % self.ii) self.output('ended.') self.plt.close('all')
[docs] def poll_draw(self): while 1: self.ii = 0 while 1: if not self.pipe.poll(): break command = self.pipe.recv() can_break = False if command is None: self.terminate() return False elif command[0] == 'continue': can_break = True else: self.process_command(command) if (self.ii >= self.aggregate) and can_break: break self.ii += 1 if self.ii: self.apply_commands() self.fig.canvas.draw() self.output('processed %d commands' % self.ii) self.fig.canvas.flush_events() time.sleep(self.sleep) return True
[docs] def make_axes(self): from sfepy.linalg import cycle self.fig.clf() self.ax = [] n_col = min(5.0, nm.fix(nm.sqrt(self.n_gr))) if int(n_col) == 0: n_row = 0 else: n_row = int(nm.ceil(self.n_gr / n_col)) n_col = int(n_col) for ii, (ir, ic) in enumerate(cycle((n_col, n_row))): if ii == self.n_gr: break self.ax.append(self.fig.add_subplot(n_row, n_col, ii + 1)) self.vlines.setdefault(ii, [])
def __call__(self, pipe, log_file, data_names, yscales, xlabels, ylabels, plot_kwargs): """ Sets-up the non-blocking plotting window and calls self.poll_draw() that does the actual plotting, taking commands out of `pipe`. """ import matplotlib.pyplot as plt self.plt = plt self.output.set_output(filename=log_file) self.output('starting plotter...') self.pipe = pipe self.data_names = data_names self.yscales = yscales self.xlabels = xlabels self.ylabels = ylabels self.plot_kwargs = plot_kwargs self.n_gr = len(data_names) self.vlines = {} self.fig = self.plt.figure() self.make_axes() self.output('...done') self.plt.show(block=False) self.poll_draw()