为多个数组实现 numpy.in1d 的最有效方法

Most efficient way to implement numpy.in1d for muliple arrays

实现一个函数的最佳方法是什么,该函数采用任意数量的一维数组和 returns 包含匹配值索引(如果有)的元组。

这是我想做的一些伪代码:

a = np.array([1, 0, 4, 3, 2])
b = np.array([1, 2, 3, 4, 5])
c = np.array([4, 2])

(ind_a, ind_b, ind_c) = return_equals(a, b, c)
# ind_a = [2, 4]
# ind_b = [1, 3]
# ind_c = [0, 1]

(ind_a, ind_b, ind_c) = return_equals(a, b, c, sorted_by=a)
# ind_a = [2, 4]
# ind_b = [3, 1]
# ind_c = [0, 1]

def return_equals(*args, sorted_by=None):
    ...

首先,我会尝试:

def return_equals(*args):
    x=[]
    c=args[-1]
    for a in args:
        x.append(np.nonzero(np.in1d(a,c))[0])
    return x

如果我添加一个d=np.array([1,0,4,3,0])(它只有1个匹配;如果没有匹配怎么办?)

然后

return_equals(a,b,d,c)

产生:

[array([2, 4], dtype=int32),
 array([1, 3], dtype=int32),
 array([2], dtype=int32),
 array([0, 1], dtype=int32)]

由于输入数组和返回数组的长度可能不同,因此您确实无法将问题向量化。也就是说,需要一些特殊的体操才能同时对所有输入执行操作。如果数组的数量与其典型长度相比较小,我就不会担心速度。迭代几次并不昂贵。它迭代了 100 个昂贵的值。

当然,您可以将关键字参数传递给 in1d

不清楚您尝试使用 sorted_by 参数做什么。在将数组传递给此函数之前,您是否可以轻松地将其应用于数组?


本次迭代的列表理解版本:

 [np.nonzero(np.in1d(x,c))[0] for x in [a,b,d,c]]

我可以想象将数组连接成一个较长的数组,应用 in1d,然后将其拆分为子数组。有一个np.split,但它需要你告诉它每个子列表要放多少元素。这意味着,以某种方式确定每个参数有多少匹配项。在没有循环的情况下这样做可能很棘手。

这个(仍然需要打包为函数)的部分是:

args=[a,b,d,c]
lens=[len(x) for x in args]
abc=np.concatenate(args)
C=np.cumsum(lens)
I=np.nonzero(np.in1d(abc,c))[0]
S=np.split(I,(2,4,5))
[S[0],S[1]-C[0],S[2]-C[1],S[3]-C[2]]

I
# array([ 2,  4,  6,  8, 12, 15, 16], dtype=int32)
C
# array([ 5, 10, 15, 17], dtype=int32)

(2,4,5)C 的连续值之间的 I 的元素数,即每个 a、[= 匹配的元素数24=],...

您可以将 numpy.intersect1dreduce 结合使用:

def return_equals(*arrays):
    matched = reduce(np.intersect1d, arrays)
    return np.array([np.where(np.in1d(array, matched))[0] for array in arrays])

reduce 在这里可能有点慢,因为我们在这里创建中间 NumPy 数组(对于大量输入,它可能非常慢),如果我们使用 Python 可以防止这种情况set 及其 .intersection() 方法:

matched = np.array(list(set(arrays[0]).intersection(*arrays[1:])))

相关GitHub 工单:n-array versions of set operations, especially intersect1d

在Python中:

def return_equal(*args):
    rtr=[]
    for i, arr in enumerate(args):
        rtr.append([j for j, e in enumerate(arr) if 
                    all(e in a for a in args[0:i]) and 
                    all(e in a for a in args[i+1:])])
    return rtr    

>>> return_equal(a,b,c) 
[[2, 4], [1, 3], [0, 1]]

此解决方案基本上将所有输入 1D 数组连接成一个大 1D 数组,目的是在 vectorized manner 中执行所需的操作。它使用循环的唯一地方是在开始时它获取输入数组的长度,这在运行时成本必须是最小的。

这里是函数实现 -

import numpy as np

def return_equals(*argv):
    # Concatenate input arrays into one big array for vectorized processing
    A = np.concatenate((argv[:]))

    # lengths of input arrays
    narr = len(argv)
    lens = np.zeros((1,narr),int).ravel()
    for i in range(narr):
        lens[i] = len(argv[i])  

    N = A.size

    # Start indices of each group of identical elements from different input arrays
    # in a sorted version of the huge concatenated input array
    start_idx = np.where(np.append([True],np.diff(np.sort(A))!=0))[0]

    # Runlengths of islands of identical elements
    runlens = np.diff(np.append(start_idx,N))

    # Starting and all indices of the positions in concatenate array that has 
    # islands of identical elements which are present across all input arrays
    good_start_idx = start_idx[runlens==narr]
    good_all_idx = good_start_idx[:,None] + np.arange(narr)

    # Get offsetted indices and sort them to get the desired output
    idx = np.argsort(A)[good_all_idx] - np.append([0],lens[:-1].cumsum())
    return np.sort(idx.T,1)