# Anything vaguely portable from the ipython notebook is here in plaintext import pandas import stumpy import numpy import matplotlib.pyplot as plot import matplotlib.dates as dates from matplotlib.patches import Rectangle import datetime as dt ## Process wrapper, time reporting from functools import wraps import time def timeit(fn): def timed(*args, **kw): print (f'>>> {fn.__name__} >>>') ts = time.time() result = fn(*args, **kw) te = time.time() print (f'<<< {fn.__name__} <<< {(te-ts):.3f}s') return result return timed @timeit @wraps(stumpy.stump) def timed_stump(dataset, w): l = len(dataset) ws = l - w + 1 print(f'Processing a matrix profile over {ws} windows...') result = stumpy.stump(dataset, w) return result # Basic feature extraction def threshold_extraction(op, cmp, profile, motif_order, window_width, threshold): assert(0 < threshold < 1.0) closest = profile[motif_order[0], 0] cutoff = op(closest, (closest * threshold)) rv = [] for obs in motif_order: if cmp(profile[obs,0], cutoff): if not (any(map(lambda i: abs(i - obs) < window_width, rv))): rv.append(obs) else: # If two motifs overlap it's possible the window is too small? # Further criteria needed for that... pass else: break return rv def get_motifs(profile, motif_order, window_width, threshold): return threshold_extraction(lambda a, b: a + b ,lambda a, b: a < b ,profile ,motif_order ,window_width ,threshold) def get_discords(profile, motif_order, window_width, threshold): return threshold_extraction(lambda a, b: a - b ,lambda a, b: a > b ,profile ,numpy.flip(motif_order) ,window_width ,threshold) def mark_discovered(motif_list, data_fig, profile_fig, window_width, fig_height): for motif in motif_list: rect = Rectangle((motif, 0), window_width, fig_height, facecolor='lightgrey') data_fig.add_patch(rect) profile_fig.axvline(x=motif, linestyle='dashed') # Basic helper functions def plot_matrix_profile(plot_, profile): plot_.set_xlabel('Time', fontsize='15') plot_.set_ylabel('Distance', fontsize='15') plot_.plot(profile[:, 0]) from string import capwords plot.style.use("data/stumpy.mplstyle") def plot_pandas_import (plot_, data, axis_name): plot_.set_ylabel(capwords(axis_name), fontsize='15') plot_.plot(data[axis_name].values)