根据 numpy 数组中的行生成唯一值

Question

我有一个 3D numpy 数组 arr，形状为 m*n*k。

对于沿m 轴的每组值（例如arr[:, 0, 0]），我想生成一个值来表示该组，以便最终得到一个二维矩阵n*k。 如果沿m 轴重复一组值，那么我们应该每次生成相同的值。

即这是一个哈希问题。

我使用字典创建了该问题的解决方案，但它大大降低了性能。对于每组值，我调用这个函数：

 def getCellId(self, valueSet):

     # Turn the set of values (a numpy vector) to a tuple so it can be hashed
     key = tuple(valueSet)

     # Try and simply return an existing ID for this key
     try:
       return self.attributeDict[key]
     except KeyError:

       # If the key was new (and didnt exist), try and generate a new Id by adding one to the max of all current Id's. This will fail the very first time we do this (as there will be no Id's yet), so in that case, just assign the value '1' to the newId
       try:
         newId = max(self.attributeDict.values()) +1
       except ValueError:
         newId = 1
       self.attributeDict[key] = newId
       return newId

数组本身的大小通常为 30*256*256，因此一组值将有 30 个值。 我随时都有数百个这样的数组要处理。 目前，完成所有需要完成的处理以计算哈希 100 个数组的块需要 1.3 秒。 包括高达 75 秒的散列颠簸。

有没有更快的方法来生成单个代表值？

Answer 1

这可能是使用基本 numpy 函数的一种方法 -

import numpy as np

# Random input for demo
arr = np.random.randint(0,3,[2,5,4])

# Get dimensions for later usage
m,n,k = arr.shape

# Reshape arr to a 2D array that has each slice arr[:, n, k] in each row
arr2d = np.transpose(arr,(1,2,0)).reshape([-1,m])

# Perform lexsort & get corresponding indices and sorted array 
sorted_idx = np.lexsort(arr2d.T)
sorted_arr2d =  arr2d[sorted_idx,:]

# Differentiation along rows for sorted array
df1 = np.diff(sorted_arr2d,axis=0)

# Look for changes along df1 that represent new labels to be put there
df2 = np.append([False],np.any(df1!=0,1),0)

# Get unique labels
labels = df2.cumsum(0)

# Store those unique labels in a n x k shaped 2D array
pos_labels = np.zeros_like(labels)
pos_labels[sorted_idx] = labels
out = pos_labels.reshape([n,k])

示例运行 -

In [216]: arr
Out[216]: 
array([[[2, 1, 2, 1],
        [1, 0, 2, 1],
        [2, 0, 1, 1],
        [0, 0, 1, 1],
        [1, 0, 0, 2]],

       [[2, 1, 2, 2],
        [0, 0, 2, 1],
        [2, 1, 0, 0],
        [1, 0, 1, 0],
        [0, 1, 1, 0]]])

In [217]: out
Out[217]: 
array([[6, 4, 6, 5],
       [1, 0, 6, 4],
       [6, 3, 1, 1],
       [3, 0, 4, 1],
       [1, 3, 3, 2]], dtype=int32)

Answer 2

根据需要生成多少新密钥和旧密钥，很难说什么是最佳的。但是使用您的逻辑，以下应该相当快：

import collections
import hashlib

_key = 0

def _get_new_key():
    global _key
    _key += 1
    return _key

attributes = collections.defaultdict(_get_new_key)

def get_cell_id(series):                             
    global attributes
    return attributes[hashlib.md5(series.tostring()).digest()]

编辑：

我现在更新了根据您的问题使用 strides 循环所有数据系列：

In [99]: import numpy as np

In [100]: A = np.random.random((30, 256, 256))

In [101]: A_strided = np.lib.stride_tricks.as_strided(A, (A.shape[1] * A.shape[2], A.shape[0]), (A.itemsize, A.itemsize * A.shape[1] * A.shape[2]))

In [102]: %timeit tuple(get_cell_id(S) for S in A_strided)
10 loops, best of 3: 169 ms per loop

上面每个 30 个元素数组进行 256x256 次查找/分配。 当然不能保证 md5 哈希不会发生冲突。如果这应该是一个问题，您当然可以更改为同一个库中的其他哈希。

编辑 2：

鉴于您似乎在 3D 阵列的第一个轴上执行了大部分昂贵的操作，我建议您重新组织您的阵列：

In [254]: A2 = np.random.random((256, 256, 30))

In [255]: A2_strided = np.lib.stride_tricks.as_strided(A2, (A2.shape[0] * A2.shape[1], A2.shape[2]), (A2.itemsize * A2.shape[2], A2.itemsize))

In [256]: %timeit tuple(get_cell_id(S) for S in A2_strided)
10 loops, best of 3: 126 ms per loop

不必在内存中长距离跳转，速度提高了大约 25%

编辑 3：

如果实际上不需要将哈希缓存到int 查找，但您只需要实际的哈希，并且如果 3D 数组是 int8 类型，则给定 A2 和 @987654327 @组织，时间可以再减少一些。这 15 毫秒是元组循环。

In [9]: from hashlib import md5

In [10]: %timeit tuple(md5(series.tostring()).digest() for series in A2_strided) 
10 loops, best of 3: 72.2 ms per loop

Answer 3

如果只是散列，试试这个

import numpy as np
import numpy.random

# create random data
a = numpy.random.randint(10,size=(5,3,3))

# create some identical 0-axis data
a[:,0,0] = np.arange(5)
a[:,0,1] = np.arange(5)

# create matrix with the hash values
h = np.apply_along_axis(lambda x: hash(tuple(x)),0,a)

h[0,0]==h[0,1]
# Output: True

但是，请谨慎使用它并首先使用您的代码测试此代码。 ...我只能说它适用于这个简单的例子。

此外，虽然两个值不同，但它们可能具有相同的哈希值。这是一个使用散列函数总是会发生的问题，但可能性很小

编辑：为了与其他解决方案进行比较

timeit(np.apply_along_axis(lambda x: hash(tuple(x)),0,a))
# output: 1 loops, best of 3: 677 ms per loop