jiateng_ws/widgets/serial_settings_widget.py

324 lines
14 KiB
Python
Raw Normal View History

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.mdz_refresh_btn.clicked.connect(self.refresh_ports)
self.cz_refresh_btn.clicked.connect(self.refresh_ports)
self.test_mdz_btn.clicked.connect(self.test_mdz_port)
self.test_cz_btn.clicked.connect(self.test_cz_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()
current_cz_port = self.cz_port_combo.currentData()
# 清空列表
self.mdz_port_combo.clear()
self.cz_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)
self.cz_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)
if current_cz_port:
index = self.cz_port_combo.findData(current_cz_port)
if index >= 0:
self.cz_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)
# 加载称重设置
cz_config = self.config.get_config('cz')
if cz_config:
# 设置串口
cz_port = cz_config.get('ser', '')
index = self.cz_port_combo.findData(cz_port)
if index >= 0:
self.cz_port_combo.setCurrentIndex(index)
# 设置波特率
cz_baud = str(cz_config.get('port', '9600'))
index = self.cz_baud_combo.findText(cz_baud)
if index >= 0:
self.cz_baud_combo.setCurrentIndex(index)
# 设置数据位
cz_data_bits = str(cz_config.get('data_bits', '8'))
index = self.cz_data_bits_combo.findText(cz_data_bits)
if index >= 0:
self.cz_data_bits_combo.setCurrentIndex(index)
# 设置停止位
cz_stop_bits = str(cz_config.get('stop_bits', '1'))
index = self.cz_stop_bits_combo.findText(cz_stop_bits)
if index >= 0:
self.cz_stop_bits_combo.setCurrentIndex(index)
# 设置校验位
cz_parity = cz_config.get('parity', 'N')
index = self.cz_parity_combo.findData(cz_parity)
if index >= 0:
self.cz_parity_combo.setCurrentIndex(index)
# 设置稳定阈值
cz_stable_threshold = cz_config.get('stable_threshold', 10)
self.cz_stable_threshold.setValue(cz_stable_threshold)
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_config = {}
mdz_config['ser'] = self.mdz_port_combo.currentData()
mdz_config['port'] = int(self.mdz_baud_combo.currentText())
mdz_config['data_bits'] = int(self.mdz_data_bits_combo.currentText())
mdz_config['stop_bits'] = float(self.mdz_stop_bits_combo.currentText())
mdz_config['parity'] = self.mdz_parity_combo.currentData()
mdz_config['query_cmd'] = self.mdz_query_cmd.text()
mdz_config['query_interval'] = self.mdz_query_interval.value()
mdz_config['code'] = 'mdz'
mdz_config['bit'] = 10
mdz_config['timeout'] = 1
self.config.set_config('mdz', mdz_config)
# 保存称重设置
cz_config = {}
cz_config['ser'] = self.cz_port_combo.currentData()
cz_config['port'] = int(self.cz_baud_combo.currentText())
cz_config['data_bits'] = int(self.cz_data_bits_combo.currentText())
cz_config['stop_bits'] = float(self.cz_stop_bits_combo.currentText())
cz_config['parity'] = self.cz_parity_combo.currentData()
cz_config['stable_threshold'] = self.cz_stable_threshold.value()
cz_config['code'] = 'cz'
cz_config['bit'] = 10
cz_config['timeout'] = 1
self.config.set_config('cz', cz_config)
# 保存键盘设置
keyboard_config = {}
keyboard_config['enabled'] = enable_keyboard
keyboard_config['trigger_key'] = 'Key.page_up'
self.config.set_config('keyboard', keyboard_config)
# 保存配置文件
self.config.save_config()
# 重新加载串口管理器的配置
self.serial_manager.reload_config()
# 根据键盘监听设置立即启动或停止键盘监听
if enable_keyboard:
logging.info("键盘监听已启用,正在启动键盘监听...")
self.serial_manager.start_keyboard_listener()
# 添加测试触发,确认键盘监听是否正常工作
if hasattr(self.serial_manager, 'keyboard_listener') and self.serial_manager.keyboard_listener:
if self.serial_manager.keyboard_listener.is_active():
logging.info("键盘监听已成功启动,尝试手动触发测试事件")
self.serial_manager.keyboard_listener.trigger_test_event()
else:
logging.warning("键盘监听启动失败,未处于活动状态")
else:
logging.info("键盘监听已禁用,正在停止键盘监听...")
self.serial_manager.stop_keyboard_listener()
# 发送设置改变信号
self.settings_changed.emit()
logging.info("已保存串口设置")
QMessageBox.information(self, "保存成功", "串口设置已保存")
except Exception as e:
logging.error(f"保存串口设置失败: {e}")
QMessageBox.warning(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_cz_port(self):
"""测试线径串口"""
try:
port = self.cz_port_combo.currentData()
baud = int(self.cz_baud_combo.currentText())
data_bits = int(self.cz_data_bits_combo.currentText())
stop_bits = float(self.cz_stop_bits_combo.currentText())
parity = self.cz_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, 'cz', 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}")