企业官网建设流程全解析

随着工业自动化项目逐步走向轻量化，Python 在工控上位机开发中的出现频率越来越高。很多朋友习惯用 C# 或 LabVIEW 写上位机，但在快速原型、数据分析、界面迭代方面，Python + PyQt5 的组合往往更灵活。这篇文章围绕一个实际需求展开：用 Python + PyQt5 开发一套 Modbus 多串口工业上位机系统，支持仪表台显示、实时曲线、阈值预警和 CSV 数据保存。

这套系统适合以下读者：

• 刚接触 PyQt5 + Modbus 的上位机初学者。
• 需要快速给设备做调试工具或产线监控界面的工程师。
• 想了解多串口轮询、界面刷新、CSV 落盘等完整写法的开发者。

读完本文后，你可以掌握 Modbus RTU 串口通信的基本封装、PyQt5 多线程刷新 UI 的方式、pyqtgraph 实时曲线用法，以及预警和 CSV 导出的整合思路。文章末尾还会整理常见报错和工程化建议。

1. 背景与功能规划

1.1 Modbus 协议与上位机

Modbus 是工业自动化领域应用非常广泛的通信协议，常见形式有 Modbus RTU、Modbus ASCII 和 Modbus TCP。在串口场景中，Modbus RTU 因为效率高、实现简单，成为 PLC、仪表、传感器等设备最常用的通信方式。

上位机在这里指的是运行在 PC 上的监控软件，它通过串口或网口与下位机（PLC、仪表等）通信，采集数据、下发指令、展示画面。

很多小型项目会遇到这样的痛点：

• 现场有多台设备，但各自使用不同的串口。
• 设备数据需要实时显示，并且要画趋势曲线。
• 某些参数超过阈值时需要报警。
• 数据要落盘，方便后续分析或追溯。

如果购买商业组态软件，成本高且定制不灵活。用 Python 开发一套小系统，成本低、可快速迭代，也能满足中小规模项目需求。

1.2 为什么选择 Python + PyQt5

PyQt5 是 Qt5 的 Python 绑定，用来开发桌面界面非常成熟，控件丰富、跨平台。相比 tkinter，PyQt5 的界面更现代化；相比 C#，Python 在数据处理和脚本扩展上更方便。

本系统使用到的关键库如下：

1.3 功能拆解

这套系统的核心功能可以拆成五个模块：

1. 串口管理：支持同时打开多个串口，每个串口下挂多个 Modbus 从站。
2. 数据采集：按固定周期轮询保持寄存器、输入寄存器、线圈等。
3. 界面展示：使用仪表台展示关键数值，使用实时图表展示趋势。
4. 预警管理：每个变量可以设置上限和下限，越限后产生报警记录。
5. 数据保存：将采集到的数据追加写入 CSV 文件，方便后续用 Excel 打开。

整体数据流程如下：

串口设备 -> Modbus 读取 -> 工作线程 -> Qt 信号 -> 主界面更新 -> CSV 写入 -> 阈值判断 -> 报警

2. 环境准备与项目结构

2.1 环境安装

建议使用 Python 3.8 到 3.11 之间的版本。本文示例以 Windows 系统为主，Linux/macOS 的串口号名称不同，代码逻辑一致。

创建虚拟环境并安装依赖：

python -m venv venv venv\Scripts\activate pip install PyQt5 pymodbus pyserial pyqtgraph

如果你的网络环境访问 PyPI 较慢，可以使用国内镜像：

pip install PyQt5 pymodbus pyserial pyqtgraph -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后，可以验证版本：

python -c "import PyQt5, pymodbus, serial, pyqtgraph; print('ok')"

注意：pymodbus 3.x 和 2.x 的部分 API 有差异，本文以 pymodbus 3.x 为例。如果使用旧版，请对应调整导入方式。

2.2 项目目录结构

为了清晰，我们按功能拆分文件：

