PyQt5开发Modbus多串口上位机：架构设计与工程实现

在实际工业数据采集项目中用 Python PyQt5 开发 Modbus 多串口上位机系统是很多设备调试、产线监控和小型 SCADA 数据采集团队的常见需求。Modbus 协议本身并不复杂真正的复杂度集中在三个方面多串口如何并行轮询不冲突、界面如何实时刷新不卡顿、数据如何稳定落盘方便复盘。本文围绕一个可运行的完整工程拆解通信层、界面层和存储层的设计思路并给出关键源码示例。这套结构适用于温度压力采集、电力参数监测、设备状态巡检等场景也可以作为后续扩展 TCP 或从站功能的基础模板。文章采用的软件架构主线是PyQt5 负责界面和事件循环每个串口对应一个 QSerialPort 通道用 QTimer 驱动 Modbus RTU 轮询收到响应后通过 Qt 信号分发到仪表台、实时曲线和 CSV 存储模块。这样既避免了多线程串口读写的复杂性又能满足中小规模采集系统的实时性要求。1. 先理解这套系统的三个核心模块在动手写代码之前先明确系统的边界。一套多串口 Modbus 上位机不是简单地把串口数据打印到文本框而是由通信采集、人机交互、数据沉淀三部分组成。这三部分如果耦合在一起后期维护会很痛苦。1.1 通信采集层负责什么通信采集层负责与 Modbus 从站设备对话。它的职责包括按配置打开串口、设置波特率和数据位、构造 Modbus RTU 请求帧、等待从站响应、解析响应数据、判断设备是否离线。因为是多串口通信层必须做到“每个串口独立管理”。一个串口一条总线总线上可能挂多个从站地址。每个串口的轮询节奏、从站地址列表、寄存器映射都可能不同所以不能把所有串口塞进同一个收发循环。1.2 界面层负责什么界面层负责把采集到的数据变成操作者能看懂的信息。本文系统包含仪表台、实时图表、预警状态三块。仪表台用于展示关键数值和运行状态实时图表用于观察趋势变化预警用于在数值越限时提醒现场人员。界面层必须响应及时。如果串口读取或 CSV 写入操作阻塞了 Qt 主线程界面会出现假死、曲线不刷新、按钮无响应等现象。因此在设计时串口事件应通过 Qt 信号槽机制分发到界面组件不要让界面代码直接操作串口读写。1.3 数据存储层负责什么数据存储层负责把采集结果按时间顺序保存方便后续分析和追溯。CSV 是最简单直接的方式通用性强能用 Excel、Python、Pandas 直接读取。但 CSV 保存也要注意文件滚动、写入性能、中文编码等问题。在本系统中CSV 保存模块独立成一个类不直接在串口回调里写文件而是通过信号统一接收数据。这样即使未来把 CSV 换成 SQLite 或 MySQL界面和通信层都不需要改动。2. 环境准备和项目结构开发本系统需要 Python 3.9 以上版本主要依赖 PyQt5、pyqtgraph、pyserial。其中 QSerialPort 是 PyQt5 自带的串口模块底层封装了不同操作系统的串口访问能力能直接使用 Qt 的信号槽机制。pyserial 作为可选方案保留适合在无界面脚本中调试。2.1 依赖清单和安装命令推荐创建独立的虚拟环境避免污染系统 Python。在项目根目录执行以下命令# 创建虚拟环境 python -m venv venv # Windows 激活 venv\Scripts\activate # Linux / macOS 激活 source venv/bin/activate # 升级 pip 后安装依赖 python -m pip install --upgrade pip pip install PyQt5 pyqtgraph pyserial各依赖用途如下依赖包用途说明PyQt5GUI 框架提供窗口、控件、信号槽、QSerialPortpyqtgraph实时图表基于 Qt 的高性能绘图库适合高频刷新pyserial串口调试可选用于无界面脚本或排查串口参数安装完成后可以用下面的方式验证 PyQt5 可正常导入python -c from PyQt5.QtCore import QTimer; print(PyQt5 OK) python -c import pyqtgraph; print(pyqtgraph OK)如果出现缺少 Qt 平台插件的报错通常是 PyQt5 安装不完整或系统缺少相关图形库在 Linux 环境下需要安装 libgl 等基础库。2.2 项目目录规划建议按模块拆分的目录结构如下modbus_gui/ ├── main.py # 程序入口 ├── config.json # 串口、从站、预警参数 ├── core/ │ ├── __init__.py │ ├── modbus_rtu.py # CRC、报文构造、响应解析 │ ├── serial_channel.py # 串口通道与轮询调度 │ └── data_packet.py # 数据打包和信号定义 ├── ui/ │ ├── __init__.py │ ├── main_window.py # 主窗口 │ ├── dashboard.py # 仪表台控件 │ ├── trend_chart.py # 实时曲线 │ └── alarm_panel.py # 预警面板 └── storage/ ├── __init__.py └── csv_writer.py # CSV 滚动保存这里的核心思路是通信层不依赖 UI存储层只接收数据对象。这样每个模块都可以单独测试。下面先实现通信层。3. 通信层Modbus RTU 报文构造与解析Modbus RTU 是请求响应式协议主站发送请求从站返回响应。上位机作为主站需要按功能码构造请求帧并解析从站响应。3.1 常用功能码实际项目中读保持寄存器最常用功能码为 0x03。如果需要读开关量输入可以用功能码 0x02。本系统中的报文构造可以做成通用函数按功能码区分。功能码含义典型用途0x01读线圈读取继电器输出状态0x02读离散输入读取开关量输入0x03读保持寄存器读取温度、压力、频率等参数0x04读输入寄存器读取只读测量值3.2 CRC16 校验实现Modbus RTU 每个报文都带 CRC16 校验低字节在前高字节在后。计算时采用查表法或按位计算。下面给出一个干净的按位实现# core/modbus_rtu.py def crc16_modbus(data: bytes) - bytes: crc 0xFFFF for byte in data: crc ^ byte for _ in range(8): if crc 0x0001: crc (crc 1) ^ 0xA001 else: crc 1 # Modbus CRC 低字节在前 return bytes([crc 0xFF, (crc 8) 0xFF])校验算法是 Modbus 协议的一部分不能随便换。如果 CRC 计算错误从站会直接丢弃报文。调试时可以先构造一帧已知报文例如读从站地址 1、起始地址 0、数量 2确认 CRC 字节是否与标准工具一致。3.3 构造读保持寄存器请求读保持寄存器请求帧格式为从站地址、功能码、起始地址高字节、起始地址低字节、寄存器数量高字节、寄存器数量低字节、CRC 低字节、CRC 高字节。# core/modbus_rtu.py def build_read_holding_request(slave_id: int, start_addr: int, quantity: int) - bytes: if quantity 1 or quantity 125: raise ValueError(quantity must be 1..125) frame struct.pack(B B H H, slave_id, 0x03, start_addr, quantity) return frame crc16_modbus(frame)使用struct.pack可以避免手动处理大小端。B B H H表示大端序先打包从站地址、功能码再打包两个无符号短整型。3.4 解析响应从站正常响应格式为从站地址、功能码、字节数、数据字节、CRC。数据长度等于寄存器数量乘以 2。# core/modbus_rtu.py def parse_read_holding_response(response: bytes, quantity: int) - list: if len(response) 5: raise ValueError(response too short) expect_len 3 quantity * 2 2 if len(response) ! expect_len: raise ValueError(fresponse length error: {len(response)} ! {expect_len}) byte_count response[2] if byte_count ! quantity * 2: raise ValueError(byte count mismatch) # 跳过从站地址、功能码、字节数然后去掉 CRC 两字节 data response[3:3 byte_count] values [] for i in range(quantity): values.append(struct.unpack(H, data[i * 2:i * 2 2])[0]) return values解析时要注意字节序。多数设备寄存器值是大端存储但部分定制设备可能是小端或字序反转。遇到读数明显不合理时优先检查字节序而不是怀疑设备故障。4. 多串口管理器轮询、超时与异常状态Modbus RTU 是半双工协议同一时刻一条串口总线上只能有一个主站发起请求。多串口场景下每个串口是独立总线因此可以并行处理但每个串口内部必须串行轮询。4.1 为什么选择 QSerialPort 而不是 pyserial在 PyQt5 程序中QSerialPort 是更合适的选择。它支持readyRead信号数据到达时 Qt 事件循环会自动触发槽函数不需要额外开线程接收数据。pyserial 是阻塞读取如果直接放在 GUI 线程里可能因为等待数据导致界面卡死。对比一下两种方案方案线程模型优点缺点pyserial 单独线程需要手动管理线程和队列灵活、适合复杂协议线程同步容易出错QSerialPort 信号槽使用 Qt 事件循环无需手动线程、代码简单需要理解 Qt 信号机制本系统采用 QSerialPort配合 QTimer 轮询。核心数据结构是SerialChannel它负责打开串口、维护从站轮询列表、发送请求、等待响应、处理超时。4.2 SerialChannel 通道实现SerialChannel是一个 QObject 子类每个实例管理一个串口并通过信号向外发送采集结果。# core/serial_channel.py import json import struct from PyQt5.QtCore import QObject, QTimer, pyqtSignal, QByteArray from PyQt5.QtSerialPort import QSerialPort, QSerialPortInfo from core.modbus_rtu import build_read_holding_request, parse_read_holding_response class SerialChannel(QObject): # 发送状态、从站地址、寄存器值列表 data_ready pyqtSignal(int, int, list) # 从站地址、错误消息 error_occurred pyqtSignal(int, str) # 串口名称、是否打开 status_changed pyqtSignal(str, bool) def __init__(self, port_name: str, config: dict, parentNone): super().__init__(parent) self.port_name port_name self.config config self.serial QSerialPort(self) self.serial.setPortName(port_name) self.serial.readyRead.connect(self._on_ready_read) self.serial.errorOccurred.connect(self._on_error) self.poll_timer QTimer(self) self.poll_timer.timeout.connect(self._poll_once) self.slave_list config.get(slaves, []) self.current_slave_index 0 self.busy False self.pending_slave None self.pending_quantity 0 self.response_buffer QByteArray() def open(self) - bool: params self.config.get(serial_params, {}) self.serial.setBaudRate(params.get(baud_rate, 9600)) self.serial.setDataBits(QSerialPort.Data8) self.serial.setParity(QSerialPort.NoParity) self.serial.setStopBits(QSerialPort.OneStop) ok self.serial.open(QSerialPort.ReadWrite) if ok: interval self.config.get(poll_interval_ms, 1000) self.poll_timer.start(interval) self.status_changed.emit(self.port_name, ok) return ok def _poll_once(self): if self.busy: # 上一轮请求未完成跳过本轮防止请求堆积 return if not self.serial.isOpen(): return if not self.slave_list: return slave_cfg self.slave_list[self.current_slave_index] self.current_slave_index (self.current_slave_index 1) % len(self.slave_list) slave_id slave_cfg[slave_id] start_addr slave_cfg[start_addr] quantity slave_cfg[quantity] request build_read_holding_request(slave_id, start_addr, quantity) self.serial.write(request) self.busy True self.pending_slave slave_cfg self.pending_quantity quantity self.response_buffer.clear() # 启动超时检查 QTimer.singleShot(self.config.get(timeout_ms, 500), self._check_timeout) def _on_ready_read(self): if not self.busy: self.serial.readAll() return self.response_buffer.append(self.serial.readAll()) if self._is_response_complete(): self._handle_response() def _is_response_complete(self) - bool: buf bytes(self.response_buffer) if len(buf) 3: return False # 根据字节数字段判断长度 byte_count buf[2] expect_len 3 byte_count 2 return len(buf) expect_len def _handle_response(self): buf bytes(self.response_buffer) if len(buf) 5: self._finish_poll() return try: values parse_read_holding_response(buf, self.pending_quantity) slave_id self.pending_slave[slave_id] self.data_ready.emit(self.pending_slave.get(tag_id, slave_id), slave_id, values) except ValueError as exc: self.error_occurred.emit(self.pending_slave.get(slave_id, 0), str(exc)) self._finish_poll() def _check_timeout(self): if self.busy: slave_id self.pending_slave.get(slave_id, 0) self.error_occurred.emit(slave_id, response timeout) self._finish_poll() def _finish_poll(self): self.busy False self.pending_slave None self.response_buffer.clear() def _on_error(self, error): if error QSerialPort.ResourceError: self.serial.close() self.status_changed.emit(self.port_name, False)这个通道类的关键点有三个用busy标志防止在上一次请求未完成时发送新请求。用response_buffer接收零散数据因为串口数据可能分多次到达不能假设一次读完。超时通过QTimer.singleShot实现超时后直接标记当前轮询结束避免影响下一轮。注意QTimer.singleShot不是线程安全的完美超时方案但对这种单事件循环、请求间隔较长的场景足够。如果未来要支持数百个从站、毫秒级轮询应该改用状态机加精确超时管理。4.3 多串口并行启动主程序可以根据配置文件中的串口列表创建多个SerialChannel实例。每个通道有独立的 QSerialPort 和 QTimer因此多个串口之间天然并行。# ui/main_window.py 中初始化的核心逻辑 self.channels [] for port_config in self.app_config.get(ports, []): channel SerialChannel(port_config[port_name], port_config) channel.data_ready.connect(self._on_data_ready) channel.error_occurred.connect(self._on_channel_error) channel.status_changed.connect(self._on_status_changed) if channel.open(): self.channels.append(channel) else: print(fopen {port_config[port_name]} failed)这里要注意如果配置文件里写了一个不存在的串口open()会返回 False。生产环境中应该让程序继续运行并在界面上提示哪些串口打开失败而不是直接退出。5. 界面层仪表台、实时图表和预警界面层是整个系统的直观展示部分。为了避免界面代码与通信逻辑耦合所有模块只通过信号接收数据。下面分别介绍仪表台、实时图表和预警面板的实现思路。5.1 主窗口布局主窗口采用左右结构左侧是仪表台卡片区域右侧上方是实时图表右侧下方是预警和日志面板。这样在 1920 分辨率下可以同时看到实时数值、趋势和告警。# ui/main_window.py from PyQt5.QtWidgets import QMainWindow, QWidget, QHBoxLayout, QVBoxLayout, QSplitter from ui.dashboard import DashboardPanel from ui.trend_chart import TrendChartWidget from ui.alarm_panel import AlarmPanel class MainWindow(QMainWindow): def __init__(self, config): super().__init__() self.app_config config self.setWindowTitle(Modbus 多串口上位机系统) self.resize(1440, 800) self.dashboard DashboardPanel(config) self.trend_chart TrendChartWidget(config) self.alarm_panel AlarmPanel() splitter QSplitter(self) splitter.addWidget(self.dashboard) splitter.addWidget(self.trend_chart) splitter.addWidget(self.alarm_panel) splitter.setSizes([320, 800, 320]) self.setCentralWidget(splitter)QSplitter允许用户自由调整面板宽度适合现场工控机不同分辨率的显示需求。注意仪表台面板内部要支持动态添加卡片因为配置的从站数量可能变化。5.2 卡片式仪表台仪表台采用卡片列表每个卡片显示一个标签、当前值、单位和状态灯。相比复杂圆形仪表盘卡片方式信息密度更高开发成本低适合多路数据同时展示。# ui/dashboard.py from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QScrollArea from PyQt5.QtCore import Qt class DashboardPanel(QScrollArea): def __init__(self, config): super().__init__() self.container QWidget() self.layout QVBoxLayout(self.container) self.setWidget(self.container) self.setWidgetResizable(True) self.cards {} for tag in self._collect_tags(config): card QLabel(f{tag} --) card.setAlignment(Qt.AlignCenter) card.setMinimumHeight(60) card.setStyleSheet(QLabel { background: #f5f5f5; border: 1px solid #ccc; }) self.layout.addWidget(card) self.cards[tag] card def update_value(self, tag, value_text): if tag in self.cards: self.cards[tag].setText(f{tag} {value_text}) def _collect_tags(self, config): tags [] for port in config.get(ports, []): for slave in port.get(slaves, []): tags.append(fT{slave.get(tag_id, slave[slave_id])}) return tags这里先用 QLabel 做简化卡片。在实际项目中可以把卡片改成 QFrame里面放标签、数值、单位、状态灯四个子控件实现类似仪表台的效果。如果确实需要半圆表盘效果可以重写paintEvent使用 QPainter 画刻度线和指针。但要注意绘制频率过高会占用 CPU建议画面刷新频率控制在每秒 5 到 10 次。5.3 实时曲线图表实时曲线使用 pyqtgraph。它在 Qt 中绘制效率高支持大量数据点不需要频繁调用 update 重绘整个控件。# ui/trend_chart.py import pyqtgraph as pg from PyQt5.QtWidgets import QWidget, QVBoxLayout class TrendChartWidget(QWidget): def __init__(self, config): super().__init__() layout QVBoxLayout(self) self.plot_widget pg.PlotWidget() layout.addWidget(self.plot_widget) self.plot_widget.showGrid(xTrue, yTrue, alpha0.3) self.plot_widget.setLabel(bottom, 时间) self.plot_widget.setLabel(left, 数值) self.curves {} colors [r, g, b, c, m, y] idx 0 for port in config.get(ports, []): for slave in port.get(slaves, []): tag slave.get(tag_id, slave[slave_id]) pen pg.mkPen(colors[idx % len(colors)], width2) self.curves[tag] self.plot_widget.plot([], [], penpen, namefTag {tag}) idx 1 def append_point(self, tag, value): if tag not in self.curves: return curve self.curves[tag] x, y curve.getData() # 这里 x 简化使用序列号实际项目应使用时间戳 if x is None: x [0] y [value] else: x list(x) [len(x)] y list(y) [value] # 只保留最近 200 个点避免内存增长 x x[-200:] y y[-200:] curve.setData(x, y)生产环境下建议把 X 轴改成时间戳并定期清理历史数据。上图示例每收到一个值就调用append_point刷新频率受采集周期控制不会造成性能压力。5.4 预警判断与界面变化预警机制可以按标签配置上限和下限。为了避免个别抖动引发误报采用“连续越限 3 次”才触发告警的策略。# ui/alarm_panel.py from PyQt5.QtWidgets import QWidget, QVBoxLayout, QTextEdit from PyQt5.QtCore import QDateTime class AlarmPanel(QWidget): def __init__(self): super().__init__() layout QVBoxLayout(self) self.log QTextEdit() self.log.setReadOnly(True) layout.addWidget(self.log) self._counter {} def check_alarm(self, tag_id, value, alarm_cfg): ident tag_id high alarm_cfg.get(high) low alarm_cfg.get(low) triggered False if high is not None and value high: triggered True if low is not None and value low: triggered True self._counter[ident] self._counter.get(ident, 0) 1 if triggered else 0 if self._counter[ident] 3: now QDateTime.currentDateTime().toString(yyyy-MM-dd HH:mm:ss) self.log.append(f[{now}] 预警: 标签 {tag_id} 数值 {value} 越限)预警发出后还应该在仪表台卡片上改变背景色。这个可以通过DashboardPanel暴露一个set_alarm_state(tag, state)方法由主窗口统一调用。不要把预警逻辑写进通信层。6. CSV 保存与数据落盘CSV 保存是系统的数据沉淀环节。如果直接把数据写入同一个文件长时间运行会导致文件过大打开和写入都很慢。因此需要按时间或大小滚动。6.1 按文件大小滚动保存CSV 写入器维护当前文件句柄当文件大小超过阈值时关闭并创建新文件。文件命名中加入时间戳方便按时间段查找。# storage/csv_writer.py import csv import os from datetime import datetime class CsvRollingWriter: def __init__(self, directory: str, max_size_mb: int 10): self.directory directory self.max_size max_size_mb * 1024 * 1024 os.makedirs(directory, exist_okTrue) self.file None self.writer None self.current_size 0 self.headers [timestamp, tag_id, slave_id, value] self._open_new_file() def _open_new_file(self): now datetime.now().strftime(%Y%m%d_%H%M%S) path os.path.join(self.directory, fdata_{now}.csv) self.file open(path, w, newline, encodingutf-8-sig) self.writer csv.writer(self.file) self.writer.writerow(self.headers) self.current_size 0 def write_record(self, timestamp: str, tag_id, slave_id: int, value): row [timestamp, tag_id, slave_id, value] self.writer.writerow(row) self.file.flush() self.current_size self.file.tell() if self.current_size self.max_size: self.rollover() def rollover(self): if self.file: self.file.close() self._open_new_file() def close(self): if self.file: self.file.close()6.2 为什么使用 utf-8-sig 编码如果直接使用encodingutf-8Windows 下用 Excel 打开 CSV 时中文表头和内容可能出现乱码。utf-8-sig会在文件开头写入 BOM 标记Excel 能正确识别为 UTF-8。这是一个高频踩坑点。同时要注意newline。在 Windows 平台如果不指定空换行符csv 模块写入时可能出现空行。这个参数习惯性写上可以避免跨平台表现不一致。主窗口接收到data_ready信号后把数据同时分发到 dashboard、trend_chart、alarm_panel 和 csv_writer 四部分。这样可以保证界面显示和落盘数据来自同一条数据流不会出现记录不一致。7. 运行验证用模拟从站跑通全流程在真实设备到手之前可以用 Modbus Slave 模拟从站验证上位机逻辑。Modbus Slave 是常用的 Modbus 从站模拟工具也可以使用国产的虚拟串口配合调试软件完成测试。7.1 准备模拟串口和从站如果只有一台电脑需要先创建一对虚拟串口例如 COM3 和 COM4。上位机打开 COM3Modbus Slave 打开 COM4。两个软件通过虚拟串口连接实现单机联调。注意不要用真实串口直接对接自己因为串口不能自发自收。必须经过虚拟串口软件或真实的串口环回设备。在 Modbus Slave 中创建一个从站Slave ID1Function03 Holding Registers寄存器数量10将几个寄存器填入测试值比如地址 0 填 250地址 1 填 500。7.2 配置文件写法在上位机项目的config.json中配置串口为 COM3从站地址为 1采集起始地址 0数量 2。配置示例如下{ ports: [ { port_name: COM3, serial_params: { baud_rate: 9600, data_bits: 8, parity: N, stop_bits: 1 }, poll_interval_ms: 1000, timeout_ms: 500, slaves: [ { slave_id: 1, start_addr: 0, quantity: 2, tag_id: 1, alarm: { high: 600, low: 100 } } ] } ] }注意真实项目中串口号、波特率、从站地址不应写死在代码里应从外部 JSON 或数据库读取方便现场人员修改。7.3 启动和验证结果运行主程序python main.py正常情况下可以看到串口状态显示打开成功。仪表台卡片显示寄存器 0 和寄存器 1 的值。实时图表按采集周期生成曲线。CSV 目录下生成带时间戳的文件且内容与界面显示一致。当把从站寄存器值改为超过 600 时连续 3 次后预警面板出现告警。如果打开失败先检查串口号是否正确再确认波特率等参数与从站一致。Modbus 通讯没有响应时优先抓取报文或输出原始字节排查 CRC 和地址配置。8. 常见问题排查现象、原因、处理方式下面把开发调试中高频出现的问题整理成表。按“先检查配置再检查报文最后怀疑硬件”的顺序排查效率最高。问题现象常见原因检查方式处理建议串口打开失败串口号不存在或被占用在设备管理器确认串口号关闭其他串口工具重新插拔设备或更换串口号请求发送后无响应从站地址错误、波特率不匹配、线序错误检查 Modbus Slave 配置使用串口监视工具抓包先用手动发送工具发送固定报文验证响应 CRC 校验错误线路干扰、从站协议栈异常、响应解析长度判断错误打印原始响应字节用 CRC 工具核对检查接线和地线降低波特率试验数据全部是 0 或大数寄存器地址错误、字节序错误、数据类型错误对比设备寄存器表修改struct.unpack格式确认是大端还是小端必要时支持配置字节序界面卡顿、图表不刷新在 UI 线程做了阻塞操作或写入 CSV 未 flush检查代码里是否有time.sleep、大循环将耗时操作移入工作线程CSV 采用批量写入CSV 中文乱码编码不是 utf-8-sig用文本编辑器打开查看编码写入时指定encodingutf-8-sig多串口之间互相干扰共用全局变量或未独立管理端口对象检查代码是否把多个串口放在同一个 receive 函数里每个串口一个 QSerialPort 实例打包成 exe 后找不到串口PyInstaller 未打包 QtSerialPort 插件用--hidden-import PyQt5.QtSerialPort重新打包打包后用干净环境验证其中最简单也最容易忽视的坑是串口数据分片。很多设备返回的数据不是一次到达如果代码里直接readAll()并立刻解析可能得到不完整的帧。解决方式是维护缓冲区按长度字段判断当前帧是否完整。本文的SerialChannel已经处理了这一点实际项目扩展协议时应沿用这个思路。9. 生产环境落地建议和可复用清单学习环境跑通后进入生产环境之前还有一些工程化改动要做。下面区分两者差异并给出一份上线前检查清单。9.1 学习环境与生产环境的差异维度学习环境生产环境从站来源模拟软件真实工业设备配置文件相对路径、写死外部配置文件、热更新日志控制台输出文件日志、按天滚动串口故障重启程序即可自动重连、状态看板提示异常处理捕获后打印记录上下文、报警、恢复策略数据保存单文件按时间滚动、定期归档、备份部署方式源码运行PyInstaller 打包工控机安装在真实项目中建议给每个串口加入自动重连机制。当ResourceError触发时关闭串口并延时 3 到 5 秒尝试重新打开。重连期间保留界面布局并标记设备离线。9.2 上线前检查清单上线前按以下清单逐项确认配置文件中的串口号、波特率、数据位、校验位、停止位是否与现场设备一致。从站地址、寄存器起始地址、寄存器数量是否与设备手册一致。字节序是否适配设备的 Modbus 寄存器数据类型。预警上限和下限是否已按工艺要求设置。CSV 保存目录是否有写权限磁盘剩余空间是否充足。工控机是否安装了必要的串口驱动是否预留了恢复串口权限的账号。程序是否支持异常退出后重新启动并从断点继续保存数据。是否配置了文件日志日志目录和保留天数是否合理。上位机与设备之间的接地和屏蔽是否做好避免通信干扰。正式运行前是否进行了 24 小时连续采集测试并核对过数据准确性。9.3 下一步扩展方向这套架构可以继续扩展的方向包括支持 Modbus TCP将通道层抽象成RtuChannel和TcpChannel对外暴露相同的数据信号。把 CSV 存储替换为 SQLite方便按时间范围查询和统计。增加设备离线探活机制当连续多次超时后切到备用通道。在仪表台加入自定义组态功能让用户拖拽标签到任意布局。使用 PyInstaller 打包将配置、调试、部署一体化。如果刚开始接触这个方向建议先把本文的通信层跑通用模拟从站观察原始报文和解析结果再逐步添加界面和存储。通信采集如果不稳定界面做得再花哨也没有实际意义。先把最小链路跑通再考虑工程化增强是这类系统最稳妥的开发顺序。