import sys
import time
import socket
import threading
import asyncio
import os
import configparser
import json
import re
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QComboBox, QPushButton, QTextEdit, QTabWidget,
QGridLayout, QLineEdit, QGroupBox, QCheckBox, QSpinBox,
QFontDialog, QAction, QMenu, QMenuBar, QStatusBar, QMessageBox,
QFrame, QSplitter, QDialog, QDialogButtonBox, QFormLayout,
QToolButton, QSizePolicy, QColorDialog, QRadioButton, QButtonGroup)
from PyQt5.QtCore import (Qt, QTimer, pyqtSignal, QByteArray, QEvent, QObject, QSettings,
QSize, QPoint, QRect, QMimeData)
from PyQt5.QtGui import QFont, QDrag, QColor, QPalette, QPainter, QPen
from PyQt5.QtSerialPort import QSerialPort, QSerialPortInfo
import bleak
from bleak.backends.characteristic import BleakGATTCharacteristic
# 自定义事件类型，用于从线程安全地更新UI
class NetworkDataEvent(QEvent):
EVENT_TYPE = QEvent.Type(QEvent.registerEventType())
def __init__(self, data, is_status=False, rx_count=0):
super().__init__(self.EVENT_TYPE)
self.data = data
self.is_status = is_status
self.rx_count = rx_count
# 自定义信号对象，用于从异步蓝牙操作中更新UI
class BluetoothSignals(QObject):
device_discovered = pyqtSignal(object)
scan_finished = pyqtSignal()
connected = pyqtSignal()
disconnected = pyqtSignal()
error = pyqtSignal(str)
data_received = pyqtSignal(bytes)
# 异步事件循环处理类
class AsyncioEventLoopThread(threading.Thread):
def __init__(self):
super().__init__(daemon=True)
self.loop = None
self._running = True
self._ready = threading.Event()
def run(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self._ready.set()
try:
self.loop.run_forever()
finally:
# 关闭循环前取消所有挂起的任务
pending = asyncio.all_tasks(self.loop)
for task in pending:
task.cancel()
# 等待任务取消
if pending:
self.loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
self.loop.close()
def stop(self):
if self.loop and self._running:
self._running = False
self.loop.call_soon_threadsafe(self.loop.stop)
def wait_for_ready(self):
self._ready.wait()
def run_coroutine(self, coro):
if not self.loop or not self._running:
raise RuntimeError("Event loop is not running")
future = asyncio.run_coroutine_threadsafe(coro, self.loop)
return future
# 自定义按钮编辑对话框
class ButtonEditDialog(QDialog):
def __init__(self, button_data=None, parent=None):
super().__init__(parent)
self.setWindowTitle("编辑按钮")
self.resize(400, 350)
# 初始化按钮数据
if button_data is None:
self.button_data = {
"name": "按钮",
"press_text": "",
"release_text": "",
"long_press_enabled": False,
"long_press_interval": 1000,
"color": "#f0f0f0",
"text_color": "#000000",
"width": 80,
"height": 40
}
else:
self.button_data = button_data
# 创建表单布局
layout = QFormLayout(self)
# 按钮名称
self.name_edit = QLineEdit(self.button_data["name"])
layout.addRow("按钮名称:", self.name_edit)
# 按下时发送的文本
self.press_text_edit = QTextEdit(self.button_data["press_text"])
self.press_text_edit.setMaximumHeight(60)
layout.addRow("按下发送:", self.press_text_edit)
# 添加特殊字符说明
layout.addRow("", QLabel("支持特殊字符: \\r (回车), \\n (换行), \\t (制表符)"))
# 松开时发送的文本
self.release_text_edit = QTextEdit(self.button_data["release_text"])
self.release_text_edit.setMaximumHeight(60)
layout.addRow("松开发送:", self.release_text_edit)
# 长按循环发送
self.long_press_check = QCheckBox("启用长按循环发送")
self.long_press_check.setChecked(self.button_data["long_press_enabled"])
layout.addRow("", self.long_press_check)
# 长按发送间隔
self.interval_spin = QSpinBox()
self.interval_spin.setRange(10, 10000)
self.interval_spin.setValue(self.button_data["long_press_interval"])
self.interval_spin.setSuffix(" ms")
layout.addRow("发送间隔:", self.interval_spin)
# 按钮大小
size_layout = QHBoxLayout()
self.width_spin = QSpinBox()
self.width_spin.setRange(20, 500)
self.width_spin.setValue(self.button_data["width"])
self.width_spin.setSuffix(" px")
self.height_spin = QSpinBox()
self.height_spin.setRange(20, 500)
self.height_spin.setValue(self.button_data["height"])
self.height_spin.setSuffix(" px")
size_layout.addWidget(QLabel("宽:"))
size_layout.addWidget(self.width_spin)
size_layout.addWidget(QLabel("高:"))
size_layout.addWidget(self.height_spin)
layout.addRow("按钮大小:", size_layout)
# 按钮颜色
color_layout = QHBoxLayout()
self.color_button = QPushButton()
self.color_button.setStyleSheet(f"background-color: {self.button_data['color']};")
self.color_button.setFixedSize(30, 20)
self.color_button.clicked.connect(self.choose_button_color)
self.text_color_button = QPushButton()
self.text_color_button.setStyleSheet(f"background-color: {self.button_data['text_color']};")
self.text_color_button.setFixedSize(30, 20)
self.text_color_button.clicked.connect(self.choose_text_color)
color_layout.addWidget(QLabel("背景:"))
color_layout.addWidget(self.color_button)
color_layout.addWidget(QLabel("文字:"))
color_layout.addWidget(self.text_color_button)
layout.addRow("按钮颜色:", color_layout)
# 对话框按钮
self.button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
self.button_box.accepted.connect(self.accept)
self.button_box.rejected.connect(self.reject)
layout.addRow(self.button_box)
def choose_button_color(self):
color = QColorDialog.getColor(QColor(self.button_data["color"]), self, "选择按钮颜色")
if color.isValid():
self.button_data["color"] = color.name()
self.color_button.setStyleSheet(f"background-color: {color.name()};")
def choose_text_color(self):
color = QColorDialog.getColor(QColor(self.button_data["text_color"]), self, "选择文字颜色")
if color.isValid():
self.button_data["text_color"] = color.name()
self.text_color_button.setStyleSheet(f"background-color: {color.name()};")
def get_button_data(self):
self.button_data["name"] = self.name_edit.text()
self.button_data["press_text"] = self.press_text_edit.toPlainText()
self.button_data["release_text"] = self.release_text_edit.toPlainText()
self.button_data["long_press_enabled"] = self.long_press_check.isChecked()
self.button_data["long_press_interval"] = self.interval_spin.value()
self.button_data["width"] = self.width_spin.value()
self.button_data["height"] = self.height_spin.value()
return self.button_data
# 自定义按钮类
class CustomButton(QToolButton):
def __init__(self, parent=None, button_data=None, edit_mode=False):
super().__init__(parent)
self.parent = parent
self.edit_mode = edit_mode
self.long_press_timer = QTimer(self)
self.long_press_timer.timeout.connect(self.on_long_press_timeout)
if button_data is None:
self.button_data = {
"name": "按钮",
"press_text": "",
"release_text": "",
"long_press_enabled": False,
"long_press_interval": 1000,
"color": "#f0f0f0",
"text_color": "#000000",
"width": 80,
"height": 40,
"position": [0, 0]
}
else:
self.button_data = button_data
self.update_appearance()
# 设置拖放功能
self.setMouseTracking(True)
def update_appearance(self):
self.setText(self.button_data["name"])
self.setFixedSize(self.button_data["width"], self.button_data["height"])
if "position" in self.button_data:
self.move(self.button_data["position"][0], self.button_data["position"][1])
# 设置样式
self.setStyleSheet(f"""
QToolButton {{
background-color: {self.button_data["color"]};
color: {self.button_data["text_color"]};
border: 1px solid #888888;
border-radius: 4px;
padding: 4px;
}}
QToolButton:pressed {{
background-color: {self.darken_color(self.button_data["color"])};
}}
""")
def darken_color(self, color_str):
# 简单的颜色变暗函数
color = QColor(color_str)
h, s, v, a = color.getHsv()
color.setHsv(h, s, max(0, v - 20), a)
return color.name()
def mousePressEvent(self, event):
if self.edit_mode:
if event.button() == Qt.LeftButton:
self.drag_start_position = event.pos()
elif event.button() == Qt.RightButton:
self.show_context_menu(event.pos())
else:
# 正常模式下，发送按下命令
if event.button() == Qt.LeftButton:
if self.button_data["press_text"]:
# 处理特殊字符
text = self.process_special_chars(self.button_data["press_text"])
self.parent.send_button_command(text)
# 如果启用了长按功能，启动定时器
if self.button_data["long_press_enabled"]:
self.long_press_timer.start(self.button_data["long_press_interval"])
super().mousePressEvent(event)
def mouseReleaseEvent(self, event):
if not self.edit_mode and event.button() == Qt.LeftButton:
# 停止长按定时器
self.long_press_timer.stop()
# 发送松开命令
if self.button_data["release_text"]:
# 处理特殊字符
text = self.process_special_chars(self.button_data["release_text"])
self.parent.send_button_command(text)
super().mouseReleaseEvent(event)
def process_special_chars(self, text):
# 处理特殊字符: \r, \n, \t
return text.replace('\\r', '\r').replace('\\n', '\n').replace('\\t', '\t')
def mouseMoveEvent(self, event):
if self.edit_mode and event.buttons() & Qt.LeftButton:
if (event.pos() - self.drag_start_position).manhattanLength() >= QApplication.startDragDistance():
self.start_drag()
super().mouseMoveEvent(event)
def start_drag(self):
drag = QDrag(self)
mime_data = QMimeData()
mime_data.setText("custom_button")
drag.setMimeData(mime_data)
# 执行拖动
drop_action = drag.exec_(Qt.MoveAction)
# 更新位置
if drop_action == Qt.MoveAction:
# 获取网格对齐后的位置
grid_pos = self.parent.get_grid_position(self.pos())
self.move(grid_pos)
self.button_data["position"] = [grid_pos.x(), grid_pos.y()]
def show_context_menu(self, pos):
menu = QMenu(self)
edit_action = menu.addAction("编辑按钮")
delete_action = menu.addAction("删除按钮")
action = menu.exec_(self.mapToGlobal(pos))
if action == edit_action:
self.edit_button()
elif action == delete_action:
self.delete_button()
def edit_button(self):
dialog = ButtonEditDialog(self.button_data, self)
if dialog.exec_() == QDialog.Accepted:
self.button_data = dialog.get_button_data()
self.update_appearance()
self.parent.save_buttons()
def delete_button(self):
reply = QMessageBox.question(self, "确认删除",
f"确定要删除按钮 '{self.button_data['name']}' 吗?",
QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.Yes:
self.parent.remove_button(self)
def on_long_press_timeout(self):
# 长按定时器触发，发送按下命令
if self.button_data["press_text"]:
# 处理特殊字符
text = self.process_special_chars(self.button_data["press_text"])
self.parent.send_button_command(text)
# 按钮面板类
class ButtonPanel(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.parent = parent
self.setAcceptDrops(True)
self.buttons = []
self.edit_mode = False
# 网格设置
self.grid_size = 20 # 网格大小，可以根据需要调整
# 设置布局
self.layout = QVBoxLayout(self)
# 编辑模式控制
self.edit_controls = QHBoxLayout()
self.edit_mode_check = QCheckBox("编辑模式")
self.edit_mode_check.stateChanged.connect(self.toggle_edit_mode)
self.edit_controls.addWidget(self.edit_mode_check)
self.add_button = QPushButton("添加按钮")
self.add_button.clicked.connect(self.add_new_button)
self.add_button.setEnabled(False) # 初始禁用
self.edit_controls.addWidget(self.add_button)
# 网格设置
self.grid_controls = QHBoxLayout()
self.grid_controls.addWidget(QLabel("网格大小:"))
self.grid_size_spin = QSpinBox()
self.grid_size_spin.setRange(5, 50)
self.grid_size_spin.setValue(self.grid_size)
self.grid_size_spin.setSuffix(" px")
self.grid_size_spin.valueChanged.connect(self.update_grid_size)
self.grid_controls.addWidget(self.grid_size_spin)
self.show_grid_check = QCheckBox("显示网格")
self.show_grid_check.setChecked(False)
self.show_grid_check.stateChanged.connect(self.update)
self.grid_controls.addWidget(self.show_grid_check)
self.edit_controls.addLayout(self.grid_controls)
self.edit_controls.addStretch()
self.layout.addLayout(self.edit_controls)
# 按钮区域
self.button_area = QWidget()
self.button_area.setMinimumHeight(200)
self.button_area.setStyleSheet("background-color: #f5f5f5; border: 1px solid #cccccc;")
self.layout.addWidget(self.button_area)
def paintEvent(self, event):
super().paintEvent(event)
# 如果启用了网格显示并且处于编辑模式
if self.show_grid_check.isChecked() and self.edit_mode:
painter = QPainter(self)
painter.setPen(QPen(QColor("#dddddd"), 1, Qt.DotLine))
# 绘制垂直线
for x in range(0, self.button_area.width(), self.grid_size):
painter.drawLine(x, 0, x, self.button_area.height())
# 绘制水平线
for y in range(0, self.button_area.height(), self.grid_size):
painter.drawLine(0, y, self.button_area.width(), y)
def update_grid_size(self, size):
self.grid_size = size
self.update() # 重绘界面以更新网格
def get_grid_position(self, pos):
"""将位置对齐到网格"""
x = round(pos.x() / self.grid_size) * self.grid_size
y = round(pos.y() / self.grid_size) * self.grid_size
return QPoint(x, y)
def toggle_edit_mode(self, state):
self.edit_mode = (state == Qt.Checked)
self.add_button.setEnabled(self.edit_mode)
self.grid_size_spin.setEnabled(self.edit_mode)
self.show_grid_check.setEnabled(self.edit_mode)
# 更新所有按钮的编辑模式
for button in self.buttons:
button.edit_mode = self.edit_mode
# 更新按钮区域的样式
if self.edit_mode:
self.button_area.setStyleSheet("background-color: #e0e0e0; border: 2px dashed #888888;")
else:
self.button_area.setStyleSheet("background-color: #f5f5f5; border: 1px solid #cccccc;")
self.update() # 重绘以显示/隐藏网格
def add_new_button(self):
dialog = ButtonEditDialog(parent=self)
if dialog.exec_() == QDialog.Accepted:
button_data = dialog.get_button_data()
# 设置初始位置
button_data["position"] = [10, 10]
self.add_button_from_data(button_data)
self.parent.save_buttons()
def add_button_from_data(self, button_data):
button = CustomButton(self.parent, button_data, self.edit_mode)
button.setParent(self.button_area)
button.show()
self.buttons.append(button)
return button
def remove_button(self, button):
if button in self.buttons:
self.buttons.remove(button)
button.deleteLater()
self.parent.save_buttons()
def dragEnterEvent(self, event):
if self.edit_mode and event.mimeData().hasText():
event.acceptProposedAction()
def dropEvent(self, event):
if self.edit_mode and event.mimeData().hasText():
# 获取拖放位置
pos = event.pos()
# 调整位置到按钮区域的相对位置
button_area_pos = self.button_area.mapFrom(self, pos)
# 找到被拖动的按钮
source_button = event.source()
if source_button in self.buttons:
# 获取网格对齐后的位置
grid_pos = self.get_grid_position(button_area_pos)
# 更新按钮位置
source_button.move(grid_pos)
source_button.button_data["position"] = [grid_pos.x(), grid_pos.y()]
self.parent.save_buttons()
event.acceptProposedAction()
def get_buttons_data(self):
return [button.button_data for button in self.buttons]
def load_buttons(self, buttons_data):
# 清除现有按钮
for button in self.buttons:
button.deleteLater()
self.buttons = []
# 加载新按钮
for button_data in buttons_data:
self.add_button_from_data(button_data)
class SerialDebugger(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("多功能串口调试工具")
# 初始化计数器
self.rx_bytes = 0
self.tx_bytes = 0
self.serial_rx_bytes = 0
self.serial_tx_bytes = 0
self.bt_rx_bytes = 0
self.bt_tx_bytes = 0
self.net_rx_bytes = 0
self.net_tx_bytes = 0
# 初始化设置
self.settings = QSettings("config.ini", QSettings.IniFormat)
self.load_settings()
# 设置窗口大小
if self.settings.contains("window/size"):
size = self.settings.value("window/size")
if isinstance(size, QSize):
self.resize(size)
else:
self.resize(800, 600)
else:
self.resize(800, 600)
# 创建主窗口部件和布局
self.central_widget = QWidget()
self.setCentralWidget(self.central_widget)
self.main_layout = QVBoxLayout(self.central_widget)
# 创建菜单栏
self.create_menu_bar()
# 创建状态栏
self.statusBar = QStatusBar()
self.setStatusBar(self.statusBar)
# 创建字节计数标签
self.rx_label = QLabel("接收: 0 字节")
self.tx_label = QLabel("发送: 0 字节")
self.statusBar.addPermanentWidget(self.rx_label)
self.statusBar.addPermanentWidget(self.tx_label)
self.statusBar.showMessage("就绪")
# 创建选项卡
self.tabs = QTabWidget()
self.main_layout.addWidget(self.tabs)
# 创建三种串口类型的选项卡
self.serial_tab = QWidget()
self.bluetooth_tab = QWidget()
self.network_tab = QWidget()
self.buttons_tab = QWidget() # 新增按钮选项卡
self.tabs.addTab(self.serial_tab, "普通串口")
self.tabs.addTab(self.bluetooth_tab, "蓝牙串口")
self.tabs.addTab(self.network_tab, "网络串口")
self.tabs.addTab(self.buttons_tab, "自定义按钮")
# 设置当前选项卡
if self.settings.contains("tabs/current"):
current_tab = int(self.settings.value("tabs/current", 0))
if 0 <= current_tab < 4: # 现在有4个选项卡
self.tabs.setCurrentIndex(current_tab)
# 连接选项卡切换信号
self.tabs.currentChanged.connect(self.on_tab_changed)
# 初始化各个选项卡
self.init_serial_tab()
self.init_bluetooth_tab()
self.init_network_tab()
self.init_buttons_tab() # 初始化按钮选项卡
# 初始化串口对象
self.serial_port = QSerialPort()
self.bluetooth_client = None
self.network_socket = None
# 蓝牙信号
self.bt_signals = BluetoothSignals()
self.bt_signals.device_discovered.connect(self.add_bluetooth_device)
self.bt_signals.scan_finished.connect(self.on_bluetooth_scan_finished)
self.bt_signals.connected.connect(self.on_bluetooth_connected)
self.bt_signals.disconnected.connect(self.on_bluetooth_disconnected)
self.bt_signals.error.connect(self.on_bluetooth_error)
self.bt_signals.data_received.connect(self.on_bluetooth_data_received)
# 连接信号和槽
self.serial_port.readyRead.connect(self.read_serial_data)
# 定时刷新串口列表
self.refresh_timer = QTimer(self)
self.refresh_timer.timeout.connect(self.refresh_serial_ports)
self.refresh_timer.start(1000) # 每秒刷新一次
# 定时发送定时器
self.serial_send_timer = QTimer(self)
self.serial_send_timer.timeout.connect(self.send_serial_data_timed)
self.bt_send_timer = QTimer(self)
self.bt_send_timer.timeout.connect(self.send_bluetooth_data_timed)
self.net_send_timer = QTimer(self)
self.net_send_timer.timeout.connect(self.send_network_data_timed)
# 蓝牙设备列表
self.bluetooth_devices = []
# 创建并启动异步事件循环线程
self.async_thread = AsyncioEventLoopThread()
self.async_thread.start()
self.async_thread.wait_for_ready()
# 蓝牙特性
self.write_characteristic = None
# 加载自定义按钮
self.load_buttons()
# 应用字体设置
self.apply_font_settings()
def create_menu_bar(self):
menubar = self.menuBar()
# 文件菜单
file_menu = menubar.addMenu('文件')
save_settings_action = QAction('保存设置', self)
save_settings_action.triggered.connect(self.save_settings)
file_menu.addAction(save_settings_action)
clear_counters_action = QAction('清除计数器', self)
clear_counters_action.triggered.connect(self.clear_counters)
file_menu.addAction(clear_counters_action)
exit_action = QAction('退出', self)
exit_action.triggered.connect(self.close)
file_menu.addAction(exit_action)
# 设置菜单
settings_menu = menubar.addMenu('设置')
font_action = QAction('字体设置', self)
font_action.triggered.connect(self.change_font)
settings_menu.addAction(font_action)
# 帮助菜单
help_menu = menubar.addMenu('帮助')
about_action = QAction('关于', self)
about_action.triggered.connect(self.show_about)
help_menu.addAction(about_action)
def show_about(self):
QMessageBox.about(self, "关于", "多功能串口调试工具\n\n支持普通串口、蓝牙串口和网络串口\n\n版本: 1.1")
def change_font(self):
font, ok = QFontDialog.getFont(self.font(), self, "选择字体")
if ok:
self.settings.setValue("font/family", font.family())
self.settings.setValue("font/size", font.pointSize())
self.settings.setValue("font/bold", font.bold())
self.settings.setValue("font/italic", font.italic())
self.apply_font_settings()
def apply_font_settings(self):
family = self.settings.value("font/family", "Arial")
size = int(self.settings.value("font/size", 10))
bold = self.settings.value("font/bold", False) == "true"
italic = self.settings.value("font/italic", False) == "true"
font = QFont(family, size)
font.setBold(bold)
font.setItalic(italic)
# 应用字体到整个应用
QApplication.setFont(font)
# 应用字体到所有文本编辑框
if hasattr(self, 'receive_text'):
self.receive_text.setFont(font)
if hasattr(self, 'send_text'):
self.send_text.setFont(font)
if hasattr(self, 'bt_receive_text'):
self.bt_receive_text.setFont(font)
if hasattr(self, 'bt_send_text'):
self.bt_send_text.setFont(font)
if hasattr(self, 'net_receive_text'):
self.net_receive_text.setFont(font)
if hasattr(self, 'net_send_text'):
self.net_send_text.setFont(font)
if hasattr(self, 'command_log'):
self.command_log.setFont(font)
def load_settings(self):
# 加载字体设置
if not self.settings.contains("font/family"):
# 默认字体设置
self.settings.setValue("font/family", "Arial")
self.settings.setValue("font/size", 10)
self.settings.setValue("font/bold", False)
self.settings.setValue("font/italic", False)
# 加载串口设置
self.last_serial_port = self.settings.value("serial/port", "")
self.last_baud_rate = self.settings.value("serial/baud_rate", "9600")
self.last_data_bits = self.settings.value("serial/data_bits", "8")
self.last_stop_bits = self.settings.value("serial/stop_bits", "1")
self.last_parity = self.settings.value("serial/parity", "无")
self.last_flow_control = self.settings.value("serial/flow_control", "无")
self.last_serial_hex_display = self.settings.value("serial/hex_display", False) == "true"
self.last_serial_hex_send = self.settings.value("serial/hex_send", False) == "true"
self.last_serial_send_newline = self.settings.value("serial/send_newline", False) == "true"
self.last_serial_auto_send = self.settings.value("serial/auto_send", False) == "true"
self.last_serial_send_interval = int(self.settings.value("serial/send_interval", 1000))
# 加载蓝牙设置
self.last_bt_hex_display = self.settings.value("bluetooth/hex_display", False) == "true"
self.last_bt_hex_send = self.settings.value("bluetooth/hex_send", False) == "true"
self.last_bt_send_newline = self.settings.value("bluetooth/send_newline", False) == "true"
self.last_bt_auto_send = self.settings.value("bluetooth/auto_send", False) == "true"
self.last_bt_send_interval = int(self.settings.value("bluetooth/send_interval", 1000))
# 加载网络设置
self.last_protocol = self.settings.value("network/protocol", "TCP客户端")
self.last_ip = self.settings.value("network/ip", "127.0.0.1")
self.last_port = self.settings.value("network/port", 8080)
self.last_net_hex_display = self.settings.value("network/hex_display", False) == "true"
self.last_net_hex_send = self.settings.value("network/hex_send", False) == "true"
self.last_net_send_newline = self.settings.value("network/send_newline", False) == "true"
self.last_net_auto_send = self.settings.value("network/auto_send", False) == "true"
self.last_net_send_interval = int(self.settings.value("network/send_interval", 1000))
# 加载按钮设置
self.last_grid_size = int(self.settings.value("buttons/grid_size", 20))
self.last_show_grid = self.settings.value("buttons/show_grid", False) == "true"
def save_settings(self):
# 保存窗口设置
self.settings.setValue("window/size", self.size())
self.settings.setValue("tabs/current", self.tabs.currentIndex())
# 保存串口设置
if hasattr(self, 'port_combo'):
self.settings.setValue("serial/port", self.port_combo.currentText())
if hasattr(self, 'baud_combo'):
self.settings.setValue("serial/baud_rate", self.baud_combo.currentText())
if hasattr(self, 'data_bits_combo'):
self.settings.setValue("serial/data_bits", self.data_bits_combo.currentText())
if hasattr(self, 'stop_bits_combo'):
self.settings.setValue("serial/stop_bits", self.stop_bits_combo.currentText())
if hasattr(self, 'parity_combo'):
self.settings.setValue("serial/parity", self.parity_combo.currentText())
if hasattr(self, 'flow_control_combo'):
self.settings.setValue("serial/flow_control", self.flow_control_combo.currentText())
if hasattr(self, 'hex_display_check'):
self.settings.setValue("serial/hex_display", self.hex_display_check.isChecked())
if hasattr(self, 'hex_send_check'):
self.settings.setValue("serial/hex_send", self.hex_send_check.isChecked())
if hasattr(self, 'send_newline_check'):
self.settings.setValue("serial/send_newline", self.send_newline_check.isChecked())
if hasattr(self, 'auto_send_check'):
self.settings.setValue("serial/auto_send", self.auto_send_check.isChecked())
if hasattr(self, 'send_interval_spin'):
self.settings.setValue("serial/send_interval", self.send_interval_spin.value())
# 保存蓝牙设置
if hasattr(self, 'bt_hex_display_check'):
self.settings.setValue("bluetooth/hex_display", self.bt_hex_display_check.isChecked())
if hasattr(self, 'bt_hex_send_check'):
self.settings.setValue("bluetooth/hex_send", self.bt_hex_send_check.isChecked())
if hasattr(self, 'bt_send_newline_check'):
self.settings.setValue("bluetooth/send_newline", self.bt_send_newline_check.isChecked())
if hasattr(self, 'bt_auto_send_check'):
self.settings.setValue("bluetooth/auto_send", self.bt_auto_send_check.isChecked())
if hasattr(self, 'bt_send_interval_spin'):
self.settings.setValue("bluetooth/send_interval", self.bt_send_interval_spin.value())
# 保存网络设置
if hasattr(self, 'protocol_combo'):
self.settings.setValue("network/protocol", self.protocol_combo.currentText())
if hasattr(self, 'ip_edit'):
self.settings.setValue("network/ip", self.ip_edit.text())
if hasattr(self, 'port_spin'):
self.settings.setValue("network/port", self.port_spin.value())
if hasattr(self, 'net_hex_display_check'):
self.settings.setValue("network/hex_display", self.net_hex_display_check.isChecked())
if hasattr(self, 'net_hex_send_check'):
self.settings.setValue("network/hex_send", self.net_hex_send_check.isChecked())
if hasattr(self, 'net_send_newline_check'):
self.settings.setValue("network/send_newline", self.net_send_newline_check.isChecked())
if hasattr(self, 'net_auto_send_check'):
self.settings.setValue("network/auto_send", self.net_auto_send_check.isChecked())
if hasattr(self, 'net_send_interval_spin'):
self.settings.setValue("network/send_interval", self.net_send_interval_spin.value())
# 保存按钮设置
if hasattr(self, 'button_panel'):
self.settings.setValue("buttons/grid_size", self.button_panel.grid_size)
self.settings.setValue("buttons/show_grid", self.button_panel.show_grid_check.isChecked())
self.save_buttons()
self.settings.sync()
self.statusBar.showMessage("设置已保存", 3000)
def save_buttons(self):
if hasattr(self, 'button_panel'):
buttons_data = self.button_panel.get_buttons_data()
self.settings.setValue("buttons/data", json.dumps(buttons_data))
self.settings.sync()
def load_buttons(self):
if self.settings.contains("buttons/data"):
try:
buttons_data = json.loads(self.settings.value("buttons/data"))
if hasattr(self, 'button_panel'):
self.button_panel.load_buttons(buttons_data)
except Exception as e:
print(f"加载按钮数据失败: {str(e)}")
def on_tab_changed(self, index):
# 更新状态栏显示的字节计数
if index == 0: # 普通串口
self.rx_bytes = self.serial_rx_bytes
self.tx_bytes = self.serial_tx_bytes
elif index == 1: # 蓝牙串口
self.rx_bytes = self.bt_rx_bytes
self.tx_bytes = self.bt_tx_bytes
elif index == 2: # 网络串口
self.rx_bytes = self.net_rx_bytes
self.tx_bytes = self.net_tx_bytes
elif index == 3: # 按钮选项卡
# 显示当前活动连接的字节计数
if self.tabs.currentIndex() == 0:
self.rx_bytes = self.serial_rx_bytes
self.tx_bytes = self.serial_tx_bytes
elif self.tabs.currentIndex() == 1:
self.rx_bytes = self.bt_rx_bytes
self.tx_bytes = self.bt_tx_bytes
elif self.tabs.currentIndex() == 2:
self.rx_bytes = self.net_rx_bytes
self.tx_bytes = self.net_tx_bytes
self.update_byte_counters()
def update_byte_counters(self):
self.rx_label.setText(f"接收: {self.rx_bytes} 字节")
self.tx_label.setText(f"发送: {self.tx_bytes} 字节")
def clear_counters(self):
self.rx_bytes = 0
self.tx_bytes = 0
self.serial_rx_bytes = 0
self.serial_tx_bytes = 0
self.bt_rx_bytes = 0
self.bt_tx_bytes = 0
self.net_rx_bytes = 0
self.net_tx_bytes = 0
self.update_byte_counters()
self.statusBar.showMessage("计数器已清零", 3000)
def init_buttons_tab(self):
layout = QVBoxLayout(self.buttons_tab)
# 创建按钮面板
self.button_panel = ButtonPanel(self)
# 设置网格大小和显示
if hasattr(self, 'last_grid_size'):
self.button_panel.grid_size = self.last_grid_size
self.button_panel.grid_size_spin.setValue(self.last_grid_size)
if hasattr(self, 'last_show_grid'):
self.button_panel.show_grid_check.setChecked(self.last_show_grid)
layout.addWidget(self.button_panel)
# 创建命令日志区域
log_group = QGroupBox("命令日志")
log_layout = QVBoxLayout()
# 日志头部控件
log_header = QHBoxLayout()
self.clear_log_button = QPushButton("清空日志")
self.clear_log_button.clicked.connect(self.clear_command_log)
log_header.addWidget(self.clear_log_button)
log_layout.addLayout(log_header)
# 日志文本框
self.command_log = QTextEdit()
self.command_log.setReadOnly(True)
log_layout.addWidget(self.command_log)
log_group.setLayout(log_layout)
layout.addWidget(log_group)
def clear_command_log(self):
self.command_log.clear()
def send_button_command(self, command):
# 记录到命令日志
self.command_log.append(f"发送: {command}")
# 根据当前选项卡发送命令
current_tab = self.tabs.currentIndex()
if current_tab == 0 or (current_tab == 3 and self.serial_port.isOpen()): # 普通串口或按钮选项卡+串口已打开
if self.serial_port.isOpen():
try:
data = command.encode()
self.serial_port.write(data)
self.serial_tx_bytes += len(data)
self.tx_bytes = self.serial_tx_bytes
self.update_byte_counters()
self.statusBar.showMessage(f"已发送 {len(data)} 字节", 3000)
except Exception as e:
self.command_log.append(f"错误: {str(e)}")
else:
self.command_log.append("错误: 串口未打开")
elif current_tab == 1 or (current_tab == 3 and self.bluetooth_client and hasattr(self.bluetooth_client, 'is_connected') and self.bluetooth_client.is_connected): # 蓝牙串口或按钮选项卡+蓝牙已连接
if self.bluetooth_client and hasattr(self.bluetooth_client, 'is_connected') and self.bluetooth_client.is_connected:
if self.write_characteristic:
try:
data = command.encode()
self.bt_tx_bytes += len(data)
self.tx_bytes = self.bt_tx_bytes
self.update_byte_counters()
self.async_thread.run_coroutine(self.bluetooth_send(data))
except Exception as e:
self.command_log.append(f"错误: {str(e)}")
else:
self.command_log.append("错误: 未找到可写的蓝牙特性")
else:
self.command_log.append("错误: 蓝牙未连接")
elif current_tab == 2 or (current_tab == 3 and self.network_socket): # 网络串口或按钮选项卡+网络已连接
if self.network_socket:
try:
data = command.encode()
protocol = self.protocol_combo.currentText()
data_len = len(data)
self.net_tx_bytes += data_len
self.tx_bytes = self.net_tx_bytes
self.update_byte_counters()
if protocol == "TCP客户端":
self.network_socket.sendall(data)
elif protocol == "TCP服务器":
if hasattr(self, 'client_socket') and self.client_socket:
self.client_socket.sendall(data)
else:
self.command_log.append("错误: 没有客户端连接")
return
else: # UDP
if hasattr(self, 'udp_client_addr') and self.udp_client_addr:
self.network_socket.sendto(data, self.udp_client_addr)
else:
# 如果没有接收过数据，则发送到指定地址
self.network_socket.sendto(data, (self.ip_edit.text(), self.port_spin.value()))
except Exception as e:
self.command_log.append(f"错误: {str(e)}")
else:
self.command_log.append("错误: 网络未连接")
else:
self.command_log.append("错误: 没有活动的连接")
def init_serial_tab(self):
layout = QVBoxLayout(self.serial_tab)
# 串口设置区域
settings_group = QGroupBox("串口设置")
settings_layout = QGridLayout()
# 串口选择
settings_layout.addWidget(QLabel("串口:"), 0, 0)
self.port_combo = QComboBox()
self.refresh_serial_ports()
settings_layout.addWidget(self.port_combo, 0, 1)
# 波特率
settings_layout.addWidget(QLabel("波特率:"), 1, 0)
self.baud_combo = QComboBox()
for baud in ["1200", "2400", "4800", "9600", "19200", "38400", "57600", "115200"]:
self.baud_combo.addItem(baud)
self.baud_combo.setCurrentText(self.last_baud_rate)
settings_layout.addWidget(self.baud_combo, 1, 1)
# 数据位
settings_layout.addWidget(QLabel("数据位:"), 2, 0)
self.data_bits_combo = QComboBox()
for bits in ["5", "6", "7", "8"]:
self.data_bits_combo.addItem(bits)
self.data_bits_combo.setCurrentText(self.last_data_bits)
settings_layout.addWidget(self.data_bits_combo, 2, 1)
# 停止位
settings_layout.addWidget(QLabel("停止位:"), 0, 2)
self.stop_bits_combo = QComboBox()
self.stop_bits_combo.addItems(["1", "1.5", "2"])
self.stop_bits_combo.setCurrentText(self.last_stop_bits)
settings_layout.addWidget(self.stop_bits_combo, 0, 3)
# 校验位
settings_layout.addWidget(QLabel("校验位:"), 1, 2)
self.parity_combo = QComboBox()
self.parity_combo.addItems(["无", "奇校验", "偶校验", "空格", "标记"])
self.parity_combo.setCurrentText(self.last_parity)
settings_layout.addWidget(self.parity_combo, 1, 3)
# 流控制
settings_layout.addWidget(QLabel("流控制:"), 2, 2)
self.flow_control_combo = QComboBox()
self.flow_control_combo.addItems(["无", "硬件", "软件"])
self.flow_control_combo.setCurrentText(self.last_flow_control)
settings_layout.addWidget(self.flow_control_combo, 2, 3)
# 打开/关闭串口按钮
self.serial_open_button = QPushButton("打开串口")
self.serial_open_button.clicked.connect(self.toggle_serial_port)
settings_layout.addWidget(self.serial_open_button, 3, 0, 1, 4)
settings_group.setLayout(settings_layout)
layout.addWidget(settings_group)
# 数据收发区域
data_group = QGroupBox("数据收发")
data_layout = QVBoxLayout()
# 接收区域
receive_layout = QVBoxLayout()
receive_header = QHBoxLayout()
receive_header.addWidget(QLabel("接收区:"))
self.hex_display_check = QCheckBox("HEX显示")
self.hex_display_check.setChecked(self.last_serial_hex_display)
receive_header.addWidget(self.hex_display_check)
self.auto_scroll_check = QCheckBox("自动滚动")
self.auto_scroll_check.setChecked(True)
receive_header.addWidget(self.auto_scroll_check)
self.clear_receive_button = QPushButton("清空接收")
self.clear_receive_button.clicked.connect(lambda: self.receive_text.clear())
receive_header.addWidget(self.clear_receive_button)
receive_layout.addLayout(receive_header)
self.receive_text = QTextEdit()
self.receive_text.setReadOnly(True)
receive_layout.addWidget(self.receive_text)
# 发送区域
send_layout = QVBoxLayout()
send_header = QHBoxLayout()
send_header.addWidget(QLabel("发送区:"))
self.hex_send_check = QCheckBox("HEX发送")
self.hex_send_check.setChecked(self.last_serial_hex_send)
send_header.addWidget(self.hex_send_check)
self.send_newline_check = QCheckBox("发送新行")
self.send_newline_check.setChecked(self.last_serial_send_newline)
send_header.addWidget(self.send_newline_check)
# 自动发送
self.auto_send_check = QCheckBox("自动发送")
self.auto_send_check.setChecked(self.last_serial_auto_send)
self.auto_send_check.stateChanged.connect(self.toggle_serial_auto_send)
send_header.addWidget(self.auto_send_check)
# 发送间隔
send_header.addWidget(QLabel("间隔(ms):"))
self.send_interval_spin = QSpinBox()
self.send_interval_spin.setRange(10, 60000) # 10ms到60s
self.send_interval_spin.setValue(self.last_serial_send_interval)
self.send_interval_spin.valueChanged.connect(self.update_serial_send_interval)
send_header.addWidget(self.send_interval_spin)
self.send_button = QPushButton("发送")
self.send_button.clicked.connect(self.send_serial_data)
send_header.addWidget(self.send_button)
send_layout.addLayout(send_header)
self.send_text = QTextEdit()
self.send_text.setMaximumHeight(100)
send_layout.addWidget(self.send_text)
data_layout.addLayout(receive_layout, 7)
data_layout.addLayout(send_layout, 3)
data_group.setLayout(data_layout)
layout.addWidget(data_group)
def init_bluetooth_tab(self):
layout = QVBoxLayout(self.bluetooth_tab)
# 蓝牙设置区域
settings_group = QGroupBox("蓝牙设置")
settings_layout = QGridLayout()
# 蓝牙设备选择
settings_layout.addWidget(QLabel("设备:"), 0, 0)
self.bt_device_combo = QComboBox()
settings_layout.addWidget(self.bt_device_combo, 0, 1)
# 扫描按钮
self.bt_scan_button = QPushButton("扫描设备")
self.bt_scan_button.clicked.connect(self.scan_bluetooth_devices)
settings_layout.addWidget(self.bt_scan_button, 0, 2)
# 连接按钮
self.bt_connect_button = QPushButton("连接")
self.bt_connect_button.clicked.connect(self.toggle_bluetooth_connection)
settings_layout.addWidget(self.bt_connect_button, 1, 0, 1, 3)
settings_group.setLayout(settings_layout)
layout.addWidget(settings_group)
# 数据收发区域
data_group = QGroupBox("数据收发")
data_layout = QVBoxLayout()
# 接收区域
receive_layout = QVBoxLayout()
receive_header = QHBoxLayout()
receive_header.addWidget(QLabel("接收区:"))
self.bt_hex_display_check = QCheckBox("HEX显示")
self.bt_hex_display_check.setChecked(self.last_bt_hex_display)
receive_header.addWidget(self.bt_hex_display_check)
self.bt_auto_scroll_check = QCheckBox("自动滚动")
self.bt_auto_scroll_check.setChecked(True)
receive_header.addWidget(self.bt_auto_scroll_check)
self.bt_clear_receive_button = QPushButton("清空接收")
self.bt_clear_receive_button.clicked.connect(lambda: self.bt_receive_text.clear())
receive_header.addWidget(self.bt_clear_receive_button)
receive_layout.addLayout(receive_header)
self.bt_receive_text = QTextEdit()
self.bt_receive_text.setReadOnly(True)
receive_layout.addWidget(self.bt_receive_text)
# 发送区域
send_layout = QVBoxLayout()
send_header = QHBoxLayout()
send_header.addWidget(QLabel("发送区:"))
self.bt_hex_send_check = QCheckBox("HEX发送")
self.bt_hex_send_check.setChecked(self.last_bt_hex_send)
send_header.addWidget(self.bt_hex_send_check)
self.bt_send_newline_check = QCheckBox("发送新行")
self.bt_send_newline_check.setChecked(self.last_bt_send_newline)
send_header.addWidget(self.bt_send_newline_check)
# 自动发送
self.bt_auto_send_check = QCheckBox("自动发送")
self.bt_auto_send_check.setChecked(self.last_bt_auto_send)
self.bt_auto_send_check.stateChanged.connect(self.toggle_bt_auto_send)
send_header.addWidget(self.bt_auto_send_check)
# 发送间隔
send_header.addWidget(QLabel("间隔(ms):"))
self.bt_send_interval_spin = QSpinBox()
self.bt_send_interval_spin.setRange(10, 60000) # 10ms到60s
self.bt_send_interval_spin.setValue(self.last_bt_send_interval)
self.bt_send_interval_spin.valueChanged.connect(self.update_bt_send_interval)
send_header.addWidget(self.bt_send_interval_spin)
self.bt_send_button = QPushButton("发送")
self.bt_send_button.clicked.connect(self.send_bluetooth_data)
send_header.addWidget(self.bt_send_button)
send_layout.addLayout(send_header)
self.bt_send_text = QTextEdit()
self.bt_send_text.setMaximumHeight(100)
send_layout.addWidget(self.bt_send_text)
data_layout.addLayout(receive_layout, 7)
data_layout.addLayout(send_layout, 3)
data_group.setLayout(data_layout)
layout.addWidget(data_group)
def init_network_tab(self):
layout = QVBoxLayout(self.network_tab)
# 网络设置区域
settings_group = QGroupBox("网络设置")
settings_layout = QGridLayout()
# 协议选择
settings_layout.addWidget(QLabel("协议:"), 0, 0)
self.protocol_combo = QComboBox()
self.protocol_combo.addItems(["TCP客户端", "TCP服务器", "UDP"])
self.protocol_combo.setCurrentText(self.last_protocol)
self.protocol_combo.currentIndexChanged.connect(self.update_network_ui)
settings_layout.addWidget(self.protocol_combo, 0, 1)
# IP地址
settings_layout.addWidget(QLabel("IP地址:"), 1, 0)
self.ip_edit = QLineEdit(self.last_ip)
settings_layout.addWidget(self.ip_edit, 1, 1)
# 端口
settings_layout.addWidget(QLabel("端口:"), 2, 0)
self.port_spin = QSpinBox()
self.port_spin.setRange(1, 65535)
self.port_spin.setValue(int(self.last_port))
settings_layout.addWidget(self.port_spin, 2, 1)
# 连接按钮
self.net_connect_button = QPushButton("连接")
self.net_connect_button.clicked.connect(self.toggle_network_connection)
settings_layout.addWidget(self.net_connect_button, 3, 0, 1, 2)
settings_group.setLayout(settings_layout)
layout.addWidget(settings_group)
# 数据收发区域
data_group = QGroupBox("数据收发")
data_layout = QVBoxLayout()
# 接收区域
receive_layout = QVBoxLayout()
receive_header = QHBoxLayout()
receive_header.addWidget(QLabel("接收区:"))
self.net_hex_display_check = QCheckBox("HEX显示")
self.net_hex_display_check.setChecked(self.last_net_hex_display)
receive_header.addWidget(self.net_hex_display_check)
self.net_auto_scroll_check = QCheckBox("自动滚动")
self.net_auto_scroll_check.setChecked(True)
receive_header.addWidget(self.net_auto_scroll_check)
self.net_clear_receive_button = QPushButton("清空接收")
self.net_clear_receive_button.clicked.connect(lambda: self.net_receive_text.clear())
receive_header.addWidget(self.net_clear_receive_button)
receive_layout.addLayout(receive_header)
self.net_receive_text = QTextEdit()
self.net_receive_text.setReadOnly(True)
receive_layout.addWidget(self.net_receive_text)
# 发送区域
send_layout = QVBoxLayout()
send_header = QHBoxLayout()
send_header.addWidget(QLabel("发送区:"))
self.net_hex_send_check = QCheckBox("HEX发送")
self.net_hex_send_check.setChecked(self.last_net_hex_send)
send_header.addWidget(self.net_hex_send_check)
self.net_send_newline_check = QCheckBox("发送新行")
self.net_send_newline_check.setChecked(self.last_net_send_newline)
send_header.addWidget(self.net_send_newline_check)
# 自动发送
self.net_auto_send_check = QCheckBox("自动发送")
self.net_auto_send_check.setChecked(self.last_net_auto_send)
self.net_auto_send_check.stateChanged.connect(self.toggle_net_auto_send)
send_header.addWidget(self.net_auto_send_check)
# 发送间隔
send_header.addWidget(QLabel("间隔(ms):"))
self.net_send_interval_spin = QSpinBox()
self.net_send_interval_spin.setRange(10, 60000) # 10ms到60s
self.net_send_interval_spin.setValue(self.last_net_send_interval)
self.net_send_interval_spin.valueChanged.connect(self.update_net_send_interval)
send_header.addWidget(self.net_send_interval_spin)
self.net_send_button = QPushButton("发送")
self.net_send_button.clicked.connect(self.send_network_data)
send_header.addWidget(self.net_send_button)
send_layout.addLayout(send_header)
self.net_send_text = QTextEdit()
self.net_send_text.setMaximumHeight(100)
send_layout.addWidget(self.net_send_text)
data_layout.addLayout(receive_layout, 7)
data_layout.addLayout(send_layout, 3)
data_group.setLayout(data_layout)
layout.addWidget(data_group)
# 初始化网络UI
self.update_network_ui()
def toggle_serial_auto_send(self, state):
if state == Qt.Checked:
interval = self.send_interval_spin.value()
self.serial_send_timer.start(interval)
self.statusBar.showMessage(f"已启动自动发送，间隔 {interval} 毫秒", 3000)
else:
self.serial_send_timer.stop()
self.statusBar.showMessage("已停止自动发送", 3000)
def update_serial_send_interval(self, value):
if self.auto_send_check.isChecked():
self.serial_send_timer.start(value)
self.statusBar.showMessage(f"自动发送间隔已更新为 {value} 毫秒", 3000)
def toggle_bt_auto_send(self, state):
if state == Qt.Checked:
interval = self.bt_send_interval_spin.value()
self.bt_send_timer.start(interval)
self.statusBar.showMessage(f"已启动自动发送，间隔 {interval} 毫秒", 3000)
else:
self.bt_send_timer.stop()
self.statusBar.showMessage("已停止自动发送", 3000)
def update_bt_send_interval(self, value):
if self.bt_auto_send_check.isChecked():
self.bt_send_timer.start(value)
self.statusBar.showMessage(f"自动发送间隔已更新为 {value} 毫秒", 3000)
def toggle_net_auto_send(self, state):
if state == Qt.Checked:
interval = self.net_send_interval_spin.value()
self.net_send_timer.start(interval)
self.statusBar.showMessage(f"已启动自动发送，间隔 {interval} 毫秒", 3000)
else:
self.net_send_timer.stop()
self.statusBar.showMessage("已停止自动发送", 3000)
def update_net_send_interval(self, value):
if self.net_auto_send_check.isChecked():
self.net_send_timer.start(value)
self.statusBar.showMessage(f"自动发送间隔已更新为 {value} 毫秒", 3000)
def update_network_ui(self):
protocol = self.protocol_combo.currentText()
if protocol == "TCP服务器":
self.ip_edit.setEnabled(False)
self.ip_edit.setText("0.0.0.0")
else:
self.ip_edit.setEnabled(True)
if self.ip_edit.text() == "0.0.0.0":
self.ip_edit.setText(self.last_ip)
def refresh_serial_ports(self):
current_port = self.port_combo.currentText()
self.port_combo.clear()
for port in QSerialPortInfo.availablePorts():
self.port_combo.addItem(port.portName())
# 如果有上次使用的端口，尝试选择它
if self.last_serial_port:
index = self.port_combo.findText(self.last_serial_port)
if index >= 0:
self.port_combo.setCurrentIndex(index)
# 否则尝试恢复之前选择的端口
elif current_port:
index = self.port_combo.findText(current_port)
if index >= 0:
self.port_combo.setCurrentIndex(index)
def toggle_serial_port(self):
if self.serial_port.isOpen():
self.serial_port.close()
self.serial_open_button.setText("打开串口")
self.statusBar.showMessage("串口已关闭", 3000)
# 停止自动发送
if self.auto_send_check.isChecked():
self.auto_send_check.setChecked(False)
else:
# 配置串口
self.serial_port.setPortName(self.port_combo.currentText())
self.serial_port.setBaudRate(int(self.baud_combo.currentText()))
# 设置数据位
data_bits = int(self.data_bits_combo.currentText())
if data_bits == 5:
self.serial_port.setDataBits(QSerialPort.Data5)
elif data_bits == 6:
self.serial_port.setDataBits(QSerialPort.Data6)
elif data_bits == 7:
self.serial_port.setDataBits(QSerialPort.Data7)
else:
self.serial_port.setDataBits(QSerialPort.Data8)
# 设置停止位
stop_bits = self.stop_bits_combo.currentText()
if stop_bits == "1":
self.serial_port.setStopBits(QSerialPort.OneStop)
elif stop_bits == "1.5":
self.serial_port.setStopBits(QSerialPort.OneAndHalfStop)
else:
self.serial_port.setStopBits(QSerialPort.TwoStop)
# 设置校验位
parity = self.parity_combo.currentText()
if parity == "无":
self.serial_port.setParity(QSerialPort.NoParity)
elif parity == "奇校验":
self.serial_port.setParity(QSerialPort.OddParity)
elif parity == "偶校验":
self.serial_port.setParity(QSerialPort.EvenParity)
elif parity == "空格":
self.serial_port.setParity(QSerialPort.SpaceParity)
else:
self.serial_port.setParity(QSerialPort.MarkParity)
# 设置流控制
flow_control = self.flow_control_combo.currentText()
if flow_control == "无":
self.serial_port.setFlowControl(QSerialPort.NoFlowControl)
elif flow_control == "硬件":
self.serial_port.setFlowControl(QSerialPort.HardwareControl)
else:
self.serial_port.setFlowControl(QSerialPort.SoftwareControl)
# 打开串口
if self.serial_port.open(QSerialPort.ReadWrite):
self.serial_open_button.setText("关闭串口")
self.statusBar.showMessage(f"串口 {self.port_combo.currentText()} 已打开", 3000)
# 保存当前串口设置
self.last_serial_port = self.port_combo.currentText()
self.last_baud_rate = self.baud_combo.currentText()
self.last_data_bits = self.data_bits_combo.currentText()
self.last_stop_bits = self.stop_bits_combo.currentText()
self.last_parity = self.parity_combo.currentText()
self.last_flow_control = self.flow_control_combo.currentText()
# 如果设置了自动发送，启动定时器
if self.auto_send_check.isChecked():
self.serial_send_timer.start(self.send_interval_spin.value())
else:
error_msg = f"错误: 无法打开串口 {self.port_combo.currentText()}"
self.receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
def read_serial_data(self):
if not self.serial_port.isOpen():
return
data = self.serial_port.readAll()
data_len = len(data)
self.serial_rx_bytes += data_len
self.rx_bytes = self.serial_rx_bytes
self.update_byte_counters()
# 记录到命令日志（如果在按钮选项卡中）
if self.tabs.currentIndex() == 3: # 按钮选项卡
try:
text = data.data().decode('utf-8')
self.command_log.append(f"接收: {text}")
except UnicodeDecodeError:
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.command_log.append(f"接收: [HEX] {hex_str}")
if self.hex_display_check.isChecked():
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.receive_text.append(hex_str)
else:
try:
text = data.data().decode('utf-8')
self.receive_text.append(text)
except UnicodeDecodeError:
# 如果解码失败，显示十六进制
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.receive_text.append(f"[解码失败] {hex_str}")
if self.auto_scroll_check.isChecked():
self.receive_text.verticalScrollBar().setValue(
self.receive_text.verticalScrollBar().maximum())
def send_serial_data(self):
if not self.serial_port.isOpen():
self.receive_text.append("错误: 串口未打开")
self.statusBar.showMessage("错误: 串口未打开", 3000)
return
text = self.send_text.toPlainText()
if not text:
return
if self.hex_send_check.isChecked():
# 处理十六进制输入
try:
# 移除所有空格和换行符
clean_text = ''.join(text.split())
# 确保字符数是偶数
if len(clean_text) % 2 != 0:
error_msg = "错误: 十六进制字符数必须为偶数"
self.receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
# 转换为字节
byte_array = QByteArray.fromHex(clean_text.encode())
self.serial_port.write(byte_array)
self.serial_tx_bytes += len(byte_array)
self.tx_bytes = self.serial_tx_bytes
self.update_byte_counters()
self.statusBar.showMessage(f"已发送 {len(byte_array)} 字节", 3000)
except Exception as e:
error_msg = f"错误: {str(e)}"
self.receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
else:
# 处理特殊字符
processed_text = text.replace('\\r', '\r').replace('\\n', '\n').replace('\\t', '\t')
# 检查是否发送新行
if self.send_newline_check.isChecked():
processed_text += "\r\n"
data = processed_text.encode()
self.serial_port.write(data)
self.serial_tx_bytes += len(data)
self.tx_bytes = self.serial_tx_bytes
self.update_byte_counters()
self.statusBar.showMessage(f"已发送 {len(data)} 字节", 3000)
def send_serial_data_timed(self):
# 定时器触发的自动发送
if self.serial_port.isOpen():
self.send_serial_data()
def scan_bluetooth_devices(self):
self.bt_device_combo.clear()
self.bluetooth_devices = []
self.bt_scan_button.setEnabled(False)
self.bt_scan_button.setText("扫描中...")
self.statusBar.showMessage("正在扫描蓝牙设备...", 0)
# 使用持久的事件循环
self.async_thread.run_coroutine(self.bluetooth_scan())
async def bluetooth_scan(self):
try:
devices = await bleak.BleakScanner.discover()
for device in devices:
self.bluetooth_devices.append(device)
self.bt_signals.device_discovered.emit(device)
self.bt_signals.scan_finished.emit()
except Exception as e:
self.bt_signals.error.emit(f"扫描错误: {str(e)}")
self.bt_signals.scan_finished.emit()
def add_bluetooth_device(self, device):
device_name = device.name or "未知设备"
device_address = device.address
self.bt_device_combo.addItem(f"{device_name} ({device_address})", device)
def on_bluetooth_scan_finished(self):
self.bt_scan_button.setEnabled(True)
self.bt_scan_button.setText("扫描设备")
if self.bt_device_combo.count() == 0:
self.bt_receive_text.append("未找到蓝牙设备")
self.statusBar.showMessage("未找到蓝牙设备", 3000)
else:
self.statusBar.showMessage(f"找到 {self.bt_device_combo.count()} 个蓝牙设备", 3000)
def toggle_bluetooth_connection(self):
if self.bluetooth_client and hasattr(self.bluetooth_client, 'is_connected') and self.bluetooth_client.is_connected:
# 断开连接
self.bt_connect_button.setText("断开中...")
self.bt_connect_button.setEnabled(False)
self.statusBar.showMessage("正在断开蓝牙连接...", 0)
# 停止自动发送
if self.bt_auto_send_check.isChecked():
self.bt_auto_send_check.setChecked(False)
self.async_thread.run_coroutine(self.bluetooth_disconnect())
else:
# 连接
if self.bt_device_combo.count() == 0:
error_msg = "错误: 没有可用的蓝牙设备"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
selected_index = self.bt_device_combo.currentIndex()
if selected_index < 0 or selected_index >= len(self.bluetooth_devices):
error_msg = "错误: 无效的设备选择"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
selected_device = self.bluetooth_devices[selected_index]
self.bt_connect_button.setText("连接中...")
self.bt_connect_button.setEnabled(False)
self.bt_device_combo.setEnabled(False)
self.bt_scan_button.setEnabled(False)
self.statusBar.showMessage(f"正在连接到蓝牙设备 {selected_device.name or selected_device.address}...", 0)
# 使用持久的事件循环
self.async_thread.run_coroutine(self.bluetooth_connect(selected_device))
async def bluetooth_connect(self, device):
try:
self.bluetooth_client = bleak.BleakClient(device)
await self.bluetooth_client.connect()
# 查找可写特性
self.write_characteristic = None
for service in self.bluetooth_client.services:
for char in service.characteristics:
if "write" in char.properties:
self.write_characteristic = char
if "notify" in char.properties:
await self.bluetooth_client.start_notify(char.uuid, self.bluetooth_notification_handler)
self.bt_signals.connected.emit()
except Exception as e:
self.bt_signals.error.emit(f"连接错误: {str(e)}")
self.bluetooth_client = None
def bluetooth_notification_handler(self, sender: BleakGATTCharacteristic, data: bytearray):
self.bt_signals.data_received.emit(bytes(data)) # 转换为 bytes 类型
async def bluetooth_disconnect(self):
try:
await self.bluetooth_client.disconnect()
self.bt_signals.disconnected.emit()
except Exception as e:
self.bt_signals.error.emit(f"断开连接错误: {str(e)}")
finally:
self.bluetooth_client = None
self.write_characteristic = None
def on_bluetooth_connected(self):
self.bt_connect_button.setText("断开")
self.bt_connect_button.setEnabled(True)
self.bt_receive_text.append("已连接到蓝牙设备")
self.statusBar.showMessage("蓝牙设备已连接", 3000)
# 如果设置了自动发送，启动定时器
if self.bt_auto_send_check.isChecked():
self.bt_send_timer.start(self.bt_send_interval_spin.value())
def on_bluetooth_disconnected(self):
self.bt_connect_button.setText("连接")
self.bt_connect_button.setEnabled(True)
self.bt_device_combo.setEnabled(True)
self.bt_scan_button.setEnabled(True)
self.bt_receive_text.append("蓝牙设备已断开连接")
self.statusBar.showMessage("蓝牙设备已断开连接", 3000)
def on_bluetooth_error(self, error_msg):
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
self.bt_connect_button.setText("连接")
self.bt_connect_button.setEnabled(True)
self.bt_device_combo.setEnabled(True)
self.bt_scan_button.setEnabled(True)
def on_bluetooth_data_received(self, data):
data_len = len(data)
self.bt_rx_bytes += data_len
self.rx_bytes = self.bt_rx_bytes
self.update_byte_counters()
# 记录到命令日志（如果在按钮选项卡中）
if self.tabs.currentIndex() == 3: # 按钮选项卡
try:
text = data.decode('utf-8')
self.command_log.append(f"接收: {text}")
except UnicodeDecodeError:
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.command_log.append(f"接收: [HEX] {hex_str}")
if self.bt_hex_display_check.isChecked():
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.bt_receive_text.append(hex_str)
else:
try:
text = data.decode('utf-8')
self.bt_receive_text.append(text)
except UnicodeDecodeError:
# 如果解码失败，显示十六进制
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.bt_receive_text.append(f"[解码失败] {hex_str}")
if self.bt_auto_scroll_check.isChecked():
self.bt_receive_text.verticalScrollBar().setValue(
self.bt_receive_text.verticalScrollBar().maximum())
def send_bluetooth_data(self):
if not self.bluetooth_client or not hasattr(self.bluetooth_client, 'is_connected') or not self.bluetooth_client.is_connected:
error_msg = "错误: 蓝牙未连接"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
if not self.write_characteristic:
error_msg = "错误: 未找到可写的蓝牙特性"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
text = self.bt_send_text.toPlainText()
if not text:
return
# 使用持久的事件循环
if self.bt_hex_send_check.isChecked():
# 处理十六进制输入
try:
# 移除所有空格和换行符
clean_text = ''.join(text.split())
# 确保字符数是偶数
if len(clean_text) % 2 != 0:
error_msg = "错误: 十六进制字符数必须为偶数"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
# 转换为字节
data = bytes.fromhex(clean_text)
self.bt_tx_bytes += len(data)
self.tx_bytes = self.bt_tx_bytes
self.update_byte_counters()
self.async_thread.run_coroutine(self.bluetooth_send(data))
self.statusBar.showMessage(f"正在发送 {len(data)} 字节到蓝牙设备", 3000)
except Exception as e:
error_msg = f"错误: {str(e)}"
self.bt_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
else:
# 处理特殊字符
processed_text = text.replace('\\r', '\r').replace('\\n', '\n').replace('\\t', '\t')
# 检查是否发送新行
if self.bt_send_newline_check.isChecked():
processed_text += "\r\n"
data = processed_text.encode()
self.bt_tx_bytes += len(data)
self.tx_bytes = self.bt_tx_bytes
self.update_byte_counters()
self.async_thread.run_coroutine(self.bluetooth_send(data))
self.statusBar.showMessage(f"正在发送 {len(data)} 字节到蓝牙设备", 3000)
def send_bluetooth_data_timed(self):
# 定时器触发的自动发送
if self.bluetooth_client and hasattr(self.bluetooth_client, 'is_connected') and self.bluetooth_client.is_connected:
self.send_bluetooth_data()
async def bluetooth_send(self, data):
try:
if self.write_characteristic:
await self.bluetooth_client.write_gatt_char(self.write_characteristic.uuid, data)
else:
self.bt_signals.error.emit("错误: 未找到可写的蓝牙特性")
except Exception as e:
self.bt_signals.error.emit(f"发送错误: {str(e)}")
def toggle_network_connection(self):
protocol = self.protocol_combo.currentText()
if self.network_socket:
# 断开连接
if protocol == "TCP客户端" or protocol == "TCP服务器":
self.network_socket.close()
else: # UDP
self.network_socket.close()
self.network_socket = None
self.net_connect_button.setText("连接")
self.protocol_combo.setEnabled(True)
self.ip_edit.setEnabled(True)
self.port_spin.setEnabled(True)
self.net_receive_text.append("网络连接已断开")
self.statusBar.showMessage("网络连接已断开", 3000)
# 停止自动发送
if self.net_auto_send_check.isChecked():
self.net_auto_send_check.setChecked(False)
else:
# 建立连接
if protocol == "TCP客户端":
self.network_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
self.network_socket.connect((self.ip_edit.text(), self.port_spin.value()))
self.net_connect_button.setText("断开")
self.protocol_combo.setEnabled(False)
self.ip_edit.setEnabled(False)
self.port_spin.setEnabled(False)
self.net_receive_text.append(f"已连接到 {self.ip_edit.text()}:{self.port_spin.value()}")
self.statusBar.showMessage(f"已连接到 {self.ip_edit.text()}:{self.port_spin.value()}", 3000)
# 保存网络设置
self.last_protocol = protocol
self.last_ip = self.ip_edit.text()
self.last_port = self.port_spin.value()
# 如果设置了自动发送，启动定时器
if self.net_auto_send_check.isChecked():
self.net_send_timer.start(self.net_send_interval_spin.value())
# 启动接收线程
self.tcp_client_thread = threading.Thread(target=self.tcp_client_receive)
self.tcp_client_thread.daemon = True
self.tcp_client_thread.start()
except Exception as e:
error_msg = f"连接错误: {str(e)}"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
self.network_socket = None
elif protocol == "TCP服务器":
self.network_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
self.network_socket.bind(('0.0.0.0', self.port_spin.value()))
self.network_socket.listen(1)
self.net_connect_button.setText("停止服务器")
self.protocol_combo.setEnabled(False)
self.ip_edit.setEnabled(False)
self.port_spin.setEnabled(False)
self.net_receive_text.append(f"TCP服务器已启动，监听端口 {self.port_spin.value()}")
self.statusBar.showMessage(f"TCP服务器已启动，监听端口 {self.port_spin.value()}", 3000)
# 保存网络设置
self.last_protocol = protocol
self.last_port = self.port_spin.value()
# 启动服务器线程
self.tcp_server_thread = threading.Thread(target=self.tcp_server_run)
self.tcp_server_thread.daemon = True
self.tcp_server_thread.start()
except Exception as e:
error_msg = f"服务器启动错误: {str(e)}"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
self.network_socket = None
else: # UDP
self.network_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
self.network_socket.bind(('0.0.0.0', self.port_spin.value()))
self.net_connect_button.setText("断开")
self.protocol_combo.setEnabled(False)
self.port_spin.setEnabled(False)
self.net_receive_text.append(f"UDP已启动，监听端口 {self.port_spin.value()}")
self.statusBar.showMessage(f"UDP已启动，监听端口 {self.port_spin.value()}", 3000)
# 保存网络设置
self.last_protocol = protocol
self.last_ip = self.ip_edit.text()
self.last_port = self.port_spin.value()
# 如果设置了自动发送，启动定时器
if self.net_auto_send_check.isChecked():
self.net_send_timer.start(self.net_send_interval_spin.value())
# 启动UDP接收线程
self.udp_thread = threading.Thread(target=self.udp_receive)
self.udp_thread.daemon = True
self.udp_thread.start()
except Exception as e:
error_msg = f"UDP启动错误: {str(e)}"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
self.network_socket = None
def tcp_client_receive(self):
try:
while self.network_socket:
data = self.network_socket.recv(1024)
if not data:
break
self.handle_network_data(data)
except Exception as e:
if self.network_socket: # 只有在socket仍然存在时才报告错误
QApplication.postEvent(self, NetworkDataEvent(f"接收错误: {str(e)}"))
# 连接已关闭
if self.network_socket:
self.network_socket.close()
self.network_socket = None
QApplication.postEvent(self, NetworkDataEvent("TCP连接已关闭", is_status=True))
def tcp_server_run(self):
try:
QApplication.postEvent(self, NetworkDataEvent("等待客户端连接...", is_status=True))
self.client_socket, addr = self.network_socket.accept()
QApplication.postEvent(self, NetworkDataEvent(f"客户端 {addr[0]}:{addr[1]} 已连接", is_status=True))
# 如果设置了自动发送，启动定时器
if self.net_auto_send_check.isChecked():
QApplication.instance().postEvent(self, QEvent(QEvent.User)) # 自定义事件触发启动定时器
while self.network_socket:
data = self.client_socket.recv(1024)
if not data:
break
self.handle_network_data(data)
except Exception as e:
if self.network_socket: # 只有在socket仍然存在时才报告错误
QApplication.postEvent(self, NetworkDataEvent(f"服务器错误: {str(e)}"))
# 连接已关闭
if hasattr(self, 'client_socket') and self.client_socket:
self.client_socket.close()
QApplication.postEvent(self, NetworkDataEvent("客户端已断开连接", is_status=True))
# 继续监听新连接
if self.network_socket:
self.tcp_server_thread = threading.Thread(target=self.tcp_server_run)
self.tcp_server_thread.daemon = True
self.tcp_server_thread.start()
def udp_receive(self):
try:
while self.network_socket:
data, addr = self.network_socket.recvfrom(1024)
self.udp_client_addr = addr # 保存最后一个发送数据的客户端地址
QApplication.postEvent(self, NetworkDataEvent(f"从 {addr[0]}:{addr[1]} 接收:", is_status=True))
self.handle_network_data(data)
except Exception as e:
if self.network_socket: # 只有在socket仍然存在时才报告错误
QApplication.postEvent(self, NetworkDataEvent(f"UDP错误: {str(e)}"))
# 连接已关闭
if self.network_socket:
self.network_socket.close()
self.network_socket = None
QApplication.postEvent(self, NetworkDataEvent("UDP已关闭", is_status=True))
def handle_network_data(self, data):
data_len = len(data)
self.net_rx_bytes += data_len
self.rx_bytes = self.net_rx_bytes
# 记录到命令日志（如果在按钮选项卡中）
if self.tabs.currentIndex() == 3: # 按钮选项卡
try:
text = data.decode('utf-8')
self.command_log.append(f"接收: {text}")
except UnicodeDecodeError:
hex_str = ' '.join([f"{byte:02X}" for byte in data])
self.command_log.append(f"接收: [HEX] {hex_str}")
if self.net_hex_display_check.isChecked():
hex_str = ' '.join([f"{byte:02X}" for byte in data])
QApplication.postEvent(self, NetworkDataEvent(hex_str, rx_count=data_len))
else:
try:
text = data.decode('utf-8')
QApplication.postEvent(self, NetworkDataEvent(text, rx_count=data_len))
except UnicodeDecodeError:
# 如果解码失败，显示十六进制
hex_str = ' '.join([f"{byte:02X}" for byte in data])
QApplication.postEvent(self, NetworkDataEvent(f"[解码失败] {hex_str}", rx_count=data_len))
def send_network_data(self):
if not self.network_socket:
error_msg = "错误: 网络未连接"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
text = self.net_send_text.toPlainText()
if not text:
return
protocol = self.protocol_combo.currentText()
try:
if self.net_hex_send_check.isChecked():
# 处理十六进制输入
# 移除所有空格和换行符
clean_text = ''.join(text.split())
# 确保字符数是偶数
if len(clean_text) % 2 != 0:
error_msg = "错误: 十六进制字符数必须为偶数"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
return
# 转换为字节
data = bytes.fromhex(clean_text)
else:
# 处理特殊字符
processed_text = text.replace('\\r', '\r').replace('\\n', '\n').replace('\\t', '\t')
# 检查是否发送新行
if self.net_send_newline_check.isChecked():
processed_text += "\r\n"
data = processed_text.encode()
data_len = len(data)
self.net_tx_bytes += data_len
self.tx_bytes = self.net_tx_bytes
self.update_byte_counters()
if protocol == "TCP客户端":
self.network_socket.sendall(data)
self.statusBar.showMessage(f"已发送 {data_len} 字节", 3000)
elif protocol == "TCP服务器":
if hasattr(self, 'client_socket') and self.client_socket:
self.client_socket.sendall(data)
self.statusBar.showMessage(f"已发送 {data_len} 字节到客户端", 3000)
else:
error_msg = "错误: 没有客户端连接"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
else: # UDP
if hasattr(self, 'udp_client_addr') and self.udp_client_addr:
self.network_socket.sendto(data, self.udp_client_addr)
self.statusBar.showMessage(f"已发送 {data_len} 字节到 {self.udp_client_addr[0]}:{self.udp_client_addr[1]}", 3000)
else:
# 如果没有接收过数据，则发送到指定地址
self.network_socket.sendto(data, (self.ip_edit.text(), self.port_spin.value()))
self.statusBar.showMessage(f"已发送 {data_len} 字节到 {self.ip_edit.text()}:{self.port_spin.value()}", 3000)
except Exception as e:
error_msg = f"发送错误: {str(e)}"
self.net_receive_text.append(error_msg)
self.statusBar.showMessage(error_msg, 3000)
def send_network_data_timed(self):
# 定时器触发的自动发送
if self.network_socket:
self.send_network_data()
def closeEvent(self, event):
# 保存设置
self.save_settings()
# 关闭所有连接
if self.serial_port.isOpen():
self.serial_port.close()
# 停止所有定时器
self.serial_send_timer.stop()
self.bt_send_timer.stop()
self.net_send_timer.stop()
# 停止异步事件循环线程
if self.async_thread:
# 如果蓝牙客户端连接，先断开
if self.bluetooth_client and hasattr(self.bluetooth_client, 'is_connected') and self.bluetooth_client.is_connected:
try:
# 创建一个新的事件循环来处理断开连接
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self.bluetooth_client.disconnect())
loop.close()
except:
pass
self.async_thread.stop()
if self.network_socket:
self.network_socket.close()
event.accept()
# 重写事件处理函数
def event(self, event):
if event.type() == NetworkDataEvent.EVENT_TYPE:
if event.is_status:
self.net_receive_text.append(event.data)
self.statusBar.showMessage(event.data, 3000)
else:
self.net_receive_text.append(event.data)
if event.rx_count > 0:
self.update_byte_counters()
if self.net_auto_scroll_check.isChecked():
self.net_receive_text.verticalScrollBar().setValue(
self.net_receive_text.verticalScrollBar().maximum())
return True
elif event.type() == QEvent.User:
# 自定义事件，用于在TCP服务器接受连接后启动定时器
if self.net_auto_send_check.isChecked():
self.net_send_timer.start(self.net_send_interval_spin.value())
return True
return super(SerialDebugger, self).event(event)
if __name__ == "__main__":
app = QApplication(sys.argv)
window = SerialDebugger()
window.show()
sys.exit(app.exec_())
![[Python 原创] 用ai写的串口调试助手-政达科技](https://sccn.top/wp-content/uploads/2025/07/0eacd7268432883c943f85489313c5685bf95da5.jpg)
没有回复内容