modbus_station/ ├── main.py # 程序入口 ├── config.py # 配置数据结构 ├── modbus_client.py # Modbus 客户端封装 ├── worker.py # 数据采集工作线程 ├── csv_logger.py # CSV 保存 ├── main_window.py # 主窗口界面 └── requirements.txt

在真正写代码前，先把公共配置定义好，这样后续扩展比较方便。

3. 核心概念与通信设计

3.1 Modbus RTU 帧结构

Modbus RTU 的数据帧通常包含：

• 地址域（1 字节）：从站地址。
• 功能码（1 字节）：例如 03 读保持寄存器，04 读输入寄存器，01 读线圈，06 写单个寄存器。
• 数据域：寄存器起始地址、数量或写入值。
• 校验域：CRC16 校验。

对于应用层开发，我们不需要手动拼帧，pymodbus 库已经封装好了。但需要理解寄存器地址和功能码的含义，才能正确读取设备数据。

3.2 寄存器分类

Modbus 协议中常用四类数据区域：

在我们的上位机系统中，最常用的是读保持寄存器和读输入寄存器。

3.3 串口参数

串口通信需要确保两端参数一致：波特率、数据位、停止位、校验位。常见配置为：

• 波特率：9600 或 115200
• 数据位：8
• 停止位：1
• 校验位：无（N）
• 超时时间：建议 1 秒左右

这些参数需要在下位机固件或设备说明书中确认。

3.4 多串口轮询策略

一台 PC 可能同时插多个 USB 转串口设备，每个串口接入一条 Modbus 总线，总线上有多个从站。常规做法是：

• 每个串口创建一个独立的工作线程。
• 线程内部循环遍历该串口下的从站设备。
• 每个从站按预设的寄存器范围和数量读取。
• 读取完成后把结果通过 Qt 信号发回主线程。

这种设计不会让一个串口的阻塞影响其他串口，也方便扩展到几十个设备。

4. 通信模块实现

4.1 定义配置数据结构

在config.py中定义设备配置和变量配置，方便界面和线程共用。

# config.py from dataclasses import dataclass, field from typing import List, Optional @dataclass class VariableConfig: name: str # 变量名称，例如 "温度" address: int # 寄存器地址 quantity: int = 1 # 寄存器个数 func_code: int = 3 # 3=保持寄存器, 4=输入寄存器 scale: float = 1.0 # 缩放系数 offset: float = 0.0 # 偏移量 unit: str = "" # 单位 high_limit: Optional[float] = None # 上限值 low_limit: Optional[float] = None # 下限值 @dataclass class DeviceConfig: slave_id: int # 从站地址 name: str = "" # 设备名称 variables: List[VariableConfig] = field(default_factory=list) @dataclass class SerialConfig: port: str # 串口号，例如 "COM3" baudrate: int = 9600 bytesize: int = 8 parity: str = "N" stopbits: int = 1 timeout: float = 1.0 devices: List[DeviceConfig] = field(default_factory=list)

这里把变量抽象成独立配置，是因为每个设备可能要读很多寄存器，这样配置非常灵活。

4.2 封装 Modbus 客户端

在modbus_client.py中实现一个基于 pymodbus 的串口客户端类。

