我有一个23x23x30的空间，每个1x1x1的多维数据集代表一个点，这些23x23x30点中的一些填充了-65 to -45范围内的数字，我想确保不超过1个数字在填充点周围的5x5x30的任何给定区域中，如果在5x5x30的任何区域中有多个点，则应消除数量最少的点。我已经使用嵌套的for循环以串行方式完成了此操作，但这是非常昂贵的操作。我想并行化此操作。我有n个核心，每个核心都有它自己的23x23x30总区域子区域，没有任何重叠。我可以收集那些“子区域”并构造上面提到的23x23x30的整个区域，以便所有内核也可以同时拥有它们的“子区域”来访问23x23x30的整个区域。我不确定在python中是否有任何库可用于这种操作。在我的应用程序中，8个进程将用大约23x23x30个点填充该3500空间，现在我要对所有8个进程（即重复工作）执行此“过滤”操作，这是资源的浪费，所以我为了有效地使用可用资源，将必须并行执行此“过滤”操作。
这是序列号：self.tntatv_stdp_ids是字典，键为step1，step2 ....多达30个维度，为z。此键具有该步骤中已填充点的数字（1 to 529）。请注意，在代码的串行实现中，z维度中每个步骤中的点均来自1 to 529。self.cell_voltages是具有键step1，step2 ....最多30个维度z的词典。每个键给出一个点中存在的数字。
a_keys = self.tntatv_stdp_ids.keys()
#Filter tentative neuron ids using suppression algo to come up with final stdp neuron ids.
for i in range(0,len(a_keys)):
b_keys= list(set(a_keys) - set([a_keys[i]]))
c_keys = self.tntatv_stdp_ids[a_keys[i]]
for j in range(0,len(b_keys)):
d_keys=self.tntatv_stdp_ids[b_keys[j]]
for k in c_keys[:]:
key = k
key_row= key/(image_size-kernel+1)
key_col = key%(image_size-kernel+1)
remove =0
for l in d_keys[:]:
target = l
tar_row = target/(image_size-kernel+1)
tar_col = target%(image_size-kernel+1)
if(abs(key_row-tar_row) > kernel-1 and abs(key_col-tar_col) > kernel-1):
pass
else:
if(self.cell_voltages[a_keys[i]][key]>=self.cell_voltages[b_keys[j]][target]):
d_keys.remove(target)
else:
remove+=1
if(remove):
c_keys.remove(key)
在此操作结束时，如果
30的23x23x1区域中还有多个点，则可以通过查看剩余的填充点中的哪个来选择每个30 23x23x1区域的一个最终获胜者点的23x23x1个分数最高。这样，从30的所有点数中可以得出最大获胜者数量为23x23x30，每个23x23x1得1。也可以少于30，这取决于开始时填充了多少23x23x30个点。 最佳答案
这个问题可能不需要并行化：
# Generate a random array of appropriate size for testing
super_array = [[[None for _ in range(30)] for _ in range(529)] for _ in range(529)]
for _ in range(3500):
super_array[random.randint(0, 528)][random.randint(0, 528)][random.randint(0,29)] = random.randint(-65, -45)
第一步是构建填充节点的列表：
filled = []
for x in range(len(super_array)):
for y in range(len(super_array[0])):
for z in range(len(super_array[0][0])):
if super_array[x][y][z] is not None:
filled.append((x, y, z, super_array[x][y][z]))
然后，将列表从高到低排序：
sfill = sorted(filled, key=lambda x: x[3], reverse=True)
现在，生成一个阻塞网格：
block_array = [[None for _ in range(529)] for _ in range(529)]
遍历该列表，在找到节点时阻止邻居，并删除已经占用的邻居中的节点：
for node in sfill:
x, y, z, _ = node
if block_array[x][y] is not None:
super_array[x][y][z] = None # kill node if it's in the neighborhood of a larger node
else: # Block their neighborhood
for dx in range(5):
for dy in range(5):
cx = x + dx - 2
cy = y + dy - 2
if 529 > cx >= 0 and 529 > cy >= 0:
block_array[cx][cy] = True
一些注意事项：
这使用滑动邻域，因此它检查以每个节点为中心的5x5。从最高到最低进行检查很重要，因为这可以确保被删除的节点以前没有强制要删除其他节点。
您可以通过执行范围而不是完整的529x529数组来更有效地执行此操作，但是邻域阻塞只需不到一秒钟的时间，从生成的数组到修剪的最终列表的整个过程为1.2秒。
通过仅在任何
z堆栈中添加最高价值的节点，可以改善填充节点列表的构建。如果大量节点最终具有相同的x，y值，这将减小必须排序的列表的大小。在23x23x30上，它需要大约18ms的时间，再次包括了构建3d阵列的时间：
timeit.timeit(prune_test, number=1000)
17.61786985397339