如何优化取决于迭代中每一步的几个条件的循环？

Question

我必须编写代码来检查矩阵 2000x2000 中的每个点是否满足所有给定条件。为了加快迭代速度，我将一些计算放在循环之外。是否有任何方法可以固定这个 for 循环或使用其他方法在满足所有这些条件的地方获得值为 1 的矩阵 A ？我虽然关于 np.where 函数或创建 True 和 False 元素的矩阵 - 但条件会通过迭代发生变化，因此最终需要更长的时间。

import numpy as np

A = np.zeros(2000, 2000)

(i, j) = A.shape
x = list(range(int(-0.5 * i), int(0.5 * i), 1))
y = list(range(int(-0.5 * j), int(0.5 * j), 1))
r = 50
w = 30
h = 150

w_tr = np.array([(-0.5 * w, 0.5 * h), (0.5 * w, 0.5 * h), (0.5 * w, -0.5 * h), (-0.5 * w, -0.5 * h)]).T


x_squared_A = list((np.array(x) - w_tr[0, 0]) ** 2)  # list (x-a)^2 for  A
y_squared_A = list((np.array(y) - w_tr[1, 0]) ** 2)  # list (y-b)^2 for  A
x_squared_B = list((np.array(x) - w_tr[0, 1]) ** 2)  # list (x-a)^2 for  B
y_squared_B = list((np.array(y) - w_tr[1, 1]) ** 2)  # list (y-b)^2 for  B
x_squared_C = list((np.array(x) - w_tr[0, 2]) ** 2)  # list (x-a)^2 for  C
y_squared_C = list((np.array(y) - w_tr[1, 2]) ** 2)  # list (y-b)^2 for  C
x_squared_D = list((np.array(x) - w_tr[0, 3]) ** 2)  # list (x-a)^2 for  D
y_squared_D = list((np.array(y) - w_tr[1, 3]) ** 2)  # list (y-b)^2 for  D
r_squared = r ** 2


for m in range(i):
    for vm in range(j):
        if (w_tr[1, 2] <= y[vm] < w_tr[1, 0] and w_tr[0, 0] - r <= x[m] < w_tr[0, 1] + r) or \
                (w_tr[1, 2] - r <= y[vm] < w_tr[1, 0] + r and w_tr[0, 0] <= x[m] < w_tr[0, 1]) or \
                x_squared_A[m] + y_squared_A[vm] < r_squared or \
                x_squared_B[m] + y_squared_B[vm] < r_squared or \
                x_squared_C[m] + y_squared_C[vm] < r_squared or \
                x_squared_D[m] + y_squared_D[vm] < r_squared:
            A[vm, m] = 1
        else:
            continue

Answer 1

您的循环中似乎存在一些冗余计算。例如，w_tr[1, 2] - r 是一个常量，因为如果我没记错的话，w_tr 不会在您的循环中被修改。因此，您可以这样做一次。 w_tr[1, 2] 相同，因为计算需要访问字典。因此，您可以通过将一些不依赖于 for 循环变量的计算移出循环来执行各种简单的优化。

这给出了以下代码：

w_tr12 = w_tr[1, 2]
w_tr12_minus_r = w_tr[1, 2] - r
w_tr00 = w_tr[0, 0]
w_tr00_minus_r = w_tr[0, 0] - r
w_tr01 = w_tr[0, 1]
w_tr01_plus_r = w_tr[0, 1] + r
w_tr10 = w_tr[1, 0]
w_tr10_plus_r = w_tr[1, 0] + r

for m in range(i):
    xm = x[m]
    x_squared_Am = x_squared_A[m]
    x_squared_Bm = x_squared_B[m]
    x_squared_Cm = x_squared_C[m]
    x_squared_Dm = x_squared_D[m]

    cond1 = w_tr00_minus_r <= xm < w_tr01_plus_r
    cond2 = w_tr00 <= xm < w_tr01
    for vm in range(j):
        if (cond1 and w_tr12 <= y[vm] < w_tr10) or \
           (cond2 and w_tr12_minus_r <= y[vm] < w_tr10_plus_r) or \
                x_squared_Am + y_squared_A[vm] < r_squared or \
                x_squared_Bm + y_squared_B[vm] < r_squared or \
                x_squared_Cm + y_squared_C[vm] < r_squared or \
                x_squared_Dm + y_squared_D[vm] < r_squared:
            A[vm, m] = 1

