Python中的大小增量Numpy数组

我刚刚在Python中遇到了增量Numpy数组的需要,因为我没有找到任何实现它的东西.我只是想知道我的方式是最好的方式还是你可以提出其他想法.

所以,问题是我有一个2D数组(程序处理nD数组),其大小事先是未知的,并且可变数据量需要在一个方向上连接到数组(让我们说我要去很多次打电话给np.vstak).每次我连接数据时,我都需要获取数组,沿轴0排序并执行其他操作,因此我无法构建一个长列表数组,然后立即对列表进行np.vstak.
由于内存分配很昂贵,我转向增量数组,其中我增加数量大于我需要的数量的数组(我使用50％增量),以便最小化分配数量.

我对此进行了编码,您可以在以下代码中看到它：

class ExpandingArray:

    __DEFAULT_ALLOC_INIT_DIM = 10   # default initial dimension for all the axis is nothing is given by the user
    __DEFAULT_MAX_INCREMENT = 10    # default value in order to limit the increment of memory allocation

    __MAX_INCREMENT = []    # Max increment
    __ALLOC_DIMS = []       # Dimensions of the allocated np.array
    __DIMS = []             # Dimensions of the view with data on the allocated np.array (__DIMS <= __ALLOC_DIMS)

    __ARRAY = []            # Allocated array

    def __init__(self,initData,allocInitDim=None,dtype=np.float64,maxIncrement=None):
        self.__DIMS = np.array(initData.shape)

        self.__MAX_INCREMENT = maxIncrement
        if self.__MAX_INCREMENT == None:
            self.__MAX_INCREMENT = self.__DEFAULT_MAX_INCREMENT

        # Compute the allocation dimensions based on user's input
        if allocInitDim == None:
            allocInitDim = self.__DIMS.copy()

        while np.any( allocInitDim < self.__DIMS  ) or np.any(allocInitDim == 0):
            for i in range(len(self.__DIMS)):
                if allocInitDim[i] == 0:
                    allocInitDim[i] = self.__DEFAULT_ALLOC_INIT_DIM
                if allocInitDim[i] < self.__DIMS[i]:
                    allocInitDim[i]  = min(allocInitDim[i]/2, self.__MAX_INCREMENT)

        # Allocate memory 
        self.__ALLOC_DIMS = allocInitDim
        self.__ARRAY = np.zeros(self.__ALLOC_DIMS,dtype=dtype)

        # Set initData 
        sliceIdxs = [slice(self.__DIMS[i]) for i in range(len(self.__DIMS))]
        self.__ARRAY[sliceIdxs] = initData

    def shape(self):
        return tuple(self.__DIMS)

    def getAllocArray(self):
        return self.__ARRAY

    def getDataArray(self):
        """
        Get the view of the array with data
        """
        sliceIdxs = [slice(self.__DIMS[i]) for i in range(len(self.__DIMS))]
        return self.__ARRAY[sliceIdxs]

    def concatenate(self,X,axis=0):
        if axis > len(self.__DIMS):
            print "Error: axis number exceed the number of dimensions"
            return

        # Check dimensions for remaining axis 
        for i in range(len(self.__DIMS)):
            if i != axis:
                if X.shape[i] != self.shape()[i]:
                    print "Error: Dimensions of the input array are not consistent in the axis %d" % i
                    return

        # Check whether allocated memory is enough 
        needAlloc = False
        while self.__ALLOC_DIMS[axis] < self.__DIMS[axis]   X.shape[axis]:
            needAlloc = True
            # Increase the __ALLOC_DIMS 
            self.__ALLOC_DIMS[axis]  = min(self.__ALLOC_DIMS[axis]/2,self.__MAX_INCREMENT)

        # Reallocate memory and copy old data 
        if needAlloc:
            # Allocate 
            newArray = np.zeros(self.__ALLOC_DIMS)
            # Copy 
            sliceIdxs = [slice(self.__DIMS[i]) for i in range(len(self.__DIMS))]
            newArray[sliceIdxs] = self.__ARRAY[sliceIdxs]
            self.__ARRAY = newArray

        # Concatenate new data 
        sliceIdxs = []
        for i in range(len(self.__DIMS)):
            if i != axis:
                sliceIdxs.append(slice(self.__DIMS[i]))
            else:
                sliceIdxs.append(slice(self.__DIMS[i],self.__DIMS[i] X.shape[i]))

        self.__ARRAY[sliceIdxs] = X
        self.__DIMS[axis]  = X.shape[axis]

该代码显示出比vstack / hstack几个随机大小的连接更好的性能.

我想知道的是：这是最好的方式吗？ numpy中有没有这样做的东西？

而且这将是很好能够重载np.array切片赋值运算符,所以实际的尺寸之外,一旦用户分配什么,一个ExpandingArray.concatenate()执行.怎么做这样的重载？

测试代码：我在这里也发布了一些代码,用于比较vstack和我的方法.我添加了最大长度为100的随机数据块.

import time

N = 10000

def performEA(N):
    EA = ExpandingArray(np.zeros((0,2)),maxIncrement=1000)
    for i in range(N):
        nNew = np.random.random_integers(low=1,high=100,size=1)
        X = np.random.rand(nNew,2)
        EA.concatenate(X,axis=0)
        # Perform operations on EA.getDataArray()
    return EA

def performVStack(N):
    A = np.zeros((0,2))
    for i in range(N):
        nNew = np.random.random_integers(low=1,high=100,size=1)
        X = np.random.rand(nNew,2)
        A = np.vstack((A,X))
        # Perform operations on A
    return A

start_EA = time.clock()
EA = performEA(N)
stop_EA = time.clock()

start_VS = time.clock()
VS = performVStack(N)
stop_VS = time.clock()

print "Elapsed Time EA: %.2f" % (stop_EA-start_EA)
print "Elapsed Time VS: %.2f" % (stop_VS-start_VS)

解决方法:

我认为这些东西最常见的设计模式是只使用小数组的列表.当然你可以做动态调整大小的事情(如果你想做疯狂的事情,你也可以尝试使用resize数组方法).我认为一种典型的方法是在你真的不知道会有多大的时候总是加倍.当然,如果您知道阵列将增长到多大,那么只需预先分配完整的东西就是最简单的.

def performVStack_fromlist(N):
    l = []
    for i in range(N):
        nNew = np.random.random_integers(low=1,high=100,size=1)
        X = np.random.rand(nNew,2)
        l.append(X)
    return np.vstack(l)

我确信有一些用例,其中扩展数组可能很有用(例如当附加数组都非常小时),但是使用上述模式可以更好地处理这个循环.优化主要是关于你需要复制周围事物的频率,以及像这样的列表(除了列表本身之外),这恰好就在这里.所以它通常要快得多.

来源：https://www./content-1-266801.html