import os import json import logging import serial.tools.list_ports import time from PySide6.QtWidgets import QMessageBox from ui.serial_settings_ui import SerialSettingsUI from utils.config_loader import ConfigLoader from utils.serial_manager import SerialManager class SerialSettingsWidget(SerialSettingsUI): """串口设置组件""" def __init__(self, parent=None): super().__init__(parent) self.config = ConfigLoader.get_instance() self.serial_manager = SerialManager() # 连接信号 self.init_connections() # 加载设置 self.load_settings() def init_connections(self): """初始化信号连接""" # 米电阻串口 self.mdz_refresh_btn.clicked.connect(self.refresh_ports) self.test_mdz_btn.clicked.connect(self.test_mdz_port) # 线径串口 self.xj_refresh_btn.clicked.connect(self.refresh_ports) self.test_xj_btn.clicked.connect(self.test_xj_port) # 保存按钮 self.save_btn.clicked.connect(self.save_settings) # 初始化时刷新串口列表 self.refresh_ports() # 加载设置 self.load_settings() def refresh_ports(self): """刷新串口列表""" try: # 保存当前选择 current_mdz_port = self.mdz_port_combo.currentData() # 清空列表 self.mdz_port_combo.clear() # 获取可用串口 ports = list(serial.tools.list_ports.comports()) for port in ports: port_name = port.device port_desc = f"{port_name} ({port.description})" self.mdz_port_combo.addItem(port_desc, port_name) # 恢复之前的选择 if current_mdz_port: index = self.mdz_port_combo.findData(current_mdz_port) if index >= 0: self.mdz_port_combo.setCurrentIndex(index) logging.info(f"已刷新串口列表，找到 {len(ports)} 个串口") except Exception as e: logging.error(f"刷新串口列表失败: {e}") QMessageBox.warning(self, "刷新失败", f"刷新串口列表失败: {e}") def load_settings(self): """加载设置""" try: # 加载全局设置 enable_serial = self.config.get_value('app.features.enable_serial_ports', False) enable_keyboard = self.config.get_value('app.features.enable_keyboard_listener', False) self.enable_serial_checkbox.setChecked(enable_serial) self.enable_keyboard_checkbox.setChecked(enable_keyboard) # 加载米电阻设置 mdz_config = self.config.get_config('mdz') if mdz_config: # 设置串口 mdz_port = mdz_config.get('ser', '') index = self.mdz_port_combo.findData(mdz_port) if index >= 0: self.mdz_port_combo.setCurrentIndex(index) # 设置波特率 mdz_baud = str(mdz_config.get('port', '9600')) index = self.mdz_baud_combo.findText(mdz_baud) if index >= 0: self.mdz_baud_combo.setCurrentIndex(index) # 设置数据位 mdz_data_bits = str(mdz_config.get('data_bits', '8')) index = self.mdz_data_bits_combo.findText(mdz_data_bits) if index >= 0: self.mdz_data_bits_combo.setCurrentIndex(index) # 设置停止位 mdz_stop_bits = str(mdz_config.get('stop_bits', '1')) index = self.mdz_stop_bits_combo.findText(mdz_stop_bits) if index >= 0: self.mdz_stop_bits_combo.setCurrentIndex(index) # 设置校验位 mdz_parity = mdz_config.get('parity', 'N') index = self.mdz_parity_combo.findData(mdz_parity) if index >= 0: self.mdz_parity_combo.setCurrentIndex(index) # 设置查询指令 mdz_query_cmd = mdz_config.get('query_cmd', '01 03 00 01 00 07 55 C8') self.mdz_query_cmd.setText(mdz_query_cmd) # 设置查询间隔 mdz_query_interval = mdz_config.get('query_interval', 5) self.mdz_query_interval.setValue(mdz_query_interval) # 加载线径设置 xj_config = self.config.get_config('xj') if xj_config: # 设置串口 xj_port = xj_config.get('ser', '') index = self.xj_port_combo.findData(xj_port) if index >= 0: self.xj_port_combo.setCurrentIndex(index) # 设置波特率 xj_baud = str(xj_config.get('port', '9600')) index = self.xj_baud_combo.findText(xj_baud) if index >= 0: self.xj_baud_combo.setCurrentIndex(index) # 设置数据位 xj_data_bits = str(xj_config.get('data_bits', '8')) index = self.xj_data_bits_combo.findText(xj_data_bits) if index >= 0: self.xj_data_bits_combo.setCurrentIndex(index) # 设置停止位 xj_stop_bits = str(xj_config.get('stop_bits', '1')) index = self.xj_stop_bits_combo.findText(xj_stop_bits) if index >= 0: self.xj_stop_bits_combo.setCurrentIndex(index) # 设置校验位 xj_parity = xj_config.get('parity', 'N') index = self.xj_parity_combo.findData(xj_parity) if index >= 0: self.xj_parity_combo.setCurrentIndex(index) logging.info("已加载串口设置") except Exception as e: logging.error(f"加载串口设置失败: {e}") QMessageBox.warning(self, "加载失败", f"加载串口设置失败: {e}") def save_settings(self): """保存设置""" try: # 保存全局设置 enable_serial = self.enable_serial_checkbox.isChecked() enable_keyboard = self.enable_keyboard_checkbox.isChecked() self.config.set_value('app.features.enable_serial_ports', enable_serial) self.config.set_value('app.features.enable_keyboard_listener', enable_keyboard) # 保存米电阻设置 mdz_port = self.mdz_port_combo.currentData() mdz_baud = int(self.mdz_baud_combo.currentText()) mdz_data_bits = int(self.mdz_data_bits_combo.currentText()) mdz_stop_bits = float(self.mdz_stop_bits_combo.currentText()) mdz_parity = self.mdz_parity_combo.currentData() mdz_query_cmd = self.mdz_query_cmd.text().strip() mdz_query_interval = self.mdz_query_interval.value() mdz_config = { 'ser': mdz_port, 'port': mdz_baud, 'data_bits': mdz_data_bits, 'stop_bits': mdz_stop_bits, 'parity': mdz_parity, 'query_cmd': mdz_query_cmd, 'query_interval': mdz_query_interval } self.config.set_config('mdz', mdz_config) # 保存线径设置 xj_port = self.xj_port_combo.currentData() xj_baud = int(self.xj_baud_combo.currentText()) xj_data_bits = int(self.xj_data_bits_combo.currentText()) xj_stop_bits = float(self.xj_stop_bits_combo.currentText()) xj_parity = self.xj_parity_combo.currentData() xj_config = { 'ser': xj_port, 'port': xj_baud, 'data_bits': xj_data_bits, 'stop_bits': xj_stop_bits, 'parity': xj_parity } self.config.set_config('xj', xj_config) # 发送设置变更信号 self.settings_changed.emit() QMessageBox.information(self, "保存成功", "串口设置已保存") except Exception as e: logging.error(f"保存串口设置失败: {e}") QMessageBox.critical(self, "保存失败", f"保存串口设置失败: {e}") def test_mdz_port(self): """测试米电阻串口""" try: port = self.mdz_port_combo.currentData() baud = int(self.mdz_baud_combo.currentText()) data_bits = int(self.mdz_data_bits_combo.currentText()) stop_bits = float(self.mdz_stop_bits_combo.currentText()) parity = self.mdz_parity_combo.currentData() if not port: QMessageBox.warning(self, "测试失败", "请选择串口") return # 关闭可能已经打开的串口 if self.serial_manager.is_port_open(port): self.serial_manager.close_port(port) # 尝试打开串口 success = self.serial_manager.open_port( port, 'mdz', baud, data_bits, stop_bits, parity, 1.0 ) if success: # 尝试发送查询指令 query_cmd = self.mdz_query_cmd.text() if query_cmd: try: # 转换查询指令为字节 query_bytes = bytes.fromhex(query_cmd.replace(' ', '')) # 发送查询指令 self.serial_manager.write_data(port, query_bytes) # 等待一段时间 time.sleep(0.5) # 关闭串口 self.serial_manager.close_port(port) QMessageBox.information(self, "测试成功", f"米电阻串口 {port} 测试成功，已发送查询指令") except Exception as e: self.serial_manager.close_port(port) QMessageBox.warning(self, "测试失败", f"发送查询指令失败: {e}") else: self.serial_manager.close_port(port) QMessageBox.information(self, "测试成功", f"米电阻串口 {port} 打开成功，但未发送查询指令") else: QMessageBox.warning(self, "测试失败", f"无法打开米电阻串口 {port}") except Exception as e: logging.error(f"测试米电阻串口失败: {e}") QMessageBox.warning(self, "测试失败", f"测试米电阻串口失败: {e}") def test_xj_port(self): """测试线径串口""" try: port = self.xj_port_combo.currentData() baud = int(self.xj_baud_combo.currentText()) data_bits = int(self.xj_data_bits_combo.currentText()) stop_bits = float(self.xj_stop_bits_combo.currentText()) parity = self.xj_parity_combo.currentData() if not port: QMessageBox.warning(self, "测试失败", "请选择串口") return # 关闭可能已经打开的串口 if self.serial_manager.is_port_open(port): self.serial_manager.close_port(port) # 尝试打开串口 success = self.serial_manager.open_port( port, 'xj', baud, data_bits, stop_bits, parity, 1.0 ) if success: # 等待一段时间 time.sleep(2) # 关闭串口 self.serial_manager.close_port(port) QMessageBox.information(self, "测试成功", f"线径串口 {port} 测试成功") else: QMessageBox.warning(self, "测试失败", f"无法打开线径串口 {port}") except Exception as e: logging.error(f"测试线径串口失败: {e}") QMessageBox.warning(self, "测试失败", f"测试线径串口失败: {e}")