不是很大，但我可以在我的机器上从 19.5 秒到 8.5。如果您想要更好的优化，@mike67 提出的多线程似乎是最好的选择。

Answer 2

感谢您的提示，我想为您的答案添加评论，但有 500 个标志的限制，因此我将其发布为答案。我尝试使用多处理的方式，即 for 循环将在单一条件下同时运行 6 次，例如 x_squared_Cm + y_squared_C[vm]

import concurrent.futures
import numpy as np



def loop_1(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_1 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if w_tr[1, 2] <= y[vm] < w_tr[1, 0] and w_tr[0, 0]-r <= x[m] < w_tr[0, 1]+r:
                    A_1[vm, m] = 1
            else:
                continue
    return A_1

def loop_2(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_2 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if w_tr[1, 2]-r <= y[vm] < w_tr[1, 0]+r and w_tr[0, 0] <= x[m] < w_tr[0, 1]:
                    A_2[vm, m] = 1
            else:
                continue
    return A_2

def loop_3(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_3 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if  x_squared_A[m] + y_squared_A[vm] < r_squared:
                    A_3[vm, m] = 1
            else:
                continue
    return A_3

def loop_4(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_4 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if x_squared_B[m] + y_squared_B[vm] < r_squared:
                    A_4[vm, m] = 1
            else:
                continue
    return A_4

def loop_5(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_5 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if x_squared_C[m] + y_squared_C[vm] < r_squared:
                    A_5[vm, m] = 1
            else:
                continue
    return A_5

def loop_6(i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B, y_squared_C, y_squared_D, r_squared, max_r):
    A_6 = A
    for m in range(int(0.5 * i - max_r - 2), int(0.5 * i + max_r + 2)):
        for vm in range(int(0.5 * j - max_r - 2), int(0.5 * j + max_r + 2)):
            if x_squared_D[m] + y_squared_D[vm] < r_squared:
                    A_6[vm, m] = 1
            else:
                continue
    return A_6 


def main():
    w = 30
    h = 150
    r = 50

    A = np.zeros((2000, 2000))  

    (i, j) = A.shape  # i,j współrzędne macierzy,  x,y  współrzędne układu
    x = list(range(int(-0.5 * i), int(0.5 * i), 1))
    y = list(range(int(-0.5 * j), int(0.5 * j), 1))

    wierzcholki = np.array([(-0.5 * w, 0.5 * h), (0.5 * w, 0.5 * h), (0.5 * w, -0.5 * h), (-0.5 * w, -0.5 * h)]).T
    w_tr = wierzcholki

    x_squared_A = list((np.array(x) - w_tr[0, 0]) ** 2)
    y_squared_A = list((np.array(y) - w_tr[1, 0]) ** 2)
    x_squared_B = list((np.array(x) - w_tr[0, 1]) ** 2)
    y_squared_B = list((np.array(y) - w_tr[1, 1]) ** 2)
    x_squared_C = list((np.array(x) - w_tr[0, 2]) ** 2)
    y_squared_C = list((np.array(y) - w_tr[1, 2]) ** 2)
    x_squared_D = list((np.array(x) - w_tr[0, 3]) ** 2)
    y_squared_D = list((np.array(y) - w_tr[1, 3]) ** 2)
    r_squared = r ** 2

    max_r = np.sqrt(
        (0.5 * w + r) ** 2 + (0.5 * h + r) ** 2)

    args = [i,j, r, A, x, y, w_tr, x_squared_A, x_squared_B, x_squared_C, x_squared_D, y_squared_A, y_squared_B,
            y_squared_C, y_squared_D, r_squared, max_r]

    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = [executor.map(loop_1, args),
                   executor.map(loop_2, args),
                   executor.map(loop_3, args),
                   executor.map(loop_4, args),
                   executor.map(loop_5, args),
                   executor.map(loop_6, args)]

    for result in results:
        A += result


if __name__ == '__main__':
    main()