从二进制文件创建 Numpy 数组的有效方法

Posted 2023-02-23

【中文标题】从二进制文件创建 Numpy 数组的有效方法

【英文标题】：Efficient Way to Create Numpy Arrays from Binary Files

【发布时间】：2011-09-27 13:01:58

【问题描述】：

我有非常大的数据集，它们存储在硬盘上的二进制文件中。以下是文件结构的示例：

文件头

149 Byte ASCII Header

开始录制

4 Byte Int - Record Timestamp

样品开始

2 Byte Int - Data Stream 1 Sample
2 Byte Int - Data Stream 2 Sample
2 Byte Int - Data Stream 3 Sample
2 Byte Int - Data Stream 4 Sample

样品结束

每个记录有 122,880 个样本，每个文件有 713 个记录。这产生了 700,910,521 字节的总大小。采样率和记录数有时会有所不同，因此我必须编写代码来检测每个文件的数量。

目前我用来将这些数据导入数组的代码是这样的：

from time import clock
from numpy import zeros , int16 , int32 , hstack , array , savez
from struct import unpack
from os.path import getsize

start_time = clock()
file_size = getsize(input_file)

with open(input_file,'rb') as openfile:
  input_data = openfile.read()

header = input_data[:149]
record_size = int(header[23:31])
number_of_records = ( file_size - 149 ) / record_size
sample_rate = ( ( record_size - 4 ) / 4 ) / 2

time_series = zeros(0,dtype=int32)
t_series = zeros(0,dtype=int16)
x_series = zeros(0,dtype=int16)
y_series = zeros(0,dtype=int16)
z_series = zeros(0,dtype=int16)

for record in xrange(number_of_records):

  time_stamp = array( unpack( '<l' , input_data[ 149 + (record * record_size) : 149 + (record * record_size) + 4 ] ) , dtype = int32 )
  unpacked_record = unpack( '<' + str(sample_rate * 4) + 'h' , input_data[ 149 + (record * record_size) + 4 : 149 + ( (record + 1) * record_size ) ] ) 

  record_t = zeros(sample_rate , dtype=int16)
  record_x = zeros(sample_rate , dtype=int16)
  record_y = zeros(sample_rate , dtype=int16)
  record_z = zeros(sample_rate , dtype=int16)

  for sample in xrange(sample_rate):

    record_t[sample] = unpacked_record[ ( sample * 4 ) + 0 ]
    record_x[sample] = unpacked_record[ ( sample * 4 ) + 1 ]
    record_y[sample] = unpacked_record[ ( sample * 4 ) + 2 ]
    record_z[sample] = unpacked_record[ ( sample * 4 ) + 3 ]

  time_series = hstack ( ( time_series , time_stamp ) )
  t_series = hstack ( ( t_series , record_t ) )
  x_series = hstack ( ( x_series , record_x ) )
  y_series = hstack ( ( y_series , record_y ) )
  z_series = hstack ( ( z_series , record_z ) )

savez(output_file, t=t_series , x=x_series ,y=y_series, z=z_series, time=time_series)
end_time = clock()
print 'Total Time',end_time - start_time,'seconds'

目前每个 700 MB 文件大约需要 250 秒，这对我来说似乎非常高。有没有更有效的方法可以做到这一点？

最终解决方案

使用带有自定义 dtype 的 numpy fromfile 方法将运行时间缩短到 9 秒，比上面的原始代码快 27 倍。最终代码如下。

from numpy import savez, dtype , fromfile 
from os.path import getsize
from time import clock

start_time = clock()
file_size = getsize(input_file)

openfile = open(input_file,'rb')
header = openfile.read(149)
record_size = int(header[23:31])
number_of_records = ( file_size - 149 ) / record_size
sample_rate = ( ( record_size - 4 ) / 4 ) / 2

record_dtype = dtype( [ ( 'timestamp' , '<i4' ) , ( 'samples' , '<i2' , ( sample_rate , 4 ) ) ] )

data = fromfile(openfile , dtype = record_dtype , count = number_of_records )
time_series = data['timestamp']
t_series = data['samples'][:,:,0].ravel()
x_series = data['samples'][:,:,1].ravel()
y_series = data['samples'][:,:,2].ravel()
z_series = data['samples'][:,:,3].ravel()

savez(output_file, t=t_series , x=x_series ,y=y_series, z=z_series, fid=time_series)

end_time = clock()

