Python+PyQt6实现跨平台串口设备热插拔检测_Python

引言

在嵌入式开发、工业控制或物联网应用中，经常需要通过串口与外部设备通信。一个常见的需求是：当用户插入或拔出一个 usb 转串口适配器（如 ch340、cp2102 等）时，应用程序需要能够自动检测到设备的变化，并更新可用的串口列表，而无需重启程序。这种功能被称为“设备热插拔检测”。

本文将详细介绍如何在基于 pyqt6 的图形界面应用程序中，实现跨平台（windows、linux、macos）的串口设备热插拔检测。我们将结合 pyqt6 的信号槽机制和 python 的 pyserial 库，构建一个稳定、响应及时的解决方案。

核心思路与原理

实现热插拔检测的核心在于监听系统底层发出的设备变更事件，并在事件发生时，重新扫描可用的串口列表。不同操作系统的底层机制不同：

windows: 可以通过监听 windows 消息（如 wm_devicechange）或使用 windows api（如 registerdevicenotification）。在 pyqt6 中，我们可以利用其事件过滤器（eventfilter）来捕获这些系统级消息。
linux/macos: 通常通过监听 /dev 目录下文件（如 ttyusb*, ttyacm*）的变化来实现。我们可以使用 qfilesystemwatcher 来监控目录。

为了简化跨平台开发，我们将采用一种更通用的策略：使用一个定时器（qtimer）定期轮询当前可用的串口列表，并与之前保存的列表进行比较。如果列表发生变化，则发出信号通知界面更新。虽然轮询不如事件驱动高效，但对于串口设备变化频率不高的场景来说，实现简单且完全跨平台，是更稳妥的选择。

环境准备与依赖安装

首先，确保你的 python 环境已安装必要的库：

pip install pyqt6 pyserial

pyqt6: 用于构建图形用户界面。
pyserial: 用于列举和操作串口设备。

代码实现：热插拔检测器类

我们将创建一个名为 serialportmonitor 的类，它继承自 qobject，以便使用 pyqt6 的信号槽机制。

import sys
from pyqt6.qtcore import qobject, qtimer, pyqtsignal
from pyqt6.qtwidgets import qapplication, qmainwindow, qcombobox, qvboxlayout, qwidget
import serial.tools.list_ports

class serialportmonitor(qobject):
    """
    串口设备热插拔检测器。
    使用定时器轮询方式，跨平台兼容。
    """
    # 定义一个信号，当可用串口列表发生变化时发出，参数为新的端口名列表
    ports_changed = pyqtsignal(list)

    def __init__(self, poll_interval=2000):
        """
        初始化检测器。
        :param poll_interval: 轮询间隔，单位毫秒，默认2秒
        """
        super().__init__()
        self.poll_interval = poll_interval
        self._current_ports = []  # 保存当前检测到的端口列表
        self._timer = qtimer()
        self._timer.timeout.connect(self._check_ports)
        self._init_ports_list()

    def _init_ports_list(self):
        """初始化当前端口列表"""
        self._current_ports = [port.device for port in serial.tools.list_ports.comports()]

    def _check_ports(self):
        """定时器触发的检查函数"""
        # 获取当前所有可用串口
        available_ports = [port.device for port in serial.tools.list_ports.comports()]

        # 比较列表是否发生变化（考虑顺序无关）
        if set(available_ports) != set(self._current_ports):
            print(f"串口列表已变更。旧列表: {self._current_ports}, 新列表: {available_ports}")
            self._current_ports = available_ports
            # 发出信号，传递新的端口列表
            self.ports_changed.emit(available_ports)

    def start(self):
        """启动监控"""
        self._init_ports_list()  # 启动时先获取一次
        self._timer.start(self.poll_interval)
        print("串口热插拔监控已启动。")

    def stop(self):
        """停止监控"""
        self._timer.stop()
        print("串口热插拔监控已停止。")

    def get_current_ports(self):
        """获取当前检测到的端口列表"""
        return self._current_ports.copy()

关键点说明：

ports_changed 信号：这是连接检测逻辑和用户界面的桥梁。当端口列表变化时，此信号会被触发。
_check_ports 方法：核心轮询逻辑。使用 serial.tools.list_ports.comports() 获取当前所有串口设备，并与之前保存的列表进行对比。
start() / stop() 方法：控制监控的启停。

集成到 pyqt6 主界面

下面创建一个简单的主窗口，包含一个下拉框（qcombobox）来显示可用串口，并将监控器集成进去。

class mainwindow(qmainwindow):
    def __init__(self):
        super().__init__()
        self.setwindowtitle("pyqt6 串口热插拔检测示例")
        self.setgeometry(100, 100, 400, 200)

        # 创建中央部件和布局
        central_widget = qwidget()
        self.setcentralwidget(central_widget)
        layout = qvboxlayout(central_widget)

        # 创建下拉框用于显示串口
        self.combo_ports = qcombobox()
        layout.addwidget(self.combo_ports)

        # 初始化串口监控器
        self.monitor = serialportmonitor(poll_interval=1500)  # 每1.5秒检查一次
        # 连接监控器的信号到更新界面的槽函数
        self.monitor.ports_changed.connect(self.update_com_port_list)
        # 启动监控
        self.monitor.start()

        # 初始更新一次列表
        self.update_com_port_list(self.monitor.get_current_ports())

    def update_com_port_list(self, port_list):
        """
        更新下拉框中的串口列表。
        :param port_list: 新的可用串口设备名列表
        """
        current_selection = self.combo_ports.currenttext()
        self.combo_ports.clear()
        self.combo_ports.additems(port_list)

        # 尝试恢复之前的选择（如果该端口仍然存在）
        if current_selection in port_list:
            index = self.combo_ports.findtext(current_selection)
            if index >= 0:
                self.combo_ports.setcurrentindex(index)
        elif port_list:  # 如果列表不为空，默认选择第一个
            self.combo_ports.setcurrentindex(0)

        print(f"界面已更新，可用串口: {port_list}")

    def closeevent(self, event):
        """重写关闭事件，确保停止监控器"""
        self.monitor.stop()
        event.accept()

