使用Python实现串口数据实时监控与日志存储

实现串口数据实时监控与日志存储，核心在于使用 pyserial 建立与底层硬件接口的稳定通信，通过独立工作线程与线程安全队列（queue.Queue）解除 I/O 阻塞对数据采样的反压，并依托 Python 标准库的 logging 模块及其 TimedRotatingFileHandler 处理器实现精确到毫秒级的时间戳记录与按日/按大小自动切分的日志归档。整套方案兼顾了控制台实时观察的高响应性与长期无人值守测试中的落盘可靠性。

一、准备工作：依赖安装与串口硬件参数确认

串口（UART/RS232/RS485）属于按位传输的串行异步通信协议，上位机与下位机正常通信的前提是通信波特率、数据位、校验位与停止位完全匹配。在 Python 生态中，标准驱动库为 pyserial。

需要特别注意环境安装命令。部分初学者容易误装名为 serial 的第三方包，这会导致命名空间冲突并抛出 AttributeError: module 'serial' has no attribute 'Serial' 错误。正确的安装命令如下：

# 卸载可能冲突的伪包并安装正确的官方库
pip uninstall serial
pip install pyserial

安装完成后，首要任务是枚举并确认目标设备接入的物理端口号。在跨平台场景中，Windows 表现为 COM3、COM4，Linux 表现为 /dev/ttyUSB0 或 /dev/ttyACM0，macOS 则表现为 /dev/cu.usbserial-*。使用以下脚本可快速探测系统中所有连接的串行设备及其硬件 VID 与 PID：

import serial.tools.list_ports

def scan_available_ports():
    ports = serial.tools.list_ports.comports()
    if not ports:
        print("未检测到任何可用串口，请检查物理连线或驱动安装。")
        return []
    
    print(f"检测到 {len(ports)} 个有效串口：")
    for port in ports:
        print(f"  - 端口设备: {port.device}")
        print(f"    硬件描述: {port.description}")
        print(f"    硬件ID  : {port.hwid}")
    return ports

if __name__ == "__main__":
    scan_available_ports()

在确认端口号后，串口初始化的核心参数包含以下几项：

• 波特率 (Baudrate)：数据传输速率，工业与嵌入式最常见为 9600 或 115200。上位机设置必须与设备端固件严格一致。
• 字节大小与校验位：标准通信格式多为 8-N-1（8 位数据位、无校验位、1 位停止位）。在 pyserial 中默认即为此配置。
• 读超时 (Timeout)：必须设定明确的超时秒数（例如 timeout=1.0）。若设为 None，在下位机无数据输出时 read() 操作将永久卡死当前线程。

二、核心机制：多线程解耦与非阻塞安全读取

很多简单的测试脚本常采用单线程循环：每次 ser.readline() 读取一行后，紧接着调用 print() 和普通文件写入。但在高速数据传输场景（如 115200 波特率下传感器高频回传）中，这种做法存在严重的性能缺陷：磁盘 I/O 刷新或控制台输出会导致每次循环耗时波动，未被及时读取的数据在操作系统的底层串口接收 FIFO 缓冲区（通常仅数 KB）中积压，最终引发缓冲区溢出与静默丢包。

解决该问题的工程方案是采用生产者-消费者架构进行线程解耦：

1. I/O 采集线程（生产者）：唯一职责是以高优先级循环从串口读取原始字节，一旦获取完整数据帧，打上系统高精度时间戳后立即压入内存队列 queue.Queue，绝不进行耗时的字符串复杂解析与磁盘写操作。
2. 线程安全队列（缓冲中枢）：利用 Python 标准库的 queue.Queue(maxsize=2000)，该队列基于互斥锁实现了并发安全，在突发数据洪峰时充当蓄水池。
3. 消费调度线程（消费者）：从队列中逐条提取数据包，分别分发至实时控制台渲染器与日志持久化处理器。

三、日志存储设计：精确时间戳与分卷轮转落盘

长期运行的串口监控系统决不能使用简单的 open("log.txt", "a") 进行日志追加。原因在于：缺乏异常刷盘机制可能导致程序异常退出时缓存丢失；同时单文件体积无限膨胀会耗尽磁盘存储，并使得后续分析工具难以载入超大文件。

标准库中的 logging.handlers.TimedRotatingFileHandler 是实现串口日志落盘的最佳实践。它支持按照时间周期（天、小时、分钟）自动重命名旧日志并创建新文件，并可设置保留备份的最大份数（backupCount），自动淘汰过期文件。

对于工业协议与嵌入式调试，日志格式需满足全要素溯源要求，建议在单条记录中包含以下要素：

