#
# Copyright (c) 2016-2017 - Adjacent Link LLC, Bridgewater, New Jersey
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Adjacent Link LLC nor the names of its
# contributors may be used to endorse or promote products derived
# from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import numpy as np
import pandas as pd
import traceback
from itertools import cycle
from mpl_toolkits.axes_grid1 import make_axes_locatable
from functools import partial
import time
class Heatmap(object):
[docs] class Plot(object):
def __init__(self,*columns,**kwargs):
"""Creates a Heatmap Plot.
A heatmap plot is a colorbar chart of a time-series
measurment.
Args:
columns ([str]): One or more DataFrame column names
to plot.
Keyword Args:
title (str): Plot title. Default: ''.
ylim ((min,max)): Plot y-axis limit plot
limit. Default: None.
markers ([str]): One or more DataFrame columns to use
as markers. A marker is a vertical line that
indicates an event occurred. A marker is drawn any
time the indicated DataFrame column value is
non-zero and non-nan. Default: [].
indicators ([str]): One or more DataFrame columns to
use as indicators. An indicator is a horizontal
bar that indicates a condition is met. An
indicator is continuously drawn for time stamps
where the indicated DataFrame column value is
non-zero and non-nan. Default: [].
legend (bool): Flag indicating whether a plot legend
should be displayed. Default: False.
Raises:
KeyError: If an invalid keyword is found.
"""
self.columns = columns
self.title = kwargs.pop('title','')
self.ylim = kwargs.pop('ylim',None)
self.markers = kwargs.pop('markers',[])
self.legend = kwargs.pop('legend',False)
if not hasattr(self.markers ,'__iter__'):
self.markers = [self.markers]
if kwargs:
raise KeyError("%s unknown key(s): %s" % (self.__class__.__name__,
", ".join(kwargs.keys())))
def __init__(self,model,**kwargs):
self._model = model
self._last_events = 0
self._title = kwargs.pop('title','')
self._interval = kwargs.pop('interval',1000)
self._model_data_kwargs = {}
self._last_update_time = None
for kwarg in kwargs.keys():
if kwarg.startswith('model_'):
self._model_data_kwargs[kwarg[6:]] = kwargs.pop(kwarg)
if kwargs:
raise KeyError("%s unknown key(s): %s" % (self.__class__.__name__,
", ".join(kwargs.keys())))
def show(self,*plots,**kwargs):
self.plot(*plots,**kwargs)
ani = self.animate()
plt.show()
def plot(self,*plots,**kwargs):
plots = [Heatmap.Plot(*p.columns,**p.kwargs) for p in plots]
self._fig = kwargs.pop('figure',None)
tight = kwargs.pop('tight',False)
self._stale_timeout = kwargs.pop('stale_timeout',10)
if kwargs:
raise KeyError("%s unknown key(s): %s" % (self.__class__.__name__,
", ".join(kwargs.keys())))
if self._fig == None:
self._fig = plt.gcf()
self._fig.canvas.set_window_title(self._title)
self._fig.patch.set_facecolor('white')
plt.style.use('ggplot')
self._fig_face_color = self._fig.patch.get_facecolor()
# indicate no updates yet
self._fig.patch.set_facecolor('#ff8080')
self._axes = []
self._pcolors = []
self._columns = []
self._mark_columns = []
self._marks = []
self._ylims = []
self._dfs = []
for i in range(1,len(plots)+1):
self._axes.append(self._fig.add_subplot(len(plots),1,i))
for plot_index,(plot,ax) in enumerate(zip(plots,self._axes)):
ax.set_title(plot.title)
ax.set_xlim(0, self._model.stream().cache() - 1)
items = [('Time',[])]
for column in plot.columns:
items.append((column,[]))
df = pd.DataFrame.from_items(items)
df.reset_index(drop=True,inplace=True)
vmin = None
vmax = None
if plot.ylim != None:
vmin = plot.ylim[0]
vmax = plot.ylim[1]
else:
raise KeyError("%s missing key(s): ylim" % (self.__class__.__name__))
df = df[list(plot.columns)].transpose()
heatmap = ax.pcolor(df,
cmap=plt.cm.YlOrRd,
vmin=vmin,
vmax=vmax,
animated=True)
self._dfs.append(df)
self._pcolors.append(heatmap)
self._ylims.append((vmin,vmax))
divider = make_axes_locatable(ax)
cax = divider.append_axes("right", "2%", pad="1%")
cb = self._fig.colorbar(heatmap, cax=cax)
ax.set_yticks([x + .5 for x in range(0,len(plot.columns)+1)])
ax.set_yticklabels(plot.columns)
self._columns.append(plot.columns)
self._mark_columns.append(plot.markers)
# add markers to the legend using blank plots
color=cycle(plt.rcParams['axes.color_cycle'])
for index,mcol in enumerate(plot.markers):
ax.plot([],[],
label=mcol,
ls='--',
color=next(color),
animated=True)
ax.set_xticks(range(0, self._model.stream().cache()))
ax.set_xticklabels([""] * self._model.stream().cache())
ax.grid(True)
def format_coord(index,title,x, y):
try:
return '%s=%1.4f' % (list(self._dfs[index].index)[int(y)],
self._dfs[index].iloc[int(y),int(x)])
except:
return ''
ax.format_coord = partial(format_coord,plot_index,plot.title)
if plot.legend:
ax.legend()
if tight:
self._fig.tight_layout()
def animate(self):
def _init():
return []
def _animate(i):
df,events,_ = self._model.data(**self._model_data_kwargs)
if self._last_events != events:
if self._last_update_time == None:
self._fig.patch.set_facecolor(self._fig_face_color)
self._fig.canvas.draw()
self._last_update_time = time.time()
else:
if self._last_update_time != None and \
time.time() - self._last_update_time > self._stale_timeout:
self._fig.patch.set_facecolor('#ff8080')
self._fig.canvas.draw()
self._last_update_time = None
if not df.empty and events != self._last_events:
try:
while self._marks:
self._marks.pop(0).remove()
while self._pcolors:
self._pcolors.pop(0).remove()
for index,(ax,pcols,ylim) in enumerate(zip(self._axes,self._columns,self._ylims)):
df_pcolor = df
df_pcolor.reset_index(drop=True,inplace=True)
df_pcolor = df_pcolor[list(pcols)]
self._dfs[index] = df_pcolor.copy().transpose()
# mark na as invalid
df_pcolor = np.ma.masked_invalid(df_pcolor)
self._pcolors.append(ax.pcolor(df_pcolor.transpose(),
cmap=plt.cm.YlOrRd,
vmin=ylim[0],
vmax=ylim[1],
animated=True))
for ax,mcols in zip(self._axes,self._mark_columns):
for mcol,color in zip(mcols,cycle(plt.rcParams['axes.color_cycle'])):
for x,y in enumerate(df[mcol]):
if pd.notnull(y) and y > 0:
self._marks.append(ax.axvline(x,
ls='--',
animated=True,
label=mcol,
color=color))
except:
traceback.print_exc()
self._last_events = events
return self._pcolors + self._marks
return animation.FuncAnimation(self._fig,
_animate,
init_func=_init,
interval=self._interval,
blit=True)