Source code for sfepy.base.log_plotter
"""
Plotting class to be used by Log.
"""
import numpy as nm
from sfepy.base.base import Output, Struct
[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):
Struct.__init__(self, aggregate=aggregate)
[docs] def process_command(self, command):
from matplotlib.ticker import LogLocator, AutoLocator
self.output(command[0])
if command[0] == 'ig':
self.ig = command[1]
elif command[0] == 'plot':
xdata, ydata = command[1:]
ig = self.ig
ax = self.ax[ig]
ax.set_yscale(self.yscales[ig])
ax.yaxis.grid(True)
ax.plot(xdata, ydata)
if self.yscales[ig] == 'log':
ymajor_formatter = ax.yaxis.get_major_formatter()
ymajor_formatter.label_minor(True)
yminor_locator = LogLocator()
else:
yminor_locator = AutoLocator()
self.ax[ig].yaxis.set_minor_locator(yminor_locator)
elif command[0] == 'vline':
x, kwargs = command[1:]
self.vlines[self.ig].append((x, kwargs))
elif command[0] == 'clear':
self.ax[self.ig].cla()
elif command[0] == 'legends':
for ig, ax in enumerate(self.ax):
try:
ax.legend(self.data_names[ig])
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)
self.plt.tight_layout(pad=0.5)
elif command[0] == 'add_axis':
ig, names, yscale, xlabel, ylabel = command[1:]
self.data_names[ig] = names
self.yscales[ig] = yscale
self.xlabels[ig] = xlabel
self.ylabels[ig] = ylabel
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 terminate(self):
if self.ii:
self.output('processed %d commands' % self.ii)
self.output('ended.')
self.plt.close('all')
[docs] def poll_draw(self):
def call_back():
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.fig.canvas.draw()
self.output('processed %d commands' % self.ii)
return True
return call_back
[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):
"""
Sets-up the plotting window, sets GTK event loop timer callback to
callback() returned by self.poll_draw(). The callback does the actual
plotting, taking commands out of `pipe`, and is called every second.
Note that pyplot _must_ be imported here and not in this module so that
the import occurs _after_ the plotting process is started in that
process.
"""
import matplotlib.pyplot as plt
import gobject
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.n_gr = len(data_names)
self.vlines = {}
self.fig = self.plt.figure()
self.make_axes()
self.gid = gobject.timeout_add(1000, self.poll_draw())
self.output('...done')
self.plt.show()