小灰灰搞电子头像
PyQt+QML 实现串口助手源码分享封面图

PyQt+QML 实现串口助手源码分享

在这里插入图片描述

PyQt+QML 实现串口助手源码分享

一、效果展示

在这里插入图片描述
在这里插入图片描述

二、源码分享

1、工程结构

在这里插入图片描述

2、库安装

在这里插入图片描述

3、main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
串口调试助手 - 主入口
"""

import sys
import os

from PyQt6.QtWidgets import QApplication
from PyQt6.QtQml import QQmlApplicationEngine
from PyQt6.QtCore import QSettings, QObject, pyqtSlot

from serial_manager import SerialManager
from receive_model import ReceiveModel


class FileSaveHelper(QObject):
    """文件保存助手 - 供 QML 调用保存日志"""

    def __init__(self, parent=None):
        super().__init__(parent)

    @pyqtSlot(str, str)
    def saveFile(self, path: str, content: str):
        """保存文件"""
        try:
            with open(path, 'w', encoding='utf-8') as f:
                f.write(content)
            print(f"文件已保存: {path}")
        except Exception as e:
            print(f"保存文件失败: {e}")


class ConfigManager(QObject):
    """配置管理器 - 使用 QSettings 持久化配置"""

    def __init__(self, parent=None):
        super().__init__(parent)
        self._settings = QSettings("SerialAssistant", "SerialAssistant")

    def save_config(self, serial_mgr: SerialManager):
        """保存配置"""
        self._settings.setValue("portName", serial_mgr.portName)
        self._settings.setValue("baudRate", serial_mgr.baudRate)
        self._settings.setValue("dataBits", serial_mgr.dataBits)
        self._settings.setValue("stopBits", serial_mgr.stopBits)
        self._settings.setValue("parity", serial_mgr.parity)
        self._settings.setValue("flowControl", serial_mgr.flowControl)
        self._settings.setValue("encoding", serial_mgr.encoding)
        self._settings.sync()

    def load_config(self, serial_mgr: SerialManager):
        """加载配置"""
        serial_mgr.portName = self._settings.value("portName", "COM1")
        serial_mgr.baudRate = int(self._settings.value("baudRate", 115200))
        serial_mgr.dataBits = int(self._settings.value("dataBits", 8))
        serial_mgr.stopBits = float(self._settings.value("stopBits", 1))
        serial_mgr.parity = self._settings.value("parity", "N")
        serial_mgr.flowControl = self._settings.value("flowControl", "None")
        serial_mgr.encoding = self._settings.value("encoding", "UTF-8")


def main():
    """主函数"""
    # 高 DPI 支持
    os.environ["QT_ENABLE_HIGHDPI_SCALING"] = "1"

    # 设置 QML 样式为 Fusion（避免 Windows 样式 DLL 问题）
    os.environ["QT_QUICK_CONTROLS_STYLE"] = "Fusion"

    # 创建应用
    app = QApplication(sys.argv)
    app.setApplicationName("串口调试助手")
    app.setOrganizationName("SerialAssistant")

    # 创建 QML 引擎
    engine = QQmlApplicationEngine()

    # 设置 QML 导入路径
    if hasattr(sys, '_MEIPASS'):
        # PyInstaller 打包后的路径
        qml_import_path = os.path.join(sys._MEIPASS, 'PyQt6', 'Qt6', 'qml')
    else:
        # 开发环境路径
        import PyQt6
        pyqt6_path = os.path.dirname(PyQt6.__file__)
        qml_import_path = os.path.join(pyqt6_path, 'Qt6', 'qml')

    if os.path.exists(qml_import_path):
        engine.addImportPath(qml_import_path)

    # 创建后端对象
    serial_mgr = SerialManager()
    receive_model = ReceiveModel()
    config_mgr = ConfigManager()
    file_save_helper = FileSaveHelper()

    # 连接信号：串口接收数据 -> 接收模型
    serial_mgr._worker = None  # 临时
    # 在 openPort 时动态连接

    # 重写 SerialManager 的 _onDataReceived 来触发模型更新
    original_onDataReceived = serial_mgr._onDataReceived
    def onDataReceivedWrapper(data, timestamp):
        original_onDataReceived(data, timestamp)
        receive_model.appendData(data, timestamp)
    serial_mgr._onDataReceived = onDataReceivedWrapper

    # 注册到 QML 上下文
    ctx = engine.rootContext()
    ctx.setContextProperty("serialManager", serial_mgr)
    ctx.setContextProperty("receiveModel", receive_model)
    ctx.setContextProperty("configManager", config_mgr)
    ctx.setContextProperty("fileSaveHelper", file_save_helper)

    # 加载配置
    config_mgr.load_config(serial_mgr)

    # 加载 QML
    qml_path = os.path.join(os.path.dirname(__file__), "main.qml")
    engine.load(qml_path)

    if not engine.rootObjects():
        print("错误: 无法加载 QML")
        return -1

    # 运行应用
    result = app.exec()

    # 关闭串口
    serial_mgr.closePort()

    # 保存配置
    config_mgr.save_config(serial_mgr)

    return result


if __name__ == "__main__":
    sys.exit(main())

4、main.qml

import QtQuick
import QtQuick.Controls
import QtQuick.Layouts
import QtQuick.Dialogs

ApplicationWindow {
    id: root
    visible: true
    width: 1000
    height: 700
    minimumWidth: 800
    minimumHeight: 600
    title: "串口调试助手"

    // 现代化配色方案
    readonly property color bgColor: "#1a1a2e"
    readonly property color surfaceColor: "#16213e"
    readonly property color cardColor: "#0f3460"
    readonly property color primaryColor: "#e94560"
    readonly property color secondaryColor: "#533483"
    readonly property color accentColor: "#0ea5e9"
    readonly property color textColor: "#eaeaea"
    readonly property color textSecondary: "#a0a0a0"
    readonly property color successColor: "#10b981"
    readonly property color errorColor: "#ef4444"
    readonly property color warningColor: "#f59e0b"

    background: Rectangle {
        color: bgColor
    }

    // 应用状态
    QtObject {
        id: appState
        property bool displayMode: false  // false=ASCII, true=HEX
        property bool autoScroll: true
        property bool sendHexMode: false
        property int lineEnding: 0  // 0=none, 1=\r, 2=\n, 3=\r\n
        property bool showTimestamp: true
    }

    // 自定义按钮组件
    component StyledButton: Button {
        id: styledBtn
        property bool isPrimary: false
        property bool isDanger: false

        implicitHeight: 32

        background: Rectangle {
            color: {
                if (!styledBtn.enabled) return "#555555"
                if (styledBtn.pressed) return isDanger ? "#dc2626" : (isPrimary ? "#c7294d" : "#432868")
                if (styledBtn.hovered) return isDanger ? "#f87171" : (isPrimary ? "#ff5a75" : "#6b46a1")
                return isDanger ? errorColor : (isPrimary ? primaryColor : secondaryColor)
            }
            radius: 6
            border.color: styledBtn.hovered ? "#ffffff40" : "transparent"
            border.width: 1

            // 阴影效果
            Rectangle {
                anchors.fill: parent
                anchors.topMargin: 2
                color: "transparent"
                opacity: 0.3
                radius: parent.radius
            }
        }

        contentItem: Label {
            text: styledBtn.text
            color: styledBtn.enabled ? "white" : "#888888"
            font.pixelSize: 13
            font.weight: Font.Medium
            horizontalAlignment: Text.AlignHCenter
            verticalAlignment: Text.AlignVCenter
        }
    }

    // 自定义下拉框样式
    component StyledComboBox: ComboBox {
        id: styledCombo
        implicitHeight: 32

        background: Rectangle {
            color: cardColor
            radius: 6
            border.color: styledCombo.hovered ? accentColor : "#ffffff20"
            border.width: 1
        }

        contentItem: Label {
            text: styledCombo.displayText
            color: textColor
            font.pixelSize: 13
            verticalAlignment: Text.AlignVCenter
            leftPadding: 10
            elide: Text.ElideRight
        }

        popup: Popup {
            y: styledCombo.height
            width: styledCombo.width
            padding: 0

            contentItem: ListView {
                clip: true
                implicitHeight: contentHeight
                model: styledCombo.popup.visible ? styledCombo.delegateModel : null
                currentIndex: styledCombo.highlightedIndex

                ScrollBar.vertical: ScrollBar {
                    policy: ScrollBar.AsNeeded
                    width: 8

                    contentItem: Rectangle {
                        color: "#555555"
                        radius: 4
                    }
                }
            }

            background: Rectangle {
                color: cardColor
                radius: 6
                border.color: "#ffffff20"
                border.width: 1
            }
        }

        delegate: ItemDelegate {
            width: styledCombo.width
            height: 32
            highlighted: styledCombo.highlightedIndex === index

            background: Rectangle {
                color: highlighted ? accentColor : (hovered ? "#ffffff10" : "transparent")
            }

            contentItem: Label {
                // 支持 ListModel 和数组两种模式
                text: {
                    if (typeof model !== 'undefined') {
                        if (styledCombo.textRole && model[styledCombo.textRole] !== undefined) {
                            return model[styledCombo.textRole]
                        } else if (model.modelData !== undefined) {
                            return model.modelData
                        } else if (model.text !== undefined) {
                            return model.text
                        }
                    }
                    return ""
                }
                color: highlighted ? "white" : textColor
                font.pixelSize: 13
                verticalAlignment: Text.AlignVCenter
                leftPadding: 10
            }
        }
    }

    // 主布局
    ColumnLayout {
        anchors.fill: parent
        anchors.margins: 15
        spacing: 12

        // 串口配置栏
        Rectangle {
            Layout.fillWidth: true
            Layout.preferredHeight: 90
            color: surfaceColor
            radius: 10
            border.color: "#ffffff10"
            border.width: 1

            GridLayout {
                anchors.fill: parent
                anchors.margins: 15
                columns: 8
                rowSpacing: 8
                columnSpacing: 12

                // 端口选择
                Label {
                    text: "端口:"
                    color: textSecondary
                    font.pixelSize: 13
                    Layout.preferredWidth: 60
                }

                StyledComboBox {
                    id: portCombo
                    Layout.preferredWidth: 130
                    model: ListModel {
                        id: portModel
                    }
                    textRole: "text"

                    Component.onCompleted: refreshPorts()

                    function refreshPorts() {
                        var ports = serialManager.refreshPorts()
                        console.log("发现端口数量:", ports.length)
                        portModel.clear()
                        for (var i = 0; i < ports.length; i++) {
                            console.log("端口", i, ":", ports[i].name, "-", ports[i].description)
                            portModel.append({
                                text: ports[i].name + " - " + ports[i].description,
                                name: ports[i].name
                            })
                        }
                        if (ports.length > 0) {
                            currentIndex = 0
                        }
                    }
                }

                // 波特率
                Label {
                    text: "波特率:"
                    color: textSecondary
                    font.pixelSize: 13
                }

                StyledComboBox {
                    id: baudCombo
                    Layout.preferredWidth: 100
                    model: ["1200", "2400", "4800", "9600", "19200", "38400", "57600", "115200", "230400", "460800", "921600"]
                    currentIndex: 7  // 115200

                    onCurrentTextChanged: {
                        serialManager.baudRate = parseInt(currentText)
                    }
                }

                // 数据位
                Label {
                    text: "数据位:"
                    color: textSecondary
                    font.pixelSize: 13
                }

                StyledComboBox {
                    id: dataBitsCombo
                    Layout.preferredWidth: 70
                    model: ["5", "6", "7", "8"]
                    currentIndex: 3  // 8

                    onCurrentTextChanged: {
                        serialManager.dataBits = parseInt(currentText)
                    }
                }

                // 停止位
                Label {
                    text: "停止位:"
                    color: textSecondary
                    font.pixelSize: 13
                }

                StyledComboBox {
                    id: stopBitsCombo
                    Layout.preferredWidth: 70
                    model: ["1", "1.5", "2"]
                    currentIndex: 0

                    onCurrentTextChanged: {
                        serialManager.stopBits = parseFloat(currentText)
                    }
                }

                // 校验位
                Label {
                    text: "校验位:"
                    color: textSecondary
                    font.pixelSize: 13
                }

                StyledComboBox {
                    id: parityCombo
                    Layout.preferredWidth: 80
                    model: ["None", "Even", "Odd", "Mark", "Space"]
                    currentIndex: 0

                    onCurrentTextChanged: {
                        var map = {"None": "N", "Even": "E", "Odd": "O", "Mark": "M", "Space": "S"}
                        serialManager.parity = map[currentText]
                    }
                }

                // 流控
                Label {
                    text: "流控:"
                    color: textSecondary
                    font.pixelSize: 13
                }

                StyledComboBox {
                    id: flowControlCombo
                    Layout.preferredWidth: 90
                    model: ["None", "RTS/CTS", "XON/XOFF"]
                    currentIndex: 0

                    onCurrentTextChanged: {
                        serialManager.flowControl = currentText
                    }
                }

                // 打开/关闭按钮
                StyledButton {
                    id: openCloseBtn
                    Layout.preferredWidth: 90
                    Layout.columnSpan: 2
                    text: serialManager.isOpen ? "关闭" : "打开"
                    isDanger: serialManager.isOpen
                    isPrimary: !serialManager.isOpen

                    onClicked: {
                        if (serialManager.isOpen) {
                            serialManager.closePort()
                        } else {
                            if (portCombo.currentIndex >= 0) {
                                serialManager.portName = portModel.get(portCombo.currentIndex).name
                                serialManager.openPort()
                            }
                        }
                    }
                }
            }
        }

        // 工具栏
        RowLayout {
            Layout.fillWidth: true
            Layout.preferredHeight: 45
            spacing: 10

            StyledButton {
                text: appState.displayMode ? "HEX" : "ASCII"
                Layout.preferredWidth: 80
                checkable: true
                checked: appState.displayMode
                isPrimary: checked

                onClicked: appState.displayMode = !appState.displayMode
            }

            CheckBox {
                text: "时间戳"
                checked: appState.showTimestamp
                onCheckedChanged: appState.showTimestamp = checked

                indicator: Rectangle {
                    width: 16
                    height: 16
                    x: parent.leftPadding
                    y: parent.topPadding + (parent.availableHeight - height) / 2
                    radius: 3
                    color: parent.checked ? accentColor : "transparent"
                    border.color: parent.checked ? accentColor : "#ffffff50"
                    border.width: 2

                    // 勾选标记
                    Rectangle {
                        visible: parent.parent.checked
                        width: 8
                        height: 2
                        anchors.centerIn: parent
                        color: "white"
                        rotation: 45
                    }
                }

                contentItem: Label {
                    text: parent.text
                    color: textColor
                    font.pixelSize: 13
                    leftPadding: parent.indicator.width + 8
                }
            }

            CheckBox {
                text: "自动滚动"
                checked: appState.autoScroll
                onCheckedChanged: appState.autoScroll = checked

                indicator: Rectangle {
                    width: 16
                    height: 16
                    x: parent.leftPadding
                    y: parent.topPadding + (parent.availableHeight - height) / 2
                    radius: 3
                    color: parent.checked ? accentColor : "transparent"
                    border.color: parent.checked ? accentColor : "#ffffff50"
                    border.width: 2

                    // 勾选标记
                    Rectangle {
                        visible: parent.parent.checked
                        width: 8
                        height: 2
                        anchors.centerIn: parent
                        color: "white"
                        rotation: 45
                    }
                }

                contentItem: Label {
                    text: parent.text
                    color: textColor
                    font.pixelSize: 13
                    leftPadding: parent.indicator.width + 8
                }
            }

            Item { Layout.fillWidth: true }

            Label {
                text: "RX: " + serialManager.rxCount + " B"
                color: accentColor
                font.pixelSize: 13
                font.weight: Font.Medium
            }

            Label {
                text: "TX: " + serialManager.txCount + " B"
                color: successColor
                font.pixelSize: 13
                font.weight: Font.Medium
            }

            StyledButton {
                text: "清零"
                Layout.preferredWidth: 60
                onClicked: serialManager.resetCounters()
            }

            StyledButton {
                text: "清空"
                Layout.preferredWidth: 60
                onClicked: receiveModel.clear()
            }

            StyledButton {
                text: "保存"
                Layout.preferredWidth: 60
                isPrimary: true
                onClicked: saveFileDialog.open()
            }
        }

        // 接收区
        Rectangle {
            Layout.fillWidth: true
            Layout.fillHeight: true
            color: "#0a0a0a"
            radius: 10
            border.color: "#ffffff10"
            border.width: 1

            ListView {
                id: receiveView
                anchors.fill: parent
                anchors.margins: 15
                model: receiveModel
                clip: true
                spacing: 4

                ScrollBar.vertical: ScrollBar {
                    policy: ScrollBar.AsNeeded
                    width: 10

                    contentItem: Rectangle {
                        color: receiveView.ScrollBar.pressed ? accentColor : (receiveView.ScrollBar.hovered ? "#555555" : "#333333")
                        radius: 5
                    }
                }

                delegate: ItemDelegate {
                    width: receiveView.width
                    height: contentLayout.implicitHeight + 8

                    background: Rectangle {
                        color: index % 2 === 0 ? "transparent" : '#3b3b21'
                    }

                    contentItem: RowLayout {
                        id: contentLayout
                        width: receiveView.width
                        spacing: 12

                        // 时间戳
                        Label {
                            height:contentLayout.height
                            visible: appState.showTimestamp
                            text: model.timestamp
                            color: "#666666"
                            font.pixelSize: 12
                            font.family: "Consolas, Monaco, monospace"
                            Layout.preferredWidth: 100
                            Layout.alignment: Qt.AlignVCenter
                            verticalAlignment: Text.AlignVCenter
                        }

                        // 数据内容
                        Label {
                            height:contentLayout.height
                            text: appState.displayMode ? model.hexText : model.asciiText
                            color: appState.displayMode ? accentColor : textColor
                            font.pixelSize: 12
                            font.family: "Consolas, Monaco, monospace"
                            Layout.fillWidth: true
                            Layout.alignment: Qt.AlignVCenter
                            wrapMode: Text.WrapAnywhere
                            verticalAlignment: Text.AlignVCenter
                        }
                    }
                }

                // 自动滚动
                onContentHeightChanged: {
                    if (appState.autoScroll) {
                        receiveView.contentY = receiveView.contentHeight - receiveView.height
                    }
                }
            }
        }

        // DTR/RTS 控制
        RowLayout {
            Layout.fillWidth: true
            Layout.preferredHeight: 45
            spacing: 20

            Label {
                text: "引脚控制:"
                color: textSecondary
                font.pixelSize: 13
            }

            CheckBox {
                text: "DTR"
                checked: serialManager.dtr
                enabled: serialManager.isOpen
                onCheckedChanged: serialManager.dtr = checked

                indicator: Rectangle {
                    width: 16
                    height: 16
                    x: parent.leftPadding
                    y: parent.topPadding + (parent.availableHeight - height) / 2
                    radius: 3
                    color: parent.checked ? accentColor : "transparent"
                    border.color: parent.checked ? accentColor : (parent.enabled ? "#ffffff50" : "#ffffff30")
                    border.width: 2

                    // 勾选标记
                    Rectangle {
                        visible: parent.parent.checked
                        width: 8
                        height: 2
                        anchors.centerIn: parent
                        color: "white"
                        rotation: 45
                    }
                }

                contentItem: Label {
                    text: parent.text
                    color: parent.enabled ? textColor : "#555555"
                    font.pixelSize: 13
                    leftPadding: parent.indicator.width + 8
                }
            }

            CheckBox {
                text: "RTS"
                checked: serialManager.rts
                enabled: serialManager.isOpen
                onCheckedChanged: serialManager.rts = checked

                indicator: Rectangle {
                    width: 16
                    height: 16
                    x: parent.leftPadding
                    y: parent.topPadding + (parent.availableHeight - height) / 2
                    radius: 3
                    color: parent.checked ? accentColor : "transparent"
                    border.color: parent.checked ? accentColor : (parent.enabled ? "#ffffff50" : "#ffffff30")
                    border.width: 2

                    // 勾选标记
                    Rectangle {
                        visible: parent.parent.checked
                        width: 8
                        height: 2
                        anchors.centerIn: parent
                        color: "white"
                        rotation: 45
                    }
                }

                contentItem: Label {
                    text: parent.text
                    color: parent.enabled ? textColor : "#555555"
                    font.pixelSize: 13
                    leftPadding: parent.indicator.width + 8
                }
            }

            Item { Layout.fillWidth: true }

            // 状态指示器
            Rectangle {
                Layout.preferredWidth: 120
                Layout.preferredHeight: 30
                color: serialManager.isOpen ? successColor : "#333333"
                radius: 15
                opacity: serialManager.isOpen ? 0.9 : 0.5

                RowLayout {
                    anchors.centerIn: parent
                    spacing: 8

                    Rectangle {
                        width: 8
                        height: 8
                        radius: 4
                        color: serialManager.isOpen ? "white" : "#666666"

                        SequentialAnimation on opacity {
                            running: serialManager.isOpen
                            loops: Animation.Infinite
                            NumberAnimation { to: 0.3; duration: 1000 }
                            NumberAnimation { to: 1.0; duration: 1000 }
                        }
                    }

                    Label {
                        text: serialManager.isOpen ? "已连接" : "未连接"
                        color: "white"
                        font.pixelSize: 12
                        font.weight: Font.Medium
                    }
                }
            }
        }

        // 发送区
        Rectangle {
            Layout.fillWidth: true
            Layout.preferredHeight: 130
            color: surfaceColor
            radius: 10
            border.color: "#ffffff10"
            border.width: 1

            ColumnLayout {
                anchors.fill: parent
                anchors.margins: 15
                spacing: 12

                RowLayout {
                    Layout.fillWidth: true
                    spacing: 10

                    StyledButton {
                        text: appState.sendHexMode ? "HEX" : "ASCII"
                        Layout.preferredWidth: 80
                        checkable: true
                        checked: appState.sendHexMode
                        isPrimary: checked

                        onClicked: appState.sendHexMode = !appState.sendHexMode
                    }

                    Label {
                        text: "换行符:"
                        color: textSecondary
                        font.pixelSize: 13
                    }

                    StyledComboBox {
                        id: lineEndingCombo
                        Layout.preferredWidth: 100
                        model: ["无", "\\r", "\\n", "\\r\\n"]
                        currentIndex: appState.lineEnding
                        onCurrentIndexChanged: appState.lineEnding = currentIndex
                    }

                    Item { Layout.fillWidth: true }

                    StyledButton {
                        text: "发送"
                        Layout.preferredWidth: 90
                        Layout.preferredHeight: 35
                        enabled: serialManager.isOpen
                        isPrimary: true

                        onClicked: sendData()
                    }
                }

                TextField {
                    id: sendInput
                    Layout.fillWidth: true
                    Layout.fillHeight: true
                    placeholderText: appState.sendHexMode ? "输入 HEX: AA BB 01 02" : "输入 ASCII 文本"
                    font.family: "Consolas, Monaco, monospace"
                    font.pixelSize: 13
                    color: textColor

                    background: Rectangle {
                        color: "#0a0a0a"
                        radius: 6
                        border.color: sendInput.activeFocus ? accentColor : "#ffffff20"
                        border.width: sendInput.activeFocus ? 2 : 1
                    }

                    Keys.onReturnPressed: sendData()
                    Keys.onEnterPressed: sendData()
                }
            }
        }
    }

    // 发送数据函数
    function sendData() {
        if (!serialManager.isOpen) return

        var text = sendInput.text

        if (appState.sendHexMode) {
            // HEX 模式：将换行符转换为 HEX 格式
            if (appState.lineEnding === 1) {
                text += " 0D"  // \r
            } else if (appState.lineEnding === 2) {
                text += " 0A"  // \n
            } else if (appState.lineEnding === 3) {
                text += " 0D 0A"  // \r\n
            }
            serialManager.sendHex(text)
        } else {
            // ASCII 模式：直接添加换行符
            if (appState.lineEnding === 1) {
                text += "\r"
            } else if (appState.lineEnding === 2) {
                text += "\n"
            } else if (appState.lineEnding === 3) {
                text += "\r\n"
            }
            serialManager.sendData(text)
        }
    }

    // 保存文件对话框
    FileDialog {
        id: saveFileDialog
        title: "保存日志文件"
        fileMode: FileDialog.SaveFile
        nameFilters: ["文本文件 (*.txt)", "HEX 文件 (*.hex)", "所有文件 (*)"]
        onAccepted: {
            saveLogToFile(selectedFile)
        }
    }

    // 错误对话框
    MessageDialog {
        id: errorDialog
        title: "错误"
        buttons: MessageDialog.Ok
    }

    // 保存日志到文件
    function saveLogToFile(filePath) {
        // 移除 file:// 前缀
        var path = filePath.toString()
        if (path.startsWith("file:///")) {
            path = path.substring(8)
        }

        var content = ""
        var count = receiveModel.count

        for (var i = 0; i < count; i++) {
            var index = receiveModel.index(i, 0)
            var timestamp = receiveModel.data(index, 256 + 1)  // TimestampRole
            var ascii = receiveModel.data(index, 256 + 2)      // AsciiRole
            var hex = receiveModel.data(index, 256 + 3)        // HexRole

            content += "[" + timestamp + "] "
            if (appState.displayMode) {
                content += hex
            } else {
                content += ascii
            }
            content += "\n"
        }

        // 使用 Python 保存文件
        fileSaveHelper.saveFile(path, content)
    }

    // 错误提示
    Connections {
        target: serialManager

        function onPortError(msg) {
            errorDialog.text = msg
            errorDialog.open()
        }
    }

    // 组件完成时
    Component.onCompleted: {
        // 初始化端口选择
        serialManager.portName = portCombo.currentText
    }
}

5、receive_model.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
接收数据模型 - 用于 QML ListView 显示
"""