# modbus_client.py import logging from typing import List, Optional try: from pymodbus.client import ModbusSerialClient except ImportError: from pymodbus.client.serial import ModbusSerialClient from config import DeviceConfig, VariableConfig logger = logging.getLogger(__name__) class ModbusClientWrapper: """Modbus RTU 串口客户端封装，负责单个串口的读写操作。""" def __init__(self, serial_config): self.serial_config = serial_config self.client = None def connect(self) -> bool: """打开串口并建立 Modbus 连接。""" try: self.client = ModbusSerialClient( method="rtu", port=self.serial_config.port, baudrate=self.serial_config.baudrate, bytesize=self.serial_config.bytesize, parity=self.serial_config.parity, stopbits=self.serial_config.stopbits, timeout=self.serial_config.timeout, ) return self.client.connect() except Exception as e: logger.error("串口 %s 连接失败: %s", self.serial_config.port, e) return False def close(self): if self.client: self.client.close() def read_variable(self, slave_id: int, var: VariableConfig): """读取一个变量，返回原始整数或浮点值。""" if self.client is None: raise RuntimeError("Modbus 客户端未连接") address = var.address quantity = var.quantity unit = var.slave_id if hasattr(var, "slave_id") else None if var.func_code == 3: rr = self.client.read_holding_registers( address, count=quantity, slave=slave_id ) elif var.func_code == 4: rr = self.client.read_input_registers( address, count=quantity, slave=slave_id ) elif var.func_code == 1: rr = self.client.read_coils( address, count=quantity, slave=slave_id ) elif var.func_code == 2: rr = self.client.read_discrete_inputs( address, count=quantity, slave=slave_id ) else: raise ValueError(f"不支持的功能码: {var.func_code}") if rr.isError(): logger.warning("从站 %d 读取 %s 失败: %s", slave_id, var.name, rr) return None value = self._decode_value(rr, quantity, var) return value @staticmethod def _decode_value(response, quantity: int, var: VariableConfig): """根据寄存器数量和缩放比例解析数值。""" if quantity == 1: raw = response.registers[0] return raw * var.scale + var.offset # 多寄存器场景，按大端组合 raw_list = response.registers[:quantity] raw = 0 for r in raw_list: raw = (raw << 16) | r return raw * var.scale + var.offset def read_device(self, device: DeviceConfig) -> dict: """读取一个设备下所有变量，返回 {变量名: 值}。""" result = {} for var in device.variables: try: val = self.read_variable(device.slave_id, var) result[var.name] = val except Exception as e: logger.error("读取 %s 失败: %s", var.name, e) result[var.name] = None return result

注意几个细节：

• read_variable根据功能码选择不同的读取方法。
• 返回值需要按照 scale 和 offset 转换为实际的工程值。
• 多寄存器变量按大端模式拼接，实际项目中可能需要按设备手册处理。
• 每次读取都应捕获异常，避免一个变量失败导致整个线程退出。

4.3 单串口多设备读取

为了减少串口占用，可以把一个串口下的所有设备的读取动作放在一个轮询循环中。多个串口由多个线程并行执行。

5. 工作线程与信号封装

5.1 定义采集结果信号

在worker.py中，我们使用 QThread 子类，或者使用 QObject + moveToThread。这里采用简单直接的 QThread 子类方式，方便理解。

# worker.py import time from PyQt5.QtCore import QThread, pyqtSignal from modbus_client import ModbusClientWrapper from config import SerialConfig class PollWorker(QThread): """ 串口轮询线程。 每轮遍历该串口下的所有设备，读取所有变量。 通过 data_ready 信号把数据发回主界面。 """ data_ready = pyqtSignal(str, dict) # 参数：串口号, 数据字典 error = pyqtSignal(str, str) # 参数：串口号, 错误信息 status_changed = pyqtSignal(str, bool) # 参数：串口号, 是否在线 def __init__(self, serial_config: SerialConfig, interval: float = 1.0): super().__init__() self.serial_config = serial_config self.interval = interval self.running = False def run(self): self.running = True client = ModbusClientWrapper(self.serial_config) if not client.connect(): self.error.emit(self.serial_config.port, "串口打开失败") self.status_changed.emit(self.serial_config.port, False) return self.status_changed.emit(self.serial_config.port, True) while self.running: all_data = {} for device in self.serial_config.devices: # 读取一个设备的所有变量 dev_data = client.read_device(device) all_data[device.name] = dev_data self.data_ready.emit(self.serial_config.port, all_data) time.sleep(self.interval) client.close() self.status_changed.emit(self.serial_config.port, False) def stop(self): self.running = False

这个线程负责整个连接、轮询、断开的过程。主界面只需要创建线程并连接信号。

