Как «сгладить» дискретный/ступенчатый сигнал векторизованным способом с помощью numpy/scipy?

У меня есть сигнал, подобный оранжевому на следующем графике, который может иметь только целые значения:

Как вы можете видеть, оранжевый сигнал немного зашумлен и иногда «колеблется» между уровнями, когда он вот-вот перейдет в новое устойчивое состояние. Я хотел бы «сгладить» этот эффект и добиться синего сигнала. Синий сигнал — это оранжевый сигнал, отфильтрованный таким образом, что переходы не происходят до тех пор, пока 3 сэмпла подряд не перейдут на следующий шаг. Это довольно просто, если я просматриваю каждый образец вручную и использую пару переменных состояния, чтобы отслеживать, сколько раз подряд я перепрыгивал на новый шаг, но это также медленно. Я хотел бы найти способ векторизовать это в numpy. Есть идеи?

Вот пример невекторизованного способа, который, кажется, делает то, что я хочу:

up_count = 0
dn_count = 0
out = x.copy()
for i in range(len(out)-1):
    if out[i+1] > out[i]:
        up_count += 1
        dn_count = 0

        if up_count == 3:
            up_count = 0
            out[i+1] = out[i+1]
        else:
            out[i+1] = out[i]
    elif out[i+1] < out[i]:
        up_count = 0
        dn_count += 1

        if dn_count == 3:
            dn_count = 0
            out[i+1] = out[i+1]
        else:
            out[i+1] = out[i]
    else:
        dn_count = 0
        up_count = 0

Обновлено:
Спасибо @Богдану Шевченко за это решение. У меня уже есть numpy и scipy, поэтому вместо того, чтобы привлекать панд, вот моя numpy/scipy версия его ответа:

def ffill(arr, mask):
    idx = np.where(~mask, np.arange(mask.shape[0])[:, None], 0)
    np.maximum.accumulate(idx, axis=0, out=idx)
    return arr[idx, np.arange(idx.shape[1])]

x_max = scipy.ndimage.maximum_filter1d(x, 3, axis=0, origin=1, mode = "nearest")
x_min = scipy.ndimage.minimum_filter1d(x, 3, axis=0, origin=1, mode = "nearest")
x_smooth = ffill(x, x_max!=x_min)

🤔 А знаете ли вы, что...
Python используется в научных вычислениях и обработке изображений с использованием библиотеки OpenCV.


2
61
1

Ответ:

Решено

Может быть много разных стратегий, основанных на DataFrame.rolling обработке. В примере:

t = pd.Series([
    1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 
    1, 2, 1, 2, 1, 1, 2, 2, 3, 1, 2, 
    2, 1, 1, 2, 1, 1, 1, 1, 2, 2, 2, 
    2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2])
min_at_row = 3

t_smoothed = t.copy()
t_smoothed[
    t.rolling(min_at_row, min_periods=1).max() != 
    t.rolling(min_at_row, min_periods=1).min()
] = None
t_smoothed = t_smoothed.ffill()

Результат выглядит так, как вы хотите получить:

Если вы хотите использовать только numpy, без панд, используйте np.lib.stride_tricks.sliding_window_view(t, min_at_row).max(axis=1) вместо t.rolling(min_at_row).max() (и аналогично с min), но имейте в виду, что он вернет массив без первых (min_at_row - 1) значений.

Вы также можете использовать t_smoothed.shift(-min_at_row + 1), чтобы удалить задержку сглаженного сигнала.