from typing import List, Dict
from datetime import datetime

from PyQt6.QtCore import (
    Qt, QAbstractListModel, QModelIndex, pyqtSlot, pyqtProperty
)


class ReceiveModel(QAbstractListModel):
    """接收数据模型 - 环形缓冲区，上限 10000 条"""

    # 角色名称
    TimestampRole = Qt.ItemDataRole.UserRole + 1
    AsciiRole = Qt.ItemDataRole.UserRole + 2
    HexRole = Qt.ItemDataRole.UserRole + 3
    RawRole = Qt.ItemDataRole.UserRole + 4

    MAX_ENTRIES = 10000

    def __init__(self, parent=None):
        super().__init__(parent)
        self._entries: List[Dict] = []
        self._encoding = 'UTF-8'

    def roleNames(self):
        """定义 QML 可访问的角色"""
        return {
            self.TimestampRole: b"timestamp",
            self.AsciiRole: b"asciiText",
            self.HexRole: b"hexText",
            self.RawRole: b"rawData",
        }

    def rowCount(self, parent=QModelIndex()):
        """返回行数"""
        return len(self._entries)

    def data(self, index, role=Qt.ItemDataRole.DisplayRole):
        """返回指定角色数据"""
        if not index.isValid() or index.row() >= len(self._entries):
            return None

        entry = self._entries[index.row()]

        if role == self.TimestampRole:
            return entry["timestamp"]
        elif role == self.AsciiRole:
            return entry["ascii"]
        elif role == self.HexRole:
            return entry["hex"]
        elif role == self.RawRole:
            return entry["raw"]
        else:
            return None

    @pyqtSlot(bytes, float)
    def appendData(self, data: bytes, timestamp: float):
        """添加接收数据"""
        # 格式化时间戳
        dt = datetime.fromtimestamp(timestamp)
        time_str = dt.strftime("%H:%M:%S.%f")[:-3]  # 毫秒精度

        # 解码为 ASCII
        try:
            ascii_text = data.decode(self._encoding, errors='replace')
        except:
            ascii_text = data.decode('UTF-8', errors='replace')

        # 转换为 HEX
        hex_text = data.hex(' ').upper()

        # 创建条目
        entry = {
            "timestamp": time_str,
            "raw": data,
            "ascii": ascii_text,
            "hex": hex_text,
        }

        # 环形缓冲：超过上限时移除最旧的
        if len(self._entries) >= self.MAX_ENTRIES:
            self.beginRemoveRows(QModelIndex(), 0, 0)
            self._entries.pop(0)
            self.endRemoveRows()

        # 插入新行
        self.beginInsertRows(QModelIndex(), len(self._entries), len(self._entries))
        self._entries.append(entry)
        self.endInsertRows()

    @pyqtSlot()
    def clear(self):
        """清空所有数据"""
        self.beginResetModel()
        self._entries.clear()
        self.endResetModel()

    def setEncoding(self, encoding: str):
        """设置编码"""
        self._encoding = encoding

    @pyqtProperty(int)
    def count(self):
        """返回条目数"""
        return len(self._entries)