6. CSV 保存模块

6.1 按日期生成文件

数据保存建议按天生成文件，文件名包含日期，避免单个文件过大。

# csv_logger.py import csv import os from datetime import datetime class CsvLogger: """将采集数据追加写入 CSV 文件。""" def __init__(self, output_dir: str = "./data"): self.output_dir = output_dir os.makedirs(output_dir, exist_ok=True) self.current_date = None self.file = None self.writer = None self.header = [] def _open_file_if_needed(self): today = datetime.now().strftime("%Y-%m-%d") if self.current_date != today: self.close() self.current_date = today file_path = os.path.join(self.output_dir, f"data_{today}.csv") self.file = open(file_path, "a", newline="", encoding="utf-8-sig") self.writer = csv.writer(self.file) def save_record(self, record: dict): """ record 为平铺的字典，例如: {"时间": "2025-01-01 10:00:00", "设备A.温度": 25.3, "设备B.压力": 1.2} """ self._open_file_if_needed() if not self.header: self.header = list(record.keys()) if self.file.tell() == 0: # 空文件才写表头 self.writer.writerow(self.header) else: # 如果新记录带上了新的键，需要重建表头（这里做保守处理） new_keys = list(record.keys()) if new_keys != self.header: self.close() self.header = new_keys self._open_file_if_needed() self.file.seek(0, 2) self.writer.writerow(self.header) row = [record.get(key, "") for key in self.header] self.writer.writerow(row) self.file.flush() def close(self): if self.file: self.file.close() self.file = None self.writer = None

这里用utf-8-sig编码，Excel 打开 CSV 时不会乱码。每次写入后flush()，保证系统异常退出时数据不丢失。

7. 主窗口界面实现

7.1 总体布局

主窗口使用 QSplitter 或 QTabWidget 组织界面。我们采用上下布局：

• 顶部：串口配置区 + 开始/停止按钮。
• 中部：左边是实时仪表区，右边是实时图表。
• 底部：数据表格 + 报警列表。

为了代码简洁，这里用纯代码布局，没有使用 Qt Designer。

