Nat*_*ate 9 python optimization numpy binary-data
我有一个包含多个数据通道的文件.文件以基本速率进行采样,每个通道以基本速率除以某个数字进行采样 - 它似乎总是2的幂,但我认为这不重要.
所以,如果我有通道a,b和c,在1,2和4的分隔处采样,我的流将如下所示:
a0 b0 c0 a1 a2 b1 a3 a4 b2 c1 a5 ...
为了增加乐趣,通道可以独立地浮动或整数(虽然我知道每个通道),并且数据流不一定以2的幂结束:示例流在没有进一步扩展的情况下是有效的.虽然我知道我正在处理的是什么,但这些价值有时很大,有时甚至是小端.
我有一些代码可以正确地解包这些并用正确的值填充numpy数组,但它很慢:它看起来像(希望我不会过多地掩盖;只是给出算法的概念):
for sample_num in range(total_samples):
channels_to_sample = [ch for ch in all_channels if ch.samples_for(sample_num)]
format_str = ... # build format string from channels_to_sample
data = struct.unpack( my_file.read( ... ) ) # read and unpack the data
# iterate over data tuple and put values in channels_to_sample
for val, ch in zip(data, channels_to_sample):
ch.data[sample_num / ch.divider] = val
它很慢 - 在我的笔记本电脑上读取20MB文件几秒钟.Profiler告诉我,我花了很多时间Channel#samples_for()- 这是有道理的; 那里有一些条件逻辑.
我的大脑觉得有一种方法可以一举做到而不是嵌套循环 - 也许使用索引技巧来读取每个数组中我想要的字节？建立一个巨大的,疯狂的格式字符串的想法似乎也是一个值得怀疑的道路.
感谢那些回应的人.对于它的价值,numpy索引技巧将读取我的测试数据所需的时间从大约10秒减少到大约0.2秒,加速为50倍.
真正提高性能的最好方法是摆脱所有样本的Python循环,让NumPy在编译的C代码中完成这个循环.这有点难以实现,但它是可能的.
首先,你需要做一些准备.正如Justin Peel所指出的,样品排列的模式在经过一些步骤后重复进行.如果d_1,...,d_k是k数据流的除数,b_1,...,b_k是以字节为单位的流的样本大小,而lcm是这些除数的最小公倍数,则
N = lcm*sum(b_1/d_1+...+b_k/d_k)
将是流模式之后重复的字节数.如果您已经确定了前N个字节中的每个字段属于哪个流,则可以简单地重复此模式.
您现在可以通过类似的内容为前N个字节构建流索引数组
stream_index = []
for sample_num in range(lcm):
stream_index += [i for i, ch in enumerate(all_channels)
if ch.samples_for(sample_num)]
repeat_count = [b[i] for i in stream_index]
stream_index = numpy.array(stream_index).repeat(repeat_count)
这里d是序列d_1,...,d_k,并且b是序列b_1,...,b_k.
现在你可以做到
data = numpy.fromfile(my_file, dtype=numpy.uint8).reshape(-1, N)
streams = [data[:,stream_index == i].ravel() for i in range(k)]
您可能需要在最后填充数据以进行reshape()工作.
现在,您将所有属于每个流的字节放在单独的NumPy数组中.您可以通过简单地分配dtype每个流的属性来重新解释数据.如果您希望将第一个流解释为大端整数,只需编写即可
streams[0].dtype = ">i"
这不会以任何方式改变数组中的数据,只是解释它的方式.
这可能看起来有点神秘,但性能要好得多.