PyQt5 实现串口接数据波形显示工具_串口数据波形显示

安装模块

#pip install pyqt5 -i https://pypi.tuna.tsinghua.edu.cn/simple
#pip install pyqtgraph -i https://pypi.tuna.tsinghua.edu.cn/simple
#pip install serial -i https://pypi.tuna.tsinghua.edu.cn/simple
#pip install pyserial -i https://pypi.tuna.tsinghua.edu.cn/simple

工具简述

基于PyQt5开发UI界面使用QtDesigner设计，需要使用到serial模块（串口库）和pyqtgraph(图形库)。上位机通过串口接收来自MCU发送数据，解析出每一个数据项并以波形图的方式显示。本例程下位机是Raspberry Pi Pico发送HMC5883L地磁模块数据，数据项有x,y,z,h等，数据格式’$$:x,y,z,h’。

主程序代码

import sys
import numpy as np

from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *

import serial
import serial.tools.list_ports
from ui_main import Ui_MainWindow
import pyqtgraph as pg


class AppMainWindow(QMainWindow, Ui_MainWindow):
    def __init__(self, parent=None):
        super(AppMainWindow, self).__init__(parent)
        self.setupUi(self)
        self.setWindowTitle("串口数据波形显示工具")
        self.PosX = 0
        self.dataIndex = 0          # 数据列表当前索引
        self.dataMaxLength = 10000  # 数据列表最大长度
        self.dataheader = b'$$:'    # 数据帧开头
        self.dataX = np.zeros(self.dataMaxLength, dtype=float)
        self.dataY = np.zeros(self.dataMaxLength, dtype=float)
        self.dataZ = np.zeros(self.dataMaxLength, dtype=float)
        self.dataH = np.zeros(self.dataMaxLength, dtype=float)

        self.mSerial = serial.Serial()
        self.ScanComPort()
        self.SelectComPort()
        
        self.btnScanPort.clicked.connect(self.ScanComPort)
        self.btnOpenPort.clicked.connect(self.OpenComPort)
        self.btnClosePort.clicked.connect(self.CloseComPort)
        self.cmbComPort.currentIndexChanged.connect(self.SelectComPort)

        self.mTimer = QTimer()
        self.mTimer.timeout.connect(self.ReceiverPortData)

        self.plotM.setAntialiasing(True)
        # self.plotM.setBackground()
        self.canvasX = self.plotM.plot(self.dataX, pen=pg.mkPen(color='r', width=1))
        self.canvasY = self.plotM.plot(self.dataY, pen=pg.mkPen(color='g', width=1))
        self.canvasZ = self.plotM.plot(self.dataZ, pen=pg.mkPen(color='b', width=1))
        self.canvasH = self.plotM.plot(self.dataH, pen=pg.mkPen(color='y', width=1))

    def ScanComPort(self):
        self.cmbComPort.clear()
        self.portDict = {}
        portlist = list(serial.tools.list_ports.comports())
        for port in portlist:
            self.portDict["%s" % port[0]] = "%s" % port[1]
            self.cmbComPort.addItem(port[0])
        if len(self.portDict) == 0:
            QMessageBox.critical(self, "警告", "未找到串口设备！", QMessageBox.Cancel, QMessageBox.Cancel)
        pass
    
    def SelectComPort(self):
        if len(self.portDict) > 0 :
            self.labComPortName.setText(self.portDict[self.cmbComPort.currentText()])
        else:
            self.labComPortName.setText("未检测到串口设备！")
        pass

    def OpenComPort(self):
        self.mSerial.port = self.cmbComPort.currentText()
        self.mSerial.baudrate = int(self.cmbBaudrate.currentText())

        if self.mSerial.isOpen():
            QMessageBox.warning(self, "警告", "串口已打开", QMessageBox.Cancel, QMessageBox.Cancel)
        else:
            try:
                self.btnOpenPort.setEnabled(False)
                self.mSerial.open()
                self.mSerial.flushInput()
                self.mSerial.flushOutput()
                self.mTimer.start(1)
            except:
                QMessageBox.critical(self, "警告", "串口打开失败！", QMessageBox.Cancel, QMessageBox.Cancel)
                self.btnOpenPort.setEnabled(True)
        print(self.mSerial)
        pass

    def CloseComPort(self):
        self.mTimer.stop()
        if self.mSerial.isOpen():
            self.btnOpenPort.setEnabled(True)
            self.mSerial.flushInput()
            self.mSerial.flushOutput()
            self.mSerial.close()
        pass

    def ReceiverPortData(self):
        '''
        接收串口数据，并解析出每一个数据项更新到波形图
        数据帧格式'$$:95.68,195.04,-184.0\r\n'
        每个数据帧以b'$$:'开头，每个数据项以','分割
        '''
        try:
            n = self.mSerial.inWaiting()
        except:
            self.CloseComPort()

        if n > 0:
            # 端口缓存内有数据
            try:
                self.recvdata = self.mSerial.readline(1024) # 读取一行数据最大长度1024字节

                if self.recvdata.decode('UTF-8').startswith(self.dataheader.decode('UTF-8')):
                    rawdata = self.recvdata[len(self.dataheader) : len(self.recvdata) - 2]
                    data = rawdata.split(b',')
                    # print(rawdata)
                    
                    if self.dataIndex < self.dataMaxLength:
                        # 接收到的数据长度小于最大数据缓存长度，直接按索引赋值，索引自增1
                        # print(rawdata, rawdata[0], rawdata[1], rawdata[2], self.dataX[self.dataIndex], self.dataY[self.dataIndex], self.dataZ[self.dataIndex])
                        self.dataX[self.dataIndex] = float(data[0])
                        self.dataY[self.dataIndex] = float(data[1])
                        self.dataZ[self.dataIndex] = float(data[2])
                        self.dataH[self.dataIndex] = float(data[3])
                        self.dataIndex = self.dataIndex + 1
                    else:
                        # 寄收到的数据长度大于或等于最大数据缓存长度，丢弃最前一个数据新数据添加到数据列尾
                        self.dataX[:-1] = self.dataX[1:]
                        self.dataY[:-1] = self.dataY[1:]
                        self.dataZ[:-1] = self.dataZ[1:]
                        self.dataH[:-1] = self.dataH[1:]

                        self.dataX[self.dataIndex - 1] = float(data[0])
                        self.dataY[self.dataIndex - 1] = float(data[1])
                        self.dataZ[self.dataIndex - 1] = float(data[2])
                        self.dataH[self.dataIndex - 1] = float(data[3])
                    
                    # 更新波形数据
                    self.canvasX.setData(self.dataX)
                    self.canvasY.setData(self.dataY)
                    self.canvasZ.setData(self.dataZ)
                    self.canvasH.setData(self.dataH)

                    # self.canvasX.setPos(self.PosX, 0)
                    # self.canvasY.setPos(self.PosX, 0)
                    # self.canvasZ.setPos(self.PosX, 0)
            except:
                pass
        pass

    def SendPortData(self):
        pass


if __name__ == "__main__":
    app = QApplication(sys.argv)
    win = AppMainWindow()
    win.show()
    sys.exit(app.exec_())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

Qt Designer设计UI界面

程序运行效果