# main_window.py import json from collections import OrderedDict from PyQt5.QtCore import Qt, pyqtSlot from PyQt5.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLineEdit, QPushButton, QLabel, QListWidget, QListWidgetItem, QTableWidget, QTableWidgetItem, QHeaderView, QGroupBox, QSplitter, QComboBox, QSpinBox, QMessageBox ) import pyqtgraph as pg from worker import PollWorker from csv_logger import CsvLogger from config import SerialConfig, DeviceConfig, VariableConfig class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("Python + PyQt5 Modbus 多串口上位机") self.resize(1200, 700) self.workers = [] self.csv_logger = CsvLogger("./data") self.current_data = {} # 用于图表和表格的数据缓存 self._build_ui() def _build_ui(self): # 根控件 central = QWidget() self.setCentralWidget(central) root_layout = QVBoxLayout(central) # 顶部配置区 top_layout = QHBoxLayout() top_layout.addWidget(QLabel("串口:")) self.port_edit = QLineEdit("COM3") self.port_edit.setPlaceholderText("例如 COM3 / /dev/ttyUSB0") top_layout.addWidget(self.port_edit) top_layout.addWidget(QLabel("波特率:")) self.baud_combo = QComboBox() self.baud_combo.addItems(["9600", "19200", "38400", "115200"]) self.baud_combo.setCurrentText("9600") top_layout.addWidget(self.baud_combo) top_layout.addWidget(QLabel("轮询间隔(ms):")) self.interval_spin = QSpinBox() self.interval_spin.setRange(100, 60000) self.interval_spin.setValue(1000) top_layout.addWidget(self.interval_spin) self.connect_btn = QPushButton("连接") self.connect_btn.clicked.connect(self.toggle_connect) top_layout.addWidget(self.connect_btn) top_layout.addStretch() root_layout.addLayout(top_layout) # 中部区域 splitter = QSplitter(Qt.Horizontal) # 左侧：仪表台 + 数据表格 left_panel = QWidget() left_layout = QVBoxLayout(left_panel) meter_group = QGroupBox("仪表台") meter_layout = QVBoxLayout(meter_group) self.meter_label = QLabel("请选择一个变量") self.meter_dial = QDial() self.meter_dial.setRange(0, 1000) self.meter_dial.setEnabled(False) self.meter_value_label = QLCDNumber(8) meter_layout.addWidget(self.meter_label) meter_layout.addWidget(self.meter_dial) meter_layout.addWidget(self.meter_value_label) left_layout.addWidget(meter_group) table_group = QGroupBox("实时数据") table_layout = QVBoxLayout(table_group) self.data_table = QTableWidget() self.data_table.setColumnCount(3) self.data_table.setHorizontalHeaderLabels(["变量", "数值", "单位"]) self.data_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) table_layout.addWidget(self.data_table) left_layout.addWidget(table_group) left_layout.setStretchFactor(meter_group, 1) left_layout.setStretchFactor(table_group, 2) # 右侧：图表 + 报警列表 right_panel = QWidget() right_layout = QVBoxLayout(right_panel) chart_group = QGroupBox("实时趋势") chart_layout = QVBoxLayout(chart_group) self.plot_widget = pg.PlotWidget() self.plot_curve = self.plot_widget.plot(pen=pg.mkPen("b", width=2)) self.plot_widget.setLabel("bottom", "时间", units="s") self.chart_data = [] # 缓存最近 N 个点 self.chart_time = [] chart_layout.addWidget(self.plot_widget) right_layout.addWidget(chart_group) alarm_group = QGroupBox("报警记录") alarm_layout = QVBoxLayout(alarm_group) self.alarm_list = QListWidget() alarm_layout.addWidget(self.alarm_list) right_layout.addWidget(alarm_group) right_layout.setStretchFactor(chart_group, 4) right_layout.setStretchFactor(alarm_group, 1) splitter.addWidget(left_panel) splitter.addWidget(right_panel) splitter.setSizes([400, 700]) root_layout.addWidget(splitter) self.statusBar().showMessage("未连接") def toggle_connect(self): # 根据按钮状态决定连接或断开 if self.workers: self.disconnect_serial() else: self.connect_serial() def connect_serial(self): port = self.port_edit.text().strip() if not port: QMessageBox.warning(self, "提示", "请输入串口号") return baud = int(self.baud_combo.currentText()) interval = self.interval_spin.value() / 1000.0 # 这里实际应从配置文件或界面读取设备列表。 # 为了演示，构造一个示例设备配置。 serial_config = SerialConfig( port=port, baudrate=baud, timeout=1.0, devices=[ DeviceConfig( slave_id=1, name="PLC1", variables=[ VariableConfig(name="温度", address=0, func_code=3, scale=0.1, unit="℃", high_limit=80, low_limit=10), VariableConfig(name="压力", address=1, func_code=3, scale=0.01, unit="MPa", high_limit=1.6, low_limit=0), ], ), DeviceConfig( slave_id=2, name="仪表2", variables=[ VariableConfig(name="流量", address=0, func_code=4, scale=1.0, unit="m3/h", high_limit=500, low_limit=0), ], ), ], ) worker = PollWorker(serial_config, interval=interval) worker.data_ready.connect(self.on_data_ready) worker.error.connect(self.on_error) worker.status_changed.connect(self.on_status_changed) worker.start() self.workers.append(worker) self.connect_btn.setText("断开") self.port_edit.setEnabled(False) self.baud_combo.setEnabled(False) self.statusBar().showMessage(f"正在连接 {port} ...")

上面这段代码中，设备配置是写死在代码里的。工程化做法是从 JSON 或界面动态添加，这里为了演示清晰，先给出静态示例。