if __name__ == "__main__":
    app = qapplication(sys.argv)
    window = mainwindow()
    window.show()
    sys.exit(app.exec())

运行效果：运行上述程序后，窗口会显示当前可用的串口（如 com3, /dev/ttyusb0）。当你插入或拔出一个 usb 串口设备时，控制台会打印变更日志，并且下拉框中的列表会自动刷新。

进阶优化与平台特定方法

减少轮询开销

如果担心定时轮询对性能有影响，可以：

动态调整轮询间隔：设备稳定时延长间隔（如 5 秒），检测到一次变化后，在短时间内提高频率（如 500 毫秒）连续检查几次，以应对系统枚举设备较慢的情况。
使用线程：将端口扫描放在单独的线程中，避免阻塞主界面。

windows 事件驱动方法（可选）

对于 windows 平台，可以实现更高效的事件驱动监听。这需要重写主窗口的 nativeevent 方法或使用事件过滤器来捕获 wm_devicechange 消息。此方法更复杂，但响应更及时。

# 简要示例：在mainwindow中增加事件过滤（仅windows有效）
from pyqt6.qtcore import qevent
import ctypes
import win32con  # 需要 pip install pywin32

class mainwindow(qmainwindow):
    # ... 之前的初始化代码 ...

    def nativeevent(self, eventtype, message):
        """windows 原生事件处理"""
        try:
            msg = ctypes.wintypes.msg.from_address(message.__int__())
            if msg.message == win32con.wm_devicechange:
                # 收到设备变更消息，通知监控器立即检查（或直接在这里处理）
                print("收到设备变更系统消息")
                # 可以在这里直接调用 self.monitor._check_ports() 或发一个信号
                self.monitor._check_ports()
        except:
            pass
        return super().nativeevent(eventtype, message)

linux/macos 目录监控方法（可选）

在 linux/macos 上，可以使用 qfilesystemwatcher 监控 /dev 目录。

from pyqt6.qtcore import qfilesystemwatcher

class serialportmonitor(qobject):
    def __init__(self):
        super().__init__()
        self.watcher = qfilesystemwatcher()
        # 注意：可能需要监控 /dev 目录，但权限和事件粒度需要处理
        # self.watcher.addpath('/dev')
        # self.watcher.directorychanged.connect(self._check_ports)
        # 更简单的做法是监控一个特定的符号链接目录或结合轮询
        # 此处作为思路展示，不展开实现

总结

本文提供了一种在 pyqt6 应用中实现串口设备热插拔检测的可靠、跨平台的解决方案。核心是使用 qtimer 定期轮询并结合 pyserial 库来检测串口列表变化。虽然轮询方式在实时性上略有妥协，但其实现简单、稳定性高，非常适合大多数桌面应用场景。

完整代码

定时器扫描

import sys
from pyqt6.qtwidgets import qapplication, qmainwindow, qcombobox, qvboxlayout, qwidget, qlabel
from pyqt6.qtcore import qtimer, qt
from pyqt6.qtserialport import qserialportinfo

class serialhotplugwindow(qmainwindow):
    def __init__(self):
        super().__init__()
        self.setwindowtitle("串口热插拔检测 - 轮询方案")
        self.resize(400, 200)

        # ui
        central = qwidget()
        self.setcentralwidget(central)
        layout = qvboxlayout(central)
        self.cbx_port = qcombobox()
        self.label_tip = qlabel("插拔usb串口会自动刷新下拉框")
        layout.addwidget(self.label_tip)
        layout.addwidget(self.cbx_port)

        # 保存上一次串口列表
        self.last_port_names = set()

        # 定时刷新 500ms一次
        self.timer = qtimer(self)
        self.timer.setinterval(500)
        self.timer.timeout.connect(self.scan_serial_ports)
        self.timer.start()

    def scan_serial_ports(self):
        # 获取当前所有串口名称
        current_ports = {info.portname() for info in qserialportinfo.availableports()}
        # 新增
        new_ports = current_ports - self.last_port_names
        # 移除
        lost_ports = self.last_port_names - current_ports

        if new_ports:
            print(f"【插入串口】{new_ports}")
        if lost_ports:
            print(f"【拔出串口】{lost_ports}")


        # 更新下拉框
        self.cbx_port.clear()
        self.cbx_port.additems(sorted(current_ports))

        # 更新历史列表
        self.last_port_names = current_ports

if __name__ == "__main__":
    app = qapplication(sys.argv)
    win = serialhotplugwindow()
    win.show()
    sys.exit(app.exec())