使用掩码将 2D 阵列的部分替换为较小的 2D 阵列

Question

如何用较小的 2D numpy 数组替换大型 2D numpy 数组中的多个模式实例？

我正在寻找一个使用布尔掩码的矢量化解决方案，以尽量减少对性能的影响，因为我正在处理的大型数组将有数百万行长。

例如：

#Large array
largeArr = np.array([
    [0, 1, 1],
    [0, 1, 1],
    [0, 1, 1],
    [0, 0, 0],
    [0, 0, 0],
    [0, 0, 0],
    [0, 1, 1],
    [0, 1, 1],
    [0, 1, 1],
    [0, 0, 0],
    [0, 0, 0],
    [3, 2, 0],
    [3, 2, 0],
    [3, 2, 0],
    [3, 2, 0],
    [0, 0, 0],
    [0, 0, 0],
    [3, 2, 0],
    [3, 2, 0],
    [3, 2, 0],
    [3, 2, 0],
    [0, 0, 0]
])

我想用包含 [0, 1, 1] 的连续 3 行替换部分

pattern1 = [
    [0, 2, 1],
    [0, 2, 2],
    [0, 2, 3]
]

然后我想用包含[3, 2, 0]的连续4行替换部分

pattern2 = [
    [5, 2, 1],
    [5, 3, 2],
    [5, 4, 3],
    [5, 5, 4]
]

预期结果：

[[0, 2, 1],
 [0, 2, 2],
 [0, 2, 3],
 [0, 0, 0],
 [0, 0, 0],
 [0, 0, 0],
 [0, 2, 1],
 [0, 2, 2],
 [0, 2, 3],
 [0, 0, 0],
 [0, 0, 0],
 [5, 2, 1],
 [5, 3, 2],
 [5, 4, 3],
 [5, 5, 4],
 [0, 0, 0],
 [0, 0, 0],
 [5, 2, 1],
 [5, 3, 2],
 [5, 4, 3],
 [5, 5, 4],
 [0, 0, 0]]

将有多个模式要查找和替换，每个模式都有自己的替换数组。目的是一次循环遍历提供的搜索行和替换模式。

搜索行始终是单行，重复次数与替换模式中的行数一样多。

Answer 1

我假设您所有的数量都是数组，而不是列表。如果不是这样，请将它们包装在np.array：

search = np.array([0, 1, 1])
pattern = np.array(pattern1)

块的大小由下式给出

n = len(pattern)  # or pattern.shape[0]

我假设您只想替换不重叠的段。因此，虽然 6 行 search 在输出中恰好对应两个 pattern 实例，但七行对应两个 pattern 实例和一个 search 实例。

搜索模式很简单。首先创建行与模式匹配的掩码：

mask = (largeArr == search).all(1)

在这个网站上寻找连续运行的面具的成语已经被打死了。要点是使用np.diff 查找掩码符号更改的位置，然后使用np.flatnonzero 获取索引，并再次使用np.diff 计算运行长度。首先填充掩码以确保结果正确包含端点：

indices = np.flatnonzero(np.diff(np.r_[False, mask, False])).reshape(-1, 2)
runs = np.diff(indices, axis=1).squeeze()

请注意，为方便起见，indices 已重新调整为两列。传递的 l 填充保证这是可能的。第一列是每次运行的开始（包括），而第二列是结束（不包括）。这使得计算runs 中的运行长度变得微不足道。

现在您可以调整 indices 以仅包含大小为 n 或更长的运行，并将结束元素修剪为远离开始元素的 n 的倍数：

# runs = n * (runs // n), but for huge arrays
np.floor_divide(runs, n, out=runs)
np.multiply(runs, n, out=runs)

indices[:, 1] = indices[:, 0] + runs

您可以使用indices = indices[np.flatnonzero(runs)] 修剪零长度的indices，但这不是必需的。下一步是将调整后的indices 转换回掩码：

mask = np.zeros_like(mask, dtype=np.int8)

np.uint8 dtype 允许您在掩码中存储 +1 和 -1，并且与 np.bool_ 的大小相同，这意味着如果处理得当，最终结果可以无缝地查看为布尔掩码：

starts, ends = indices.T
if ends[-1] == mask.size:
    ends = ends[:-1]
mask[starts] = 1
mask[ends] -= 1  # This takes care of zero-length segments automatically
mask = np.cumsum(mask, out=mask).view(bool)

ends 的额外处理，被解包为indices 的第二列，负责处理掩码运行到数组末尾的情况。由于结束索引是独占的，这将超过数组的末尾，但这也意味着根本不需要终止该运行。

现在您的蒙版已被过滤和修剪，您可以分配给largeArr 中的正确行。最简单的方法是根据需要多次重复pattern：

largeArr[mask, :] = np.tile(pattern, [runs.sum() // n, 1])

如果你把它打包成一个函数，你可以为多个模式运行它：

def replace_pattern(arr, search, pattern):
    n = len(pattern)
    mask = (arr == search).all(1)
    indices = np.flatnonzero(np.diff(np.r_[False, mask, False])).reshape(-1, 2)
    runs = np.diff(indices, axis=1).squeeze()
    np.floor_divide(runs, n, out=runs)
    np.multiply(runs, n, out=runs)
    indices[:, 1] = indices[:, 0] + runs
    mask = np.zeros_like(mask, dtype=np.int8)
    starts, ends = indices.T
    if ends[-1] == mask.size:
        ends = ends[:-1]
    mask[starts] = 1
    mask[ends] -= 1
    mask = np.cumsum(mask, out=mask).view(bool)
    arr[mask, :] = np.tile(pattern, [runs.sum() // n, 1])

replacements = [
    ([0, 1, 1], [[0, 2, 1],
                 [0, 2, 2],
                 [0, 2, 3]]),
    ([3, 2, 0], [[5, 2, 1],
                 [5, 3, 2],
                 [5, 4, 3],
                 [5, 5, 4]])
]

largeArr = np.array(...)

for search, pattern in replacements:
    replace_pattern(largeArr, search, pattern)