print 'It took',end_time - start_time,'seconds'

【问题讨论】：

是医学数据吗？法国电力公司？如果您不知道我在说什么，没关系... ;o) 无论如何，看看我的答案，我用它来根据这个问题打开医疗数据二进制文件：***.com/q/5804052/401828。那里有一个有趣的讨论。

不是地球物理数据。我在发布前研究时看到了您的问题。您的数据仅由短整数组成，不幸的是，我将 4 字节整数时间戳分散在整个流中。

不管怎样，对 numpy 结构化数组的许多操作都比在常规 numpy 数组上慢得多。导入时间可能更快，但计算时间可能要长 10-100 倍 :(

【参考方案1】：

一些提示：

不要使用结构模块。相反，使用 Numpy 的结构化数据类型和fromfile。在这里查看：http://scipy-lectures.github.com/advanced/advanced_numpy/index.html#example-reading-wav-files

您可以一次读取所有记录，只需将合适的 count= 传递给 fromfile。

类似这样的东西（未经测试，但你明白了）：

将 numpy 导入为 np

文件 = 打开（输入文件，'rb'）
标头 = 文件.read(149)

# ... 像你一样解析标题...

record_dtype = np.dtype([
    ('时间戳', '

【讨论】：

现在总时间是 9.07 秒，包括 savez！谢谢你。我将使用最终代码更新问题。

很好的答案，很好地使用了 numpy 内置功能！ +1

你怎么知道你是否有一个标题，它有多长？

【参考方案2】：

一个明显的低效是在循环中使用hstack：

  time_series = hstack ( ( time_series , time_stamp ) )
  t_series = hstack ( ( t_series , record_t ) )
  x_series = hstack ( ( x_series , record_x ) )
  y_series = hstack ( ( y_series , record_y ) )
  z_series = hstack ( ( z_series , record_z ) )

在每次迭代中，这会为每个系列分配一个稍大的数组，并将到目前为止读取的所有数据复制到其中。这涉及 大量 不必要的复制，并可能导致不良的内存碎片。

我会将time_stamp 的值累积到一个列表中，并在最后执行一个hstack，并且对record_t 等执行完全相同的操作。

如果这不能带来足够的性能改进，我会注释掉循环的主体，并开始一次将事情带回来，看看时间到底花在了哪里。

【讨论】：

好的，很好地实现了这一点，将时间缩短到了 110 秒！谢谢，我也将尝试实施其他一些建议。

现在还有 110 秒，大约 40 秒用于我无法优化的 savez 功能。虽然为了比较加载 .npz 只需要 20 秒。

我一定错了 savez 因为使用自定义 dtype 的 fromfile 方法，包括 savez 在内的时间下降到 9 秒。

【参考方案3】：

Numpy 支持通过numpy.memmap 将二进制数据从数据直接映射到类似数组的对象中。您也许可以对文件进行内存映射并通过偏移量提取所需的数据。

对于字节序的正确性，只需对您读入的内容使用 numpy.byteswap。您可以使用条件表达式来检查主机系统的字节序：

if struct.pack('=f', np.pi) == struct.pack('>f', np.pi):
  # Host is big-endian, in-place conversion
  arrayName.byteswap(True)

【讨论】：

我已经看过了，但似乎无法指定数据的字节序。代码需要在 windows 和 unix 下工作，因此需要明确说明字节序。

如果需要，您可以使用numpy.byteswap 就地设置正确的数据字节序。见编辑。

【参考方案4】：

通过使用array 和struct.unpack，对于类似的问题（多分辨率多通道二进制数据文件），我得到了令人满意的结果。在我的问题中，我想要每个通道的连续数据，但文件具有面向间隔的结构，而不是面向通道的结构。

“秘密”是首先读取整个文件，然后才将已知大小的切片分发到所需的容器（在下面的代码中，self.channel_content[channel]['recording'] 是 array 类型的对象）：

f = open(somefilename, 'rb')    
fullsamples = array('h')
fullsamples.fromfile(f, os.path.getsize(wholefilename)/2 - f.tell())
position = 0
for rec in xrange(int(self.header['nrecs'])):
    for channel in self.channel_labels:
        samples = int(self.channel_content[channel]['nsamples'])
        self.channel_content[channel]['recording'].extend(fullsamples[position:position+samples])
            position += samples

当然，我不能说这比提供的其他答案更好或更快，但至少你可以评估一下。

希望对你有帮助！