使用给定的步幅/步长从numpy数组中获取子数组

Question

假设我有一个 Python Numpy 数组 a。

a = numpy.array([1,2,3,4,5,6,7,8,9,10,11])

我想从这个长度为5的数组中创建一个步幅为3的子序列矩阵。因此，结果矩阵将如下所示：

numpy.array([[1,2,3,4,5],[4,5,6,7,8],[7,8,9,10,11]])

实现这个的一种可能的方法是使用 for 循环。

result_matrix = np.zeros((3, 5))
for i in range(0, len(a), 3):
  result_matrix[i] = a[i:i+5]

有更简洁的方法在Numpy中实现这个吗？

- Stackd

Answer 1

方法一：使用broadcasting -

def broadcasting_app(a, L, S ):  # Window len = L, Stride len/stepsize = S
    nrows = ((a.size-L)//S)+1
    return a[S*np.arange(nrows)[:,None] + np.arange(L)]

方法二：使用更高效的NumPy strides。

def strided_app(a, L, S ):  # Window len = L, Stride len/stepsize = S
    nrows = ((a.size-L)//S)+1
    n = a.strides[0]
    return np.lib.stride_tricks.as_strided(a, shape=(nrows,L), strides=(S*n,n))

示例运行 -

In [143]: a
Out[143]: array([ 1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11])

In [144]: broadcasting_app(a, L = 5, S = 3)
Out[144]: 
array([[ 1,  2,  3,  4,  5],
       [ 4,  5,  6,  7,  8],
       [ 7,  8,  9, 10, 11]])

In [145]: strided_app(a, L = 5, S = 3)
Out[145]: 
array([[ 1,  2,  3,  4,  5],
       [ 4,  5,  6,  7,  8],
       [ 7,  8,  9, 10, 11]])

- Divakar

---

我之前使用过 as_strided，但发现它会导致非常严重的内存泄漏问题。这对于小数组来说不是问题，但即使在服务器上使用了64GB的内存，我的Python程序也会出现 MemoryError 错误。强烈建议使用 broadcasting_app 方法。 - pacificgilly1992

老兄，这太神奇了！我正在实现Shi-Tomasi角点检测算法，需要为每个像素创建一个窗口并计算一些复杂的内容。这种方法立即给我提供了所有窗口！！！ - insanely_sin

@kkawabat他们只是在说我们在使用它时需要小心，了解它的作用。writeable标志可以添加到更安全的一侧。像scikit-image这样的模块也使用as_strided。 - Divakar

@Divakar 我在链接的参考文献底部看到了一行字，"因此，尽可能避免使用 as_strided。" 我想 "通用用法" 这个词有点模糊，但如果一个方法在使用时需要注意事项，那么应该为可能会遇到这段代码的普通用户做出说明。 - kkawabat

@AndyL。好的，输入数组是一维的，所以 n = a.strides[0] 是正确的。 - Divakar

Answer 2

从 Numpy 1.20 开始，我们可以使用新的sliding_window_view 来滑动/滚动元素的窗口。

再加上一个步长 [::3] ，它就变得非常简单：

from numpy.lib.stride_tricks import sliding_window_view

# values = np.array([1,2,3,4,5,6,7,8,9,10,11])
sliding_window_view(values, window_shape = 5)[::3]
# array([[ 1,  2,  3,  4,  5],
#        [ 4,  5,  6,  7,  8],
#        [ 7,  8,  9, 10, 11]])

---

滑动的中间结果为：

sliding_window_view(values, window_shape = 5)
# array([[ 1,  2,  3,  4,  5],
#        [ 2,  3,  4,  5,  6],
#        [ 3,  4,  5,  6,  7],
#        [ 4,  5,  6,  7,  8],
#        [ 5,  6,  7,  8,  9],
#        [ 6,  7,  8,  9, 10],
#        [ 7,  8,  9, 10, 11]])

- Xavier Guihot

Answer 3

在 @Divakar 的代码基础上进行了修改，添加了检查以确保内存是连续的，并且返回的数组不能被修改。（变量名称已更改为我的 DSP 应用程序）。

def frame(a, framelen, frameadv):
"""frame - Frame a 1D array
a - 1D array
framelen - Samples per frame
frameadv - Samples between starts of consecutive frames
   Set to framelen for non-overlaping consecutive frames

Modified from Divakar's 10/17/16 11:20 solution:
https://dev59.com/7FkS5IYBdhLWcg3wSk3b

CAVEATS:
Assumes array is contiguous
Output is not writable as there are multiple views on the same memory

"""

if not isinstance(a, np.ndarray) or \
   not (a.flags['C_CONTIGUOUS'] or a.flags['F_CONTIGUOUS']):
    raise ValueError("Input array a must be a contiguous numpy array")

# Output
nrows = ((a.size-framelen)//frameadv)+1
oshape = (nrows, framelen)

# Size of each element in a
n = a.strides[0]

# Indexing in the new object will advance by frameadv * element size
ostrides = (frameadv*n, n)
return np.lib.stride_tricks.as_strided(a, shape=oshape,
                                       strides=ostrides, writeable=False)

- Marie R