根据键翻译numpy数组中的每个元素

问题描述：

我试图numpy.array根据给定的键翻译 a 的每个元素：

例如：

a = np.array([[1,2,3],
              [3,2,4]])

my_dict = {1:23, 2:34, 3:36, 4:45}

我想要得到：

array([[ 23.,  34.,  36.],
       [ 36.,  34.,  45.]])

我可以看到如何使用循环来实现这一点：

def loop_translate(a, my_dict):
    new_a = np.empty(a.shape)
    for i,row in enumerate(a):
        new_a[i,:] = map(my_dict.get, row)
    return new_a

有没有更有效和/或纯粹的 numpy 方法？

编辑：

我对其进行了计时，np.vectorize对于较大的数组，DSM 提出的方法要快得多：

In [13]: def loop_translate(a, my_dict):
   ....:     new_a = np.empty(a.shape)
   ....:     for i,row in enumerate(a):
   ....:         new_a[i,:] = map(my_dict.get, row)
   ....:     return new_a
   ....: 

In [14]: def vec_translate(a, my_dict):    
   ....:     return np.vectorize(my_dict.__getitem__)(a)
   ....: 

In [15]: a = np.random.randint(1,5, (4,5))

In [16]: a
Out[16]: 
array([[2, 4, 3, 1, 1],
       [2, 4, 3, 2, 4],
       [4, 2, 1, 3, 1],
       [2, 4, 3, 4, 1]])

In [17]: %timeit loop_translate(a, my_dict)
10000 loops, best of 3: 77.9 us per loop

In [18]: %timeit vec_translate(a, my_dict)
10000 loops, best of 3: 70.5 us per loop

In [19]: a = np.random.randint(1, 5, (500,500))

In [20]: %timeit loop_translate(a, my_dict)
1 loops, best of 3: 298 ms per loop

In [21]: %timeit vec_translate(a, my_dict)
10 loops, best of 3: 37.6 ms per loop

In [22]:  %timeit loop_translate(a, my_dict)

解决方案 1：

我不知道效率如何，但你可以使用字典的np.vectorize方法.get：

>>> a = np.array([[1,2,3],
              [3,2,4]])
>>> my_dict = {1:23, 2:34, 3:36, 4:45}
>>> np.vectorize(my_dict.get)(a)
array([[23, 34, 36],
       [36, 34, 45]])

解决方案 2：

这是另一种方法，使用numpy.unique：

>>> a = np.array([[1,2,3],[3,2,1]])
>>> a
array([[1, 2, 3],
       [3, 2, 1]])
>>> d = {1 : 11, 2 : 22, 3 : 33}
>>> u,inv = np.unique(a,return_inverse = True)
>>> np.array([d[x] for x in u])[inv].reshape(a.shape)
array([[11, 22, 33],
       [33, 22, 11]])

np.vectorize这种方法比数组中唯一元素数量较少时的方法
要快得多。说明： Python 速度很慢，在这种方法中，使用 Python 内部循环来转换唯一元素，之后我们依靠高度优化的 numpy 索引操作（用 C 完成）进行映射。因此，如果唯一元素的数量与数组的整体大小相当，则不会加速。另一方面，如果只有几个唯一元素，那么您可以观察到高达 100 倍的加速。

解决方案 3：

我认为最好遍历字典并“一次性”设置所有行和列的值：

>>> a = np.array([[1,2,3],[3,2,1]])
>>> a
array([[1, 2, 3],
       [3, 2, 1]])
>>> d = {1 : 11, 2 : 22, 3 : 33}
>>> for k,v in d.iteritems():
...     a[a == k] = v
... 
>>> a
array([[11, 22, 33],
       [33, 22, 11]])

编辑：

虽然它可能不像DSM（非常好）的答案那样性感，但numpy.vectorize我对所有提出的方法的测试表明，这种方法（使用@jamylak 的建议）实际上更快一些：

from __future__ import division
import numpy as np
a = np.random.randint(1, 5, (500,500))
d = {1 : 11, 2 : 22, 3 : 33, 4 : 44}

def unique_translate(a,d):
    u,inv = np.unique(a,return_inverse = True)
    return np.array([d[x] for x in u])[inv].reshape(a.shape)

def vec_translate(a, d):    
    return np.vectorize(d.__getitem__)(a)

def loop_translate(a,d):
    n = np.ndarray(a.shape)
    for k in d:
        n[a == k] = d[k]
    return n

def orig_translate(a, d):
    new_a = np.empty(a.shape)
    for i,row in enumerate(a):
        new_a[i,:] = map(d.get, row)
    return new_a


if __name__ == '__main__':
    import timeit
    n_exec = 100
    print 'orig'
    print timeit.timeit("orig_translate(a,d)", 
                        setup="from __main__ import np,a,d,orig_translate",
                        number = n_exec) / n_exec
    print 'unique'
    print timeit.timeit("unique_translate(a,d)", 
                        setup="from __main__ import np,a,d,unique_translate",
                        number = n_exec) / n_exec
    print 'vec'
    print timeit.timeit("vec_translate(a,d)",
                        setup="from __main__ import np,a,d,vec_translate",
                        number = n_exec) / n_exec
    print 'loop'
    print timeit.timeit("loop_translate(a,d)",
                        setup="from __main__ import np,a,d,loop_translate",
                        number = n_exec) / n_exec

输出：

orig
0.222067718506
unique
0.0472617006302
vec
0.0357889199257
loop
0.0285375618935

解决方案 4：

numpy_indexed包（免责声明：我是它的作者）为此类问题提供了一种优雅而高效的矢量化解决方案：

import numpy_indexed as npi
remapped_a = npi.remap(a, list(my_dict.keys()), list(my_dict.values()))

