YuanHongbin/GateWay_ConfigureAPP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d325910bfe8976f8ba36db6467f3a716ad61a0f2
GateWay_ConfigureAPP/gateway_config.py
T
YuanHongbin d325910bfe feat(ota): 新增TCP(Cat1)升级通道(hex收发+窗口流控+断点续传) 
- TCP服务端监听(默认8080)等待网关Cat1模组连入, 协议复用0x7D帧
- 升级帧转ASCII hex文本下发, 上行hex流式解码(兼容旧格式ACK)
- 窗口流控: 收到0x02请求后连发8包/批; 断点续传ACK扩展{Status,ResumedIdx}
- 握手阶段超时重发0x81(修复原重发数据批次无效问题)
- 版本V1.3.0(宽带调试版, 含Cat1远程升级)
2026-09-24 17:14:17 +08:00

1329 lines
50 KiB
Python
Raw Blame History

"""
网关配置工具
串口配置：波特率500000(可修改)，数据位8，停止位1，校验位None
"""
import sys
import os
import re
import time
import struct
import zlib
import threading
import socket
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('<H', crc)
def parse_mac(s):
"""'81-00-00-06-00-00' -> 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 TcpServerThread(QThread):
"""TCP服务线程: 监听端口, 等待网关Cat1模组(AT+LIPOPEN)主动连入后透传升级帧。"""
client_connected = pyqtSignal(str)
client_disconnected = pyqtSignal()
data_received_raw = pyqtSignal(bytes)
listen_error = pyqtSignal(str)
def __init__(self, port):
super().__init__()
self.port = port
self.running = True
self._hexbuf = b"" # 设备上行ASCII hex文本的奇数尾巴缓冲
self.sock = None # 监听socket
self.conn = None # 已连接的客户端(网关模组)
self.conn_lock = threading.Lock()
def run(self):
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind(('0.0.0.0', self.port))
self.sock.listen(1)
self.sock.settimeout(0.5)
except Exception as e:
self.listen_error.emit(str(e))
return
while self.running:
try:
conn, addr = self.sock.accept()
except socket.timeout:
continue
except OSError:
break
conn.settimeout(0.5)
try:
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except Exception:
pass
with self.conn_lock:
self.conn = conn
self.client_connected.emit(addr[0])
while self.running:
try:
data = conn.recv(4096)
except socket.timeout:
continue
except OSError:
break
if not data:
break
self.data_received_raw.emit(self._decode_hex(data))
with self.conn_lock:
try:
conn.close()
except Exception:
pass
self.conn = None
self._hexbuf = b""
self.client_disconnected.emit()
try:
self.sock.close()
except Exception:
pass
def _decode_hex(self, data):
"""设备上行是ASCII hex文本(LIPSEND hex-ascii), 流式解码为二进制(保留奇数尾巴)。"""
buf = self._hexbuf + data
n = len(buf) - (len(buf) % 2)
self._hexbuf = buf[n:]
try:
return bytes.fromhex(buf[:n].decode('ascii'))
except ValueError:
self._hexbuf = b"" # 非法字符: 丢弃重同步
return b""
def send_data(self, data):
with self.conn_lock:
if not self.conn:
return False
try:
self.conn.sendall(data)
return True
except Exception:
return False
def is_client_connected(self):
with self.conn_lock:
return self.conn is not None
def stop(self):
self.running = False
self.wait(3000)
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("网关配置工具(宽带版 V1.3.0)")
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 信道：433.1MHz ~ 492.6MHz，步进0.5MHz，共120信道（信道0起始）
lora_freqs = []
fc_hz = 433100000
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('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
def log_send(self, text):
self.text_log.append(f'<span style="color:#569CD6;">{self._escape_html(text)}</span>')
def log_recv(self, text):
clean = self._escape_html(text).replace('\r', '').replace('\n', '<br>')
self.text_log.append(f'<span style="color:#CE9178;">&lt;&lt;&lt; {clean}</span>')
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 (串口/Cat1 TCP)")
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.channel = 'serial' # 升级通道: serial / tcp
self.tcp_thread = None
self._window = 8 # TCP模式每批包数(窗口流控)
self._window_start = 0 # 当前批次首包序号(超时重发用)
self._retry_cnt = 0 # TCP模式超时重发计数
self._resume_base = 0 # 断点续传基数(设备已收包数)
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)
# 通道选择 + TCP监听
cfg_layout.addWidget(QLabel("通道:"), 2, 0)
self.combo_channel = QComboBox()
self.combo_channel.addItems(["串口", "TCP(Cat1)"])
self.combo_channel.currentIndexChanged.connect(self._on_channel_changed)
cfg_layout.addWidget(self.combo_channel, 2, 1)
tcp_row = QHBoxLayout()
self.tcp_lbl = QLabel("TCP端口:")
tcp_row.addWidget(self.tcp_lbl)
self.tcp_port_edit = QLineEdit("8080")
self.tcp_port_edit.setFixedWidth(80)
tcp_row.addWidget(self.tcp_port_edit)
self.btn_tcp_listen = QPushButton("开始监听")
self.btn_tcp_listen.setFixedWidth(90)
self.btn_tcp_listen.clicked.connect(self._toggle_tcp_listen)
tcp_row.addWidget(self.btn_tcp_listen)
self.tcp_status_lbl = QLabel("未监听")
self.tcp_status_lbl.setStyleSheet("color:#909090;")
tcp_row.addWidget(self.tcp_status_lbl)
tcp_row.addStretch()
cfg_layout.addLayout(tcp_row, 2, 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, 3, 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, 4, 0, 1, 2)
self.status_lbl = QLabel("就绪")
cfg_layout.addWidget(self.status_lbl, 4, 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._window_start = 0
self._retry_cnt = 0
self._resume_base = 0
self.start_btn.setEnabled(True)
self.progress.setValue(0)
self.status_lbl.setText("就绪")
def _on_channel_changed(self, idx):
self.channel = 'serial' if idx == 0 else 'tcp'
tcp_on = (self.channel == 'tcp')
self.tcp_lbl.setEnabled(tcp_on)
self.tcp_port_edit.setEnabled(tcp_on)
self.btn_tcp_listen.setEnabled(tcp_on)
def _toggle_tcp_listen(self):
if self.tcp_thread and self.tcp_thread.isRunning():
self.tcp_thread.stop()
self.tcp_thread = None
self.btn_tcp_listen.setText("开始监听")
self.tcp_status_lbl.setText("未监听")
self.tcp_status_lbl.setStyleSheet("color:#909090;")
return
try:
port = int(self.tcp_port_edit.text().strip())
assert 1 <= port <= 65535
except Exception:
QMessageBox.warning(self, "提示", "TCP端口无效")
return
self.tcp_thread = TcpServerThread(port)
self.tcp_thread.client_connected.connect(self._on_tcp_client)
self.tcp_thread.client_disconnected.connect(self._on_tcp_disconnect)
self.tcp_thread.data_received_raw.connect(self.feed_data)
self.tcp_thread.listen_error.connect(self._on_tcp_error)
self.tcp_thread.start()
self.btn_tcp_listen.setText("停止监听")
self.tcp_status_lbl.setText(f"监听中(0.0.0.0:{port})")
self.tcp_status_lbl.setStyleSheet("color:#1565C0;")
def _on_tcp_client(self, ip):
self.tcp_status_lbl.setText(f"已连接: {ip}")
self.tcp_status_lbl.setStyleSheet("color:#2E7D32;")
def _on_tcp_disconnect(self):
if self.tcp_thread and self.tcp_thread.isRunning():
self.tcp_status_lbl.setText("客户端断开,等待重连...")
self.tcp_status_lbl.setStyleSheet("color:#D32F2F;")
if self.running and self.channel == 'tcp':
self._log("TCP连接断开,等待网关模组重连...", "err")
def _on_tcp_error(self, msg):
self.tcp_status_lbl.setText("监听失败")
self.tcp_status_lbl.setStyleSheet("color:#D32F2F;")
self._log(f"TCP监听失败: {msg}", "err")
if self.tcp_thread:
self.tcp_thread.stop()
self.tcp_thread = None
self.btn_tcp_listen.setText("开始监听")
def _send(self, frame):
"""按当前通道发送升级帧(TCP通道按协议约定发ASCII hex文本, 设备端AsciiToHex还原)。"""
if self.channel == 'tcp':
if not (self.tcp_thread and self.tcp_thread.is_client_connected()):
self._log("TCP未连接,发送失败", "err")
return False
return self.tcp_thread.send_data(frame.hex().upper().encode('ascii'))
return self.main.send_raw(frame)
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'<span style="color:{color};">{GatewayConfigTool._escape_html(str(msg))}</span>')
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 self.channel == 'tcp':
if not (self.tcp_thread and self.tcp_thread.is_client_connected()):
QMessageBox.warning(self, "提示", "TCP未监听或网关未连接, 请先监听并等待网关接入")
return
elif 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
if len(fw) > 232 * 1024:
QMessageBox.warning(self, "提示", f"固件大小{len(fw)}字节超出A区上限(237568字节), 无法升级")
return
self._mac = mac
self._fw_data = fw
self._got_ack = False
self._announce_frame = None
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._window_start = 0
self._retry_cnt = 0
self._resume_base = 0
self.status_lbl.setText("等待握手...")
app_crc = zlib.crc32(fw) & 0xFFFFFFFF
payload = struct.pack('<H', self._pkg_num) + struct.pack('<I', len(fw)) + struct.pack('<I', app_crc)
frame = build_frame(mac, CMD_ANNOUNCE, payload)
self._log(f"下发APP信息: 包数={self._pkg_num}, 大小={len(fw)}字节, CRC32=0x{app_crc:08X}")
self._announce_frame = frame
self._send(frame)
self._log("等待设备握手后接收数据请求...")
self._timer.start(10000 if self.channel == 'tcp' else 5000)
def feed_data(self, data):
if not self.running:
return
self._buf += data
while True:
frame = self._try_next_frame()
if frame is None:
break
cmd, payload = frame
self._handle(cmd, payload)
def _try_next_frame(self):
while len(self._buf) >= 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(3000 if self.channel == 'tcp' else 5000)
self._retry_cnt = 0
if cmd == CMD_ACK:
self._got_ack = True
# 兼容旧格式(1字节0x20)与新格式{Status, ResumedIdx}: 0x20=全新, 0x21=断点续传
if len(payload) >= 3:
status, resumed = struct.unpack('<BH', payload[0:3])
if status == 0x21 and resumed > 0:
self._resume_base = resumed
self._log(f"设备断点续传: 已接收 {resumed} 包, 从断点继续", "ok")
else:
self._resume_base = 0
self._log("收到从机应答(0x01), 全新升级", "ok")
else:
self._resume_base = 0
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('<HH', payload[0:4])
if idx >= self._pkg_num:
self._log(f"从机请求包序号越界({idx})，视为完成", "ok")
self._finish()
return
batch = self._window if self.channel == 'tcp' else 1
self._window_start = idx
self._send_window(idx, batch)
return
self._log(f"收到其他帧 Cmd={cmd:#04x}", "err")
def _send_window(self, start, batch):
"""从start包起连发batch包(TCP窗口流控; 串口停等batch=1)。"""
for k in range(batch):
idx = start + k
if idx >= self._pkg_num:
break
s = idx * UPG_PACKET_DATA
data = self._fw_data[s:s + UPG_PACKET_DATA]
p = struct.pack('<HHB', idx, self._pkg_num, len(data)) + data
if not self._send(build_frame(self._mac, CMD_SEND_PKG, p)):
self._log("发送失败, 通道异常", "err")
return
self._sent.add(idx)
done = max(self._sent) + 1
self.progress.setValue(min(int(done * 100.0 / self._pkg_num), 100))
self.status_lbl.setText(f"{done}/{self._pkg_num} 包")
if len(self._sent) >= self._pkg_num:
self._finish()
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
if self.channel == 'tcp' and not self._got_ack and self._retry_cnt < 5:
# 握手阶段(未收到ACK): 重发0x81升级请求而非数据批次
self._retry_cnt += 1
self._log(f"握手超时, 重发升级请求({self._retry_cnt}/5)", "err")
self._timer.start(10000)
self._send(self._announce_frame)
return
if self.channel == 'tcp' and self._retry_cnt < 5:
# Cat1链路抖动: 重发当前批次(设备按期望包号丢弃重复包)
self._retry_cnt += 1
self._log(f"超时, 从第{self._window_start}包起重发批次({self._retry_cnt}/5)", "err")
self._timer.start(3000)
self._send_window(self._window_start, self._window)
return
self.running = False
self._timer.stop()
self.start_btn.setEnabled(True)
self._log("升级超时(无响应)", "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._parsed = False # 本次查询是否已完成解析（防止数据流中途误触发）
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._parsed = False
self._quiet.stop()
self.table.setRowCount(0) # 清空旧数据，避免新旧混淆
self.status_lbl.setText("查询中...")
self.main.send_command("culs")
self._quiet.start(1200) # 首次等待稍长，给设备响应时间
def feed_text(self, text):
# 已完成本次解析后，忽略后续到来的残留数据，直到下一次手动查询
if self._parsed:
return
self._buf += text
self._quiet.start(600) # 数据流仍在到达，重置静默定时器
def _remove_row_by_mac(self, mac):
"""从表格中移除指定MAC的设备行, 返回是否移除成功。"""
mac_u = mac.upper()
for row in range(self.table.rowCount()):
item = self.table.item(row, 1)
if item and mac_u in item.text():
self.table.removeRow(row)
return True
return False
def _on_quiet(self):
# 仅解析一次：避免数据流间歇触发的中途解析把表格清零
if self._parsed:
return
self._parsed = True
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)
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
# 网关cudel无应答, 直接本地移除该行
self.main.send_command(f"cudel {mac}")
if self._remove_row_by_mac(mac):
self.status_lbl.setText(f"已删除 {mac} (共 {self.table.rowCount()} 个设备)")
else:
self.status_lbl.setText(f"删除命令已发送 ({mac})")
def _del_all(self):
if not self._serial_ok():
return
ret = QMessageBox.question(self, "确认删除", "确定删除全部已注册设备吗？",
QMessageBox.Yes | QMessageBox.No)
if ret != QMessageBox.Yes:
return
# 网关cudel无应答, 直接清空表格
self.main.send_command("cudel all")
self.table.setRowCount(0)
self.status_lbl.setText("已删除全部设备")
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()
Reference in New Issue View Git Blame Copy Permalink