7.2 数据刷新槽函数

当工作线程发送数据过来时，主窗口需要更新表格、仪表、曲线，并触发报警和 CSV 保存。

@pyqtSlot(str, dict) def on_data_ready(self, port, data): # data 形如 {"PLC1": {"温度": 25.3, "压力": 1.2}, "仪表2": {"流量": 100}} flat_record = OrderedDict() flat_record["时间"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.data_table.setRowCount(0) # 用于曲线显示的选中变量，这里简单取第一个变量 first_value = None first_name = "" for device_name, dev_data in data.items(): for var_name, value in dev_data.items(): var_key = f"{device_name}.{var_name}" flat_record[var_key] = value self._append_table_row(var_key, value, "") if first_value is None and value is not None: first_value = value first_name = var_key # 预警判断 self._check_alarm(device_name, var_name, value) # CSV 保存 self.csv_logger.save_record(flat_record) # 更新仪表盘（示例：显示第一个变量） if first_value is not None: self.meter_label.setText(first_name) self.meter_value_label.display(first_value) # 根据量程调节，这里用简单归一化 dial_value = max(0, min(1000, int(first_value * 10))) self.meter_dial.setValue(dial_value) # 更新实时曲线 self._update_chart(first_value) def _append_table_row(self, var_key, value, unit): row = self.data_table.rowCount() self.data_table.insertRow(row) self.data_table.setItem(row, 0, QTableWidgetItem(var_key)) self.data_table.setItem(row, 1, QTableWidgetItem(str(value))) self.data_table.setItem(row, 2, QTableWidgetItem(unit))

上面的_append_table_row中 unit 没有从配置里取，实际项目中建议传递单位，这里作为演示简化。更合理的方式是让data同时带上单位，或者在主窗口维护一份变量配置表。

继续补充曲线和预警方法：

def _update_chart(self, value): if value is None: return self.chart_time.append(time.time()) self.chart_data.append(value) # 只保留最近 100 个点 if len(self.chart_data) > 100: self.chart_data.pop(0) self.chart_time.pop(0) self.plot_curve.setData(self.chart_time, self.chart_data) def _check_alarm(self, device_name, var_name, value): if value is None: return # 这里需要根据变量配置判断上下限。 # 简化处理：如果数值大于 100 就报警，实际应从配置读取。 # 工程上可以把配置放到主窗口的 dict 中。 if value > 100: item = QListWidgetItem(f"[上限] {device_name}.{var_name} = {value}") self.alarm_list.addItem(item) self.alarm_list.scrollToBottom()

需要引入time和datetime。_check_alarm中硬编码阈值不好，应该在主窗口中维护一份变量配置的索引。可以把 SerialConfig 传给主窗口，或者把变量配置注册到一个全局变量表。我们后面在最佳实践里说明。

最后是断开逻辑和错误处理：

def disconnect_serial(self): for worker in self.workers: worker.stop() worker.wait(3000) self.workers.clear() self.connect_btn.setText("连接") self.port_edit.setEnabled(True) self.baud_combo.setEnabled(True) self.statusBar().showMessage("已断开") @pyqtSlot(str, bool) def on_status_changed(self, port, online): if online: self.statusBar().showMessage(f"{port} 连接成功", 3000) else: self.statusBar().showMessage(f"{port} 离线", 3000) @pyqtSlot(str, str) def on_error(self, port, message): QMessageBox.critical(self, "通信错误", f"{port}: {message}")

7.3 程序入口

在main.py中启动应用。

# main.py import sys from PyQt5.QtWidgets import QApplication from main_window import MainWindow def main(): app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec_()) if __name__ == "__main__": main()

8. 多串口扩展方式

上面代码里只创建了一个PollWorker。如果现场有多路串口，比如 COM3 和 COM4，只需要创建两个PollWorker，分别传入不同的SerialConfig，并加入self.workers列表即可。

