使用 numpy 构造一个数组，其中从另一个 2D 数组中提取行作为 2x2 块

Question

假设我有以下2D数组：

x = np.array([[10,20,30,40], [50,60,70,80],[90,100,110,120]])  
print(x)

array([[ 10,  20,  30,  40],
       [ 50,  60,  70,  80],
       [ 90, 100, 110, 120]])

我想构造一个新数组， y ，其中每一行的值来自 x 以顺时针顺序：

print(y)
array([[ 10,  20,  60,  50],
       [ 20,  30,  70,  60],
       [ 30,  40,  80,  70],
       [ 50,  60,  100, 90],
       [ 60,  70,  110, 100],
       [ 70,  80,  120, 110]])

i可以使用python进行循环，如下所示：

n_rows, n_cols = x.shape
y = []
for i in range(n_rows-1): 
     for j in range(n_cols-1): 
         row = [x[i,j],x[i,j+1],x[i+1, j+1],x[i+1,j]] 
         y.append(row) 
y = np.array(y)

我想知道是否有一种更快的方法可以利用Numpy函数并避免使用Python循环。

Answer 1

首先，创建一个进入 x ，带有您要查看的2x2框：

b = np.lib.stride_tricks.sliding_window_view(x, (2, 2))

最内在的2x2数组中的每个数组都包含您想要的内容的拆开版本，但是随着数组的第二部分逆转。到目前为止，我们还没有复制任何数据。现在，通过将最后一个维度删除来复制。重塑将始终在此处制作副本，因为 b 是高度连接的：

c = b.reshape(*b.shape[:2], 4)

交换最后两列：

c[..., 2:] = c[..., -1:1:-1]

现在ravel ravel the Leading Dimensions：

y = c.reshape(-1, c.shape[-1])

如果您有一个numpy的版本，那是较旧的whan 1.20，则可以用 b 的定义用

b = np.lib.stride_tricks.as_strided(x, shape=(x.shape[0] - 1, x.shape[1] - 1, 2, 2), strides=x.strides * 2)

Answer 2

您可以缓存您的代码，因为循环主要是一次又一次地迭代相同的矩阵（如果您想通过循环保留相同的代码）。我对你的代码在缓存之前和之后进行了速度比较。

# Before caching
def loop_before_cache():
    n_rows, n_cols = x.shape
    y = []
    for i in range(n_rows-1): 
        for j in range(n_cols-1): 
            row = [x[i,j],x[i,j+1],x[i+1, j+1],x[i+1,j]] 
            y.append(row) 
    return np.array(y)


%timeit loop_before_cache()
11.6 µs ± 318 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

现在，通过

# After caching
from functools import lru_cache

@lru_cache()
def loop_after_cache():
    n_rows, n_cols = x.shape
    y = []
    for i in range(n_rows-1): 
        for j in range(n_cols-1): 
            row = [x[i,j],x[i,j+1],x[i+1, j+1],x[i+1,j]] 
            y.append(row) 
    return np.array(y)

%timeit loop_after_cache()
83.6 ns ± 2.42 ns per loop (mean ± std. dev. of 7 runs, 10000000 loops each)

额外的

缓存，我使用 range 添加了带有 (1000,5000) 数组的模拟数据，以显示缓存的效率。

x = np.array([i for i in range(1,5000001)])
x = np.reshape(x, (1000,5000))

# Before caching
def loop_before_cache():
    n_rows, n_cols = x.shape
    y = []
    for i in range(n_rows-1): 
        for j in range(n_cols-1): 
            row = [x[i,j],x[i,j+1],x[i+1, j+1],x[i+1,j]] 
            y.append(row) 
    return np.array(y)

%timeit loop_before_cache()
8.58 s ± 113 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

# After caching
@lru_cache(maxsize = 256)
def loop_after_cache():
    n_rows, n_cols = x.shape
    y = []
    for i in range(n_rows-1): 
        for j in range(n_cols-1): 
            row = [x[i,j],x[i,j+1],x[i+1, j+1],x[i+1,j]] 
            y.append(row) 
    return np.array(y)

%timeit loop_after_cache()
82.2 ns ± 5.58 ns per loop (mean ± std. dev. of 7 runs, 10000000 loops each)