<jupyter_start><jupyter_text>Adaptive Scale Selection AMPD (ASS-AMPD)Here, we look into signals that are quasi-periodic locally, but have variable scale across the timeseries<jupyter_code>%load_ext autoreload
%autoreload 2
import numpy as np
from numpy import sin, pi
from scipy.signal import find_peaks_cwt
from pyampd.ampd import find_peaks, find_peaks_original, find_peaks_adaptive
import matplotlib.pyplot as plt<jupyter_output><empty_output><jupyter_text>Create fake signal<jupyter_code>def fake_timeseries(tmax=30, seed=None, dilation=1.):
    """Generate fake data for testing peak detection algorithms"""
    np.random.seed(seed)
    fs, f1, f2, f3 = 20, dilation * 1, dilation * 2, dilation * 0.1
    a, b, c, d = 1, 0.6, 1, 1.
    N = int(tmax*fs)
    t = np.linspace(0, tmax, N)
    x = a*sin(2*pi*f1*t) + b*sin(2*pi*f2*t) + c*sin(2*pi*f3*t) + d*np.random.rand(N)
    return t, x<jupyter_output><empty_output><jupyter_text>Compare peak detect methods<jupyter_code>methods = [
    ('ass_ampd_50', lambda x: find_peaks_adaptive(x, window=50)),
    ('ass_ampd_100', lambda x: find_peaks_adaptive(x, window=100)),
    ('ass_ampd_200', lambda x: find_peaks_adaptive(x, window=200)),
    ('ampd', find_peaks)
]
styles = ['rs', 'go', 'bx', 'cs', 'mo']<jupyter_output><empty_output><jupyter_text>Create a signal with distinct regions characterised by different frequency/scale content<jupyter_code>t1, x1 = fake_timeseries(tmax=10, seed=0)
t2, x2 = fake_timeseries(tmax=10, seed=1, dilation=3.1)
t = np.hstack([t1, t1[-1]+ t2 + 1/20])
x = np.hstack([x1, x2 - x2[0] + x1[1]])
all_pks = [method(x) for name, method in methods]
plt.figure(figsize=(25, 7))
plt.plot(t, x, '-')
for idx in range(len(methods)):
    pks = all_pks[idx]
    offset = idx * 0.2
    style = styles[idx]
    name = methods[idx][0]
    plt.plot(t[pks], offset + x[pks], style, label=name)
plt.legend()
plt.xlabel('time/s')
plt.ylabel('signal')<jupyter_output><empty_output><jupyter_text>Compare timing<jupyter_code>t, x = fake_timeseries(tmax=1000, seed=0)
print('{} samples over {} s'.format(len(x), t[-1]))
methods = [
    ('ass_ampd_200', lambda x: find_peaks_adaptive(x, window=200)),
    ('ass_ampd', lambda x: find_peaks_adaptive(x, window=None)),
    ('ampd', find_peaks),
]
for name, method in methods:
    print('timing "{}":'.format(name.upper()))
    %time pks = method(x)
    print()<jupyter_output>timing "ASS_AMPD_200":
CPU times: user 126 ms, sys: 28 ms, total: 154 ms
Wall time: 40 ms

timing "ASS_AMPD":
CPU times: user 5.72 s, sys: 1.92 s, total: 7.64 s
Wall time: 7.31 s

timing "AMPD":
CPU times: user 728 ms, sys: 98.3 ms, total: 826 ms
Wall time: 478 ms<jupyter_text>Adaptive LSM<jupyter_code>t1, x1 = fake_timeseries(tmax=15, seed=0)
t2, x2 = fake_timeseries(tmax=15, seed=1, dilation=3.1)
t = np.hstack([t1, t1[-1]+ t2 + 1/20])
x = np.hstack([x1, x2 - x2[0] + x1[1]])
pks, LSM, adaptive_scale = find_peaks_adaptive(x, window=100, debug=True)
L, N = LSM.shape
fig, ax0 = plt.subplots(ncols=1, figsize=(20, 7))
ax = ax0
ax.imshow(LSM)
ax.set_title('LSM')
ax.set_xlabel('sample index')
ax.set_ylabel('scale')
plt.plot(adaptive_scale)
plt.figure(figsize=(25, 7))
plt.plot(t, x, '-')
plt.plot(t[pks], x[pks], 'rs', label='a-ampd')
plt.legend()
plt.xlabel('time/s')
plt.ylabel('signal')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16631387964922892, 'fraction_numerical': 0.04777744179014212, 'mean_word_length': 5.398452611218569, 'pattern_counts': {'":': 4, '<': 24, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '349010', 'license': 'MIT', 'n_tokens_mistral': 1477, 'n_tokens_neox': 1370, 'n_words': 314, 'path': 'notebooks/adaptive-ampd.ipynb', 'repo_name': 'luigiluz/pyampd', 'size': '3307'}