numpy中的滑动窗口

Question

我有一个numpy形状(6,2)

[[00,01],
 [10,11],
 [20,21],
 [30,31],
 [40,41],
 [50,51]]

我需要一个步长为1的滑动窗口,窗口大小为3喜欢这个:

[[00,01,10,11,20,21],
 [10,11,20,21,30,31],
 [20,21,30,31,40,41],
 [30,31,40,41,50,51]]

我正在寻找一个numpy解决方案.如果您的解决方案可以参数化原始数组的形状以及窗口大小和步长,那就太好了.

我发现这个相关的答案使用步幅有效的移动平均滤波器,但我没有看到如何指定那里的步长以及如何将窗口从3d折叠到连续的2d数组.这个滚动或滑动窗口迭代器在Python中,但是在Python中,我不确定它的效率如何.此外,它支持元素,但如果每个元素具有多个特征,则最终不会将它们连接在一起.

Answer 1

您可以使用花式索引在numpy中执行矢量化滑动窗口.

>>> import numpy as np

>>> a = np.array([[00,01], [10,11], [20,21], [30,31], [40,41], [50,51]])

>>> a
array([[ 0,  1],
       [10, 11],
       [20, 21],                      #define our 2d numpy array
       [30, 31],
       [40, 41],
       [50, 51]])

>>> a = a.flatten()

>>> a
array([ 0,  1, 10, 11, 20, 21, 30, 31, 40, 41, 50, 51])    #flattened numpy array

>>> indexer = np.arange(6)[None, :] + 2*np.arange(4)[:, None]

>>> indexer
array([[ 0,  1,  2,  3,  4,  5],
       [ 2,  3,  4,  5,  6,  7],            #sliding window indices
       [ 4,  5,  6,  7,  8,  9],
       [ 6,  7,  8,  9, 10, 11]])

>>> a[indexer]
array([[ 0,  1, 10, 11, 20, 21],
       [10, 11, 20, 21, 30, 31],            #values of a over sliding window
       [20, 21, 30, 31, 40, 41],
       [30, 31, 40, 41, 50, 51]])

>>> np.sum(a[indexer], axis=1)
array([ 63, 123, 183, 243])         #sum of values in 'a' under the sliding window.

解释此代码正在执行的操作.

在np.arange(6)[None, :]通过6创建一个行向量0,并且np.arange(4)[:, None]通过4.这导致了4x6的矩阵,其中每行(其中六)表示一个窗口,和行(其中四个)的数量创建一个列向量0表示的数视窗.2的倍数使滑动窗口每次滑动2个单位,这是在每个元组上滑动所必需的.使用numpy数组切片,您可以将滑动窗口传递到展平的numpy数组,并像sum一样对它们进行聚合.

Answer 2

In [1]: import numpy as np

In [2]: a = np.array([[00,01], [10,11], [20,21], [30,31], [40,41], [50,51]])

In [3]: w = np.hstack((a[:-2],a[1:-1],a[2:]))

In [4]: w
Out[4]: 
array([[ 0,  1, 10, 11, 20, 21],
       [10, 11, 20, 21, 30, 31],
       [20, 21, 30, 31, 40, 41],
       [30, 31, 40, 41, 50, 51]])

您可以将其写为函数,如下所示:

def window_stack(a, stepsize=1, width=3):
    n = a.shape[0]
    return np.hstack( a[i:1+n+i-width:stepsize] for i in range(0,width) )

这并不真正取决于原始阵列的形状,只要a.ndim = 2.请注意,我从不在交互式版本中使用任何长度.形状的第二个维度是无关紧要的; 每行可以随意长.感谢@ Jaime的建议,你可以在不检查形状的情况下完成:

def window_stack(a, stepsize=1, width=3):
    return np.hstack( a[i:1+i-width or None:stepsize] for i in range(0,width) )

Answer 3

解决方案是

np.lib.stride_tricks.as_strided(a, shape=(4,6), strides=(8,4)).

当您开始考虑指针/地址时,使用步幅是直观的.

该as_strided()方法有3个参数.

1. 数据
2. 形状
3. 进步

data是我们要操作的数组.

要as_strided()用于实现滑动窗口函数,我们必须事先计算输出的形状.在问题中,(4,6)是输出的形状.如果尺寸不正确,我们最终会读取垃圾值.这是因为我们通过将指针移动几个字节来访问数据(取决于数据类型).

确定步幅的正确值对于获得预期结果至关重要.在计算步幅之前,找出每个元素占用的内存.我们可以使用它arr.strides[-1].在此示例中,一个元素占用的内存为4个字节.Numpy数组以行主要方式创建.因此,下一行的第一个元素紧邻当前行的最后一个元素.

