加速Python的struct.unpack函数

Question

我试图加快我的脚本速度。它基本上读取一个带有Velodyne的Lidar HDL-32信息的pcap文件，并使我能够获取X、Y、Z和Intensity值。我使用python -m cProfile ./spTestPcapToLas.py对我的脚本进行了剖析，发现它在我的readDataPacket()函数调用中花费了大量的时间。在一个小测试（80 MB文件）中，解包部分占用了执行时间的约56%。

我像这样调用readDataPacket函数（chunk指pcap文件）：

packets = []
for packet in chunk:
    memoryView = memoryview(packet.raw())
    udpDestinationPort = unpack('!h', memoryView[36:38].tobytes())[0]

    if udpDestinationPort == 2368:
        packets += readDataPacket(memoryView)

readDataPacket()函数本身是这样定义的：

def readDataPacket(memoryView):
    firingData = memoryView[42:]    
    firingDataStartingByte = 0    
    laserBlock = []

    for i in xrange(firingBlocks):
        rotational = unpack('<H', firingData[firingDataStartingByte+2:firingDataStartingByte+4])[0]        
        startingByte = firingDataStartingByte+4
        laser = []
        for j in xrange(lasers):   
            distanceInformation = unpack('<H', firingData[startingByte:(startingByte + 2)])[0] * 0.002
            intensity = unpack('<B', firingData[(startingByte + 2)])[0]   
            laser.append([distanceInformation, intensity])
            startingByte += 3
        firingDataStartingByte += 100
        laserBlock.append([rotational, laser])

    return laserBlock

你有什么想法可以加速这个过程吗？顺便说一下，我正在使用numpy进行X、Y、Z和强度的计算。

- Xavier Merino

Answer 1

Numpy可以让您快速完成此操作。在这种情况下，我认为最简单的方法是直接使用ndarray构造函数：

import numpy as np

def with_numpy(buffer):
    # Construct ndarray with: shape, dtype, buffer, offset, strides.
    rotational = np.ndarray((firingBlocks,), '<H', buffer, 42+2, (100,))
    distance = np.ndarray((firingBlocks,lasers), '<H', buffer, 42+4, (100,3))
    intensity = np.ndarray((firingBlocks,lasers), '<B', buffer, 42+6, (100,3))
    return rotational, distance*0.002, intensity

这将返回单独的数组，而不是嵌套列表，这样进一步处理应该更容易。作为输入，它接受一个buffer对象（在Python 2中）或任何暴露缓冲区接口的对象。不幸的是，它取决于您的Python版本（2/3）可以使用哪些对象。但这种方法非常快：

import numpy as np

firingBlocks = 10**4
lasers = 32
packet_raw = np.random.bytes(42 + firingBlocks*100)

%timeit readDataPacket(memoryview(packet_raw))
# 1 loop, best of 3: 807 ms per loop
%timeit with_numpy(packet_raw)
# 100 loops, best of 3: 10.8 ms per loop

- user2379410

---

这导致该特定函数的速度增加了约30倍。非常感谢。 :D - Xavier Merino

从200秒减少到3秒！ - brthornbury

Answer 2

提前编译Struct，以避免使用模块级方法包装Python级别的代码。在循环之外执行此操作，以避免重复付出构建成本。

unpack_ushort = struct.Struct('<H').unpack
unpack_ushort_byte = struct.Struct('<HB').unpack

Struct 方法本身是由 CPython 中的 C 实现的（模块级别方法在解析格式字符串后最终委派给相同的工作），因此仅需构建一次 Struct 并存储绑定方法即可节省大量工作，尤其是在解包少量值时。

您还可以通过一次性解包多个值而不是一个一个地解包来节省一些工作：

distanceInformation, intensity = unpack_ushort_byte(firingData[startingByte:startingByte + 3])
distanceInformation *= 0.002

如Dan指出，您可以使用iter_unpack进一步改进此操作，这将进一步减少字节码执行和小片操作的数量。

- ShadowRanger

---

我建议在确定它是否提高了性能之前，先测试一下我的iter_unpack方法 - 我认为它会创建大量临时对象。你的方法听起来更可靠。 - Dan Getz

Answer 3

对于你的具体情况，如果可以将循环嵌入numpy调用中，则速度最快。

话虽如此，对于仅涉及struct.unpack部分的内容——如果你的数据恰好符合本机字节顺序，可以使用memoryview.cast。以short为例，它比朴素的struct.unpack快约3倍，而逻辑没有任何变化。

In [20]: st = struct.Struct("<H")

In [21]: %timeit struct.unpack("<H", buf[20:22])
1.45 µs ± 26.5 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

In [22]: %timeit st.unpack(buf[20:22])
778 ns ± 10.8 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

In [23]: %timeit buf.cast("H")[0]
447 ns ± 4.16 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

- kizzx2

Answer 4

您可以在一次调用中解包原始的distanceInformation和intensity值。特别是因为您只是将它们放在一起放入列表中：这就是当解包多个值时unpack()所做的。在您的情况下，您需要将distanceInformation乘以0.002，但是您可以稍后再做这个，因为您可以使用iter_unpack()来解析整个原始对的列表。该函数会给您一个生成器，可以使用itertools.islice()进行切片，然后转换为列表。类似于以下内容：

laser_iter = struct.iter_unpack('<HB', firingData[firingDataStartingByte + 4])
laser = [[d * 0.002, i] for d, i in itertools.islice(laser_iter, lasers)]

不幸的是，这段代码有点难以阅读，因此您可能需要找到一种方法将其分散成更多行的代码，使用更具描述性的变量名称，或者添加注释以供将来查看，以便您忘记为什么编写了这个……

- Dan Getz

---

很遗憾，我不能使用Python 3。我正在使用Python 2.7.11。你知道其他的解决方案吗？ - Xavier Merino