У меня есть сигнал, подобный оранжевому на следующем графике, который может иметь только целые значения:
Как вы можете видеть, оранжевый сигнал немного зашумлен и иногда «колеблется» между уровнями, когда он вот-вот перейдет в новое устойчивое состояние. Я хотел бы «сгладить» этот эффект и добиться синего сигнала. Синий сигнал — это оранжевый сигнал, отфильтрованный таким образом, что переходы не происходят до тех пор, пока 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.
Может быть много разных стратегий, основанных на 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)
, чтобы удалить задержку сглаженного сигнала.