numpy.frombuffer
-
numpy.frombuffer(buffer, dtype=float, count=-1, offset=0) -
Interpret a buffer as a 1-dimensional array.
Parameters: - buffer : buffer_like
-
An object that exposes the buffer interface.
- dtype : data-type, optional
-
Data-type of the returned array; default: float.
- count : int, optional
-
Number of items to read.
-1means all data in the buffer. - offset : int, optional
-
Start reading the buffer from this offset (in bytes); default: 0.
Notes
If the buffer has data that is not in machine byte-order, this should be specified as part of the data-type, e.g.:
>>> dt = np.dtype(int) >>> dt = dt.newbyteorder('>') >>> np.frombuffer(buf, dtype=dt)The data of the resulting array will not be byteswapped, but will be interpreted correctly.
Examples
>>> s = 'hello world' >>> np.frombuffer(s, dtype='S1', count=5, offset=6) array(['w', 'o', 'r', 'l', 'd'], dtype='|S1')>>> np.frombuffer(b'\x01\x02', dtype=np.uint8) array([1, 2], dtype=uint8) >>> np.frombuffer(b'\x01\x02\x03\x04\x05', dtype=np.uint8, count=3) array([1, 2, 3], dtype=uint8)
NumPy的ndarray數組對象不能像list一樣動態地改變其大小，在做數據采集時很不方便。本文介紹如何通過np.frombuffer()實現動態數組。
列表對象的內存動態分配
Python的列表對象實際上是一個動態指針數組。當列表中沒有空間儲存新的元素時，列表會動態地改變其大小，以容納新的元素。每次改變大小時，它都會預留一部分空間，以降低改變大小的頻率。下面的程序可以觀察列表的這一行為。
import sys import pylab as pl size = [] for i in xrange(10000): size.append(sys.getsizeof(size)) pl.plot(size, lw="2") pl.show()
程序的輸出如下圖所示，圖中每個階梯跳變的位置都表示一次內存分配，而每個階梯的高度表示額外分配的內存的大小。
因此由於往列表中添加新元素時，基本上時間復雜度都為O(1)，只有在重新分配內存時，時間復雜度才變為O(n)。由於每次額外分配的內存和列表的長度成正比，因此隨着列表的增大，重新分配內存的次數會減少，從而整體上append()方法的平均時間復雜度為O(1)。這種動態數組很適合用來做數據采集，然而由於列表中的每個元素都是對象，比較浪費內存，因此用列表做大量數據的采集並不划算。我們希望通過類似NumPy數組的對象采集數據。
NumPy數組的動態分配
NumPy的數組沒有這種動態改變大小的功能，numpy.append()函數每次都會重新分配整個數組，並把原來的數組復制到新數組中。下面的程序模擬列表的動態分配，從而實現動態數組：
import numpy as np class DynamicArray(object): def __init__(self, item_type): self._data = np.zeros(10, dtype=item_type) self._size = 0 def get_data(self): return self._data[:self._size] def append(self, value): if len(self._data) == self._size: self._data = np.resize(self._data, int(len(self._data)*1.25)) self._data[self._size] = value self._size += 1 item_type = np.dtype({ "names":["id", "x", "y", "z"], "formats":["i4", "f8", "f8", "f8"]}) da = DynamicArray(item_type) for i in xrange(100): da.append((i, i*0.1, i*0.2, i*0.3)) data = da.get_data()
用array數組采集數據
Python標准庫中的array數組也提供了動態分配內存的功能，而且它和NumPy數組一樣直接將數值的二進制數據保存在一塊內存中，因此我們可以先用array數組收集數組，然后通過np.frombuffer()將array數組的數據內存直接轉換為一個NumPy數組。下面是一個例子：
>>> import numpy as np >>> from array import array >>> a = array("d", [1,2,3,4]) # 創建一個array數組 >>> a array('d', [1.0, 2.0, 3.0, 4.0]) >>> na = np.frombuffer(a, dtype=np.float) # 通過np.frombuffer()創建一個和a共享內存的NumPy數組 >>> na array([ 1., 2., 3., 4.]) >>> na[1] = 20 # 修改NumPy數組中的第一個元素 >>> a array('d', [1.0, 20.0, 3.0, 4.0]) # array數組中的第一個元素也同時改變
array數組只支持一維，如果我們需要采集多個頻道的數據，可以將這些數據依次添加進array數組，然后通過reshape()方法將np.frombuffer()所創建的NumPy數組改為二維數組。下面是一個例子：
buf = array("d") for i in range(100): buf.append(math.sin(i*0.1)) ❶ buf.append(math.cos(i*0.1)) data = np.frombuffer(buf, dtype=np.float).reshape(-1, 2) ❷ print data
在這個例子中，❶我們通過array數組buf采集兩個頻道的數據，數據采集完畢之后，我們通過np.frombuffer()將其轉換為NumPy數組，並通過reshape()將其形狀改為(100,2)。
用bytearray采集數據
當每個頻道的數據類型不同時，就不能采用上節所介紹的方法了。這時我們可以使用bytearray收集數據。bytearray是字節數組，因此我們首先需要通過struct模塊將Python的數值轉換成其字節表示形式。如果數據來自二進制文件或者硬件，那么我們得到得已經是字節數據，這個步驟可以省略。下面是使用bytearray進行數據采集的例子：
buf = bytearray() for i in range(100): buf.extend(struct.pack("=hdd", i, math.sin(i*0.1), math.cos(i*0.1))) ❶ dtype = np.dtype({"names":["id","sin","cos"], "formats":["h", "d", "d"]}) ❷ data = np.frombuffer(buf, dtype=dtype) ❸ print data
❶采集三個頻道的數據，其中頻道1是短整型整數，其類型符號為”h”，頻道2和3為雙精度浮點數，其類型符號為”d”。類型格式字符串中的”=”表示輸出得字節數據不進行內存對齊。即一條數據的字節數為2+8+8=16，如果沒有”=”，那么一條數據的字節數則為8+8+8=24。
❷定義一個dtype對象表示一條數據的結構，dtype對象缺省不進行內存對齊，如果采集數據用的bytearray中的數據是內存對齊的話，只需要設置dtype()的align參數為True即可。
❸最后通過np.frombuffer()將bytearray轉換為NumPy的結構數組。然后我們就可以通過data[“id”]、data[“sin”]和data[“cos”]訪問三個頻道的數據了。