实现的方法与 John Vinyard 提到的方法类似，但更加通用。例如，数组的项不必是 int，可以是任何类型，甚至是 nd-subarrays 本身。

如果将可选的“missing”kwarg 设置为“raise”（默认为“ignore”），性能会稍微好一些，并且如果键中不存在“a”的所有元素，您将收到 KeyError。

解决方案 5：

假设你的字典键是正整数，没有巨大的差距（类似于从 0 到 N 的范围），你最好将你的翻译字典转换为这样的数组my_array[i] = my_dict[i]，然后使用 numpy 索引进行翻译。

使用此方法的代码是：

def direct_translate(a, d):
    src, values = d.keys(), d.values()
    d_array = np.arange(a.max() + 1)
    d_array[src] = values
    return d_array[a]

使用随机数组测试：

N = 10000
shape = (5000, 5000)
a = np.random.randint(N, size=shape)
my_dict = dict(zip(np.arange(N), np.random.randint(N, size=N)))

对于这些尺寸，我采用140 ms这种方法。np.get 矢量化需要大约5.8 s和unique_translate大约8 s。

可能的概括：

• 如果您需要翻译负值，则可以将a字典中的值和键中的值移动一个常数，以将它们映射回正整数：

def direct_translate(a, d): # handles negative source keys
    min_a = a.min()
    src, values = np.array(d.keys()) - min_a, d.values()
    d_array = np.arange(a.max() - min_a + 1)
    d_array[src] = values
    return d_array[a - min_a]

• 如果源键之间存在巨大差距，则初始数组创建会浪费内存。我会求助于 cython 来加速该功能。

解决方案 6：

如果您确实不必使用字典作为替换表，那么简单的解决方案是（对于您的示例）：

a = numpy.array([your array])
my_dict = numpy.array([0, 23, 34, 36, 45])     # your dictionary as array

def Sub (myarr, table) :
    return table[myarr] 

values = Sub(a, my_dict)

当然，只有当的索引d涵盖了的所有可能值时，这才会起作用a，换句话说，只有对于a无符号整数，这才会起作用。

解决方案 7：

def dictonarize(np_array, dictonary, el_type='float'):
    
    final_array = np.zeros_like(np_array).astype(el_type)
    for x in dictonary:
        x_layer = (np_array == x)
        x_layer = (x_layer* dictonary[x]).astype(el_type)
        final_array += x_layer
        
    return final_array

解决方案 8：

结合@DSM和@John Vinyard 的解决方案：

• dict.__getitem__仅对唯一值进行矢量化。

• 使用 numpy 优化索引进行映射。

代码：

>>> a = np.array([[1,2,3],[3,2,1]])
>>> a
array([[1, 2, 3],
       [3, 2, 1]])
>>> d = {1 : 11, 2 : 22, 3 : 33}

>>> u, inv = np.unique(a, return_inverse=True)
>>> np.vectorize(d.get)(u)[inv].reshape(a.shape)
array([[11, 22, 33],
       [33, 22, 11]])

这具有与@DSM答案相同的优点，同时还避免了数组中唯一元素的 python 循环。

解决方案 9：

Pandas 为此目的提供了一个 map 函数，它比迄今为止发布的所有其他解决这个问题的解决方案都要快，也就是说，它可以将任何值映射到字典中定义的另一个值。

import numpy as np
import pandas as pd

def map_pandas(x, mapping_dict):
    return pd.Series(x.flatten()).map(mapping_dict).values.reshape(x.shape)

period = 6
x = np.random.randint(0, period, (5000,5000))
mapping = {i: i*11 for i in np.unique(x)}
map_pandas(x, mapping)

速度比较

x.shape = (5000, 5000)使用6 个映射键值对对所有提出的解决方案进行测试。

*每个函数调用 3 次运行，每次循环 3 次，都是

**非泛化函数，即需要已经索引的整数数组作为输入，因此仅适用于少数情况。

速度测试代码

import numpy as np
import pandas as pd
import numpy_indexed as npi

def map_npi(x, mapping_dict):
    return npi.remap(x.flatten(), list(mapping.keys()), list(mapping.values())).reshape(x.shape)

def map_np_unique(x, mapping_dict, default=np.nan):
    u, inv = np.unique(x, return_inverse = True)
    return np.array([mapping_dict.get(x, default) for x in u])[inv].reshape(x.shape)

def map_np_vectorize(x, mapping_dict):
    return np.vectorize(mapping.get)(x)

def map_np_unique_vectorize(x, mapping_dict):
    u, inv = np.unique(x, return_inverse = True)
    return np.vectorize(mapping.get)(u)[inv].reshape(x.shape)

def map_np_iteritems(x, mapping_dict):
    x_new = np.array(x)
    for k, v in mapping_dict.items():
        x_new[x == k] = v
    return x_new

def map_np_direct(x, mapping_dict):
    d_array = np.arange(x.max() + 1)
    d_array[list(mapping_dict.keys())] = list(mapping_dict.values())
    return d_array[x]

def map_np_direct2(x, mapping_dict):
    return np.array(list(mapping_dict.values()))[x]

period = 6
x = np.random.randint(0, period, (5000,5000))
# x = np.random.random((5000,5000)).round(1) # non generalising functions will fail
mapping = {i: i*11 for i in np.unique(x)}
result_probe = map_np_unique(x, mapping)
    
for f in [map_npi, map_np_vectorize, map_np_unique, map_np_unique_vectorize, 
          map_np_iteritems, map_pandas, map_np_direct, map_np_direct2]:
    print(f.__name__)
    try:
        assert (result_probe == f(x, mapping)).all()
    except AssertionError:
        print('Wrong result')
    except Exception as e:
        print(f'{e.__class__.__name__}: {e}')
    else:
        %timeit -n 3 -r 3 f(x, mapping)
    print()