Python PyQt 上位机设计与实战指南

四个关键角色

1. QApplication：整个 GUI 程序的'心脏'，负责事件循环、消息分发。
2. MainWindow (继承自 QWidget)：主窗口本身，放置各种控件。
3. QPushButton：最常见的交互元素之一，点击后发出 clicked 信号。

信号与槽机制：这是 PyQt 的灵魂。通过 self.btn.clicked.connect(self.on_start_clicked)，当用户点击按钮时自动调用函数，让界面和逻辑解耦。

串口通信：解决阻塞问题

在主线程直接读串口会导致界面卡死。正确做法是将通信放入独立线程。

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

class SerialWorker(QObject):
    data_received = pyqtSignal(str)  # 自定义信号

    def __init__(self, port, baudrate=115200):
        super().__init__()
        self.port = port
        self.baudrate = baudrate
        self.is_running = False

    def start(self):
        self.is_running = True
        self.thread = threading.Thread(target=self._read_loop)
        self.thread.start()

    def _read_loop(self):
        try:
            ser = serial.Serial(self.port, self.baudrate, timeout=1)
            while self.is_running:
                if ser.in_waiting > 0:
                    line = ser.readline().decode('utf-8').strip()
                    self.data_received.emit(line)
            ser.close()
        except Exception as e:
            self.data_received.emit(f"ERROR: {str(e)}")

    def stop(self):
        self.is_running = False

重点在于 data_received = pyqtSignal(str) 是跨线程安全的通信通道。子线程发射信号，由 Qt 内部机制投递到主线程处理。

接入主程序：

# 在主窗口中初始化串口模块
self.serial_worker = SerialWorker('COM3', 115200)
self.serial_worker.data_received.connect(self.handle_serial_data)
self.serial_worker.start()

# 定义数据处理函数
def handle_serial_data(self, data):
    print("收到:", data)
    # 可以在这里更新文本框、触发绘图、解析协议...

数据显示：表格与状态指示

假设传感器返回格式为 TEMP:23.5,HUMI:45.2，需结构化展示。

方案一：表格显示历史记录

from PyQt5.QtWidgets import QTableWidget, QTableWidgetItem
self.table = QTableWidget(100, 2)  # 100 行 2 列
self.table.setHorizontalHeaderLabels(['时间', '数值'])

def add_table_row(self, value):
    row = self.table.rowCount()
    for i in range(row - 1):  # 向上滚动
        for col in range(2):
            item = self.table.item(i+1, col)
            if item:
                self.table.setItem(i, col, QTableWidgetItem(item.text()))
    # 插入新行
    self.table.setItem(row-1, 0, QTableWidgetItem(time.strftime("%H:%M:%S")))
    self.table.setItem(row-1, 1, QTableWidgetItem(value))

方案二：状态指示灯

温度超过 30℃变红：

from PyQt5.QtGui import QColor
from PyQt5.QtWidgets import QLabel
self.status_label = QLabel("正常")
self.status_label.setStyleSheet("background-color: green; color: white; padding: 5px")

def update_status(self, temp):
    if float(temp) > 30:
        self.status_label.setText("高温报警！")
        self.status_label.setStyleSheet("background-color: red; color: white; padding: 5px")
    else:
        self.status_label.setText("正常")
        self.status_label.setStyleSheet("background-color: green; color: white; padding: 5px")

实时波形图：使用 pyqtgraph

Matplotlib 不适合实时监控，推荐使用专为科学计算优化的 pyqtgraph。

import pyqtgraph as pg
import numpy as np

# 初始化绘图区域
plot_widget = pg.PlotWidget()
plot_curve = plot_widget.plot(pen='y')  # 黄色曲线

# 设置坐标轴
plot_widget.setLabel('left', '电压', units='V')
plot_widget.setLabel('bottom', '时间', units='s')
plot_widget.setTitle('实时采样波形')

# 缓冲区（固定长度，形成环形队列）
buffer_size = 1000
x_data = np.linspace(0, 10, buffer_size)
y_data = np.zeros(buffer_size)

def update_plot(new_value):
    global y_data
    y_data = np.append(y_data[1:], new_value)  # 移除旧值，添加新值
    plot_curve.setData(x_data, y_data)  # 快速更新

# 每 50ms 刷新一次
from PyQt5.QtCore import QTimer
timer = QTimer()
timer.timeout.connect(lambda: update_plot(np.random.rand()))
timer.start(50)  # 20 FPS

关键点：使用 NumPy 数组存储数据，setData() 增量更新，配合定时器形成稳定动画节奏。

复杂系统架构设计

标准四层架构：

┌─────────────────────┐
│ 用户界面层 (UI)      │ ← 用户看到的一切
├─────────────────────┤
│ 控制逻辑层 (Controller)│ ← 参数处理、状态切换
├─────────────────────┤
│ 数据通信层 (Serial)   │ ← 串口/网络收发
├─────────────────────┤
│ 数据模型层 (Model)    │ ← 缓冲区、协议解析、文件保存
└─────────────────────┘

每一层只和相邻层交互。好处包括修改通信协议不影响界面、更换连接方式只需替换底层模块、单元测试更容易。

常见开发陷阱

坑 1：中文乱码 or 数据断包

原因：编码不统一 + 未处理粘包问题。 ✅ 正确做法：

# 统一使用 UTF-8
line = ser.readline().decode('utf-8', errors='ignore').strip()
# 或者采用定长帧接收
data = ser.read(32)  # 固定每次读 32 字节

更高级的做法是定义帧头帧尾，例如 $DATA,23.5,*7F\n。

坑 2：关闭程序后串口仍被占用

原因：异常退出时没有正确释放资源。 ✅ 解决方案：

def closeEvent(self, event):
    if hasattr(self, 'serial_worker'):
        self.serial_worker.stop()
        self.serial_worker.thread.join(timeout=1.0)  # 等待线程结束
    event.accept()

记得绑定 closeEvent，确保优雅退出。

坑 3：长时间运行内存暴涨

原因：不断往列表追加数据却不清理。 ✅ 解决办法：限制缓冲区大小、定期清理日志文本框、使用 weakref 防止循环引用。

提升产品感的加分项

✅ 配置记忆功能

下次打开自动填上次的串口号和波特率：

from PyQt5.QtCore import QSettings
settings = QSettings("MyCompany", "SensorMonitor")
settings.setValue("port", "COM3")
settings.setValue("baudrate", 115200)
# 启动时读取
last_port = settings.value("port", "COM1")
last_baud = int(settings.value("baudrate", 9600))

✅ 日志输出面板

加个 QTextEdit 实时显示通信日志，故障排查效率翻倍。

✅ 数据导出按钮

一键保存当前波形为 CSV 文件：

with open('data.csv', 'w') as f:
    for t, v in zip(x_data, y_data):
        f.write(f"{t:.3f},{v:.3f}\n")

✅ 图形美化

用 QSS 给按钮加渐变背景、圆角边框：

btn.setStyleSheet("""
    QPushButton {
        background: qlineargradient(x1:0, y1:0, x2:0, y2:1, stop:0 #5a9, stop:1 #488);
        color: white;
        border-radius: 8px;
        padding: 10px;
    }
""")