jiateng_ws/widgets/serial_settings_widget.py

310 lines
12 KiB
Python

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}")