Python PyQt 上位机开发：温控系统实时监控实例

Python+PyQt 工业级温控监控系统开发

在工业现场，温控系统的可视化监控至关重要。本文介绍使用 Python 和 PyQt 搭建一套功能完整、稳定可靠的工业监控界面的方法。项目涵盖界面搭建、串口通信、实时绘图及多线程处理。

为什么选择 PyQt 做上位机？

相比专业组态软件，Python 配合 PyQt 适合中小型项目及快速原型开发：

• 开发效率高：语法简洁，代码量少
• 跨平台运行：Windows/Linux/macOS 通吃
• 成本几乎为零：开源免费，无需授权
• 易于集成 AI 与数据分析模块

本项目实现一个典型的温控监控系统：PC 通过串口读取下位机上传的温度数据，实时显示并绘图，支持报警、参数设置和数据存储。

界面搭建

首先搭建基础窗口。PyQt5 基于事件驱动模型，核心是 QApplication 实例。

import sys
from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QLabel, QPushButton

class TemperatureMonitor(QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()

    def init_ui(self):
        layout = QVBoxLayout()
        # 当前温度显示
        self.temp_label = QLabel("当前温度：--℃")
        self.temp_label.setStyleSheet("font-size: 18px; color: #333;")
        
        # 报警提示按钮（初始隐藏）
        self.alert_button = QPushButton("⚠️ 超温报警！")
        self.alert_button.setStyleSheet("background-color: red; color: white; font-weight: bold;")
        self.alert_button.hide()
        
        layout.addWidget(self.temp_label)
        layout.addWidget(self.alert_button)
        self.setLayout(layout)
        self.setWindowTitle("温控系统监控 - 上位机")
        self.resize(400, 200)
        self.show()

if __name__ == '__main__':
    app = QApplication(sys.argv)
    window = TemperatureMonitor()
    sys.exit(app.exec_())

该代码创建了带标签和按钮的窗口，并使用 QVBoxLayout 布局管理器，控件会自动对齐排列。

串口通信

工业现场常用串口通信。使用 pyserial 库读取数据流。

安装命令：

pip install pyserial

将串口读取放入子线程，避免卡住主界面：

import serial
import threading
from PyQt5.QtCore import QObject, pyqtSignal

class SerialWorker(QObject):
    data_received = pyqtSignal(float)  # 自定义信号，用于传温度值

    def __init__(self, port='COM3', baudrate=9600):
        super().__init__()
        try:
            self.ser = serial.Serial(port, baudrate, timeout=1)
            self.running = True
        except Exception as e:
            print(f"无法打开串口 {port}: {e}")
            self.running = False

    def start_reading(self):
        while self.running and self.ser.is_open:
            if self.ser.in_waiting > 0:
                line = self.ser.readline().decode('utf-8').strip()
                try:
                    temp = float(line)
                    self.data_received.emit(temp)  # 发射信号给主线程
                except ValueError:
                    continue  # 忽略非法数据

    def send_command(self, cmd):
        """向下位机发送指令"""
        if self.ser.is_open:
            self.ser.write(f"{cmd}\r\n".encode())

    def stop(self):
        self.running = False
        if self.ser.is_open:
            self.ser.close()

关键点：

• 多线程安全：串口读取不能放在主线程。
• 信号传递：使用 pyqtSignal 在线程间通信。
• 异常处理：串口可能被占用、断开或收到乱码，必须加 try-except 保护。
• 协议容错：实际项目中建议增加帧头识别、CRC 校验等机制。

初始化示例：

worker = SerialWorker('COM3', 115200)
thread = threading.Thread(target=worker.start_reading, daemon=True)
thread.start()

实时绘图

使用 pyqtgraph 进行实时绘图。相比 Matplotlib，pyqtgraph 基于 OpenGL 加速，刷新率高。

安装命令：

pip install pyqtgraph

封装滚动波形图组件：

import pyqtgraph as pg
from PyQt5.QtCore import QTimer

class RealTimePlot:
    def __init__(self, plot_widget: pg.PlotWidget):
        self.plot_widget = plot_widget
        self.plot_widget.setLabel('left', '温度 (°C)')
        self.plot_widget.setLabel('bottom', '时间 (s)')
        self.plot_widget.setTitle('实时温度曲线')
        self.plot_widget.setYRange(0, 100)
        self.plot_widget.showGrid(x=True, y=True)
        self.curve = self.plot_widget.plot(pen='g')
        self.buffer_size = 100
        self.x_data = list(range(self.buffer_size))
        self.y_data = [0] * self.buffer_size
        self.timer = QTimer()
        self.timer.timeout.connect(self.update_plot)

    def update_data(self, new_temp):
        """接收新数据"""
        self.y_data = self.y_data[1:] + [new_temp]
        min_y = max(0, min(self.y_data) - 5)
        max_y = max(self.y_data) + 5
        self.plot_widget.setYRange(min_y, max_y)

    def update_plot(self):
        """定时刷新图像"""
        self.curve.setData(self.x_data, self.y_data)

    def start(self):
        self.timer.start(100)  # 每 100ms 刷新一次，即 10FPS

在 UI 中嵌入 PlotWidget：

from pyqtgraph import PlotWidget
# 在 init_ui 中添加
plot_widget = PlotWidget()
layout.addWidget(plot_widget)
self.plotter = RealTimePlot(plot_widget)
self.plotter.start()

调用 update_data 即可刷新曲线。

完整交互逻辑

系统包含以下核心模块：

1. GUI 界面（PyQt）
2. 数据接收（pyserial + 多线程）
3. 数据展示（pyqtgraph）
4. 用户控制（按钮、输入框）

通过 信号与槽机制 协同工作。

连接信号示例：

# 连接信号
worker.data_received.connect(self.on_temperature_update)

def on_temperature_update(self, temp):
    # 更新 LCD 显示
    self.temp_label.setText(f"当前温度：{temp:.1f}℃")
    # 检查是否超限
    if temp > 80 or temp < 30:
        self.alert_button.show()
    else:
        self.alert_button.hide()
    # 更新图表
    self.plotter.update_data(temp)
    # 存入日志文件
    self.log_data(temp)

用户点击'设置目标温度'按钮时：

def set_target_temp(self):
    target = self.target_input.text()  # 来自 QLineEdit
    try:
        value = float(target)
        worker.send_command(f"SET_TEMP:{value}")
    except ValueError:
        QMessageBox.warning(self, "输入错误", "请输入有效数值")

工程级细节

1. 资源释放问题

确保退出前关闭资源。

def closeEvent(self, event):
    if hasattr(self, 'worker'):
        self.worker.stop()
    event.accept()

2. 数据稳定性

可能因帧不完整或干扰导致。 建议：

• 添加帧头检测，如以 $TEMP: 开头
• 使用环形缓冲区重组数据包
• 对关键指令启用 CRC 校验

3. 内存管理

避免列表无限增长。 建议：

• 使用固定长度的 deque 缓冲区
• 或定期清理旧数据

from collections import deque
self.y_data = deque([0]*100, maxlen=100)

优化建议

• 使用 QSettings 保存配置
• 日志按日期命名，便于后期分析
• 增加静音功能

系统架构

系统结构如下：

[下位机] -> (UART) -> [SerialWorker] -> (signal) -> [Central Logic]
                                              |-> [UI Update]
                                              |-> [Data Logging]
                                              |-> [RealTimePlot]

构建数据采集与处理中枢：采集 → 解析 → 分发 → 展示 → 存储。

扩展方向

• 接数据库（SQLite/MySQL），支持海量查询
• 加网络服务，用 Flask 或 WebSocket 实现远程监控
• 引入机器学习模型，预测温度趋势提前预警
• 打包成 exe，发给客户一键安装