6、serial_manager.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
串口管理器 - 处理所有串口通信逻辑
"""

import sys
import time
import queue
from typing import Optional, List, Dict
from datetime import datetime

from PyQt6.QtCore import (
    QObject, QThread, pyqtSignal, pyqtProperty, pyqtSlot, QSettings
)

import serial
import serial.tools.list_ports


class SerialWorker(QThread):
    """串口工作线程 - 在独立线程中运行串口读写"""

    dataReceived = pyqtSignal(bytes, float)  # 数据, 时间戳
    errorOccurred = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)
        self._ser: Optional[serial.Serial] = None
        self._stop_event = False
        self._send_queue = queue.Queue()

    def set_serial(self, ser: serial.Serial):
        """设置串口对象"""
        self._ser = ser

    def request_send(self, data: bytes):
        """请求发送数据（线程安全）"""
        self._send_queue.put(data)

    def request_stop(self):
        """请求停止线程"""
        self._stop_event = True

    def run(self):
        """线程主循环"""
        while not self._stop_event:
            try:
                # 发送数据
                try:
                    while True:
                        data = self._send_queue.get_nowait()
                        if self._ser and self._ser.is_open:
                            self._ser.write(data)
                except queue.Empty:
                    pass

                # 接收数据
                if self._ser and self._ser.is_open:
                    if self._ser.in_waiting > 0:
                        data = self._ser.read(self._ser.in_waiting)
                        timestamp = time.time()
                        self.dataReceived.emit(data, timestamp)
                    else:
                        time.sleep(0.001)  # 1ms 轮询

            except serial.SerialException as e:
                self.errorOccurred.emit(str(e))
                break
            except Exception as e:
                self.errorOccurred.emit(f"未知错误: {e}")
                break


class SerialManager(QObject):
    """串口管理器 - 暴露给 QML 的后端对象"""

    # 信号
    portOpened = pyqtSignal()
    portClosed = pyqtSignal()
    portError = pyqtSignal(str)
    countersChanged = pyqtSignal()
    stateChanged = pyqtSignal()  # 用于通知 isOpen 状态变化

    def __init__(self, parent=None):
        super().__init__(parent)

        # 串口参数
        self._port_name = ""
        self._baud_rate = 115200
        self._data_bits = 8
        self._stop_bits = 1
        self._parity = 'N'
        self._flow_control = 'None'
        self._encoding = 'UTF-8'

        # 状态
        self._is_open = False
        self._dtr = False
        self._rts = False
        self._rx_count = 0
        self._tx_count = 0

        # 串口对象
        self._ser: Optional[serial.Serial] = None
        self._worker: Optional[SerialWorker] = None

    # === Q_PROPERTY 属性 ===

    @pyqtProperty(str)
    def portName(self):
        return self._port_name

    @portName.setter
    def portName(self, value):
        if self._port_name != value:
            self._port_name = value

    @pyqtProperty(int)
    def baudRate(self):
        return self._baud_rate

    @baudRate.setter
    def baudRate(self, value):
        if self._baud_rate != value:
            self._baud_rate = value

    @pyqtProperty(int)
    def dataBits(self):
        return self._data_bits

    @dataBits.setter
    def dataBits(self, value):
        if self._data_bits != value:
            self._data_bits = value

    @pyqtProperty(float)
    def stopBits(self):
        return self._stop_bits

    @stopBits.setter
    def stopBits(self, value):
        if self._stop_bits != value:
            self._stop_bits = value

    @pyqtProperty(str)
    def parity(self):
        return self._parity

    @parity.setter
    def parity(self, value):
        if self._parity != value:
            self._parity = value

    @pyqtProperty(str)
    def flowControl(self):
        return self._flow_control

    @flowControl.setter
    def flowControl(self, value):
        if self._flow_control != value:
            self._flow_control = value

    @pyqtProperty(str)
    def encoding(self):
        return self._encoding

    @encoding.setter
    def encoding(self, value):
        if self._encoding != value:
            self._encoding = value

    @pyqtProperty(bool, notify=stateChanged)
    def isOpen(self):
        return self._is_open

    @pyqtProperty(bool)
    def dtr(self):
        return self._dtr

    @dtr.setter
    def dtr(self, value):
        if self._dtr != value:
            self._dtr = value
            if self._ser and self._ser.is_open:
                self._ser.dtr = value

    @pyqtProperty(bool)
    def rts(self):
        return self._rts

    @rts.setter
    def rts(self, value):
        if self._rts != value:
            self._rts = value
            if self._ser and self._ser.is_open:
                self._ser.rts = value

    @pyqtProperty(int, notify=countersChanged)
    def rxCount(self):
        return self._rx_count

    @pyqtProperty(int, notify=countersChanged)
    def txCount(self):
        return self._tx_count

    # === Q_INVOKABLE 方法 ===

    @pyqtSlot(result=list)
    def refreshPorts(self) -> List[Dict[str, str]]:
        """刷新串口列表"""
        try:
            ports = serial.tools.list_ports.comports()
            return [
                {
                    "name": p.device,
                    "description": p.description,
                    "hwid": p.hwid
                }
                for p in ports
            ]
        except Exception as e:
            self.portError.emit(f"刷新串口失败: {e}")
            return []

    @pyqtSlot(result=bool)
    def openPort(self) -> bool:
        """打开串口"""
        if self._is_open:
            self.portError.emit("串口已打开")
            return False

        try:
            # Windows COM 端口号 > 9 时需要特殊格式
            port_name = self._port_name
            if port_name.startswith("COM") and int(port_name[3:]) > 9:
                port_name = f"\\\\.\\{port_name}"

            # 创建串口对象
            self._ser = serial.Serial(
                port=port_name,
                baudrate=self._baud_rate,
                bytesize=self._data_bits,
                stopbits=self._stop_bits,
                parity=self._parity,
                timeout=0.1
            )

            # 设置流控
            if self._flow_control == 'RTS/CTS':
                self._ser.rtscts = True
            elif self._flow_control == 'XON/XOFF':
                self._ser.xonxoff = True

            # 设置 DTR/RTS
            self._ser.dtr = self._dtr
            self._ser.rts = self._rts

            # 启动工作线程
            self._worker = SerialWorker()
            self._worker.set_serial(self._ser)
            self._worker.dataReceived.connect(self._onDataReceived)
            self._worker.errorOccurred.connect(self.portError.emit)
            self._worker.start()

            self._is_open = True
            self.portOpened.emit()
            self.stateChanged.emit()
            return True

        except Exception as e:
            self.portError.emit(f"打开串口失败: {e}")
            self._cleanup()
            return False

    @pyqtSlot()
    def closePort(self):
        """关闭串口"""
        if not self._is_open:
            return

        # 停止工作线程
        if self._worker:
            self._worker.request_stop()
            self._worker.wait(1000)
            self._worker = None

        self._cleanup()
        self._is_open = False
        self.portClosed.emit()
        self.stateChanged.emit()

    @pyqtSlot(str, result=bool)
    def sendData(self, text: str) -> bool:
        """发送 ASCII 文本"""
        if not self._is_open or not self._ser:
            self.portError.emit("串口未打开")
            return False

        try:
            data = text.encode(self._encoding)
            self._worker.request_send(data)
            self._tx_count += len(data)
            self.countersChanged.emit()
            return True
        except Exception as e:
            self.portError.emit(f"发送失败: {e}")
            return False

    @pyqtSlot(str, result=bool)
    def sendHex(self, hex_str: str) -> bool:
        """发送 HEX 数据"""
        if not self._is_open or not self._ser:
            self.portError.emit("串口未打开")
            return False

        try:
            # 清理输入
            cleaned = hex_str.replace(" ", "").replace("-", "").upper()
            if len(cleaned) % 2 != 0:
                self.portError.emit("HEX 长度必须为偶数")
                return False

            data = bytes.fromhex(cleaned)
            self._worker.request_send(data)
            self._tx_count += len(data)
            self.countersChanged.emit()
            return True
        except ValueError as e:
            self.portError.emit(f"无效的 HEX 字符: {e}")
            return False
        except Exception as e:
            self.portError.emit(f"发送失败: {e}")
            return False

    @pyqtSlot()
    def resetCounters(self):
        """重置计数器"""
        self._rx_count = 0
        self._tx_count = 0
        self.countersChanged.emit()

    def _onDataReceived(self, data: bytes, timestamp: float):
        """接收数据回调"""
        self._rx_count += len(data)
        self.countersChanged.emit()
        # 这里会通过信号传递给 QML 的接收模型
        # 在 main.py 中会连接这个信号

    def _cleanup(self):
        """清理资源"""
        if self._ser and self._ser.is_open:
            try:
                self._ser.close()
            except:
                pass
        self._ser = None

三、实现原理

本串口调试助手采用 PyQt6 + QML 的前后端分离架构，将界面渲染与业务逻辑解耦，既保留了 QML 声明式 UI 的高效开发体验，又借助 Python 生态轻松接入 pyserial 完成底层串口通信。整体数据流如下：

1、整体架构

  • 前端（QML）：负责界面展示与用户交互，包括串口参数配置、收发数据显示、HEX/ASCII 切换、日志保存等，全部通过声明式组件实现。
  • 后端（Python）：由 main.py 负责程序装配，serial_manager.py 封装串口通信逻辑，receive_model.py 作为数据模型向 QML 的 ListView 提供接收数据。
  • 桥接层：通过 QQmlApplicationEngine 的 rootContext().setContextProperty() 将后端对象注册为 QML 全局上下文属性，前端可直接调用其 @pyqtSlot 方法并绑定其 @pyqtProperty 属性。

2、串口通信线程模型

为避免串口读写阻塞 UI 线程，程序采用 QThread 工作线程 处理收发：

  • SerialWorker 继承 QThread，在 run() 中循环执行发送队列消费与接收轮询。
  • 发送时，UI 线程通过 request_send() 将数据放入线程安全的 queue.Queue，工作线程取出后调用 serial.write() 写入。
  • 接收时，工作线程以 1ms 间隔轮询 in_waiting，一旦有数据便读取并通过 dataReceived 信号携带 (bytes, timestamp) 发回主线程。
  • 信号跨线程自动排队，保证 UI 线程安全更新。

3、数据接收与模型刷新

接收数据到达主线程后，经 _onDataReceived 回调进入 ReceiveModel.appendData()：

  • 将原始 bytes 同时解码为 ASCII 文本与 HEX 字符串，并附带毫秒级时间戳。
  • 模型采用 环形缓冲区 设计，超过 10000 条时自动移除最旧记录，避免内存无限增长。
  • 通过 beginInsertRows / endInsertRows 通知 QML 的 ListView 增量刷新，保证滚动流畅。

4、配置持久化

程序使用 QSettings 实现串口参数的记忆功能：关闭时由 ConfigManager.save_config() 写入注册表/配置文件，启动时由 load_config() 自动恢复上次的端口、波特率、校验位等设置，提升日常使用体验。

5、关键流程时序

ReceiveModel SerialWorker(线程) SerialManager QML 界面 ReceiveModel SerialWorker(线程) SerialManager QML 界面 openPort() 启动线程 轮询串口数据 dataReceived(bytes, ts) appendData(bytes, ts) 数据行插入 ListView sendData(text) request_send(bytes) serial.write()

在这里插入图片描述

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/qq_15181569/article/details/164299417

文章来源转载

评论

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--