""" 网关配置工具 串口配置：波特率500000(可修改)，数据位8，停止位1，校验位None """ import sys import os import re import time import struct import zlib import threading from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QComboBox, QPushButton, QLineEdit, QTextEdit, QMessageBox, QGridLayout, QFileDialog, QDialog, QProgressBar, QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView ) from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer from PyQt5.QtGui import QFont, QColor, QPalette, QIcon import serial import serial.tools.list_ports def resource_path(relative_path): if hasattr(sys, '_MEIPASS'): return os.path.join(sys._MEIPASS, relative_path) return os.path.join(os.path.abspath(os.path.dirname(__file__)), relative_path) # ── 传感器通用串口升级协议 V1.1 ───────────────────────────── FRAME_HEADER = 0x7D UPG_PACKET_DATA = 200 # 每包数据长度上限 CMD_ANNOUNCE = 0x81 # 主机->从机：下发APP请求 CMD_ACK = 0x01 # 从机->主机：应答(Response=0x14) CMD_REQ_PKG = 0x02 # 从机->主机：请求包序号 CMD_SEND_PKG = 0x82 # 主机->从机：下发包 PKG_DATALEN_MAX = 200 def crc16_modbus(data): """MODBUS CRC16，多项式0x8005(反射)，返回16位整数。""" crc = 0xFFFF for b in data: crc ^= b for _ in range(8): if crc & 1: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc def build_frame(dev_mac, cmd, payload): """0x7D + DevMac(6) + Cmd + PayloadLen + Payload + CRC16(小端)。""" body = bytes([FRAME_HEADER]) + bytes(dev_mac) + bytes([cmd, len(payload)]) + bytes(payload) crc = crc16_modbus(body) return body + struct.pack(' bytes(6)，无效返回 None。""" if not s: return None parts = s.strip().replace('-', ':').split(':') parts = [p for p in parts if p] if len(parts) != 6: return None try: return bytes(int(p, 16) for p in parts) except ValueError: return None class SerialThread(QThread): data_received = pyqtSignal(str) data_received_raw = pyqtSignal(bytes) def __init__(self, ser): super().__init__() self.ser = ser self.running = True self._buffer = b"" def run(self): while self.running: try: if self.ser and self.ser.is_open and self.ser.in_waiting > 0: data = self.ser.read(self.ser.in_waiting) self.data_received_raw.emit(data) buf = self._buffer + data text = None # 优先 UTF-8；若结尾是不完整字符序列则保留尾部待下次拼接，避免乱码缺字 for enc in ('utf-8', 'gbk'): try: text = buf.decode(enc) self._buffer = b"" break except UnicodeDecodeError: keep = b"" for tail in range(1, min(4, len(buf)) + 1): try: text = buf[:-tail].decode(enc) keep = buf[-tail:] break except UnicodeDecodeError: continue if text is not None: self._buffer = keep break if text is None: text = data.decode('utf-8', errors='replace') self._buffer = b"" if text: self.data_received.emit(text) else: time.sleep(0.01) except Exception: break def stop(self): self.running = False self.wait() class GatewayConfigTool(QMainWindow): def __init__(self): super().__init__() self.serial_port = None self.serial_thread = None self._recv_buffer = "" self._para_mode = False # para 大数据量返回时，用定时器延迟解析，避免每段数据全量解析阻塞UI导致串口缓冲溢出丢包 self._para_timer = QTimer(self) self._para_timer.setSingleShot(True) self._para_timer.setInterval(250) self._para_timer.timeout.connect(self._on_para_parse) self.upgrade_dialog = None self.dev_dialog = None self.init_ui() self.refresh_ports() def init_ui(self): self.setWindowTitle("网关配置工具") self.setWindowIcon(QIcon(resource_path("物联网网关.png"))) self.setMinimumSize(940, 640) self.resize(1000, 700) central = QWidget() self.setCentralWidget(central) main_layout = QVBoxLayout(central) main_layout.setSpacing(4) main_layout.setContentsMargins(4, 4, 4, 4) # === 串口连接区 === conn_group = QGroupBox("串口连接") conn_layout = QHBoxLayout(conn_group) conn_layout.addWidget(QLabel("串口:")) self.combo_port = QComboBox() self.combo_port.setMinimumWidth(120) conn_layout.addWidget(self.combo_port) self.btn_refresh = QPushButton("刷新") self.btn_refresh.setFixedWidth(60) self.btn_refresh.clicked.connect(self.refresh_ports) conn_layout.addWidget(self.btn_refresh) conn_layout.addSpacing(20) conn_layout.addWidget(QLabel("波特率:")) self.combo_baud = QComboBox() self.combo_baud.setEditable(True) self.combo_baud.addItems(["115200", "230400", "460800", "500000", "921600", "1000000"]) self.combo_baud.setCurrentText("500000") self.combo_baud.setFixedWidth(90) conn_layout.addWidget(self.combo_baud) conn_layout.addWidget(QLabel("数据位: 8")) conn_layout.addWidget(QLabel("停止位: 1")) conn_layout.addWidget(QLabel("校验: None")) conn_layout.addStretch() self.btn_connect = QPushButton("打开串口") self.btn_connect.setFixedWidth(100) self.btn_connect.setStyleSheet("font-weight:bold;") self.btn_connect.clicked.connect(self.toggle_serial) conn_layout.addWidget(self.btn_connect) main_layout.addWidget(conn_group) # === 参数配置区 === config_group = QGroupBox("参数配置") config_group_layout = QVBoxLayout(config_group) config_group_layout.setSpacing(4) # Lora 参数（子分组） lora_group = QGroupBox("Lora 参数") lora_layout = QHBoxLayout(lora_group) lora_layout.setSpacing(6) lora_layout.addWidget(QLabel("信道频率(fc):")) self.combo_lora_fc = QComboBox() self.combo_lora_fc.setFixedWidth(200) lora_freqs = [] fc_hz = 470100000 ch_idx = 0 while fc_hz <= 492600000: mhz = fc_hz / 1000000 lora_freqs.append((str(fc_hz), f"信道{ch_idx}: {mhz:.1f}MHz")) fc_hz += 500000 ch_idx += 1 for freq, desc in lora_freqs: self.combo_lora_fc.addItem(desc, freq) self.combo_lora_fc.setMaxVisibleItems(20) self.combo_lora_fc.setStyleSheet( "QComboBox { combobox-popup: 0; }" "QComboBox QAbstractItemView { max-height: 500px; }" ) lora_layout.addWidget(self.combo_lora_fc) self.btn_lora_fc = QPushButton("发送") self.btn_lora_fc.setFixedWidth(60) self.btn_lora_fc.setStyleSheet("font-weight:bold;") self.btn_lora_fc.clicked.connect(self.on_lora_send) lora_layout.addWidget(self.btn_lora_fc) lora_hint = QLabel("设置LoRa通信信道频率") lora_hint.setStyleSheet("color:#909090;") lora_layout.addWidget(lora_hint) lora_layout.addStretch() config_group_layout.addWidget(lora_group) # 查看配置按钮 + 提示 btn_row = QHBoxLayout() btn_row.setSpacing(6) self.btn_para = QPushButton("查看配置(para)") self.btn_para.setFixedWidth(130) self.btn_para.clicked.connect(self.on_para_clicked) btn_row.addWidget(self.btn_para) self.btn_upgrade = QPushButton("设备升级") self.btn_upgrade.setFixedWidth(100) self.btn_upgrade.setStyleSheet("background-color:#1976D2; color:white; font-weight:bold;") self.btn_upgrade.clicked.connect(self.on_upgrade_clicked) btn_row.addWidget(self.btn_upgrade) self.btn_dev_mgmt = QPushButton("设备管理") self.btn_dev_mgmt.setFixedWidth(100) self.btn_dev_mgmt.setStyleSheet("background-color:#388E3C; color:white; font-weight:bold;") self.btn_dev_mgmt.clicked.connect(self.on_dev_mgmt_clicked) btn_row.addWidget(self.btn_dev_mgmt) para_hint = QLabel("配置后可通过para命令验证是否成功") para_hint.setStyleSheet("color:#909090;") btn_row.addWidget(para_hint) btn_row.addStretch() config_group_layout.addLayout(btn_row) main_layout.addWidget(config_group) # === 设备参数信息区 === info_group = QGroupBox("设备参数信息") info_grid = QGridLayout(info_group) info_grid.setSpacing(8) info_grid.setContentsMargins(12, 8, 12, 8) def _val_label(color="#757575"): lb = QLabel("-") lb.setStyleSheet(f"color:{color};") return lb row = 0 info_grid.addWidget(QLabel("软件版本:"), row, 0) self.label_sw_ver = _val_label() info_grid.addWidget(self.label_sw_ver, row, 1) info_grid.addWidget(QLabel("硬件版本:"), row, 2) self.label_hw_ver = _val_label() info_grid.addWidget(self.label_hw_ver, row, 3) row = 1 info_grid.addWidget(QLabel("电池电量:"), row, 0) self.label_battery = _val_label() info_grid.addWidget(self.label_battery, row, 1) info_grid.addWidget(QLabel("注册状态:"), row, 2) self.label_reg_status = _val_label() info_grid.addWidget(self.label_reg_status, row, 3) row = 2 info_grid.addWidget(QLabel("通讯方式:"), row, 0) self.label_cur_comm = _val_label() info_grid.addWidget(self.label_cur_comm, row, 1) info_grid.addWidget(QLabel("Lora频率:"), row, 2) self.label_cur_lora_fc = _val_label() info_grid.addWidget(self.label_cur_lora_fc, row, 3) row = 3 info_grid.addWidget(QLabel("网关MAC:"), row, 0) self.label_gw_mac = _val_label() info_grid.addWidget(self.label_gw_mac, row, 1) info_grid.addWidget(QLabel("通讯单元读取间隔:"), row, 2) self.label_comm_interval = _val_label() info_grid.addWidget(self.label_comm_interval, row, 3) row = 4 info_grid.addWidget(QLabel("Cat1 IMEI:"), row, 0) self.label_imei = _val_label() info_grid.addWidget(self.label_imei, row, 1) info_grid.addWidget(QLabel("Cat1 SIM:"), row, 2) self.label_sim = _val_label() info_grid.addWidget(self.label_sim, row, 3) main_layout.addWidget(info_group) # === 下方：通讯日志 + 自定义指令 === bottom_widget = QWidget() bottom_layout = QHBoxLayout(bottom_widget) bottom_layout.setContentsMargins(0, 0, 0, 0) bottom_layout.setSpacing(4) log_group = QGroupBox("通讯日志") log_inner = QVBoxLayout(log_group) self.text_log = QTextEdit() self.text_log.setReadOnly(True) self.text_log.setFont(QFont("Consolas", 9)) self.text_log.setStyleSheet( "QTextEdit { background-color: #1e1e1e; color: #d4d4d4; }" ) log_inner.addWidget(self.text_log) log_btn_layout = QHBoxLayout() self.btn_clear_log = QPushButton("清空日志") self.btn_clear_log.clicked.connect(self.text_log.clear) log_btn_layout.addWidget(self.btn_clear_log) log_btn_layout.addStretch() log_inner.addLayout(log_btn_layout) custom_group = QGroupBox("自定义指令") custom_inner = QVBoxLayout(custom_group) custom_inner.setSpacing(6) custom_inner.addWidget(QLabel("指令:")) self.edit_custom = QLineEdit() self.edit_custom.setPlaceholderText("输入指令...") self.edit_custom.returnPressed.connect(self.send_custom_command) custom_inner.addWidget(self.edit_custom) self.btn_custom_send = QPushButton("发送") self.btn_custom_send.setStyleSheet("font-weight:bold;") self.btn_custom_send.clicked.connect(self.send_custom_command) custom_inner.addWidget(self.btn_custom_send) custom_inner.addStretch() bottom_layout.addWidget(log_group, 4) bottom_layout.addWidget(custom_group, 1) main_layout.addWidget(bottom_widget, 1) self.statusBar().showMessage("就绪 - 请选择串口并连接") def refresh_ports(self): self.combo_port.clear() ports = serial.tools.list_ports.comports() if not ports: self.combo_port.addItem("无可用串口") return for p in ports: self.combo_port.addItem(f"{p.device} - {p.description}") def get_port_name(self): text = self.combo_port.currentText() if not text or text.startswith("无"): return None return text.split(" - ")[0] def toggle_serial(self): if self.serial_port and self.serial_port.is_open: self.close_serial() else: self.open_serial() def open_serial(self): port_name = self.get_port_name() if not port_name: QMessageBox.warning(self, "错误", "请选择有效的串口！") return try: baud = int(self.combo_baud.currentText().strip()) except ValueError: QMessageBox.warning(self, "错误", "波特率无效，请输入数字！") return try: self.serial_port = serial.Serial( port=port_name, baudrate=baud, bytesize=serial.EIGHTBITS, stopbits=serial.STOPBITS_ONE, parity=serial.PARITY_NONE, timeout=0.1 ) self.serial_thread = SerialThread(self.serial_port) self.serial_thread.data_received.connect(self.on_data_received) self.serial_thread.data_received_raw.connect(self.on_upgrade_data) self.serial_thread.start() self.btn_connect.setText("关闭串口") self.btn_connect.setStyleSheet("background-color:#f44336; color:white; font-weight:bold;") self.log_send(f"已连接 {port_name} @ {baud}") self.statusBar().showMessage(f"已连接: {port_name}") except Exception as e: QMessageBox.critical(self, "串口错误", f"无法打开串口:\n{e}") def close_serial(self): if self.serial_thread: self.serial_thread.stop() self.serial_thread = None if self.serial_port and self.serial_port.is_open: port_name = self.serial_port.port self.serial_port.close() self.log_send(f"已断开 {port_name}") self.serial_port = None self.btn_connect.setText("打开串口") self.btn_connect.setStyleSheet("font-weight:bold;") self.statusBar().showMessage("已断开") def on_para_clicked(self): self._recv_buffer = "" self._para_mode = True self.send_command("para") self._para_timer.start() def _on_para_parse(self): if not self._para_mode: return self._try_parse_para() if not self._para_mode: self._para_timer.stop() def on_lora_send(self): value = self.combo_lora_fc.currentData() self.send_command(f"lora fc {value}") def send_command(self, cmd): if not self.serial_port or not self.serial_port.is_open: QMessageBox.warning(self, "错误", "请先打开串口！") return try: data = (cmd + "\r\n").encode('utf-8') self.serial_port.write(data) self.log_send(f">>> {cmd}") except Exception as e: self.log_send(f"[发送错误] {e}") def send_raw(self, data): if not self.serial_port or not self.serial_port.is_open: QMessageBox.warning(self, "错误", "请先打开串口！") return False try: self.serial_port.write(data) self.log_send(f">>> {data.hex(' ').upper()}") return True except Exception as e: self.log_send(f"[发送错误] {e}") return False # ── 设备升级 ───────────────────────────────────────── def on_upgrade_clicked(self): if self.upgrade_dialog is None: self.upgrade_dialog = UpgradeDialog(self) else: self.upgrade_dialog._reset_ui() self.upgrade_dialog.show() self.upgrade_dialog.raise_() self.upgrade_dialog.activateWindow() def upgrade_active(self): return self.upgrade_dialog is not None and self.upgrade_dialog.isUpgrading() # ── 设备管理（查询/删除） ──────────────────────────── def on_dev_mgmt_clicked(self): if not self.serial_port or not self.serial_port.is_open: QMessageBox.warning(self, "错误", "请先打开串口！") return if self.dev_dialog is None: self.dev_dialog = DeviceDialog(self) self.dev_dialog.show() self.dev_dialog.raise_() self.dev_dialog.activateWindow() self.dev_dialog.query() def on_upgrade_data(self, data): if self.upgrade_active(): self.upgrade_dialog.feed_data(data) def send_custom_command(self): cmd = self.edit_custom.text().strip() if not cmd: return self.send_command(cmd) self.edit_custom.clear() def on_data_received(self, text): if self.upgrade_active(): return self._recv_buffer += text self.log_recv(text) if self.dev_dialog is not None and self.dev_dialog.isVisible(): self.dev_dialog.feed_text(text) if self._para_mode: self._para_timer.start() def _try_parse_para(self): buf = self._recv_buffer.replace('\r', '') lines_all = [l.strip() for l in buf.split('\n')] # 查找 *** 包围的块 star_indices = [i for i, l in enumerate(lines_all) if l and all(c == '*' for c in l)] if len(star_indices) < 2: return start_idx = star_indices[0] end_idx = star_indices[1] lines = lines_all[start_idx + 1:end_idx] def _update(label, value, ok_color="#2E7D32"): label.setText(value) label.setStyleSheet(f"font-weight:bold; color:{ok_color};") section = "" for line in lines: if line.startswith('==='): section = line.strip('= ').strip() continue # SoftWare V:1.2, HardWare V:1.2 m = re.match(r'SoftWare\s+V:([\d.]+),\s*HardWare\s+V:([\d.]+)', line) if m: _update(self.label_sw_ver, m.group(1)) _update(self.label_hw_ver, m.group(2)) continue # Key: Value 行 m = re.match(r'([\w\s/()]+?):\s*(.*)', line) if not m: continue key = m.group(1).strip() val = m.group(2).strip() if key == 'GateWay MAC': _update(self.label_gw_mac, val) elif key == 'Battery': _update(self.label_battery, val) elif key == 'Much': _update(self.label_cur_comm, val) elif key == 'Cat1 IMEI': _update(self.label_imei, val) elif key == 'Cat1 SIM': _update(self.label_sim, val) elif key == 'CommUnitReadInterval': _update(self.label_comm_interval, f"{val} s") elif key == 'MuchReg': status_map = {"true": "已注册", "false": "未注册", "null": "未注册"} _update(self.label_reg_status, status_map.get(val, val)) # --- LTENet (Cat1) section --- elif key == 'DestAddr' and 'Cat1' in section: pass # Cat1远端地址暂不单独显示 elif key == 'DestPort' and 'Cat1' in section: pass elif key == 'LocalAddr': pass elif key == 'LocalPort' and 'Cat1' in section: pass # --- Lora --- elif key == 'Lora': fc_m = re.search(r'fc\s+(\d+)', val) if fc_m: fc_val = fc_m.group(1) _update(self.label_cur_lora_fc, fc_val) idx = self.combo_lora_fc.findData(fc_val) if idx >= 0: self.combo_lora_fc.setCurrentIndex(idx) self._para_mode = False self._recv_buffer = "" self.statusBar().showMessage("配置读取完成") @staticmethod def _escape_html(s): return s.replace('&', '&').replace('<', '<').replace('>', '>') def log_send(self, text): self.text_log.append(f'{self._escape_html(text)}') def log_recv(self, text): clean = self._escape_html(text).replace('\r', '').replace('\n', '
') self.text_log.append(f'<<< {clean}') def closeEvent(self, event): self.close_serial() event.accept() class UpgradeDialog(QDialog): """设备升级对话框：参照通用串口升级协议V1.1。""" def __init__(self, main_window): super().__init__(main_window) self.main = main_window self.setWindowTitle("设备升级 - 通用串口升级协议V1.1") self.setModal(False) self.resize(720, 520) self.running = False # 升级进行中 self._mac = None self._fw_data = b"" self._pkg_num = 0 self._sent = set() self._buf = b"" self._timer = QTimer() self._timer.setSingleShot(True) self._timer.timeout.connect(self._on_timeout) root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(6) # ── 升级设置 ── cfg = QGroupBox("升级设置") cfg_layout = QGridLayout(cfg) cfg_layout.setSpacing(6) cfg_layout.addWidget(QLabel("固件文件(.bin):"), 0, 0) self.fw_edit = QLineEdit() self.fw_edit.setPlaceholderText("选择固件文件...") cfg_layout.addWidget(self.fw_edit, 0, 1) btn_browse = QPushButton("浏览") btn_browse.clicked.connect(self._choose_file) cfg_layout.addWidget(btn_browse, 0, 2) cfg_layout.addWidget(QLabel("设备MAC:"), 1, 0) default_mac = main_window.label_gw_mac.text() self.mac_edit = QLineEdit(default_mac if default_mac != "-" else "") self.mac_edit.setPlaceholderText("81-00-00-06-00-00") cfg_layout.addWidget(self.mac_edit, 1, 1) mac_hint = QLabel("6字节从机MAC地址") mac_hint.setStyleSheet("color:#909090;") cfg_layout.addWidget(mac_hint, 1, 2) btn_row = QHBoxLayout() self.start_btn = QPushButton("开始升级") self.start_btn.clicked.connect(self._start) btn_row.addWidget(self.start_btn) btn_close = QPushButton("关闭") btn_close.clicked.connect(self.close) btn_row.addWidget(btn_close) btn_row.addStretch() cfg_layout.addLayout(btn_row, 2, 0, 1, 3) self.progress = QProgressBar() self.progress.setRange(0, 100) self.progress.setValue(0) # 固定进度条颜色，避免窗口失活(点击弹窗外)时变色 self.progress.setStyleSheet(""" QProgressBar { border: 1px solid #B0B0B0; border-radius: 3px; background: #E0E0E0; text-align: center; } QProgressBar::chunk { background-color: #1976D2; } """) cfg_layout.addWidget(self.progress, 3, 0, 1, 2) self.status_lbl = QLabel("就绪") cfg_layout.addWidget(self.status_lbl, 3, 2) root.addWidget(cfg) # ── 升级日志 ── log_group = QGroupBox("升级日志") log_layout = QVBoxLayout(log_group) self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setFont(QFont("Consolas", 9)) log_btn_row = QHBoxLayout() btn_clear_log = QPushButton("清空日志") btn_clear_log.clicked.connect(self.log_text.clear) log_btn_row.addStretch() log_btn_row.addWidget(btn_clear_log) log_layout.addLayout(log_btn_row) log_layout.addWidget(self.log_text) root.addWidget(log_group, 1) def isUpgrading(self): return self.running def _reset_ui(self): """重置弹窗 UI 到初始状态，避免关闭后重新打开时界面卡住。""" self.running = False self._timer.stop() self._buf = b"" self._sent.clear() self._fw_data = b"" self._mac = None self._pkg_num = 0 self.start_btn.setEnabled(True) self.progress.setValue(0) self.status_lbl.setText("就绪") def _choose_file(self): path, _ = QFileDialog.getOpenFileName( self, "选择固件文件", "", "固件文件 (*.bin);;所有文件 (*)" ) if path: self.fw_edit.setText(path) def _log(self, msg, tag="info"): color = {"ok": "#2E7D32", "err": "#D32F2F", "info": "#1565C0"}.get(tag, "#000000") self.log_text.append(f'{GatewayConfigTool._escape_html(str(msg))}') def _log_debug(self, data): if not data: return if any(0x20 <= b <= 0x7E for b in data): text = data.decode('utf-8', errors='replace').strip('\x00\r\n ') if text: self._log(f"[设备] {text}") def _start(self): if self.running: return path = self.fw_edit.text().strip() mac = parse_mac(self.mac_edit.text()) if not path or not os.path.exists(path): QMessageBox.warning(self, "提示", "请先选择固件文件") return if mac is None: QMessageBox.warning(self, "提示", "设备MAC格式错误，应为 81-00-00-06-00-00") return if not self.main.serial_port or not self.main.serial_port.is_open: QMessageBox.warning(self, "提示", "请先打开串口") return try: with open(path, 'rb') as f: fw = f.read() except Exception as e: QMessageBox.critical(self, "错误", f"读取固件失败: {e}") return if not fw: QMessageBox.warning(self, "提示", "固件文件为空") return self._mac = mac self._fw_data = fw self._pkg_num = (len(fw) + UPG_PACKET_DATA - 1) // UPG_PACKET_DATA self._sent.clear() self._buf = b"" self.running = True self.start_btn.setEnabled(False) self.progress.setValue(0) self.status_lbl.setText("等待握手...") app_crc = zlib.crc32(fw) & 0xFFFFFFFF payload = struct.pack('= 11: idx = self._buf.find(bytes([FRAME_HEADER])) if idx < 0: # 无帧头，保留末尾字节以防跨包帧头 self._buf = self._buf[-1:] return None if idx > 0: self._log_debug(self._buf[:idx]) self._buf = self._buf[idx:] plen = self._buf[8] total = 11 + plen if len(self._buf) < total: return None frame = self._buf[:total] cmd = frame[7] payload = frame[9:9 + plen] if crc16_modbus(frame[:-2]) == (frame[-2] | (frame[-1] << 8)): self._buf = self._buf[total:] return (cmd, payload) # 误判的帧头，丢弃继续寻找 self._buf = self._buf[1:] return None def _handle(self, cmd, payload): self._timer.start(5000) if cmd == CMD_ACK: self._log("收到从机应答(0x01)", "ok") self.status_lbl.setText("设备就绪") return if cmd == CMD_REQ_PKG: if len(payload) < 4: self._log("从机请求包帧格式错误", "err") return idx, total = struct.unpack('= self._pkg_num: self._log(f"从机请求包序号越界({idx})，视为完成", "ok") self._finish() return start = idx * UPG_PACKET_DATA data = self._fw_data[start:start + UPG_PACKET_DATA] p = struct.pack('= self._pkg_num: self._finish() return self._log(f"收到其他帧 Cmd={cmd:#04x}", "err") def _finish(self): self.running = False self._timer.stop() self.start_btn.setEnabled(True) if len(self._sent) >= self._pkg_num: self._log(f"升级完成 (已发送全部 {len(self._sent)} 包)", "ok") self.progress.setValue(100) self.status_lbl.setText("完成") else: self._log(f"升级未完成，已发送 {len(self._sent)}/{self._pkg_num} 包", "err") def _on_timeout(self): if not self.running: return self.running = False self._timer.stop() self.start_btn.setEnabled(True) self._log("升级超时(5秒无响应)", "err") self.status_lbl.setText("超时") def closeEvent(self, event): self._reset_ui() event.accept() class DeviceDialog(QDialog): """设备管理对话框：通过 culs 查询已注册通讯单元，cudel 删除设备。""" def __init__(self, main_window): super().__init__(main_window) self.main = main_window self.setWindowTitle("设备管理 - 查询/删除") self.setModal(False) self.resize(860, 520) self._buf = "" self._pending_del = None # 等待删除结果的MAC，"all"表示删除全部 self._quiet = QTimer(self) self._quiet.setSingleShot(True) self._quiet.timeout.connect(self._on_quiet) root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(6) btn_row = QHBoxLayout() self.btn_refresh = QPushButton("刷新查询") self.btn_refresh.clicked.connect(self.query) btn_row.addWidget(self.btn_refresh) self.btn_del_all = QPushButton("删除全部设备") self.btn_del_all.setStyleSheet("color:#D32F2F; font-weight:bold;") self.btn_del_all.clicked.connect(self._del_all) btn_row.addWidget(self.btn_del_all) btn_close = QPushButton("关闭") btn_close.clicked.connect(self.close) btn_row.addWidget(btn_close) btn_row.addStretch() self.status_lbl = QLabel("就绪") btn_row.addWidget(self.status_lbl) root.addLayout(btn_row) self.table = QTableWidget(0, 11) self.table.setHorizontalHeaderLabels( ["DevType", "设备MAC", "传感器类型", "Laser", "电量", "RSSI", "SNR", "FPO", "LpCfg", "在线", "操作"]) self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents) self.table.horizontalHeader().setStretchLastSection(True) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table.setSelectionMode(QAbstractItemView.SingleSelection) self.table.verticalHeader().setVisible(False) root.addWidget(self.table, 1) def _serial_ok(self): if not self.main.serial_port or not self.main.serial_port.is_open: QMessageBox.warning(self, "错误", "请先打开串口！") return False return True # ── 查询 ── def query(self): if not self._serial_ok(): return self._buf = "" self._pending_del = None self._quiet.stop() self.status_lbl.setText("查询中...") self.main.send_command("culs") self._quiet.start(800) def feed_text(self, text): self._buf += text self._quiet.start(600) def _on_quiet(self): if self._pending_del is not None: buf = self._buf if "deleted" in buf.lower(): self.status_lbl.setText(f"删除成功 ({self._pending_del})") elif "not registered" in buf: self.status_lbl.setText("删除失败：设备未注册") elif "Para Error" in buf: self.status_lbl.setText("删除失败：指令参数错误") else: self.status_lbl.setText("删除失败：设备无响应") self._pending_del = None self.query() return self._parse_devices() def _parse_devices(self): devices = [] cur = None for raw in self._buf.replace('\r', '').split('\n'): line = raw.strip() # Lora通讯单元：DevType: XXXX, SensorNum: N m = re.match(r'DevType:\s*([0-9A-Fa-f]{4}),\s*SensorNum:\s*(\d+)', line) if m: cur = {"type": m.group(1).upper(), "builtin": False, "macs": [], "types": [], "bat": "", "rssi": "", "snr": "", "online": ""} devices.append(cur) continue # RS485内置通讯单元：DevType: XXXX（单独一行） m = re.match(r'DevType:\s*([0-9A-Fa-f]{4})$', line) if m: cur = {"type": m.group(1).upper(), "builtin": True, "macs": [], "types": [], "bat": "", "rssi": "", "snr": "", "online": ""} devices.append(cur) continue if cur is None: continue m = re.match(r'DevMac:\s*([0-9A-Fa-f]{2}(?:-[0-9A-Fa-f]{2}){5})', line) if m: cur["macs"].append(("Master", m.group(1).upper())) continue m = re.match( r'(Master|Slave)\s*Mac:\s*([0-9A-Fa-f]{2}(?:-[0-9A-Fa-f]{2}){5})' r'(?:\s*Type:\s*([0-9A-Fa-f]{4}))?' r'(?:\s*Laser:\s*(On|Off))?', line) if m: cur["macs"].append((m.group(1), m.group(2).upper())) if m.group(3): cur["types"].append(m.group(3).upper()) if m.group(4): cur["laser"] = m.group(4) continue m = re.match(r'Bat:(\d+)%\s*Rssi:(-?\d+)\s*Snr:(-?\d+)', line) if m: cur["bat"], cur["rssi"], cur["snr"] = m.group(1), m.group(2), m.group(3) continue # FPO Start:08:00, Dur:8h, End:16:00, FullPower m = re.match(r'FPO\s+Start:(\d+):00,\s*Dur:(\d+)h,\s*End:(\d+):00,\s*(\w+)', line) if m: cur["fpo"] = f"{m.group(1)}:00-{m.group(3)}:00 {m.group(2)}h {m.group(4)}" continue # LpCfg: CI:300sec, RI:20min, EI:2min, ET:20min m = re.match(r'LpCfg:\s*CI:(\d+)sec,\s*RI:(\d+)min,\s*EI:(\d+)min,\s*ET:(\d+)min', line) if m: cur["lpcfg"] = (f"CI:{m.group(1)}s RI:{m.group(2)}m " f"EI:{m.group(3)}m ET:{m.group(4)}m") continue m = re.match(r'Online:(Yes|No)', line) if m: cur["online"] = m.group(1) self._fill_table(devices) self._buf = "" def _fill_table(self, devices): self.table.setRowCount(0) for dev in devices: row = self.table.rowCount() self.table.insertRow(row) type_str = dev["type"] + ("(内置)" if dev["builtin"] else "") macs = "\n".join(f"{role}: {mac}" for role, mac in dev["macs"]) types = " ".join(dev["types"]) if dev["types"] else "-" online = dev["online"] or "-" vals = [type_str, macs or "-", types, dev.get("laser", "") or "-", f'{dev["bat"]}%' if dev["bat"] else "-", dev["rssi"] or "-", dev["snr"] or "-", dev.get("fpo", "") or "-", dev.get("lpcfg", "") or "-", online] for col, v in enumerate(vals): item = QTableWidgetItem(v) if col == 9 and online in ("Yes", "No"): item.setForeground(Qt.green if online == "Yes" else Qt.red) self.table.setItem(row, col, item) master = next((mac for role, mac in dev["macs"] if role == "Master"), None) if master and not dev["builtin"]: btn = QPushButton("删除") btn.setStyleSheet("color:#D32F2F;") btn.clicked.connect(lambda _, m=master: self._del_one(m)) self.table.setCellWidget(row, 10, btn) else: self.table.setItem(row, 10, QTableWidgetItem("-" if dev["builtin"] else "")) self.status_lbl.setText(f"共 {len(devices)} 个设备") # ── 删除 ── def _del_one(self, mac): if not self._serial_ok(): return ret = QMessageBox.question(self, "确认删除", f"确定删除设备 {mac} 吗？", QMessageBox.Yes | QMessageBox.No) if ret != QMessageBox.Yes: return self._buf = "" self._pending_del = mac self.status_lbl.setText(f"正在删除 {mac} ...") self.main.send_command(f"cudel {mac}") self._quiet.start(800) def _del_all(self): if not self._serial_ok(): return ret = QMessageBox.question(self, "确认删除", "确定删除全部已注册设备吗？", QMessageBox.Yes | QMessageBox.No) if ret != QMessageBox.Yes: return self._buf = "" self._pending_del = "all" self.status_lbl.setText("正在删除全部设备 ...") self.main.send_command("cudel all") self._quiet.start(800) def closeEvent(self, event): self._quiet.stop() event.accept() def main(): app = QApplication(sys.argv) app.setStyle("Fusion") palette = QPalette() palette.setColor(QPalette.Window, QColor(240, 240, 240)) palette.setColor(QPalette.WindowText, QColor(30, 30, 30)) palette.setColor(QPalette.Base, QColor(255, 255, 255)) palette.setColor(QPalette.Button, QColor(225, 225, 225)) app.setPalette(palette) window = GatewayConfigTool() window.show() sys.exit(app.exec_()) if __name__ == "__main__": main()