#!/usr/bin/env python import panel as pn import pandas as pd from bokeh.plotting import figure from bokeh.layouts import layout from bokeh.models import ( ColumnDataSource, Range1d, Slider, Button, TextInput, LabelSet, Circle, Div, ) class StumpyBokehDashboard: def __init__(self): self.sizing_mode = "stretch_both" self.window = 0 self.m = None self.df = None self.ts_cds = None self.quad_cds = None self.pattern_match_cds = None self.dist_cds = None self.circle_cds = None self.ts_plot = None self.mp_plot = None self.pm_plot = None self.logo_div = None self.heroku_div = None self.slider = None self.play_btn = None self.txt_inp = None self.pattern_btn = None self.match_btn = None self.reset_btn = None self.idx = None self.min_distance_idx = None self.animation = pn.state.add_periodic_callback( self.update_animate, 50, start=False ) def get_df_from_file(self): raw_df = pd.read_csv( "https://raw.githubusercontent.com/seanlaw/stumpy-live-demo/master/raw.csv" ) mp_df = pd.read_csv( "https://raw.githubusercontent.com/seanlaw/stumpy-live-demo/master/matrix_profile.csv" ) self.window = raw_df.shape[0] - mp_df.shape[0] + 1 self.m = raw_df.shape[0] - mp_df.shape[0] + 1 self.min_distance_idx = mp_df["distance"].argmin() df = pd.merge(raw_df, mp_df, left_index=True, how="left", right_index=True) return df.reset_index() def get_ts_dict(self, df): return self.df.to_dict(orient="list") def get_circle_dict(self, df): return self.df[["index", "y"]].to_dict(orient="list") def get_quad_dict(self, df, pattern_idx=0, match_idx=None): if match_idx is None: match_idx = df.loc[pattern_idx, "idx"].astype(int) quad_dict = dict( pattern_left=[pattern_idx], pattern_right=[pattern_idx + self.window - 1], pattern_top=[max(df["y"])], pattern_bottom=[0], match_left=[match_idx], match_right=[match_idx + self.window - 1], match_top=[max(df["y"])], match_bottom=[0], vert_line_left=[pattern_idx - 5], vert_line_right=[pattern_idx + 5], vert_line_top=[max(df["distance"])], vert_line_bottom=[0], hori_line_left=[0], hori_line_right=[max(df["index"])], hori_line_top=[df.loc[pattern_idx, "distance"] - 0.01], hori_line_bottom=[df.loc[pattern_idx, "distance"] + 0.01], ) return quad_dict def get_custom_quad_dict(self, df, pattern_idx=0, match_idx=None): if match_idx is None: match_idx = df.loc[pattern_idx, "idx"].astype(int) quad_dict = dict( pattern_left=[pattern_idx], pattern_right=[pattern_idx + self.window - 1], pattern_top=[max(df["y"])], pattern_bottom=[0], match_left=[match_idx], match_right=[match_idx + self.window - 1], match_top=[max(df["y"])], match_bottom=[0], vert_line_left=[match_idx - 5], vert_line_right=[match_idx + 5], vert_line_top=[max(df["distance"])], vert_line_bottom=[0], hori_line_left=[0], hori_line_right=[max(df["index"])], hori_line_top=[df.loc[match_idx, "distance"] - 0.01], hori_line_bottom=[df.loc[match_idx, "distance"] + 0.01], ) return quad_dict def get_pattern_match_dict(self, df, pattern_idx=0, match_idx=None): if match_idx is None: match_idx = df["idx"].loc[pattern_idx].astype(int) pattern_match_dict = dict( index=list(range(self.window)), pattern=df["y"].loc[pattern_idx : pattern_idx + self.window - 1], match=df["y"].loc[match_idx : match_idx + self.window - 1], ) return pattern_match_dict def get_ts_plot(self, color="black"): """ Time Series Plot """ ts_plot = figure( toolbar_location="above", sizing_mode=self.sizing_mode, title="Raw Time Series or Sequence", tools=["reset"], ) q = ts_plot.quad( "pattern_left", "pattern_right", "pattern_top", "pattern_bottom", source=self.quad_cds, name="pattern_quad", color="#54b847", ) q.visible = False q = ts_plot.quad( "match_left", "match_right", "match_top", "match_bottom", source=self.quad_cds, name="match_quad", color="#696969", alpha=0.5, ) q.visible = False l = ts_plot.line(x="index", y="y", source=self.ts_cds, color=color) ts_plot.x_range = Range1d( 0, max(self.df["index"]), bounds=(0, max(self.df["x"])) ) ts_plot.y_range = Range1d(0, max(self.df["y"]), bounds=(0, max(self.df["y"]))) c = ts_plot.circle( x="index", y="y", source=self.circle_cds, size=0, line_color="white" ) c.selection_glyph = Circle(line_color="white") c.nonselection_glyph = Circle(line_color="white") return ts_plot def get_dist_dict(self, df, pattern_idx=0): dist = df["distance"] max_dist = dist.max() min_dist = dist.min() x_offset = self.df.shape[0] - self.window / 2 y_offset = max_dist / 2 distance = dist.loc[pattern_idx] text = distance.round(1).astype(str) gauge_dict = dict(x=[0 + x_offset], y=[0 + y_offset], text=[text]) return gauge_dict def get_mp_plot(self): """ Matrix Profile Plot """ mp_plot = figure( x_range=self.ts_plot.x_range, toolbar_location=None, sizing_mode=self.sizing_mode, title="Matrix Profile (All Minimum Distances)", ) q = mp_plot.quad( "vert_line_left", "vert_line_right", "vert_line_top", "vert_line_bottom", source=self.quad_cds, name="pattern_start", color="#54b847", ) q.visible = False q = mp_plot.quad( "hori_line_left", "hori_line_right", "hori_line_top", "hori_line_bottom", source=self.quad_cds, name="match_dist", color="#696969", alpha=0.5, ) q.visible = False mp_plot.line(x="index", y="distance", source=self.ts_cds, color="black") # mp_plot.x_range = Range1d(0, self.df.shape[0]-self.window+1, bounds=(0, self.df.shape[0]-self.window+1)) mp_plot.x_range = Range1d( 0, self.df.shape[0] + 1, bounds=(0, self.df.shape[0] + 1) ) mp_plot.y_range = Range1d( 0, max(self.df["distance"]), bounds=(0, max(self.df["distance"])) ) label = LabelSet( x="x", y="y", text="text", source=self.dist_cds, text_align="center", name="gauge_label", text_color="black", text_font_size="30pt", ) mp_plot.add_layout(label) return mp_plot def get_pm_plot(self): """ Pattern-Match Plot """ pm_plot = figure( toolbar_location=None, sizing_mode=self.sizing_mode, title="Pattern Match Overlay", ) l = pm_plot.line( "index", "pattern", source=self.pattern_match_cds, name="pattern_line", color="#54b847", line_width=2, ) l.visible = False l = pm_plot.line( "index", "match", source=self.pattern_match_cds, name="match_line", color="#696969", alpha=0.5, line_width=2, ) l.visible = False return pm_plot def get_logo_div(self): """ STUMPY logo """ logo_div = Div( text="", sizing_mode="stretch_width" ) return logo_div def get_heroku_div(self): """ STUMPY Heroku App Link """ heroku_div = Div(text="http://tiny.cc/stumpy-demo") return heroku_div def get_slider(self, value=0): slider = Slider( start=0.0, end=max(self.df["index"]) - self.window, value=value, step=1, title="Subsequence", sizing_mode=self.sizing_mode, ) return slider def get_play_button(self): play_btn = Button(label="► Play") play_btn.on_click(self.animate) return play_btn def get_text_input(self): txt_inp = TextInput(sizing_mode=self.sizing_mode) return txt_inp def get_buttons(self): pattern_btn = Button(label="Show Motif", sizing_mode=self.sizing_mode) match_btn = Button(label="Show Nearest Neighbor", sizing_mode=self.sizing_mode) reset_btn = Button(label="Reset", sizing_mode=self.sizing_mode, button_type="primary") return pattern_btn, match_btn, reset_btn def update_plots(self, attr, new, old): self.quad_cds.data = self.get_quad_dict(self.df, self.slider.value) self.pattern_match_cds.data = self.get_pattern_match_dict( self.df, self.slider.value ) self.dist_cds.data = self.get_dist_dict(self.df, self.slider.value) def custom_update_plots(self, attr, new, old): self.quad_cds.data = self.get_custom_quad_dict( self.df, self.pattern_idx, self.slider.value ) self.pattern_match_cds.data = self.get_pattern_match_dict( self.df, self.pattern_idx, self.slider.value ) self.dist_cds.data = self.get_dist_dict(self.df, self.slider.value) dist = self.df["distance"].loc[self.slider.value] def show_hide_pattern(self): pattern_quad = self.ts_plot.select(name="pattern_quad")[0] pattern_start = self.mp_plot.select(name="pattern_start")[0] pattern_line = self.pm_plot.select(name="pattern_line")[0] if pattern_quad.visible: pattern_start.visible = False pattern_line.visible = False pattern_quad.visible = False self.pattern_btn.label = "Show Motif" else: pattern_start.visible = True pattern_line.visible = True pattern_quad.visible = True self.pattern_btn.label = "Hide Motif" def show_hide_match(self): match_quad = self.ts_plot.select(name="match_quad")[0] match_dist = self.mp_plot.select(name="match_dist")[0] match_line = self.pm_plot.select(name="match_line")[0] if match_quad.visible: match_dist.visible = False match_line.visible = False match_quad.visible = False self.match_btn.label = "Show Nearest Neighbor" else: match_dist.visible = True match_line.visible = True match_quad.visible = True self.match_btn.label = "Hide Nearest Neighbor" def update_slider(self, attr, old, new): self.slider.value = int(self.txt_inp.value) def animate(self): if self.play_btn.label == "► Play": self.play_btn.label = "❚❚ Pause" self.animation.start() else: self.play_btn.label = "► Play" self.animation.stop() def update_animate(self, shift=50): if self.window < self.m: # Probably using box select start = self.slider.value end = start + shift if self.df.loc[start:end, "distance"].min() <= 15: self.slider.value = self.df.loc[start:end, "distance"].idxmin() self.animate() elif self.slider.value + shift <= self.slider.end: self.slider.value = self.slider.value + shift else: self.slider.value = 0 elif self.slider.value + shift <= self.slider.end: self.slider.value = self.slider.value + shift else: self.slider.value = 0 def reset(self): self.sizing_mode = "stretch_both" self.window = self.m self.default_idx = self.min_distance_idx self.df = self.get_df_from_file() self.ts_cds.data = self.get_ts_dict(self.df) self.mp_plot.y_range.end = max(self.df["distance"]) self.mp_plot.title.text = "Matrix Profile (All Minimum Distances)" self.mp_plot.y_range.bounds = (0, max(self.df["distance"])) self.quad_cds.data = self.get_quad_dict(self.df, pattern_idx=self.default_idx) self.pattern_match_cds.data = self.get_pattern_match_dict( self.df, pattern_idx=self.default_idx ) self.dist_cds.data = self.get_dist_dict(self.df, pattern_idx=self.default_idx) self.circle_cds.data = self.get_circle_dict(self.df) # Remove callback and add old callback if self.custom_update_plots in self.slider._callbacks["value"]: self.slider.remove_on_change("value", self.custom_update_plots) self.slider.on_change("value", self.update_plots) self.slider.end = self.df.shape[0] - self.window self.slider.value = self.default_idx def get_data(self): self.df = self.get_df_from_file() self.default_idx = self.min_distance_idx self.ts_cds = ColumnDataSource(self.get_ts_dict(self.df)) self.quad_cds = ColumnDataSource( self.get_quad_dict(self.df, pattern_idx=self.default_idx) ) self.pattern_match_cds = ColumnDataSource( self.get_pattern_match_dict(self.df, pattern_idx=self.default_idx) ) self.dist_cds = ColumnDataSource( self.get_dist_dict(self.df, pattern_idx=self.default_idx) ) self.circle_cds = ColumnDataSource(self.get_circle_dict(self.df)) def get_plots(self, ts_plot_color="black"): self.ts_plot = self.get_ts_plot(color=ts_plot_color) self.mp_plot = self.get_mp_plot() self.pm_plot = self.get_pm_plot() def get_widgets(self): self.slider = self.get_slider(value=self.default_idx) self.play_btn = self.get_play_button() self.txt_inp = self.get_text_input() self.pattern_btn, self.match_btn, self.reset_btn = self.get_buttons() self.logo_div = self.get_logo_div() self.heroku_div = self.get_heroku_div() def set_callbacks(self): self.slider.on_change("value", self.update_plots) self.pattern_btn.on_click(self.show_hide_pattern) self.show_hide_pattern() self.match_btn.on_click(self.show_hide_match) self.show_hide_match() self.reset_btn.on_click(self.reset) self.txt_inp.on_change("value", self.update_slider) def get_layout(self): self.get_data() self.get_plots() self.get_widgets() self.set_callbacks() l = layout( [ [self.ts_plot], [self.mp_plot], [self.pm_plot], [self.slider], [self.pattern_btn, self.match_btn, self.play_btn, self.logo_div], ], sizing_mode=self.sizing_mode, ) return l def get_raw_layout(self): self.get_data() self.get_plots(ts_plot_color="#54b847") l = layout([[self.ts_plot], [self.mp_plot]], sizing_mode=self.sizing_mode) return l dashboard = StumpyBokehDashboard() def get_components(dashboard: StumpyBokehDashboard=dashboard): dashboard.get_data() dashboard.get_plots() dashboard.get_widgets() dashboard.set_callbacks() logo = dashboard.logo_div settings = layout( dashboard.pattern_btn, dashboard.match_btn, dashboard.play_btn, dashboard.slider, height=150, sizing_mode="stretch_width", ) main = layout( [ [dashboard.ts_plot], [dashboard.mp_plot], [dashboard.pm_plot], ], sizing_mode=dashboard.sizing_mode, ) return logo, settings, main pn.extension(template="fast") pn.state.template.param.update( site_url="https://awesome-panel.org", site="Awesome Panel", title="Stumpy Timeseries Analysis", favicon="https://raw.githubusercontent.com/MarcSkovMadsen/awesome-panel-assets/320297ccb92773da099f6b97d267cc0433b67c23/favicon/ap-1f77b4.ico", header_background="#459db9", theme_toggle=False, ) logo, settings, main = get_components() pn.Column( logo, settings, sizing_mode="stretch_width", ).servable(target="sidebar") pn.panel(main, sizing_mode="stretch_both", max_height=800).servable(target="main")