• 微秒级系统时间戳：使用 %(asctime)s.%(msecs)03d，精确标识上位机捕获该帧的具体物理时间。
• 端口与状态标识：注明数据源接口（如 [COM3]）与日志级别（[INFO] 或 [WARN]）。
• 十六进制 Raw Hex 快照：在调试私有协议或二进制传感器时，保留 16 进制原始表现形式（例如 AA 01 18 02）是排查帧头校验失配的唯一可靠依据。
• 可读字符串解析：对于 ASCII 形式的传感器报文，同步记录剥离回车换行符后的文本内容。

四、完整工程实现：高可靠性监控脚本

结合上述设计，以下提供一套结构完整、面向生产环境封装的串口实时监控与日志归档脚本。代码内嵌了多线程采集、队列缓冲、分卷日志与 Ctrl+C 优雅停机机制：

import os
import sys
import time
import queue
import logging
from logging.handlers import TimedRotatingFileHandler
import serial

class SerialMonitorLogger:
    def __init__(self, port, baudrate=115200, log_dir="serial_logs", timeout=1.0):
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.log_dir = log_dir
        
        # 线程控制与数据队列
        self.data_queue = queue.Queue(maxsize=5000)
        self.is_running = False
        self.serial_handle = None
        
        # 初始化日志系统
        self.logger = self._init_logger()

    def _init_logger(self):
        """配置支持按天自动滚卷与微秒级时间戳的日志器"""
        if not os.path.exists(self.log_dir):
            os.makedirs(self.log_dir, exist_ok=True)
            
        port_clean_name = os.path.basename(self.port).replace("/", "_")
        log_file_path = os.path.join(self.log_dir, f"serial_{port_clean_name}.log")
        logger = logging.getLogger(f"SerialMonitor_{self.port}")
        logger.setLevel(logging.INFO)
        logger.handlers.clear()  # 防止重复添加 Handler

        # 每天午夜切分日志，最多保留30天
        file_handler = TimedRotatingFileHandler(
            filename=log_file_path,
            when="midnight",
            interval=1,
            backupCount=30,
            encoding="utf-8"
        )
        file_handler.suffix = "%Y-%m-%d.log"
        formatter = logging.Formatter(
            fmt="[%(asctime)s.%(msecs)03d] [%(levelname)s] %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
        return logger

    def open_port(self):
        """安全打开串口连接并清空残留数据"""
        try:
            self.serial_handle = serial.Serial(
                port=self.port,
                baudrate=self.baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
                timeout=self.timeout
            )
            # 丢弃上次关闭残留在输入缓冲区的乱码或未完整帧
            self.serial_handle.reset_input_buffer()
            self.serial_handle.reset_output_buffer()
            return True
        except serial.SerialException as err:
            print(f"[错误] 无法打开串口 {self.port}: {err}", file=sys.stderr)
            return False

    def _reader_loop(self):
        """独立采集线程：只负责快速读取与入队，杜绝阻塞"""
        while self.is_running:
            try:
                if self.serial_handle and self.serial_handle.is_open:
                    raw_line = self.serial_handle.readline()
                    if raw_line:
                        capture_time = time.time()
                        try:
                            self.data_queue.put_nowait((capture_time, raw_line))
                        except queue.Full:
                            # 队列已满说明下游消费者异常滞后，丢弃最旧数据以保最新
                            try:
                                self.data_queue.get_nowait()
                                self.data_queue.put_nowait((capture_time, raw_line))
                            except queue.Empty:
                                pass
                else:
                    time.sleep(0.1)
            except (serial.SerialException, OSError) as err:
                print(f"[警告] 串口通信发生异常: {err}", file=sys.stderr)
                time.sleep(0.5)

    def _consumer_loop(self):
        """消费者处理循环：负责终端实时渲染与格式化落盘"""
        while self.is_running or not self.data_queue.empty():
            try:
                capture_time, raw_bytes = self.data_queue.get(timeout=0.2)
            except queue.Empty:
                continue

            # 容错解码与格式化
            text_line = raw_bytes.decode("utf-8", errors="replace").rstrip("\r\n")
            hex_dump = raw_bytes.hex(" ").upper()

            # 1. 终端实时高亮打印
            time_str = time.strftime("%H:%M:%S", time.localtime(capture_time))
            msec = int((capture_time % 1) * 1000)
            console_msg = f"[{time_str}.{msec:03d}] [RX] {text_line}"
            print(console_msg)

            # 2. 格式化记录写入分卷日志
            log_record = f"PORT={self.port} | HEX=[{hex_dump}] | ASCII={text_line}"
            self.logger.info(log_record)

            self.data_queue.task_done()

    def start(self):
        """启动监控工作线程并监听终止信号"""
        if not self.open_port():
            return

        import threading
        self.is_running = True
        
        # 创建并启动后台守护读取线程
        reader_thread = threading.Thread(target=self._reader_loop, daemon=True)
        reader_thread.start()
        print(f"[就绪] 正在监听串口 {self.port} @ {self.baudrate} bps. 按 Ctrl+C 停止...")

        # 主线程担任消费者，便于响应键盘中断信号
        try:
            self._consumer_loop()
        except KeyboardInterrupt:
            print("\n[提示] 收到退出信号，正在安全关闭串口并落盘...")
        finally:
            self.stop()

    def stop(self):
        """停止采集并安全释放系统底层句柄"""
        self.is_running = False
        time.sleep(0.2)
        if self.serial_handle and self.serial_handle.is_open:
            self.serial_handle.close()
            print("[完成] 串口设备已关闭，日志已完整保存。")

if __name__ == "__main__":
    TARGET_PORT = "COM3" if sys.platform.startswith("win") else "/dev/ttyUSB0"
    TARGET_BAUD = 115200
    
    monitor = SerialMonitorLogger(port=TARGET_PORT, baudrate=TARGET_BAUD)
    monitor.start()

运行与输出验证

当执行上述脚本且下位机发送测试报文时，终端将呈现低延迟的实时监控流：

同时在 serial_logs/serial_COM3.log 日志文件中，生成的持久化数据完整保留了毫秒时间、端口编号、原始十六进制序列与 ASCII 解析结果：

[2026-11-01 16:45:12.102] [INFO] PORT=COM3 | HEX=[24 45 4E 56 2C 54 45 4D 50 3D 32 34 2E 35 2C 48 55 4D 49 3D 35 38 2E 32 2A 35 42 0D 0A] | ASCII=$ENV,TEMP=24.5,HUMI=58.2*5B
[2026-11-01 16:45:13.104] [INFO] PORT=COM3 | HEX=[24 45 4E 56 2C 54 45 4D 50 3D 32 34 2E 35 2C 48 55 4D 49 3D 35 38 2E 31 2A 35 39 0D 0A] | ASCII=$ENV,TEMP=24.5,HUMI=58.1*59
[2026-11-01 16:45:14.106] [INFO] PORT=COM3 | HEX=[24 45 4E 56 2C 54 45 4D 50 3D 32 34 2E 36 2C 48 55 4D 49 3D 35 38 2E 30 2A 35 45 0D 0A] | ASCII=$ENV,TEMP=24.6,HUMI=58.0*5E

五、常见问题排查与长期运行维护

在实际工业现场或长周期测试中，串口采集系统常会遇到端口被占、数据乱码或异常断线等问题。根据工程实践，可依据以下决策树进行快速排查：

1. 权限不足与端口被占用 (PermissionError)

在 Linux/Ubuntu 系统中，普通用户直接访问 /dev/ttyUSB0 往往会触发 PermissionError: [Errno 13] Permission denied。这是由于设备节点默认仅归属 root 和 dialout 组。解决办法是将当前用户加入 dialout 组，无需使用危险的 sudo python 运行：

sudo usermod -aG dialout $USER
# 执行后需注销并重新登录终端使组权限生效

在 Windows 环境下，若出现 could not open port: PermissionError(13, 'Access is denied')，通常是因为其他上位机调试助手（如串口助手、VSCode 插件或旧的 Python 僵尸进程）未关闭端口。可使用系统资源监视器搜索被占用句柄并结束进程。

2. 成功打开端口但无任何数据输出

若端口已处于打开状态且无报错，但控制台迟迟没有数据流入，按以下优先级检查：

• TX/RX 反接问题：串行接口的物理连线必须交叉连接（上位机 TXD 连接下位机 RXD，上位机 RXD 连接下位机 TXD，且两端 GND 必须共地）。
• 流控使能与电平状态：部分微控制器或工业模块默认依赖 DTR/RTS 信号触发数据发送。在初始化 serial.Serial 后，显式调用 ser.dtr = True 与 ser.rts = True 可拉高状态电平解除静默。

3. 数据粘包与断包处理

ser.readline() 的工作机制是持续读取直到遇到 \n（ASCII 0x0A）或者达到 timeout。如果设备端发送的是不定长二进制数据或者未在帧末尾添加换行符，readline() 将在每次读取时触发完整的 1 秒超时等待，表现为数据延迟极大。对于纯二进制协议，应采用定长读取 ser.read(expected_length) 或基于特定起始标志字节（如 0xAA 0x55）配合自定义状态机进行解包。