例如:0,1 | 10,11 | ...

10紧邻1.

想象一下,将2D阵列重新整形为1D(这是可以接受的,因为数据以行主格式存储).输出中每行的第一个元素是1D数组中的奇数索引元素.0,10,20,30,..

因此,我们需要从0到10,10到20移动的内存中的步数,依此类推,是元素的2*mem大小.因此,每行的步幅为2*4bytes = 8.对于输出中的给定行,所有元素在我们想象的1D数组中彼此相邻.因此,要获得连续的下一个元素,只需要一个等于元素大小的步幅.因此,列步长为4个字节.

因此, strides=(8,4)

另一种解释:输出的形状为(4,6).列大步4.因此,第一行元素从索引开始,0有6个元素,每个元素间隔4个字节.收集第一行后,第二行从当前行的开始开始8个字节.第三行从第二行的起始点开始8个字节,依此类推.

Shape确定我们需要的行数和列数.strides定义内存步骤以开始行并收集列元素

Answer 4

¹:可以实现短列表理解:more_itertools.windowed

特定

import numpy as np
import more_itertools as mit


a = [["00","01"],
     ["10","11"],
     ["20","21"],
     ["30","31"],
     ["40","41"],
     ["50","51"]]

b = np.array(a)

码

np.array([list(mit.flatten(w)) for w in mit.windowed(a, n=3)])

要么

np.array([[i for item in w for i in item] for w in mit.windowed(a, n=3)])

要么

np.array(list(mit.windowed(b.ravel(), n=6)))

产量

array([['00', '01', '10', '11', '20', '21'],
       ['10', '11', '20', '21', '30', '31'],
       ['20', '21', '30', '31', '40', '41'],
       ['30', '31', '40', '41', '50', '51']], 
      dtype='<U2')

大小的滑动窗口n=3被创建并变平.请注意默认步长是more_itertools.windowed(..., step=1).

性能

作为一个数组,接受的答案是最快的.

%timeit np.hstack((a[:-2], a[1:-1], a[2:]))
# 37.5 µs ± 1.88 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
%timeit np.hstack((b[:-2], b[1:-1], b[2:]))
# 12.9 µs ± 166 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

%timeit np.array([list(mit.flatten(w)) for w in mit.windowed(a, n=3)])
# 23.2 µs ± 1.73 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
%timeit np.array([[i for item in w for i in item] for w in mit.windowed(a, n=3)])
# 21.2 µs ± 999 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)
%timeit np.array(list(mit.windowed(b.ravel(), n=6)))
# 43.4 µs ± 374 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

^{_{实现itertool配方和许多有用工具的第三方库.}}

Answer 5

从 开始Numpy 1.20，使用 newsliding_window_view滑动/滚动元素窗口，并基于与user42541 的回答相同的想法，我们可以这样做：

import numpy as np
from numpy.lib.stride_tricks import sliding_window_view

# values = np.array([[0,1], [10,11], [20,21], [30,31], [40,41], [50,51]])
sliding_window_view(values.flatten(), window_shape = 2*3)[::2]
# array([[ 0,  1, 10, 11, 20, 21],
#        [10, 11, 20, 21, 30, 31],
#        [20, 21, 30, 31, 40, 41],
#        [30, 31, 40, 41, 50, 51]])

其中2是子数组和3窗口的大小。

中间步骤的详细信息：

# values = np.array([[0,1], [10,11], [20,21], [30,31], [40,41], [50,51]])

# Flatten the array (concatenate sub-arrays):
values.flatten()
# array([ 0,  1, 10, 11, 20, 21, 30, 31, 40, 41, 50, 51])

# Slide through windows of size 2*3=6:
sliding_window_view(values.flatten(), 2*3)
# array([[ 0,  1, 10, 11, 20, 21],
#        [ 1, 10, 11, 20, 21, 30],
#        [10, 11, 20, 21, 30, 31],
#        [11, 20, 21, 30, 31, 40],
#        [20, 21, 30, 31, 40, 41],
#        [21, 30, 31, 40, 41, 50],
#        [30, 31, 40, 41, 50, 51]])

# Only keep even rows (1 row in 2 - if sub-arrays have a size of x, then replace 2 with x):
sliding_window_view(values.flatten(), 2*3)[::2]
# array([[ 0,  1, 10, 11, 20, 21],
#        [10, 11, 20, 21, 30, 31],
#        [20, 21, 30, 31, 40, 41],
#        [30, 31, 40, 41, 50, 51]])