""" 网关配置工具 串口配置:波特率500000,数据位8,停止位1,校验位None """ import sys import os import re import time import threading from datetime import datetime from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QComboBox, QPushButton, QLineEdit, QTextEdit, QMessageBox, QGridLayout, QSplitter ) from PyQt5.QtCore import Qt, QThread, pyqtSignal, QRegExp from PyQt5.QtGui import QFont, QColor, QPalette, QRegExpValidator, QIntValidator, 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) class SerialThread(QThread): data_received = pyqtSignal(str) def __init__(self, ser): super().__init__() self.ser = ser self.running = True self._buffer = "" 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) try: text = data.decode('utf-8', errors='replace') except Exception: text = data.decode('gbk', errors='replace') 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 self._cur_comm = "Cat1" # 当前通讯方式: Cat1 或 Eth self.init_ui() self.refresh_ports() def init_ui(self): self.setWindowTitle("网关配置工具") self.setWindowIcon(QIcon(resource_path("物联网网关.png"))) self.setMinimumSize(1000, 850) self.resize(1000, 850) 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("波特率: 500000")) 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) # === 配置区 + 日志区(上下布局)=== splitter = QSplitter(Qt.Vertical) # -- 上方:配置区 -- config_widget = QWidget() config_layout = QVBoxLayout(config_widget) config_layout.setContentsMargins(0, 0, 0, 0) config_layout.setSpacing(2) # 基本指令 + 通讯方式(一行) top_row = QHBoxLayout() top_row.setSpacing(4) self.btn_para = QPushButton("查看配置(para)") self.btn_para.clicked.connect(self.on_para_clicked) top_row.addWidget(self.btn_para) self.btn_cat1 = QPushButton("Cat1(comm cat1)") self.btn_cat1.clicked.connect(lambda: self.set_comm_mode("Cat1")) top_row.addWidget(self.btn_cat1) self.btn_eth = QPushButton("ETH(comm eth)") self.btn_eth.clicked.connect(lambda: self.set_comm_mode("Eth")) top_row.addWidget(self.btn_eth) top_row.addStretch() config_layout.addLayout(top_row) # 当前配置显示 info_group = QGroupBox("当前网关配置") info_grid = QGridLayout(info_group) info_grid.setSpacing(4) def _bold_label(color="#1565C0"): lb = QLabel("-") lb.setStyleSheet(f"font-weight:bold; color:{color};") return lb row = 0 info_grid.addWidget(QLabel("软件版本:"), row, 0) self.label_sw_ver = _bold_label() info_grid.addWidget(self.label_sw_ver, row, 1) info_grid.addWidget(QLabel("硬件版本:"), row, 2) self.label_hw_ver = _bold_label() info_grid.addWidget(self.label_hw_ver, row, 3) row = 1 info_grid.addWidget(QLabel("电池:"), row, 0) self.label_battery = _bold_label() info_grid.addWidget(self.label_battery, row, 1) info_grid.addWidget(QLabel("注册状态:"), row, 2) self.label_reg_status = _bold_label() info_grid.addWidget(self.label_reg_status, row, 3) row = 2 info_grid.addWidget(QLabel("网关MAC:"), row, 0) self.label_gw_mac = _bold_label() info_grid.addWidget(self.label_gw_mac, row, 1, 1, 3) row = 3 info_grid.addWidget(QLabel("通讯方式:"), row, 0) self.label_cur_comm = _bold_label() info_grid.addWidget(self.label_cur_comm, row, 1) info_grid.addWidget(QLabel("读取间隔:"), row, 2) self.label_read_interval = _bold_label() info_grid.addWidget(self.label_read_interval, row, 3) row = 4 info_grid.addWidget(QLabel("Cat1 IMEI:"), row, 0) self.label_imei = _bold_label() info_grid.addWidget(self.label_imei, row, 1, 1, 3) row = 5 info_grid.addWidget(QLabel("Cat1 SIM:"), row, 0) self.label_sim = _bold_label() info_grid.addWidget(self.label_sim, row, 1, 1, 3) row = 6 info_grid.addWidget(QLabel("Lora频率:"), row, 0) self.label_cur_lora_fc = _bold_label() info_grid.addWidget(self.label_cur_lora_fc, row, 1) # Eth参数(紧凑两列) eth_info_group = QGroupBox("ETH / Cat1网络参数") eth_info_grid = QGridLayout(eth_info_group) eth_info_grid.setSpacing(4) row = 0 eth_info_grid.addWidget(QLabel("网关IP:"), row, 0) self.label_cur_eth_ip = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_ip, row, 1) eth_info_grid.addWidget(QLabel("网关端口:"), row, 2) self.label_cur_eth_port = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_port, row, 3) row = 1 eth_info_grid.addWidget(QLabel("子网掩码:"), row, 0) self.label_cur_eth_mask = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_mask, row, 1) eth_info_grid.addWidget(QLabel("服务器网关:"), row, 2) self.label_cur_eth_gw = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_gw, row, 3) row = 2 eth_info_grid.addWidget(QLabel("服务器IP:"), row, 0) self.label_cur_eth_dip = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_dip, row, 1) eth_info_grid.addWidget(QLabel("服务器端口:"), row, 2) self.label_cur_eth_dport = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_dport, row, 3) row = 3 eth_info_grid.addWidget(QLabel("IP模式:"), row, 0) self.label_cur_eth_ipmode = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_ipmode, row, 1) eth_info_grid.addWidget(QLabel("工作模式:"), row, 2) self.label_cur_eth_workmode = _bold_label() eth_info_grid.addWidget(self.label_cur_eth_workmode, row, 3) # 低功耗配置 misc_group = QGroupBox("低功耗配置") misc_grid = QGridLayout(misc_group) misc_grid.setSpacing(2) misc_grid.setContentsMargins(10, 4, 4, 4) row = 0 misc_grid.addWidget(QLabel("采集间隔:"), row, 0) self.label_lpcfg_ci = _bold_label() misc_grid.addWidget(self.label_lpcfg_ci, row, 1) misc_grid.addWidget(QLabel("上报间隔:"), row, 2) self.label_lpcfg_ri = _bold_label() misc_grid.addWidget(self.label_lpcfg_ri, row, 3) row = 1 misc_grid.addWidget(QLabel("紧急上报间隔:"), row, 0) self.label_lpcfg_ei = _bold_label() misc_grid.addWidget(self.label_lpcfg_ei, row, 1) misc_grid.addWidget(QLabel("紧急上报时长:"), row, 2) self.label_lpcfg_et = _bold_label() misc_grid.addWidget(self.label_lpcfg_et, row, 3) config_layout.addWidget(info_group) config_layout.addWidget(eth_info_group) config_layout.addWidget(misc_group) # Eth参数(紧凑两列) eth_group = QGroupBox("ETH参数(需先切换到eth模式)") eth_grid = QGridLayout(eth_group) eth_grid.setSpacing(4) ip_validator = QRegExpValidator(QRegExp(r"^(\d{1,3}\.){3}\d{1,3}$")) eth_grid.addWidget(QLabel("网关IP:"), 0, 0) self.edit_eth_ip = QLineEdit("192.168.1.101") self.edit_eth_ip.setValidator(ip_validator) eth_grid.addWidget(self.edit_eth_ip, 0, 1) self.btn_eth_ip = QPushButton("发送") self.btn_eth_ip.setFixedWidth(50) self.btn_eth_ip.clicked.connect(lambda: self.send_param("eth ip", self.edit_eth_ip)) eth_grid.addWidget(self.btn_eth_ip, 0, 2) eth_grid.addWidget(QLabel("网关端口:"), 0, 3) self.edit_eth_port = QLineEdit("10000") self.edit_eth_port.setValidator(QIntValidator(1, 65535)) eth_grid.addWidget(self.edit_eth_port, 0, 4) self.btn_eth_port = QPushButton("发送") self.btn_eth_port.setFixedWidth(50) self.btn_eth_port.clicked.connect(lambda: self.send_param("eth port", self.edit_eth_port)) eth_grid.addWidget(self.btn_eth_port, 0, 5) eth_grid.addWidget(QLabel("服务器网关:"), 1, 0) self.edit_eth_gw = QLineEdit("192.168.1.1") self.edit_eth_gw.setValidator(ip_validator) eth_grid.addWidget(self.edit_eth_gw, 1, 1) self.btn_eth_gw = QPushButton("发送") self.btn_eth_gw.setFixedWidth(50) self.btn_eth_gw.clicked.connect(lambda: self.send_param("eth gw", self.edit_eth_gw)) eth_grid.addWidget(self.btn_eth_gw, 1, 2) eth_grid.addWidget(QLabel("服务器IP:"), 1, 3) self.edit_eth_dip = QLineEdit("192.168.1.100") self.edit_eth_dip.setValidator(ip_validator) eth_grid.addWidget(self.edit_eth_dip, 1, 4) self.btn_eth_dip = QPushButton("发送") self.btn_eth_dip.setFixedWidth(50) self.btn_eth_dip.clicked.connect(lambda: self.send_param("eth dip", self.edit_eth_dip)) eth_grid.addWidget(self.btn_eth_dip, 1, 5) eth_grid.addWidget(QLabel("服务器端口:"), 2, 0) self.edit_eth_dport = QLineEdit("8080") self.edit_eth_dport.setValidator(QIntValidator(1, 65535)) eth_grid.addWidget(self.edit_eth_dport, 2, 1) self.btn_eth_dport = QPushButton("发送") self.btn_eth_dport.setFixedWidth(50) self.btn_eth_dport.clicked.connect(lambda: self.send_param("eth dport", self.edit_eth_dport)) eth_grid.addWidget(self.btn_eth_dport, 2, 2) self.btn_eth_apply = QPushButton("应用ETH配置 (eth apply)") self.btn_eth_apply.setStyleSheet("background-color:#4CAF50; color:white; font-weight:bold; padding:4px;") self.btn_eth_apply.clicked.connect(self.send_eth_apply) eth_grid.addWidget(self.btn_eth_apply, 2, 3, 1, 3) config_layout.addWidget(eth_group) # Lora信道配置(独立区块,类似ETH参数) lora_group = QGroupBox("Lora信道配置") lora_layout = QHBoxLayout(lora_group) lora_layout.setSpacing(6) lora_layout.addWidget(QLabel("信道:")) self.combo_lora_fc = QComboBox() self.combo_lora_fc.setFixedWidth(200) lora_freqs = [] fc_hz = 410100000 ch_idx = 1 while fc_hz < 525000000: mhz = fc_hz / 1000000 lora_freqs.append((str(fc_hz), f"信道{ch_idx}: {mhz:.1f}MHz")) fc_hz += 200000 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_layout.addStretch() config_layout.addWidget(lora_group) # -- 下方:日志区 + 自定义指令 -- log_widget = QWidget() log_layout = QHBoxLayout(log_widget) log_layout.setContentsMargins(0, 0, 0, 0) log_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() log_layout.addWidget(log_group, 4) log_layout.addWidget(custom_group, 1) splitter.addWidget(config_widget) splitter.addWidget(log_widget) splitter.setSizes([280, 220]) main_layout.addWidget(splitter) 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: self.serial_port = serial.Serial( port=port_name, baudrate=500000, 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.start() self.btn_connect.setText("关闭串口") self.btn_connect.setStyleSheet("background-color:#f44336; color:white; font-weight:bold;") self.log_send(f"已连接 {port_name} @ 500000") 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") 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 set_comm_mode(self, mode): self._cur_comm = mode self.send_command(f"comm {mode.lower()}") def send_param(self, prefix, edit_widget): if prefix.startswith("eth") and self._cur_comm != "Eth": QMessageBox.warning(self, "提示", "当前为Cat1通讯模式,请先切换到ETH模式!") return value = edit_widget.text().strip() if not value: QMessageBox.warning(self, "错误", "参数值不能为空!") return cmd = f"{prefix} {value}" self.send_command(cmd) def send_eth_apply(self): if self._cur_comm != "Eth": QMessageBox.warning(self, "提示", "当前为Cat1通讯模式,请先切换到ETH模式!") return self.send_command("eth apply") 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): self._recv_buffer += text self.log_recv(text) if self._para_mode: self._try_parse_para() 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 == 'CommInf': _update(self.label_cur_comm, val) self._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_read_interval, f"{val}s") elif key == 'Register Status': status_map = {"0": "未注册", "1": "已注册"} _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 # --- EthNet section --- elif key == 'IpMode': _update(self.label_cur_eth_ipmode, val) elif key == 'LocalIP': _update(self.label_cur_eth_ip, val) self.edit_eth_ip.setText(val) elif key == 'LocalPort' and 'Eth' in section: _update(self.label_cur_eth_port, val) self.edit_eth_port.setText(val) elif key == 'WorkMode': _update(self.label_cur_eth_workmode, val) elif key == 'SubnetMask': _update(self.label_cur_eth_mask, val) elif key == 'Gateway': _update(self.label_cur_eth_gw, val) self.edit_eth_gw.setText(val) elif key == 'DestIP': _update(self.label_cur_eth_dip, val) self.edit_eth_dip.setText(val) elif key == 'DestPort' and 'Eth' in section: _update(self.label_cur_eth_dport, val) self.edit_eth_dport.setText(val) # --- 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) elif key == 'LpCfg': def _conv_unit(u): if u == 'sec': return '秒' if u == 'min': return '分钟' return u ci_m = re.search(r'CI:(\d+)(\w+)', val) ri_m = re.search(r'RI:(\d+)(\w+)', val) ei_m = re.search(r'EI:(\d+)(\w+)', val) et_m = re.search(r'ET:(\d+)(\w+)', val) if ci_m: _update(self.label_lpcfg_ci, f"{ci_m.group(1)}{_conv_unit(ci_m.group(2))}") if ri_m: _update(self.label_lpcfg_ri, f"{ri_m.group(1)}{_conv_unit(ri_m.group(2))}") if ei_m: _update(self.label_lpcfg_ei, f"{ei_m.group(1)}{_conv_unit(ei_m.group(2))}") if et_m: _update(self.label_lpcfg_et, f"{et_m.group(1)}{_conv_unit(et_m.group(2))}") self._para_mode = False self._recv_buffer = "" self.statusBar().showMessage("配置读取完成") def log_send(self, text): timestamp = datetime.now().strftime("%H:%M:%S") self.text_log.append(f'[{timestamp}] {text}') def log_recv(self, text): timestamp = datetime.now().strftime("%H:%M:%S") clean = text.replace('\r', '').replace('\n', '
') self.text_log.append(f'[{timestamp}] <<< {clean}') def closeEvent(self, event): self.close_serial() 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()