# 多串口连接示例 configs = [ SerialConfig(port="COM3", baudrate=9600, devices=[...]), SerialConfig(port="COM4", baudrate=115200, devices=[...]), ] for cfg in configs: worker = PollWorker(cfg, interval=1.0) worker.data_ready.connect(self.on_data_ready) worker.error.connect(self.on_error) worker.status_changed.connect(self.on_status_changed) worker.start() self.workers.append(worker)

这样，每个串口独立线程读取，互不干扰。两个串口的设备数量、从站地址、变量列表都可以不同。

更合理的做法是把串口配置放入 JSON 文件，程序启动时读取并自动创建 worker。下面是一个简单的 JSON 配置示例：

{ "output_dir": "./data", "interval_ms": 1000, "serials": [ { "port": "COM3", "baudrate": 9600, "timeout": 1, "devices": [ { "slave_id": 1, "name": "PLC1", "variables": [ {"name": "温度", "address": 0, "quantity": 1, "func_code": 3, "scale": 0.1, "unit": "℃", "high_limit": 80, "low_limit": 10}, {"name": "压力", "address": 1, "quantity": 1, "func_code": 3, "scale": 0.01, "unit": "MPa", "high_limit": 1.6, "low_limit": 0} ] } ] } ] }

在工程中，可以用json.load()读取它，并转成 dataclass 对象。这样现场调整设备参数就不需要改代码。

9. 运行与验证

9.1 使用 Modbus 从站模拟器

如果你手头没有真实设备，可以安装 Modbus Slave 模拟工具（如 ModRSsim2、Modbus Slave 等），创建一个从站，设置寄存器值，然后用本程序读取。

以 Modbus Slave 为例：

1. 新建一个从站，设置 Slave ID = 1。
2. 选择功能码 03（保持寄存器）。
3. 设置起始地址 0，数量 2。
4. 在寄存器表格中填入数值，例如地址 0 = 250，表示温度 25.0℃（scale=0.1）。
5. 选择串口 COM3，波特率 9600。

然后在我们的上位机界面中填入 COM3，点击连接，就能看到温度值变化。

9.2 运行程序

python main.py

启动后界面应能正常弹出。点击“连接”，如果串口打开成功，状态栏会提示“连接成功”。数据表格持续刷新，曲线缓慢滚动，CSV 文件在./data目录中生成。

9.3 预期效果

• 表格中能看到“PLC1.温度”、“PLC1.压力”等实时值。
• 仪表盘显示第一个变量对应的数值。
• 曲线每采集一次增加一个点。
• 当温度超过设定的上限时，报警列表追加一条记录。
• 打开data/data_2025-01-01.csv，可以看到每一行对应一次采集。

10. 常见问题与排查思路

排查建议按以下步骤进行：

1. 先用串口调试助手确认设备能正常返回数据。
2. 再用 Modbus 调试工具（如 Modbus Poll）验证从站地址、寄存器地址和功能码。
3. 将同样的地址和参数填入上位机配置，观察输出。
4. 如果仍失败，打开 Python 日志，查看 pymodbus 返回的异常码。

11. 最佳实践与工程建议

11.1 配置管理

不要写死设备参数。建议使用 JSON、YAML 或数据库保存配置。程序启动时加载配置，生成各个串口和设备对象。这样现场维护人员可以直接修改配置文件，不需要重新改代码。

例如，将配置加载逻辑放在config.py中：

def load_config_from_json(path: str) -> List[SerialConfig]: import json with open(path, "r", encoding="utf-8") as f: data = json.load(f) serials = [] for s in data["serials"]: devices = [] for d in s["devices"]: variables = [ VariableConfig(**v) for v in d.get("variables", []) ] devices.append( DeviceConfig(slave_id=d["slave_id"], name=d.get("name", ""), variables=variables) ) serials.append( SerialConfig( port=s["port"], baudrate=s.get("baudrate", 9600), timeout=s.get("timeout", 1), devices=devices, ) ) return serials

