Python工具箱系列(三十七)

news2024/12/29 9:38:48

二进制文件操作（上）

python比较擅长与文本相关的操作。但现实世界中，对于非文本消息的处理也很普遍。例如：

◆通过有线、无线传递传感器获得的测量数据。

◆卫星通过电磁波发送测量数据。

◆数据中心的数万台服务器发送当前CPU的占用率信息、内存占用量等众多指标数据。

以上数据，当然可以以文本方式发送，但是对于带宽的占用惊人。假设某个对于PM2.5进行测量的传感器，每隔一秒发送一次测量数据，以文本方式发送消息，内容如下：

"counter: 1, pm25: 170.24119426834042, timestamp: 2022-07-24 08:52:11.138894+00:00"

以上消息占用了81个字节。而如果能够按照约定的格式直接发送二进制数据，则只需要20个字节，可以大幅提升传输效率。此外，以文本方式在本地保存传感器数据，则一天就需要81*86400/1024=6834M字节，对于传感器这种体积小巧的嵌入式系统，存储空间极为有限，很快就会空间耗尽。因此，对非文本数据直接保存与读取非常有必要。下面的代码演示了传感器类，以及传感器测量数据的二进制文件存储与读取的基本操作。

import binascii
import random
import struct
from datetime import datetime
from io import BytesIO
from time import sleep

import arrow


class sensordata_v1():
    def __init__(self):
        utc = arrow.utcnow()
        self._timestamp = utc.to('Asia/Shanghai')

    @property
    def counter(self) -> int:
        """
        计数器

        Returns:
            int: 从0开始的计数器
        """
        return self._counter

    @counter.setter
    def counter(self, value: int):
        self._counter = value

    @property
    def pm25(self) -> float:
        """
        PM25测量值

        Returns:
            float: pm25的当前测量值
        """
        return self._pm25

    @pm25.setter
    def pm25(self, value: float):
        self._pm25 = value

    @property
    def timestamp(self) -> datetime:
        """
        当前时点

        Returns:
            datetime: 当前的时间
        """
        return self._timestamp.datetime

    def __str__(self):
        """
        以文字输出相关内容

        Returns:
            string: 说明性文字
        """
        return f"counter: {self.counter}, pm25: {self.pm25}, timestamp: {self.timestamp}"

    def __repr__(self):
        """
        输出字节流的16进制内容

        Returns:
            string: 16进制显示相关数值
        """
        return str(binascii.hexlify(self.toBytes()))

    def toBytes(self):
        """
        将相关数据转换成为bytes，便于网络传输或者写入文件

        Returns:
            bytes: 整合测量数据到字节流中
        """
        with BytesIO() as byio:
            byio.write(struct.pack('<i', self.counter))
            byio.write(struct.pack('<d', self.pm25))
            byio.write(struct.pack('<d', self._timestamp.timestamp()))
            return byio.getvalue()

    def fromBytes(self, data):
        """
        从字节流中解出相关值

        Args:
            data (bytes): 待解析的字节流
        """
        self.counter, self.pm25, st = struct.unpack('<idd', data)
        self._timestamp = arrow.get(st)


def toFile(filename):
    """
    向二进制文件中写入数据

    Args:
        filename (string): 文件名称
    """
    with open(filename, 'wb') as myfile:
        for index in range(10):
            sensor_obj = sensordata_v1()
            sensor_obj.counter = index
            sensor_obj.pm25 = random.uniform(0, 300)
            data = sensor_obj.toBytes()
            myfile.write(data)
            sleep(1)


def fromFile(filename):
    """
    从二进制文件中获得保存的信息，并且重建对象

    Args:
        filename (string): 文件名称
    """
    with open(filename, 'rb') as myfile:
        fmt = struct.Struct('<idd')
        datalen = fmt.size
        for index in range(10):
            data = myfile.read(datalen)
            sensor_obj = sensordata_v1()
            sensor_obj.fromBytes(data)
            print(sensor_obj)

datafilename = r"d:\dev\sensor.dat"
toFile(datafilename)
fromFile(datafilename)

其中arrow是非常值得推荐的时间处理框架。python有内置的时间处理框架，功能非常完善，但失之于乱与杂，掌握起来非常不易。而arrow则非常人性化，做到了拿来即用。其安装过程非常简单，直接pip install arrow即可。在示例程序中，为了能够方便存储，我们将时间戳直接用UNIX的timestamp来表示，转换后为了保持精度，本例使用double(8字节)存储，如果要求不高，可以改为float(4字节)。

sensordata_v1类使用@property装饰器来定义属性。总共有3个属性：

◆counter-计数器。从0开始计数累加，后续保存到数据库中时也方便检索。

◆pm25-PM2.5测量值。在示例代码中使用random.uniform模拟一个0到300的随机浮点值，没有什么太大的意义，保证每次不同即可。

◆timestamp-数据采集时的对应时间。