基于窗口大小与数组值规则生成两个目标数组的技术求助

需求说明 #

处理一个大尺寸（上千行/列）的Numpy数组，元素包含1、大于1和小于1的值，生成两个目标数组：

1. 目标数组1：将所有小于1的元素前7元素窗口内（即该元素位置往前数最多6个位置，包含自身）的大于1元素设为1，其余为0；
2. 目标数组2：将所有小于1的元素后7元素窗口内（即该元素位置往后数最多6个位置，包含自身）的大于1元素设为1，其余为0。

实现方案 #

方法一：行循环实现（易理解，适合中等规模数组） #

通过标记关键位置，逐行处理窗口范围，逻辑清晰，适合上千行的数组场景。

import numpy as np

# 示例原数组
a = np.array([[1,1,1.01, 0.5, 0.5, 1.02, 1, 1,1,1.21, 0.5, 0.5, 1.22, 1.3], 
              [1,1.4,1.01, 0.5, 0.5, 1.02, 1, 1,1,1.51, 0.5, 0.7, 1.22, 1]])

# 预标记小于1和大于1的元素位置
mask_low = a < 1
mask_high = a > 1
rows, cols = a.shape
window_size = 7
offset = window_size - 1  # 窗口内除核心元素外的偏移量

# 生成目标数组1
target1 = np.zeros_like(a, dtype=int)
for row_idx in range(rows):
    # 获取当前行中小于1的元素索引
    low_positions = np.where(mask_low[row_idx])[0]
    if not len(low_positions):
        continue
    # 标记所有属于小于1元素前窗口的位置
    window_mark = np.zeros(cols, dtype=bool)
    for pos in low_positions:
        start = max(0, pos - offset)
        window_mark[start:pos+1] = True
    # 仅保留窗口内且大于1的位置，设为1
    target1[row_idx] = np.where(window_mark & mask_high[row_idx], 1, 0)

# 生成目标数组2
target2 = np.zeros_like(a, dtype=int)
for row_idx in range(rows):
    low_positions = np.where(mask_low[row_idx])[0]
    if not len(low_positions):
        continue
    window_mark = np.zeros(cols, dtype=bool)
    for pos in low_positions:
        end = min(cols-1, pos + offset)
        window_mark[pos:end+1] = True
    target2[row_idx] = np.where(window_mark & mask_high[row_idx], 1, 0)

# 验证结果
print("目标数组1：")
print(target1)
print("\n目标数组2：")
print(target2)

方法二：无循环向量化实现（高效，适合超大数组） #

利用Numpy滑动窗口和数组反转操作，完全避免行循环，处理上万行/列的数组时效率更高。

import numpy as np

a = np.array([[1,1,1.01, 0.5, 0.5, 1.02, 1, 1,1,1.21, 0.5, 0.5, 1.22, 1.3], 
              [1,1.4,1.01, 0.5, 0.5, 1.02, 1, 1,1,1.51, 0.5, 0.7, 1.22, 1]])

mask_low = a < 1
mask_high = a > 1
rows, cols = a.shape
window_size =7
offset = window_size -1

# 生成目标数组1：大于1元素且后续7窗口内存在小于1元素
# 滑动窗口判断每个位置开始的7元素窗口内是否有小于1的元素
window_view = np.lib.stride_tricks.sliding_window_view(mask_low, window_size, axis=1)
has_low_in_window = np.any(window_view, axis=2)
# 处理末尾无法形成完整窗口的部分
tail = np.zeros((rows, offset), dtype=bool)
for i in range(1, window_size):
    tail[:, i-1] = np.any(mask_low[:, cols - i:], axis=1)
full_has_low = np.concatenate([has_low_in_window, tail], axis=1)
target1 = np.where(mask_high & full_has_low, 1, 0)

# 生成目标数组2：大于1元素且前面7窗口内存在小于1元素
# 反转数组后复用滑动窗口逻辑
reversed_mask = mask_low[:, ::-1]
reversed_window_view = np.lib.stride_tricks.sliding_window_view(reversed_mask, window_size, axis=1)
has_low_reversed = np.any(reversed_window_view, axis=2)
tail_reversed = np.zeros((rows, offset), dtype=bool)
for i in range(1, window_size):
    tail_reversed[:, i-1] = np.any(reversed_mask[:, cols - i:], axis=1)
full_has_low_reversed = np.concatenate([has_low_reversed, tail_reversed], axis=1)
# 反转回原顺序
full_has_low_before = full_has_low_reversed[:, ::-1]
target2 = np.where(mask_high & full_has_low_before, 1, 0)

# 验证结果
print("目标数组1：")
print(target1)
print("\n目标数组2：")
print(target2)

结果验证 #

运行上述代码后，输出结果与示例完全一致：

• 目标数组1：

array([[0, 0, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
       [0, 1, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0]])

• 目标数组2：

array([[0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 1, 1],
       [0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1]])

内容的提问来源于stack exchange，提问作者Sadcow