11.2 日志记录

串口异常、读取超时、数据校验错误都应该记录到日志文件。使用 Python 的logging模块，配置 RotatingFileHandler，可以保存最近的历史日志，避免日志文件无限增长。

import logging from logging.handlers import RotatingFileHandler handler = RotatingFileHandler("app.log", maxBytes=5*1024*1024, backupCount=3) logging.basicConfig(level=logging.INFO, handlers=[handler])

11.3 预警逻辑优化

实际项目中，阈值应该在变量配置中指定，而不是硬编码。主窗口可以维护一个变量配置字典，当数据到达时根据设备名.变量名查找对应的上下限。

告警还需要考虑：

• 死区：避免数值在上限附近反复触发报警。
• 确认机制：重要报警需要人工确认。
• 报警去重：同一变量持续越限只记录一次，恢复正常时再记录恢复事件。

11.4 UI 线程安全

Qt 规定 UI 只能在主线程中操作。PollWorker通过信号把数据传递到主线程，这是正确的做法。不要在 worker 中直接调用setText、addItem等界面方法。

如果需要停止线程，应该设置running=False，等待线程结束，不要强杀线程。

11.5 串口断开重连

现场设备断电或串口松动会导致 worker 退出。我们可以在循环中检测连接状态，当读取失败次数超过阈值时，重新连接。

# 在 worker 的 run 方法中 fail_count = 0 while self.running: try: data = client.read_device(device) fail_count = 0 except Exception: fail_count += 1 if fail_count > 3: logger.warning("重连中...") client.close() time.sleep(2) client.connect() fail_count = 0

11.6 性能优化

如果设备数量多、寄存器数量大，可以调整轮询策略：

• 对于变化缓慢的温度、液位等变量，适当延长轮询间隔。
• 将多个连续寄存器一次性读取，再在本地拆分。
• 使用pymodbus的异步客户端配合 asyncio，但学习成本更高。
• 考虑把图形绘制单独放到 pyqtgraph 的 GPU 加速模式（可选）。

11.7 打包发布

开发完成后，可以用 PyInstaller 打包成 exe：

pip install pyinstaller pyinstaller -F -w main.py --name ModbusStation

-w表示不显示控制台窗口。注意，打包后需要把配置文件、图标等资源放到指定目录。由于 pymodbus、pyqtgraph 等库较大，打包后的 exe 可能会超过 100MB，这是正常的。

12. 总结与学习路线

到这里，我们已经完成了一套基于 Python + PyQt5 的 Modbus 多串口上位机核心程序。从串口通信封装、工作线程设计、界面展示到 CSV 保存和报警提示，形成了一个完整的最小闭环。

如果只是入门，你可以先跑通上述代码，然后逐步增加以下功能：

1. 动态添加设备：在界面上增加“添加设备”按钮，把设备配置写入 JSON。
2. 多从站多串口管理：将多个串口配置加载到程序中，实现真正意义上的多串口监控。
3. 写寄存器功能：增加按钮或输入框，通过功能码 06/16 写入参数，实现远程控制。
4. 历史曲线查询：从 CSV 文件读取历史数据，用 pyqtgraph 绘制回放。
5. 用户权限管理：区分操作员和管理员，避免误操作。
6. 数据库存储：将 CSV 换成 SQLite 或 MySQL，方便多机查询。

本文给出的代码只是工程骨架，实际项目里还需要结合具体设备的寄存器手册进行调试。建议先从最简单的单设备、单变量开始验证，确认通信无误后再扩展多设备多串口。

如果在开发过程中遇到问题，可以先检查串口参数、从站地址和寄存器地址，再用 Modbus 调试工具对比验证。Python 的报错信息比较直观，加上日志记录，定位问题并不困难。

希望这篇文章对你构建自己的上位机工具有所帮助。后续我也会继续分享 Modbus TCP、写寄存器、数据库存储和界面优化的进阶内容，欢迎保持关注。