#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 网络调试助手 支持：串口、TCP(客户端/服务器)、UDP(客户端/服务器) 功能：收发数据（文本/十六进制）、定时发送、时间戳、清空日志 """ import sys import os import json import time import struct import copy import threading import socket from datetime import datetime import serial import serial.tools.list_ports from PySide6.QtCore import Qt, QTimer, Signal, QSize, QRect, QPoint, QEvent from PySide6.QtGui import QFont, QColor, QTextCharFormat, QTextCursor, QIcon, QPalette from PySide6.QtWidgets import ( QApplication, QMainWindow, QWidget, QTabWidget, QVBoxLayout, QHBoxLayout, QGridLayout, QLabel, QComboBox, QLineEdit, QPushButton, QDialog, QPlainTextEdit, QCheckBox, QSpinBox, QSplitter, QMessageBox, QTableWidget, QTableWidgetItem, QListWidget, QListWidgetItem, QGroupBox, QRadioButton, QHeaderView, QFileDialog, QWidgetItem, QLayout, QToolBar, QMenu, QInputDialog, QSystemTrayIcon, ) STAMP_OPTS = ['无', 'ASCII(YYYY-MM-DD HH:MM:SS)', 'ASCII(HH:MM:SS)', 'Unix秒(4字节小端)', 'Unix秒(4字节大端)', 'Unix毫秒(8字节小端)', 'Unix毫秒(8字节大端)', 'BCD(YYMMDDHHMMSS)'] CRC_OPTS = ['无', 'CRC16-MODBUS', 'CRC16-CCITT(XModem)', 'CRC16-IBM', 'CRC32', 'SUM8', 'XOR8'] FIELD_TYPES = ['固定字节', '时间戳', '长度字段', 'CRC校验', '序号', '接收数据'] TYPE_TO_NAME = {'fix': '固定字节', 'stamp': '时间戳', 'len': '长度字段', 'crc': 'CRC校验', 'seq': '序号', 'recv': '接收数据'} TYPE_FROM_NAME = {v: k for k, v in TYPE_TO_NAME.items()} WIDTH_OPTS = ['1', '2', '4'] ENDIAN_OPTS = ['小端', '大端'] def inherit_topmost(dialog): """若主窗口处于置顶状态，则让弹出的对话框也置顶，避免被主窗口遮住""" p = dialog.parentWidget() while p is not None: if getattr(p, '_topmost_active', False): dialog.setWindowFlag(Qt.WindowStaysOnTopHint, True) return p = p.parentWidget() class FlowLayout(QLayout): """可自动换行的流式布局：窗口过窄时控件自动换到下一行，不会显示不全""" def __init__(self, parent=None, h_spacing=8, v_spacing=6): super().__init__(parent) if parent is not None: self.setContentsMargins(0, 0, 0, 0) self._h_spacing = h_spacing self._v_spacing = v_spacing self._items = [] def addItem(self, item): self._items.append(item) def addWidget(self, widget): self._items.append(QWidgetItem(widget)) def count(self): return len(self._items) def itemAt(self, index): return self._items[index] if 0 <= index < len(self._items) else None def takeAt(self, index): if 0 <= index < len(self._items): return self._items.pop(index) return None def expandingDirections(self): return Qt.Orientations(Qt.Orientation(0)) def hasHeightForWidth(self): return True def heightForWidth(self, width): return self._do_layout(QRect(0, 0, width, 0), True) def setGeometry(self, rect): super().setGeometry(rect) self._do_layout(rect, False) def sizeHint(self): return self.minimumSize() def minimumSize(self): size = QSize() for item in self._items: size = size.expandedTo(item.minimumSize()) m = self.contentsMargins() size += QSize(m.left() + m.right(), m.top() + m.bottom()) return size def _smart_spacing(self, item): return self._h_spacing def _do_layout(self, rect, test_only): m = self.contentsMargins() effective = QRect(rect.x() + m.left(), rect.y() + m.top(), rect.width() - m.left() - m.right(), rect.height() - m.top() - m.bottom()) x, y = effective.x(), effective.y() line_height = 0 for item in self._items: hint = item.sizeHint() if x + hint.width() > effective.right() and line_height > 0: x = effective.x() y += line_height + self._v_spacing line_height = 0 if not test_only: item.setGeometry(QRect(QPoint(x, y), hint)) x += hint.width() + self._h_spacing line_height = max(line_height, hint.height()) return y + line_height - effective.y() + m.top() + m.bottom() class SendEdit(QPlainTextEdit): """回车即发送的发送框，支持↑↓键浏览发送历史，Ctrl+滚轮调整字号""" enterPressed = Signal() histNav = Signal(int) def __init__(self, parent=None): super().__init__(parent) self._font_size = 11 def wheelEvent(self, event): if event.modifiers() & Qt.ControlModifier: delta = event.angleDelta().y() if delta: self._font_size = max(8, min(40, self._font_size + (1 if delta > 0 else -1))) font = self.font() font.setPointSize(self._font_size) self.setFont(font) event.accept() return super().wheelEvent(event) def font_size(self): return self._font_size def keyPressEvent(self, e): if e.key() in (Qt.Key_Return, Qt.Key_Enter) and not (e.modifiers() & Qt.ShiftModifier): self.enterPressed.emit() return if e.key() in (Qt.Key_Up, Qt.Key_Down) and not (e.modifiers() & Qt.ShiftModifier) \ and self.document().blockCount() == 1: self.histNav.emit(-1 if e.key() == Qt.Key_Up else 1) return super().keyPressEvent(e) class RecvEdit(QPlainTextEdit): """接收区编辑框：Ctrl+滚轮 调整字体大小，右键菜单""" MIN_FONT = 8 MAX_FONT = 40 DEFAULT_FONT = 11 saveRequested = Signal() clearRequested = Signal() def __init__(self, parent=None): super().__init__(parent) self._font_size = self.DEFAULT_FONT def set_font_size(self, size): self._font_size = max(self.MIN_FONT, min(self.MAX_FONT, int(size))) font = self.font() font.setPointSize(self._font_size) self.setFont(font) def font_size(self): return self._font_size def contextMenuEvent(self, event): menu = self.createStandardContextMenu() menu.addSeparator() act_save = menu.addAction('另存为…') act_clear = menu.addAction('清空') ch = menu.exec(event.globalPos()) if ch == act_save: self.saveRequested.emit() elif ch == act_clear: self.clearRequested.emit() def wheelEvent(self, event): if event.modifiers() & Qt.ControlModifier: delta = event.angleDelta().y() if delta: self.set_font_size(self.font_size() + (1 if delta > 0 else -1)) event.accept() return super().wheelEvent(event) class NoWheelSpinBox(QSpinBox): def wheelEvent(self, event): event.ignore() class NoWheelComboBox(QComboBox): def wheelEvent(self, event): event.ignore() def pack_int(val: int, width: int, endian: str) -> bytes: width = int(width) if width == 1: return bytes([val & 0xFF]) prefix = '<' if endian == '小端' else '>' return struct.pack(prefix + ('H' if width == 2 else 'I'), val & 0xFFFF if width == 2 else val & 0xFFFFFFFF) def parse_match(hex_text: str): """解析匹配模板，?? 表示任意字节；返回 None 列表为通配""" s = ''.join(hex_text.split()) s = s.replace(',', '').replace(':', '') if len(s) % 2 != 0: raise ValueError('匹配模板十六进制字符个数必须为偶数') out = [] for i in range(0, len(s), 2): pair = s[i:i + 2].upper() if pair == '??': out.append(None) else: out.append(int(pair, 16)) return out def find_match(hex_text: str, data: bytes) -> int: """子串匹配：返回模板在 data 中的起始下标，未找到返回 -1。?? 为任意字节""" try: tmpl = parse_match(hex_text) except ValueError: return -1 n = len(tmpl) if n == 0 or len(data) < n: return -1 for i in range(len(data) - n + 1): ok = True for j, t in enumerate(tmpl): if t is not None and t != data[i + j]: ok = False break if ok: return i return -1 def match_rule(hex_text: str, data: bytes) -> bool: """子串匹配：接收数据中任一部分与模板一致即命中，?? 为任意字节""" return find_match(hex_text, data) != -1 APP_NAME = "网络调试助手" def app_config_path(): """返回自动保存配置的文件路径（与主程序同目录）""" if getattr(sys, 'frozen', False): base = os.path.dirname(sys.executable) else: base = os.path.dirname(os.path.abspath(__file__)) return os.path.join(base, 'app_config.json') BAUDS = ['1200', '2400', '4800', '9600', '19200', '38400', '57600', '115200'] PARITIES = {'无校验': serial.PARITY_NONE, '奇校验': serial.PARITY_ODD, '偶校验': serial.PARITY_EVEN} STOPBITS = {'1': serial.STOPBITS_ONE, '1.5': serial.STOPBITS_ONE_POINT_FIVE, '2': serial.STOPBITS_TWO} def ts(): return datetime.now().strftime('%H:%M:%S.%f')[:-3] def bytes_to_hex(data: bytes) -> str: return ' '.join(f'{b:02X}' for b in data) def hex_to_bytes(text: str) -> bytes: s = ''.join(text.split()) s = s.replace(',', '').replace(':', '') if len(s) % 2 != 0: raise ValueError('十六进制字符个数必须为偶数') return bytes.fromhex(s) def get_local_ip() -> str: """返回本机有效的非环回 IPv4 地址""" addrs = [] try: addrs = socket.gethostbyname_ex(socket.gethostname())[2] except Exception: pass if not addrs: try: addrs = [info[4][0] for info in socket.getaddrinfo( socket.gethostname(), None) if info[0] == socket.AF_INET] except Exception: addrs = [] for a in addrs: if a and not a.startswith('127.'): return a if addrs: return addrs[0] return '127.0.0.1' def make_stamp(opt: str) -> bytes: now = datetime.now() if opt == 'ASCII(YYYY-MM-DD HH:MM:SS)': return now.strftime('%Y-%m-%d %H:%M:%S').encode('ascii') if opt == 'ASCII(HH:MM:SS)': return now.strftime('%H:%M:%S').encode('ascii') if opt.startswith('Unix秒'): fmt = 'I' return struct.pack(fmt, int(now.timestamp())) if opt.startswith('Unix毫秒'): fmt = 'Q' return struct.pack(fmt, int(now.timestamp() * 1000)) if opt == 'BCD(YYMMDDHHMMSS)': s = now.strftime('%y%m%d%H%M%S') return bytes(int(s[i:i + 2]) for i in range(0, len(s), 2)) return b'' def _reflect16(x: int, bits: int = 8) -> int: r = 0 for _ in range(bits): r = (r << 1) | (x & 1) x >>= 1 return r def _crc16(data: bytes, poly: int, init: int, refin: bool, refout: bool, xorout: int) -> int: crc = init for b in data: if refin: b = _reflect16(b, 8) crc ^= b << 8 for _ in range(8): crc = ((crc << 1) ^ poly) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF if refout: crc = _reflect16(crc, 16) return (crc ^ xorout) & 0xFFFF def _crc32(data: bytes) -> int: crc = 0xFFFFFFFF for b in data: crc ^= b for _ in range(8): crc = (crc >> 1) ^ 0xEDB88320 if crc & 1 else crc >> 1 return crc ^ 0xFFFFFFFF def make_crc(opt: str, data: bytes, endian: str = '小端') -> bytes: fmt = '<' if endian == '小端' else '>' if opt == 'CRC16-MODBUS': return struct.pack(fmt + 'H', _crc16(data, 0x8005, 0xFFFF, True, True, 0)) if opt == 'CRC16-CCITT(XModem)': return struct.pack(fmt + 'H', _crc16(data, 0x1021, 0x0000, False, False, 0)) if opt == 'CRC16-IBM': return struct.pack(fmt + 'H', _crc16(data, 0x8005, 0x0000, True, True, 0)) if opt == 'CRC32': return struct.pack(fmt + 'I', _crc32(data)) if opt == 'SUM8': return bytes([sum(data) & 0xFF]) if opt == 'XOR8': x = 0 for b in data: x ^= b return bytes([x]) return b'' class CrcParam(QWidget): def __init__(self): super().__init__() h = QHBoxLayout(self) h.setContentsMargins(0, 0, 0, 0) self.algo = NoWheelComboBox() self.algo.addItems(CRC_OPTS[1:]) self.endian = NoWheelComboBox() self.endian.addItems(ENDIAN_OPTS) self.algo.currentIndexChanged.connect(self._tog) h.addWidget(self.algo) h.addWidget(QLabel('字节序')) h.addWidget(self.endian) self._tog() def _tog(self): self.endian.setEnabled(self.algo.currentText() in ('CRC16-MODBUS', 'CRC16-CCITT(XModem)', 'CRC16-IBM', 'CRC32')) def value(self): return {'algo': self.algo.currentText(), 'endian': self.endian.currentText()} class RecvParam(QWidget): def __init__(self): super().__init__() h = QHBoxLayout(self) h.setContentsMargins(0, 0, 0, 0) self.rng = NoWheelComboBox() self.rng.addItems(['全部', '指定偏移']) self.offset = NoWheelSpinBox() self.offset.setRange(0, 65535) self.length = NoWheelSpinBox() self.length.setRange(0, 65535) self.length.setSpecialValueText('到末尾') self.xform = NoWheelComboBox() self.xform.addItems(['原样', '反转', '每2字节交换']) self.rng.currentIndexChanged.connect(self._tog) for lbl, wid in (('范围', self.rng), ('偏移', self.offset), ('长度', self.length), ('变换', self.xform)): h.addWidget(QLabel(lbl)) h.addWidget(wid) self._tog() def _tog(self): part = self.rng.currentText() == '指定偏移' self.offset.setEnabled(part) self.length.setEnabled(part) def value(self): return {'range': self.rng.currentText(), 'offset': self.offset.value(), 'length': self.length.value(), 'xform': self.xform.currentText()} class LenParam(QWidget): def __init__(self): super().__init__() h = QHBoxLayout(self) h.setContentsMargins(0, 0, 0, 0) self.src = NoWheelComboBox() self.src.addItems(['接收数据', '应答已组数据']) self.width = NoWheelComboBox() self.width.addItems(WIDTH_OPTS) self.endian = NoWheelComboBox() self.endian.addItems(ENDIAN_OPTS) h.addWidget(QLabel('长度对象')) h.addWidget(self.src) h.addWidget(QLabel('宽度')) h.addWidget(self.width) h.addWidget(QLabel('字节序')) h.addWidget(self.endian) def value(self): return {'src': self.src.currentText(), 'width': int(self.width.currentText()), 'endian': self.endian.currentText()} class SeqParam(QWidget): def __init__(self): super().__init__() h = QHBoxLayout(self) h.setContentsMargins(0, 0, 0, 0) self.width = NoWheelComboBox() self.width.addItems(WIDTH_OPTS) self.start = NoWheelSpinBox() self.start.setRange(0, 0x7FFFFFFF) self.step = NoWheelSpinBox() self.step.setRange(0, 0x7FFFFFFF) self.step.setValue(1) self.endian = NoWheelComboBox() self.endian.addItems(ENDIAN_OPTS) h.addWidget(QLabel('宽度')) h.addWidget(self.width) h.addWidget(QLabel('起始')) h.addWidget(self.start) h.addWidget(QLabel('步长')) h.addWidget(self.step) h.addWidget(QLabel('字节序')) h.addWidget(self.endian) def value(self): return {'width': int(self.width.currentText()), 'start': self.start.value(), 'step': self.step.value(), 'endian': self.endian.currentText()} def make_field_widget(ftype): if ftype == '固定字节': e = QLineEdit() e.setPlaceholderText('HEX，如 AA 01 0F') return e if ftype == '时间戳': c = NoWheelComboBox() c.addItems(STAMP_OPTS[1:]) return c if ftype == '长度字段': return LenParam() if ftype == 'CRC校验': return CrcParam() if ftype == '序号': return SeqParam() if ftype == '接收数据': return RecvParam() return QWidget() def read_field(ftype, widget): if ftype == '固定字节': return {'type': 'fix', 'hex': widget.text().strip()} if ftype == '时间戳': return {'type': 'stamp', 'format': widget.currentText()} if ftype == '长度字段': v = widget.value() return {'type': 'len', **v} if ftype == 'CRC校验': if isinstance(widget, CrcParam): return {'type': 'crc', **widget.value()} return {'type': 'crc', 'algo': widget.currentText(), 'endian': '小端'} if ftype == '序号': v = widget.value() return {'type': 'seq', **v} if ftype == '接收数据': v = widget.value() return {'type': 'recv', **v} return {} def write_field(ftype, widget, field): if ftype == '固定字节': widget.setText(field.get('hex', '')) elif ftype == '时间戳': widget.setCurrentText(field.get('format', '')) elif ftype == '长度字段' and isinstance(widget, LenParam): widget.src.setCurrentText(field.get('src', '接收数据')) widget.width.setCurrentIndex(WIDTH_OPTS.index(str(field.get('width', 1)))) widget.endian.setCurrentText(field.get('endian', '小端')) elif ftype == 'CRC校验' and isinstance(widget, CrcParam): widget.algo.setCurrentText(field.get('algo', '')) widget.endian.setCurrentText(field.get('endian', '小端')) elif ftype == '序号' and isinstance(widget, SeqParam): widget.width.setCurrentIndex(WIDTH_OPTS.index(str(field.get('width', 2)))) widget.start.setValue(field.get('start', 0)) widget.step.setValue(field.get('step', 1)) widget.endian.setCurrentText(field.get('endian', '小端')) elif ftype == '接收数据' and isinstance(widget, RecvParam): widget.rng.setCurrentText(field.get('range', '全部')) widget.offset.setValue(field.get('offset', 0)) widget.length.setValue(field.get('length', 0)) widget.xform.setCurrentText(field.get('xform', '原样')) class RuleEditDialog(QDialog): def __init__(self, parent=None, rule=None): super().__init__(parent) inherit_topmost(self) self.setWindowTitle('应答规则') self.setMinimumWidth(800) v = QVBoxLayout(self) row = QHBoxLayout() row.addWidget(QLabel('应答延时')) self.spin_delay = NoWheelSpinBox() self.spin_delay.setRange(0, 600000) self.spin_delay.setValue(0) self.spin_delay.setSuffix(' ms') self.spin_delay.setToolTip('本规则命中后的应答延时，0 表示立即应答') row.addWidget(self.spin_delay) row.addSpacing(12) row.addWidget(QLabel('备注')) self.edit_note = QLineEdit() self.edit_note.setPlaceholderText('可留空') row.addWidget(self.edit_note, 1) v.addLayout(row) v.addWidget(QLabel('指令匹配模板(HEX，?? 任意字节，帧中含相同片段即匹配)：')) self.edit_match = QPlainTextEdit() self.edit_match.setMaximumHeight(80) self.edit_match.setStyleSheet('QPlainTextEdit{font-family:Consolas}') v.addWidget(self.edit_match) v.addWidget(QLabel('指令应答模板(按字节配置)：')) self.table = QTableWidget(0, 3) self.table.setHorizontalHeaderLabels(['字段类型', '参数', '操作']) header = self.table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.ResizeToContents) header.setSectionResizeMode(2, QHeaderView.ResizeToContents) header.setSectionResizeMode(1, QHeaderView.Stretch) self.table.setColumnWidth(1, 460) self.table.verticalHeader().setVisible(False) self.table.setSelectionBehavior(QTableWidget.SelectRows) self.table.setSelectionMode(QTableWidget.SingleSelection) v.addWidget(self.table) add_row = QHBoxLayout() btn_add = QPushButton('添加字段') btn_add.clicked.connect(lambda: self.add_row('固定字节', {})) add_row.addWidget(btn_add) btn_up = QPushButton('↑ 上移') btn_dn = QPushButton('↓ 下移') btn_up.setToolTip('上移选中的字段') btn_dn.setToolTip('下移选中的字段') btn_up.clicked.connect(lambda: self.move_field(self.table.currentRow(), -1)) btn_dn.clicked.connect(lambda: self.move_field(self.table.currentRow(), 1)) add_row.addWidget(btn_up) add_row.addWidget(btn_dn) add_row.addWidget(QLabel('组包顺序即字节顺序，CRC 自动计算前面已组数据')) add_row.addStretch() v.addLayout(add_row) btns = QHBoxLayout() btns.addStretch() btn_ok = QPushButton('确定') btn_cancel = QPushButton('取消') btn_ok.clicked.connect(self.on_ok) btn_cancel.clicked.connect(self.reject) btns.addWidget(btn_ok) btns.addWidget(btn_cancel) v.addLayout(btns) if rule: self.spin_delay.setValue(rule.get('delay', 0)) self.edit_note.setText(rule.get('note', '')) self.edit_match.setPlainText(rule.get('match', '')) for f in rule.get('fields', []): self.add_row(TYPE_TO_NAME.get(f['type'], '固定字节'), f) else: self.spin_delay.setValue(0) self.add_row('固定字节', {'type': 'fix', 'hex': ''}) def add_row(self, ftype, field): r = self.table.rowCount() self.table.insertRow(r) self.table.setItem(r, 0, QTableWidgetItem()) type_combo = NoWheelComboBox() type_combo.addItems(FIELD_TYPES) type_combo.setCurrentText(ftype) self.table.setCellWidget(r, 0, type_combo) param = make_field_widget(ftype) self.table.setCellWidget(r, 1, param) if field: write_field(ftype, param, field) def on_type(idx): newparam = make_field_widget(type_combo.currentText()) self.table.setCellWidget(r, 1, newparam) type_combo.currentIndexChanged.connect(on_type) ops = QWidget() ho = QHBoxLayout(ops) ho.setContentsMargins(2, 0, 2, 0) ho.setSpacing(2) btn_del = QPushButton('删除') btn_del.setToolTip('删除此字段') btn_del.clicked.connect(lambda _, rr=r: self.table.removeRow(rr)) ho.addWidget(btn_del) self.table.setCellWidget(r, 2, ops) def move_field(self, rr, delta): rows = [(self.table.cellWidget(i, 0).currentText(), read_field(self.table.cellWidget(i, 0).currentText(), self.table.cellWidget(i, 1))) for i in range(self.table.rowCount())] nr = rr + delta if rr < 0 or not (0 <= nr < len(rows)): return rows[rr], rows[nr] = rows[nr], rows[rr] self.table.setRowCount(0) for ftype, field in rows: self.add_row(ftype, field) self.table.setCurrentCell(nr, 0) def on_ok(self): try: parse_match(self.edit_match.toPlainText()) except ValueError as e: QMessageBox.warning(self, '格式错误', f'匹配模板错误:\n{e}') return fields = [] for r in range(self.table.rowCount()): ftype = self.table.cellWidget(r, 0).currentText() field = read_field(ftype, self.table.cellWidget(r, 1)) if ftype == '固定字节' and not field['hex']: continue if ftype == '固定字节': try: hex_to_bytes(field['hex']) except ValueError: QMessageBox.warning(self, '格式错误', f'固定字节「{field["hex"]}」无效：需为偶数个十六进制字符\n' f'（表示 00 需输入 00，不能只输 0）') return fields.append(field) if not fields: QMessageBox.warning(self, '提示', '至少包含一个有效应答字段') return self._result = {'note': self.edit_note.text().strip(), 'delay': self.spin_delay.value(), 'match': self.edit_match.toPlainText().strip(), 'fields': fields} self.accept() def result(self): return getattr(self, '_result', None) class AutoReplyManagerDialog(QDialog): def __init__(self, rules, parent=None): super().__init__(parent) inherit_topmost(self) self.setWindowTitle('自动应答规则') self.setMinimumSize(520, 380) self.rules = [dict(r) for r in rules] v = QVBoxLayout(self) self.list_rules = QListWidget() self.list_rules.itemDoubleClicked.connect(lambda _: self.edit_rule()) v.addWidget(self.list_rules) self._loading = False self.list_rules.itemChanged.connect(self._on_checked) btns = QHBoxLayout() btn_add = QPushButton('新增') btn_edit = QPushButton('编辑') btn_copy = QPushButton('复制') btn_del = QPushButton('删除') btn_up = QPushButton('上移') btn_down = QPushButton('下移') btn_add.clicked.connect(self.add_rule) btn_edit.clicked.connect(self.edit_rule) btn_copy.clicked.connect(self.copy_rule) btn_del.clicked.connect(self.del_rule) btn_up.clicked.connect(lambda: self.move_rule(-1)) btn_down.clicked.connect(lambda: self.move_rule(1)) for b in (btn_add, btn_edit, btn_copy, btn_del, btn_up, btn_down): btns.addWidget(b) btns.addStretch() btn_ok = QPushButton('完成') btn_ok.clicked.connect(self.accept) btns.addWidget(btn_ok) v.addLayout(btns) self._loading = False self.refresh() def refresh(self): self._loading = True try: self.list_rules.clear() for i, r in enumerate(self.rules, 1): note = r.get('note') or '(无备注)' item = QListWidgetItem(f'[{i}] 匹配: {r.get("match", "")} {note}') item.setFlags(item.flags() | Qt.ItemIsUserCheckable) item.setCheckState(Qt.Checked if r.get('enabled', True) else Qt.Unchecked) self.list_rules.addItem(item) finally: self._loading = False def _on_checked(self, item): if self._loading: return idx = self.list_rules.row(item) if 0 <= idx < len(self.rules): self.rules[idx]['enabled'] = item.checkState() == Qt.Checked def add_rule(self): dlg = RuleEditDialog(self) if dlg.exec() == QDialog.Accepted and dlg.result(): r = dlg.result() r['enabled'] = True self.rules.append(r) self.refresh() def edit_rule(self): row = self.list_rules.currentRow() if row < 0: return enabled = self.rules[row].get('enabled', True) dlg = RuleEditDialog(self, self.rules[row]) if dlg.exec() == QDialog.Accepted and dlg.result(): r = dlg.result() r['enabled'] = enabled self.rules[row] = r self.refresh() def copy_rule(self): row = self.list_rules.currentRow() if row < 0: return self.rules.insert(row + 1, copy.deepcopy(self.rules[row])) self.refresh() self.list_rules.setCurrentRow(row + 1) def del_rule(self): row = self.list_rules.currentRow() if row < 0: return del self.rules[row] self.refresh() def move_rule(self, delta): row = self.list_rules.currentRow() new = row + delta if row < 0 or not (0 <= new < len(self.rules)): return self.rules[row], self.rules[new] = self.rules[new], self.rules[row] self.refresh() self.list_rules.setCurrentRow(new) def result(self): return self.rules class BatchSendDialog(QDialog): """批量发送：每条可单独配置延时，可勾选是否循环""" batchFinished = Signal() def __init__(self, panel, parent=None): super().__init__(parent) inherit_topmost(self) self.panel = panel self.setWindowTitle('批量发送') self.setMinimumSize(560, 360) self._stop = False self._thread = None self.batchFinished.connect(self._finish) v = QVBoxLayout(self) self.table = QTableWidget(0, 4) self.table.setHorizontalHeaderLabels(['发送', '编号', '延时(ms)', '内容']) hd = self.table.horizontalHeader() hd.setSectionResizeMode(0, QHeaderView.ResizeToContents) hd.setSectionResizeMode(1, QHeaderView.ResizeToContents) hd.setSectionResizeMode(2, QHeaderView.ResizeToContents) hd.setSectionResizeMode(3, QHeaderView.Stretch) self.table.verticalHeader().setVisible(False) self.table.setSelectionBehavior(QTableWidget.SelectRows) self.table.setSelectionMode(QTableWidget.SingleSelection) v.addWidget(self.table) row = QHBoxLayout() btn_add = QPushButton('添加') btn_del = QPushButton('删除所选') btn_clear = QPushButton('清空') btn_up = QPushButton('上移') btn_down = QPushButton('下移') btn_add.clicked.connect(lambda: self.add_row(True, 0, '')) btn_del.clicked.connect(self.del_selected) btn_clear.clicked.connect(self.clear_rows) btn_up.clicked.connect(lambda: self.move_row(-1)) btn_down.clicked.connect(lambda: self.move_row(1)) row.addWidget(btn_add) row.addWidget(btn_del) row.addWidget(btn_clear) row.addWidget(btn_up) row.addWidget(btn_down) row.addStretch() self.chk_loop = QCheckBox('循环发送') row.addWidget(self.chk_loop) self.btn_send = QPushButton('开始') self.btn_send.clicked.connect(self.toggle_batch) row.addWidget(self.btn_send) v.addLayout(row) self.add_row(True, 0, '') def add_row(self, enabled, delay, text): r = self.table.rowCount() self.table.insertRow(r) chk = QCheckBox() chk.setChecked(enabled) chk.setToolTip('勾选后进入发送队列') self.table.setCellWidget(r, 0, chk) self.table.setItem(r, 1, QTableWidgetItem(str(r + 1))) self.table.item(r, 1).setTextAlignment(Qt.AlignCenter) spin = NoWheelSpinBox() spin.setRange(0, 600000) spin.setValue(delay) spin.setSuffix(' ms') self.table.setCellWidget(r, 2, spin) edit = QLineEdit() edit.setText(text) edit.setPlaceholderText('发送内容（按发送区当前 文本/HEX 格式）') self.table.setCellWidget(r, 3, edit) return r def clear_rows(self): self.table.setRowCount(0) self._renumber() def _collect_rows(self): rows = [] for i in range(self.table.rowCount()): chk = self.table.cellWidget(i, 0) spin = self.table.cellWidget(i, 2) edit = self.table.cellWidget(i, 3) rows.append((chk.isChecked() if chk else True, spin.value() if spin else 0, edit.text() if edit else '')) return rows def move_row(self, delta): rows = self._collect_rows() r = self.table.currentRow() nr = r + delta if r < 0 or not (0 <= nr < len(rows)): return rows[r], rows[nr] = rows[nr], rows[r] self.table.setRowCount(0) for enabled, delay, text in rows: self.add_row(enabled, delay, text) self._renumber() self.table.setCurrentCell(nr, 0) def del_selected(self): rows = sorted({i for i in range(self.table.rowCount()) if self.table.cellWidget(i, 0) and self.table.cellWidget(i, 0).isChecked()}, reverse=True) for r in rows: self.table.removeRow(r) self._renumber() def get_config(self): return [{'enabled': chk.isChecked() if (chk := self.table.cellWidget(i, 0)) else True, 'delay': self.table.cellWidget(i, 2).value() if self.table.cellWidget(i, 2) else 0, 'text': self.table.cellWidget(i, 3).text() if self.table.cellWidget(i, 3) else ''} for i in range(self.table.rowCount())] def import_config(self, rows): self.table.setRowCount(0) for r in rows: self.add_row(bool(r.get('enabled', True)), int(r.get('delay', 0)), str(r.get('text', ''))) self._renumber() def _renumber(self): for i in range(self.table.rowCount()): item = self.table.item(i, 1) if item: item.setText(str(i + 1)) def toggle_batch(self): if self._thread and self._thread.is_alive(): self._stop = True self.btn_send.setEnabled(False) self.btn_send.setText('停止中…') return self._stop = False rows = [] for i in range(self.table.rowCount()): chk = self.table.cellWidget(i, 0) if not (chk and chk.isChecked()): continue text = self.table.cellWidget(i, 3).text() if not text: continue rows.append((self.table.cellWidget(i, 2).value(), text)) if not rows: QMessageBox.information(self, '提示', '请至少勾选一条并填写内容') self.btn_send.setChecked(False) return self.btn_send.setText('停止') self.btn_send.setEnabled(True) self._set_busy(True) self._thread = threading.Thread(target=self._run, args=(rows,), daemon=True) self._thread.start() def _set_busy(self, busy): for i in range(self.table.rowCount()): for c in range(4): w = self.table.cellWidget(i, c) if w: w.setEnabled(not busy) self.chk_loop.setEnabled(not busy) def _run(self, rows): try: while not self._stop: for delay, text in rows: if self._stop: return try: data = self.panel.encode_line(text) except ValueError as e: self.panel.info(f'批量发送内容「{text}」无效: {e}') continue if not data: continue if self.panel.actual_send(data): self.panel.log_line('TX', data) if delay: end = time.time() + delay / 1000.0 while time.time() < end: if self._stop: return time.sleep(0.01) if not self.chk_loop.isChecked(): break finally: self.batchFinished.emit() def _finish(self): self._thread = None self.btn_send.setText('开始') self.btn_send.setEnabled(True) self._set_busy(False) self.panel.info('批量发送结束' if not self._stop else '批量发送已停止') def closeEvent(self, event): self._stop = True super().closeEvent(event) class BasePanel(QWidget): """收发数据通用面板""" sigLog = Signal(str, object, object) sigInfo = Signal(str) sigUi = Signal(object) def __init__(self, title=''): super().__init__() self.sigLog.connect(self._append_log) self.sigInfo.connect(self._append_info) self.sigUi.connect(self._exec_ui) self.timer = QTimer(self) self._bufs = {} self.timer.timeout.connect(self.doSend) self.running = False self.reply_rules = [] self.batch_rows = [] self.send_history = [] self._hist_idx = 0 self._log_fh = None self._log_path = '' self._log_dir = '' self._log_hour = None self.log_rotate_timer = QTimer(self) self.log_rotate_timer.timeout.connect(self._rotate_log) self._stat_tx = 0 self._stat_rx = 0 self._stat_tx_frames = 0 self._stat_rx_frames = 0 self._filter_enabled = False self._filter_text = '' self._append_buffer = [] self._flush_timer = QTimer(self) self._flush_timer.setSingleShot(True) self._flush_timer.timeout.connect(self._flush_pending) self.root = QVBoxLayout(self) self.top = self.build_top() self.root.addWidget(self.top) splitter = QSplitter(Qt.Vertical) splitter.setChildrenCollapsible(False) splitter.setHandleWidth(5) recv_widget = QWidget() rv = QVBoxLayout(recv_widget) rv.setContentsMargins(0, 0, 0, 0) rv.addWidget(QLabel('接收区')) self.recv = RecvEdit() self.recv.setReadOnly(True) self.recv.setMaximumBlockCount(2000) self.recv.setFont(QFont('Consolas', RecvEdit.DEFAULT_FONT)) self.recv.saveRequested.connect(self.save_received_log) self.recv.clearRequested.connect(self.recv.clear) rv.addWidget(self.recv, 1) splitter.addWidget(recv_widget) send_widget = QWidget() sv = QVBoxLayout(send_widget) sv.setContentsMargins(0, 0, 0, 0) sv.addWidget(QLabel('发送区')) self.send = SendEdit() self.send.setMaximumBlockCount(2000) self.send.setFont(QFont('Consolas', 11)) self.send.enterPressed.connect(self.on_enter_pressed) self.send.histNav.connect(self.history_nav) sv.addWidget(self.send) splitter.addWidget(send_widget) splitter.setSizes([320, 160]) splitter.setStretchFactor(0, 3) splitter.setStretchFactor(1, 1) self.root.addWidget(splitter, 1) self.chk_auto = QCheckBox('定时发送') self.chk_auto.stateChanged.connect(self.toggle_timer) self.spin_interval = QSpinBox() self.spin_interval.setRange(10, 100000) self.spin_interval.setValue(1000) self.spin_interval.setSuffix(' ms') self.chk_enter_send = QCheckBox('回车发送') self.chk_crlf = QCheckBox('回车换行') self.chk_ts = QCheckBox('时间戳') self.chk_ts.setChecked(True) self.chk_reply = QCheckBox('自动应答') self.chk_reply.toggled.connect(lambda checked: setattr(self, '_reply_enabled', checked)) self.btn_reply = QPushButton('应答配置') self.btn_reply.clicked.connect(self.open_reply_dialog) # 接收设置 grp_recv = QGroupBox('接收设置') hr = QHBoxLayout(grp_recv) hr.setContentsMargins(8, 4, 8, 4) self.rb_recv_ascii = QRadioButton('ASCII') self.rb_recv_hex = QRadioButton('HEX') self.rb_recv_hex.setChecked(True) hr.addWidget(QLabel('显示')) hr.addWidget(self.rb_recv_ascii) hr.addWidget(self.rb_recv_hex) hr.addWidget(self.chk_ts) self.chk_info = QCheckBox('INFO') self.chk_info.setChecked(True) hr.addWidget(self.chk_info) self.chk_filter = QCheckBox('过滤') self.chk_filter.toggled.connect(self._on_filter_toggled) self.edit_filter = QLineEdit() self.edit_filter.setPlaceholderText('关键字') self.edit_filter.setEnabled(False) self.edit_filter.setMaximumWidth(130) self.edit_filter.textChanged.connect(self._on_filter_text) hr.addWidget(self.chk_filter) hr.addWidget(self.edit_filter) self.chk_save_log = QCheckBox('保存日志') self.chk_save_log.toggled.connect(self.toggle_save_log) hr.addWidget(self.chk_save_log) hr.addStretch() # 发送设置 grp_send = QGroupBox('发送设置') hs = QHBoxLayout(grp_send) hs.setContentsMargins(8, 4, 8, 4) self.rb_send_text = QRadioButton('文本') self.rb_send_hex = QRadioButton('HEX') self.rb_send_text.setChecked(True) hs.addWidget(QLabel('内容')) hs.addWidget(self.rb_send_text) hs.addWidget(self.rb_send_hex) hs.addWidget(self.chk_enter_send) hs.addWidget(self.chk_crlf) hs.addWidget(self.chk_auto) hs.addWidget(self.spin_interval) hs.addStretch() # 应答设置 grp_reply = QGroupBox('应答设置') hp = QHBoxLayout(grp_reply) hp.setContentsMargins(8, 4, 8, 4) self.rb_parse_hex = QRadioButton('HEX') self.rb_parse_ascii = QRadioButton('ASCII') self.rb_parse_hex.setChecked(True) parse_box = QWidget() pb = QHBoxLayout(parse_box) pb.setContentsMargins(0, 0, 0, 0) pb.addWidget(self.rb_parse_hex) pb.addWidget(self.rb_parse_ascii) self.rb_reply_hex = QRadioButton('HEX') self.rb_reply_ascii = QRadioButton('ASCII') self.rb_reply_hex.setChecked(True) reply_box = QWidget() rb = QHBoxLayout(reply_box) rb.setContentsMargins(0, 0, 0, 0) rb.addWidget(self.rb_reply_hex) rb.addWidget(self.rb_reply_ascii) hp.addWidget(self.chk_reply) hp.addSpacing(4) hp.addWidget(QLabel('解析')) hp.addWidget(parse_box) hp.addSpacing(4) hp.addWidget(QLabel('回应')) hp.addWidget(reply_box) hp.addSpacing(4) hp.addWidget(self.btn_reply) hp.addStretch() opt_flow = FlowLayout() opt_flow.addWidget(grp_recv) opt_flow.addWidget(grp_send) opt_flow.addWidget(grp_reply) self.root.addLayout(opt_flow) btn_row = QHBoxLayout() self.btn_export = QPushButton('导出配置') self.btn_import = QPushButton('导入配置') self.btn_export.clicked.connect(self.export_config) self.btn_import.clicked.connect(self.import_config) self.btn_send = QPushButton('发送') self.btn_batch = QPushButton('批量发送') self.btn_send_file = QPushButton('发送文件') self.btn_clear_recv = QPushButton('清空接收') self.btn_clear_send = QPushButton('清空发送') self.btn_send.clicked.connect(self.doSend) self.btn_batch.clicked.connect(self.open_batch) self.btn_send_file.clicked.connect(self.send_file) self.btn_clear_recv.clicked.connect(self.recv.clear) self.btn_clear_send.clicked.connect(self.send.clear) btn_row.addWidget(self.btn_export) btn_row.addWidget(self.btn_import) btn_row.addWidget(self.btn_send) btn_row.addWidget(self.btn_batch) btn_row.addWidget(self.btn_send_file) btn_row.addWidget(self.btn_clear_recv) btn_row.addWidget(self.btn_clear_send) btn_row.addStretch() self.stat_label = QLabel('TX: 0 B / 0 帧 RX: 0 B / 0 帧') btn_row.addWidget(self.stat_label) self.root.addLayout(btn_row) self.setWindowTitle(title) def build_top(self) -> QWidget: return QWidget() def on_enter_pressed(self): if self.chk_enter_send.isChecked(): self.doSend() def open_batch(self): dlg = BatchSendDialog(self, parent=self) if self.batch_rows: dlg.import_config(self.batch_rows) dlg.exec() self.batch_rows = dlg.get_config() def toggle_timer(self): if self.chk_auto.isChecked(): self.timer.start(self.spin_interval.value()) else: self.timer.stop() def log_line(self, direction, data: bytes, hex_mode: bool = None): self.sigLog.emit(direction, data, hex_mode) def run_ui(self, fn): self.sigUi.emit(fn) def _exec_ui(self, fn): fn() def _append_log(self, direction: str, data, hex_mode: bool = None): if hex_mode is None: hex_mode = (self.rb_recv_hex.isChecked() and direction == 'RX') or \ (self.rb_send_hex.isChecked() and direction == 'TX') if hex_mode: payload = bytes_to_hex(data) else: payload = data.decode('utf-8', 'replace') if direction == 'TX': self._stat_tx += len(data) self._stat_tx_frames += 1 else: self._stat_rx += len(data) self._stat_rx_frames += 1 self._update_stats() if self._filter_enabled and self._filter_text \ and self._filter_text.lower() not in payload.lower(): return text = f'{ts()}' if self.chk_ts.isChecked() else '' color = 'blue' if direction == 'TX' else 'green' self._append_line(f'[{text}] {direction}: {payload}' if text else f'{direction}: {payload}', color) def _update_stats(self): self.stat_label.setText( f'TX: {self._stat_tx} B / {self._stat_tx_frames} 帧 ' f'RX: {self._stat_rx} B / {self._stat_rx_frames} 帧') def _flush_pending(self): if not self._append_buffer: return cur = self.recv.textCursor() cur.movePosition(QTextCursor.End) fmt_cache = {} for text, color in self._append_buffer: if color not in fmt_cache: f = QTextCharFormat() f.setForeground(QColor(color)) fmt_cache[color] = f cur.setCharFormat(fmt_cache[color]) cur.insertText(text + '\n') self._append_buffer.clear() self.recv.setTextCursor(cur) bar = self.recv.verticalScrollBar() bar.setValue(bar.maximum()) def _append_colored(self, text: str, color: str): self._append_buffer.append((text, color)) if not self._flush_timer.isActive(): self._flush_timer.start(120) def _on_filter_toggled(self, checked: bool): self._filter_enabled = checked self.edit_filter.setEnabled(checked) def _on_filter_text(self, text: str): self._filter_text = text def save_received_log(self): path, _ = QFileDialog.getSaveFileName(self, '保存接收内容', 'recv.txt', '文本文件 (*.txt);;所有文件 (*)') if not path: return try: with open(path, 'w', encoding='utf-8') as f: f.write(self.recv.toPlainText()) self.info(f'接收内容已保存: {path}') except OSError as e: self.info(f'保存失败: {e}') def _append_line(self, text: str, color: str = None): self._append_colored(text, color) if self._log_fh or self._log_dir: self._rotate_log() if self._log_fh: try: self._log_fh.write(text + '\n') self._log_fh.flush() except OSError: pass def encode_line(self, text: str) -> bytes: if not text: return b'' if self.rb_send_hex.isChecked(): return hex_to_bytes(text) data = text.encode('utf-8', 'replace') if self.chk_crlf.isChecked(): data += b'\r\n' return data def encode_send(self) -> bytes: return self.encode_line(self.send.toPlainText()) def encode_reply(self, data: bytes) -> bytes: """应答输出：回应格式选ASCII时转大写ASCII十六进制文本，HEX时原样二进制""" if self.rb_reply_ascii.isChecked(): return data.hex().upper().encode('ascii') return data def doSend(self): try: data = self.encode_send() except ValueError as e: QMessageBox.warning(self, '格式错误', str(e)) return if data: text = self.send.toPlainText() if text and (not self.send_history or self.send_history[-1] != text): self.send_history.append(text) self._hist_idx = len(self.send_history) if self.actual_send(data): self.log_line('TX', data) def history_nav(self, delta): if not self.send_history: return n = min(max(self._hist_idx + delta, 0), len(self.send_history) - 1) if n == self._hist_idx: return self._hist_idx = n self.send.setPlainText(self.send_history[n]) c = self.send.textCursor() c.movePosition(QTextCursor.End) self.send.setTextCursor(c) def actual_send(self, data: bytes) -> bool: raise NotImplementedError def send_file(self): path, _ = QFileDialog.getOpenFileName(self, '选择要发送的文件', '') if not path: return try: with open(path, 'rb') as f: data = f.read() except OSError as e: QMessageBox.warning(self, '错误', f'读取文件失败:\n{e}') return if not data: self.info('文件内容为空，未发送') return self._file_total = len(data) self.info(f'开始发送文件({len(data)} B): {path}') threading.Thread(target=self._send_file_worker, args=(data,), daemon=True).start() def _send_file_worker(self, data: bytes): sent = 0 for i in range(0, len(data), 4096): part = data[i:i + 4096] if not self.actual_send(part): self.info('发送文件已中止') return sent += len(part) self.log_line('TX', part) time.sleep(0.01) self.info(f'文件发送完成，共 {sent} B') def send_reply(self, reply: bytes, sock=None, addr=None): raise NotImplementedError def build_reply_for_rule(self, rule, recv_data: bytes): out = bytearray() for f in rule.get('fields', []): t = f['type'] if t == 'fix': try: out += hex_to_bytes(f.get('hex', '')) except ValueError: self.info(f'应答字段固定字节无效已忽略: 「{f.get("hex", "")}」') elif t == 'stamp': out += make_stamp(f.get('format', '无')) elif t == 'len': val = len(recv_data) if f.get('src') == '接收数据' else len(out) out += pack_int(val, f.get('width', 1), f.get('endian', '小端')) elif t == 'crc': out += make_crc(f.get('algo', '无'), bytes(out), f.get('endian', '小端')) elif t == 'seq': cur = f.get('cur', f.get('start', 0)) out += pack_int(cur, f.get('width', 2), f.get('endian', '小端')) f['cur'] = cur + f.get('step', 1) elif t == 'recv': if f.get('range') == '指定偏移': off = f.get('offset', 0) ln = f.get('length', 0) part = recv_data[off: off + ln if ln > 0 else None] else: part = recv_data x = f.get('xform', '原样') if x == '反转': part = part[::-1] elif x == '每2字节交换': part = b''.join(part[i + 1:i + 2] + part[i:i + 1] for i in range(0, len(part) - 1, 2)) + \ (b'' if len(part) % 2 == 0 else part[-1:]) out += part return bytes(out) def pick_rule(self, recv_data: bytes): for r in self.reply_rules: if r.get('enabled', True) is False: continue try: if match_rule(r.get('match', ''), recv_data): return r except ValueError: continue return None def feed_receive(self, data: bytes, key, sock=None, addr=None): """收包缓存拼接：把到达字节累积起来再按优先级匹配，命中则应答""" self.log_line('RX', data) if not getattr(self, '_reply_enabled', False): return if not self.reply_rules or not any(r.get('enabled', True) for r in self.reply_rules): self.info('已开启自动应答，但未选中任何应答规则(应答配置内需点“完成”保存，单选选中的规则才参与应答)') return buf = self._bufs.get(key, bytearray()) buf += data if len(buf) > 1_000_000: del buf[:len(buf) - 1_000_000] self._bufs[key] = buf # 应答解析：ASCII时把收到的ASCII十六进制文本解码为二进制再匹配；HEX时原样匹配 raw_b = bytes(buf) if self.rb_parse_ascii.isChecked(): try: b = hex_to_bytes(raw_b.decode('ascii', 'ignore')) except ValueError: b = b'' # 文本未到齐，继续缓冲等待 else: b = raw_b if not b: return for r in self.reply_rules: if r.get('enabled', True) is False: continue mtmp = r.get('match', '') try: tmpl = parse_match(mtmp) except ValueError: self.info(f'应答规则「{r.get("note") or "无备注"}」的匹配模板无法解析(请检查是否为十六进制)') continue if len(tmpl) == 0: continue st = find_match(mtmp, b) if st != -1: n = len(tmpl) window = bytes(b[st:st + n]) reply = self.build_reply_for_rule(r, window) if reply: out = self.encode_reply(reply) delay = r.get('delay', 0) self.info(f'应答规则「{r.get("note") or "无备注"}」命中模板，' f'{"延时" + str(delay) + "ms后" if delay else ""}发送: {bytes_to_hex(reply)}') if delay > 0: threading.Timer(delay / 1000.0, lambda: self.send_reply(out, sock, addr)).start() else: self.send_reply(out, sock, addr) self._bufs.pop(key, None) return self.info(f'应答无规则命中：本次收到{len(data)}B，缓冲累积{len(buf)}B（帧未到齐或与模板不匹配）') def reply_if_enabled(self, recv_data, sock=None, addr=None): if not getattr(self, '_reply_enabled', False) or not self.reply_rules: return rule = self.pick_rule(recv_data) if rule is None: return reply = self.build_reply_for_rule(rule, recv_data) if reply: self.send_reply(reply, sock, addr) def open_reply_dialog(self): dlg = AutoReplyManagerDialog(self.reply_rules, self) if dlg.exec() == QDialog.Accepted: self.reply_rules = dlg.result() def collect_config(self): return { 'recv': { 'display': 'HEX' if self.rb_recv_hex.isChecked() else 'ASCII', 'ts': self.chk_ts.isChecked(), 'info': self.chk_info.isChecked(), 'filter': self._filter_enabled, 'filter_text': self._filter_text, }, 'send': { 'format': 'HEX' if self.rb_send_hex.isChecked() else '文本', 'enter_send': self.chk_enter_send.isChecked(), 'crlf': self.chk_crlf.isChecked(), 'auto': self.chk_auto.isChecked(), 'interval': self.spin_interval.value(), 'font_size': self.send.font_size(), }, 'reply': { 'enabled': self.chk_reply.isChecked(), 'parse': 'HEX' if self.rb_parse_hex.isChecked() else 'ASCII', 'respond': 'HEX' if self.rb_reply_hex.isChecked() else 'ASCII', }, 'font_size': self.recv.font_size(), 'send_history': self.send_history[-200:], 'reply_rules': copy.deepcopy(self.reply_rules), 'batch_rows': copy.deepcopy(self.batch_rows), } def apply_config(self, cfg): if not isinstance(cfg, dict): return [] msg = [] recv = cfg.get('recv') if isinstance(recv, dict): display = recv.get('display', 'HEX') self.rb_recv_hex.setChecked(display == 'HEX') self.rb_recv_ascii.setChecked(display != 'HEX') if 'ts' in recv: self.chk_ts.setChecked(bool(recv['ts'])) if 'info' in recv: self.chk_info.setChecked(bool(recv['info'])) if 'filter' in recv: self.chk_filter.setChecked(bool(recv['filter'])) self.edit_filter.setEnabled(bool(recv['filter'])) if 'filter_text' in recv: self.edit_filter.setText(recv.get('filter_text', '')) msg.append('接收设置') send = cfg.get('send') if isinstance(send, dict): fmt = send.get('format', '文本') self.rb_send_hex.setChecked(fmt == 'HEX') self.rb_send_text.setChecked(fmt != 'HEX') if 'enter_send' in send: self.chk_enter_send.setChecked(bool(send['enter_send'])) if 'crlf' in send: self.chk_crlf.setChecked(bool(send['crlf'])) if 'auto' in send: self.chk_auto.setChecked(bool(send['auto'])) if 'interval' in send: self.spin_interval.setValue(int(send['interval'])) if 'font_size' in send: try: size = max(8, min(40, int(send['font_size']))) f = self.send.font() f.setPointSize(size) self.send.setFont(f) if hasattr(self.send, '_font_size'): self.send._font_size = size except (TypeError, ValueError): pass msg.append('发送设置') reply = cfg.get('reply') if isinstance(reply, dict): if 'enabled' in reply: self.chk_reply.setChecked(bool(reply['enabled'])) if 'parse' in reply: self.rb_parse_hex.setChecked(reply['parse'] == 'HEX') self.rb_parse_ascii.setChecked(reply['parse'] != 'HEX') if 'respond' in reply: self.rb_reply_hex.setChecked(reply['respond'] == 'HEX') self.rb_reply_ascii.setChecked(reply['respond'] != 'HEX') msg.append('应答设置') if 'font_size' in cfg: try: self.recv.set_font_size(int(cfg['font_size'])) except (TypeError, ValueError): pass msg.append('字体大小') if 'send_history' in cfg and isinstance(cfg['send_history'], list): self.send_history = [str(x) for x in cfg['send_history']][-200:] self._hist_idx = len(self.send_history) msg.append(f'发送历史 {len(self.send_history)} 条') rr = cfg.get('reply_rules') if isinstance(rr, list) and all(isinstance(x, dict) for x in rr): for r in rr: r.setdefault('enabled', True) self.reply_rules = rr msg.append(f'应答配置 {len(rr)} 条') br = cfg.get('batch_rows') if isinstance(br, list) and all(isinstance(x, dict) for x in br): self.batch_rows = [{'enabled': bool(x.get('enabled', True)), 'delay': int(x.get('delay', 0)), 'text': str(x.get('text', ''))} for x in br] msg.append(f'批量发送 {len(br)} 条') return msg def export_config(self): data = self.collect_config() default = f'config_{datetime.now():%Y%m%d_%H%M%S}.json' path, _ = QFileDialog.getSaveFileName(self, '导出配置', default, 'JSON 文件 (*.json);;所有文件 (*)') if not path: return try: with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) self.info(f'配置已导出: {path}') except OSError as e: self.info(f'导出失败: {e}') def import_config(self): path, _ = QFileDialog.getOpenFileName(self, '导入配置', '', 'JSON 文件 (*.json);;所有文件 (*)') if not path: return try: with open(path, 'r', encoding='utf-8') as f: data = json.load(f) except Exception as e: QMessageBox.warning(self, '导入失败', f'无法读取配置文件:\n{e}') return if not isinstance(data, dict): QMessageBox.warning(self, '导入失败', '配置文件格式无效') return msg = self.apply_config(data) if not msg: QMessageBox.warning(self, '导入失败', '配置文件中没有可导入的数据') return self.info('配置已导入: ' + '，'.join(msg)) def toggle_save_log(self, checked: bool): if checked: d = QFileDialog.getExistingDirectory(self, '选择日志保存目录', '') if not d: self.chk_save_log.setChecked(False) return self._log_dir = d self._log_fh = None self._log_hour = None self._rotate_log() if self._log_fh: self.log_rotate_timer.start(60000) self.info(f'日志自动保存已开启 -> {d}（每小时一个文件，保留一个月循环覆盖）') else: self.chk_save_log.setChecked(False) else: self.log_rotate_timer.stop() self._close_log() if self._log_dir: self.info('日志自动保存已停止') self._log_dir = '' def _log_path_for(self, dt) -> str: return os.path.join(self._log_dir, f'log_{dt:%Y%m%d_%H}.txt') def _close_log(self): if self._log_fh: try: self._log_fh.close() except OSError: pass self._log_fh = None self._log_hour = None def _rotate_log(self): if not self._log_dir: return now = datetime.now() key = now.strftime('%Y%m%d%H') if self._log_fh and self._log_hour == key: return self._close_log() try: self._log_fh = open(self._log_path_for(now), 'a', encoding='utf-8') self._log_hour = key except OSError as e: self.info(f'无法创建日志文件: {e}') self._log_fh = None self._cleanup_old_logs() def _cleanup_old_logs(self): try: files = [f for f in os.listdir(self._log_dir) if f.startswith('log_') and f.endswith('.txt')] files.sort() while len(files) > 720: old = files.pop(0) try: os.remove(os.path.join(self._log_dir, old)) except OSError: pass except OSError: pass def info(self, text: str): self.sigInfo.emit(text) def _append_info(self, text: str): if not self.chk_info.isChecked(): return text = f'[{ts()}] INFO: {text}' if self.chk_ts.isChecked() else f'INFO: {text}' self._append_line(text, 'gray') class SerialPanel(BasePanel): def __init__(self): super().__init__('串口') self.ser = None self.reader = None def build_top(self) -> QWidget: w = QWidget() g = QGridLayout(w) self.combox_port = QComboBox() self.combox_baud = QComboBox() self.combox_baud.addItems(BAUDS) self.combox_baud.setCurrentText('115200') self.combox_data = QComboBox() self.combox_data.addItems(['5', '6', '7', '8']) self.combox_data.setCurrentText('8') self.combox_parity = QComboBox() self.combox_parity.addItems(list(PARITIES.keys())) self.combox_stop = QComboBox() self.combox_stop.addItems(list(STOPBITS.keys())) self.combox_stop.setCurrentText('1') self.btn_refresh = QPushButton('刷新') self.btn_refresh.clicked.connect(self.refresh_ports) self.btn_open = QPushButton('打开串口') self.btn_open.clicked.connect(self.toggle_serial) self.chk_rts = QCheckBox('RTS') self.chk_dtr = QCheckBox('DTR') self.chk_rts.setToolTip('打开串口后控制 RTS 引脚电平') self.chk_dtr.setToolTip('打开串口后控制 DTR 引脚电平') g.addWidget(QLabel('串口'), 0, 0) g.addWidget(self.combox_port, 0, 1) g.addWidget(self.btn_refresh, 0, 2) g.addWidget(QLabel('波特率'), 0, 3) g.addWidget(self.combox_baud, 0, 4) g.addWidget(QLabel('数据位'), 1, 0) g.addWidget(self.combox_data, 1, 1) g.addWidget(QLabel('校验位'), 1, 3) g.addWidget(self.combox_parity, 1, 4) g.addWidget(QLabel('停止位'), 2, 0) g.addWidget(self.combox_stop, 2, 1) g.addWidget(self.chk_rts, 2, 3) g.addWidget(self.chk_dtr, 2, 4) g.addWidget(self.btn_open, 3, 3, 1, 2) g.setColumnStretch(1, 1) g.setColumnStretch(4, 1) w.setLayout(g) self.refresh_ports() return w def refresh_ports(self): self.combox_port.clear() for p in serial.tools.list_ports.comports(): name = p.device desc = p.description if p.description else '' self.combox_port.addItem(f'{name} {desc}' if desc else name) def toggle_serial(self): if self.ser and self.ser.is_open: self.close_serial() else: self.open_serial() def open_serial(self): text = self.combox_port.currentText() port = text.split(' ')[0] if text else '' if not port: QMessageBox.warning(self, '提示', '请选择串口') return try: self.ser = serial.Serial( port=port, baudrate=int(self.combox_baud.currentText()), bytesize=int(self.combox_data.currentText()), parity=PARITIES[self.combox_parity.currentText()], stopbits=STOPBITS[self.combox_stop.currentText()], timeout=0.1, ) try: self.ser.rts = self.chk_rts.isChecked() self.ser.dtr = self.chk_dtr.isChecked() except Exception: pass self.btn_open.setText('关闭串口') self.combox_port.setEnabled(False) self.info(f'串口 {port} 已打开') self.reader = threading.Thread(target=self.read_loop, daemon=True) self.reader.start() except Exception as e: QMessageBox.warning(self, '错误', f'打开串口失败:\n{e}') def close_serial(self): try: if self.ser: self.ser.close() except Exception: pass self.ser = None self.btn_open.setText('打开串口') self.combox_port.setEnabled(True) self.info('串口已关闭') def read_loop(self): while self.ser and self.ser.is_open: try: n = self.ser.in_waiting if n: data = self.ser.read(n) self.feed_receive(data, 'serial') else: time.sleep(0.01) except Exception: break def send_reply(self, reply: bytes, sock=None, addr=None): try: if self.ser and self.ser.is_open: self.ser.write(reply) self.log_line('TX', reply, hex_mode=self.rb_reply_hex.isChecked()) except Exception: pass def actual_send(self, data: bytes) -> bool: if not (self.ser and self.ser.is_open): QMessageBox.warning(self, '提示', '请先打开串口') return False try: self.ser.write(data) return True except Exception as e: QMessageBox.warning(self, '错误', f'发送失败\n{e}') return False def collect_config(self): cfg = super().collect_config() cfg['rts'] = self.chk_rts.isChecked() cfg['dtr'] = self.chk_dtr.isChecked() return cfg def apply_config(self, cfg): msg = super().apply_config(cfg) if isinstance(cfg, dict): if 'rts' in cfg: self.chk_rts.setChecked(bool(cfg['rts'])) if 'dtr' in cfg: self.chk_dtr.setChecked(bool(cfg['dtr'])) return msg class NetPanel(BasePanel): """TCP/UDP 通用面板，支持客户端与服务器模式""" def __init__(self, kind: str): self.kind = kind # 'tcp' / 'udp' self.sock = None self.clients = {} self.lock = threading.Lock() super().__init__('TCP' if kind == 'tcp' else 'UDP') def build_top(self) -> QWidget: w = QWidget() g = QGridLayout(w) self.combox_mode = QComboBox() self.combox_mode.addItems(['客户端', '服务端']) self.combox_mode.currentIndexChanged.connect(self.on_mode_changed) self.edit_ip = QLineEdit('127.0.0.1') self.spin_port = QSpinBox() self.spin_port.setRange(1, 65535) self.spin_port.setValue(8080) self.btn_conn = QPushButton('连接') self.btn_conn.clicked.connect(self.toggle_connect) self.ip_label = QLabel('目标IP') g.addWidget(QLabel('模式'), 0, 0) g.addWidget(self.combox_mode, 0, 1) g.addWidget(self.ip_label, 0, 2) g.addWidget(self.edit_ip, 0, 3) g.addWidget(QLabel('端口'), 0, 4) g.addWidget(self.spin_port, 0, 5) g.addWidget(self.btn_conn, 0, 6) if self.kind == 'tcp': self.client_bar = QWidget() cb = QHBoxLayout(self.client_bar) cb.setContentsMargins(0, 0, 0, 0) self.combox_client = QComboBox() self.btn_disconnect_all = QPushButton('断开所有') self.btn_disconnect_all.clicked.connect(self.disconnect_all) cb.addWidget(QLabel('客户端')) cb.addWidget(self.combox_client, 1) cb.addWidget(self.btn_disconnect_all) g.addWidget(self.client_bar, 1, 0, 1, 7) else: self.client_bar = None g.setColumnStretch(3, 1) w.setLayout(g) self.on_mode_changed(0) return w def on_mode_changed(self, _): server = self.combox_mode.currentText() == '服务端' if server: self.edit_ip.setText(get_local_ip()) self.edit_ip.setEnabled(False) self.ip_label.setText('本地IP') if self.btn_conn.text() not in ('停止',): pass else: self.edit_ip.setEnabled(True) self.ip_label.setText('目标IP') if self.kind == 'tcp' and self.client_bar is not None: self.client_bar.setVisible(server) def toggle_connect(self): if self.combox_mode.currentText() == '客户端': if self.sock: self.close_client() else: self.open_client() else: if self.sock: self.stop_server() else: self.start_server() def open_client(self): ip = self.edit_ip.text().strip() port = self.spin_port.value() if not ip: QMessageBox.warning(self, '提示', '请输入目标IP') return try: if self.kind == 'tcp': sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) sock.connect((ip, port)) sock.settimeout(0.5) self.sock = sock self.btn_conn.setText('断开') self.info(f'已连接{ip}:{port}') threading.Thread(target=self.tcp_read_loop, args=(sock, ip, port), daemon=True).start() else: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(0.5) self.sock = sock self.btn_conn.setText('关闭') self.info(f'UDP 客户端就绪 -> {ip}:{port}') except Exception as e: QMessageBox.warning(self, '错误', f'连接失败:\n{e}') def close_client(self): try: if self.sock: self.sock.close() except Exception: pass self.sock = None self.btn_conn.setText('连接') self.info('连接已关闭') def start_server(self): port = self.spin_port.value() local = self.edit_ip.text().strip() or get_local_ip() try: if self.kind == 'tcp': srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(('0.0.0.0', port)) srv.listen(10) srv.settimeout(0.5) self.sock = srv self.btn_conn.setText('停止') self.info(f'TCP 服务器监听{local}:{port}') threading.Thread(target=self.accept_loop, daemon=True).start() else: srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) srv.bind(('0.0.0.0', port)) srv.settimeout(0.5) self.sock = srv self.btn_conn.setText('停止') self.info(f'UDP 服务器监听{local}:{port}') threading.Thread(target=self.udp_read_loop, daemon=True).start() except Exception as e: QMessageBox.warning(self, '错误', f'启动服务器失败\n{e}') def stop_server(self): try: if self.sock: self.sock.close() except Exception: pass with self.lock: for c in list(self.clients.values()): try: c.close() except Exception: pass self.clients.clear() self.sock = None self.combox_client.clear() self.btn_conn.setText('启动') self.info('服务器已停止') def accept_loop(self): while self.sock: try: conn, addr = self.sock.accept() except socket.timeout: continue except Exception: break conn.settimeout(0.5) key = f'{addr[0]}:{addr[1]}' with self.lock: self.clients[key] = conn self.run_ui(lambda k=key: self.combox_client.addItem(k)) self.info(f'客户端接入{key}') threading.Thread(target=self.tcp_read_loop, args=(conn, addr[0], addr[1]), daemon=True).start() def tcp_read_loop(self, sock, ip, port): while True: try: data = sock.recv(4096) if not data: break self.feed_receive(data, f'{ip}:{port}', sock) except socket.timeout: continue except Exception: break self.on_disconnected(sock, ip, port) def udp_read_loop(self): while self.sock: try: data, addr = self.sock.recvfrom(65536) self.last_remote = addr self.feed_receive(data, addr, self.sock, addr) except socket.timeout: continue except Exception: break def send_reply(self, reply: bytes, sock=None, addr=None): try: if addr: sock.sendto(reply, addr) elif sock: sock.sendall(reply) else: return self.log_line('TX', reply, hex_mode=self.rb_reply_hex.isChecked()) except Exception: pass def on_disconnected(self, sock, ip, port): key = f'{ip}:{port}' with self.lock: self.clients.pop(key, None) try: sock.close() except Exception: pass self.run_ui(lambda k=key: self._ui_remove_client(k)) self.info(f'客户端{key} 断开') def _ui_remove_client(self, key): idx = self.combox_client.findText(key) if idx >= 0: self.combox_client.removeItem(idx) def disconnect_all(self): with self.lock: for c in list(self.clients.values()): try: c.close() except Exception: pass self.clients.clear() self.combox_client.clear() self.info('已断开所有客户端') def actual_send(self, data: bytes) -> bool: if self.kind == 'udp': if not self.sock: QMessageBox.warning(self, '提示', '请先连接或启动服务器') return False ip = self.edit_ip.text().strip() or '127.0.0.1' try: self.sock.sendto(data, (ip, self.spin_port.value())) return True except Exception as e: QMessageBox.warning(self, '错误', f'发送失败\n{e}') return False # TCP if self.combox_mode.currentText() == '客户端': if not self.sock: QMessageBox.warning(self, '提示', '请先连接服务器') return False try: self.sock.sendall(data) return True except Exception as e: QMessageBox.warning(self, '错误', f'发送失败\n{e}') return False # TCP 服务器 with self.lock: targets = list(self.clients.values()) if not targets: QMessageBox.warning(self, '提示', '没有已连接的客户端') return False ok = False for c in targets: try: c.sendall(data) ok = True except Exception: pass return ok class MainWindow(QMainWindow): @staticmethod def _icon_path() -> str: if getattr(sys, 'frozen', False): base = sys._MEIPASS else: base = os.path.dirname(os.path.abspath(__file__)) p = os.path.join(base, '网络调试.ico') return p if os.path.exists(p) else '' def __init__(self): super().__init__() self.setWindowTitle(APP_NAME) self.resize(900, 680) self.setMinimumSize(600, 460) icon = self._icon_path() if icon: self.setWindowIcon(QIcon(icon)) tabs = QTabWidget() self.serial_panel = SerialPanel() self.tcp_panel = NetPanel('tcp') self.udp_panel = NetPanel('udp') tabs.addTab(self.serial_panel, '串口') tabs.addTab(self.tcp_panel, 'TCP') tabs.addTab(self.udp_panel, 'UDP') self.setCentralWidget(tabs) self.top_btn = QCheckBox('窗口置顶') self.top_btn.setToolTip('选中后窗口保持在最前端') self.top_btn.toggled.connect(self._toggle_topmost) self.chk_dark = QCheckBox('深色模式') self.chk_dark.setToolTip('切换亮色/暗色主题') self.chk_dark.toggled.connect(self._toggle_dark) self.chk_tray = QCheckBox('最小化到托盘') self.chk_tray.setToolTip('最小化或关闭窗口时隐藏到系统托盘') self.chk_tray.toggled.connect(self._on_tray_toggled) tb = QToolBar('选项') tb.setMovable(False) tb.addWidget(self.top_btn) tb.addWidget(self.chk_dark) tb.addWidget(self.chk_tray) tb.addSeparator() self.combo_profile = NoWheelComboBox() self.combo_profile.setMinimumWidth(150) self.combo_profile.currentIndexChanged.connect(self._on_profile_selected) tb.addWidget(QLabel(' 预设 ')) tb.addWidget(self.combo_profile) btn_psave = QPushButton('保存预设') btn_pdel = QPushButton('删除预设') btn_psave.clicked.connect(self._save_profile) btn_pdel.clicked.connect(self._del_profile) tb.addWidget(btn_psave) tb.addWidget(btn_pdel) self.addToolBar(tb) self._profiles = {} self._tray_enabled = False self._topmost_active = False self._light_palette = QApplication.instance().palette() self.tray_icon = None self._autosave = QTimer(self) self._autosave.setInterval(30000) self._autosave.timeout.connect(self._save_app_config) self._autosave.start() self._build_tray() self._load_profiles() self._load_app_config() def _toggle_topmost(self, on: bool): self._topmost_active = on try: import ctypes user32 = ctypes.windll.user32 user32.SetWindowPos.argtypes = [ ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_uint, ] user32.SetWindowPos.restype = ctypes.c_int HWND_TOPMOST = -1 HWND_NOTOPMOST = -2 SWP_NOMOVE = 0x0002 SWP_NOSIZE = 0x0001 SWP_NOACTIVATE = 0x0010 ok = user32.SetWindowPos( ctypes.c_void_p(int(self.winId())), ctypes.c_void_p(HWND_TOPMOST if on else HWND_NOTOPMOST), 0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE) != 0 if ok: return except Exception: pass self._fallback_topmost(on) def _fallback_topmost(self, on: bool): QTimer.singleShot(0, lambda: (self.setWindowFlag(Qt.WindowStaysOnTopHint, on), self.show())) def _toggle_dark(self, on: bool): app = QApplication.instance() if on: p = QPalette() p.setColor(QPalette.Window, QColor('#2b2b2b')) p.setColor(QPalette.WindowText, QColor('#f0f0f0')) p.setColor(QPalette.Base, QColor('#232323')) p.setColor(QPalette.AlternateBase, QColor('#3c3f41')) p.setColor(QPalette.Text, QColor('#f0f0f0')) p.setColor(QPalette.Button, QColor('#3c3f41')) p.setColor(QPalette.ButtonText, QColor('#f0f0f0')) p.setColor(QPalette.BrightText, QColor('#ff5555')) p.setColor(QPalette.Highlight, QColor('#2a82da')) p.setColor(QPalette.HighlightedText, QColor('#ffffff')) p.setColor(QPalette.Link, QColor('#2a82da')) p.setColor(QPalette.ToolTipBase, QColor('#f0f0f0')) p.setColor(QPalette.ToolTipText, QColor('#2b2b2b')) p.setColor(QPalette.Disabled, QPalette.Text, QColor('#787878')) p.setColor(QPalette.Disabled, QPalette.ButtonText, QColor('#787878')) p.setColor(QPalette.Disabled, QPalette.Base, QColor('#2b2b2b')) app.setStyle('Fusion') app.setPalette(p) else: app.setStyle('Fusion') app.setPalette(self._light_palette) def _build_tray(self): icon = QIcon(self._icon_path()) if icon.isNull(): self.tray_icon = None return self.tray_icon = QSystemTrayIcon(icon, self) menu = QMenu(self) act_show = menu.addAction('显示主界面') act_show.triggered.connect(self._show_from_tray) menu.addSeparator() act_quit = menu.addAction('退出') act_quit.triggered.connect(self._quit_app) self.tray_icon.setContextMenu(menu) self.tray_icon.activated.connect(self._on_tray_activated) def _on_tray_toggled(self, checked: bool): self._tray_enabled = checked if self.tray_icon: if checked: self.tray_icon.show() else: self.tray_icon.hide() def _on_tray_activated(self, reason): if reason == QSystemTrayIcon.Trigger: self._show_from_tray() def _show_from_tray(self): self.showNormal() self.show() self.raise_() self.activateWindow() def _quit_app(self): self._tray_enabled = False if self.tray_icon: self.tray_icon.hide() self.close() def changeEvent(self, event): if event.type() == QEvent.WindowStateChange and self.isMinimized() \ and self._tray_enabled and self.tray_icon and self.tray_icon.isVisible(): QTimer.singleShot(0, self.hide) super().changeEvent(event) def info(self, text: str): self.statusBar().showMessage(text, 3000) def _collect_all_config(self): return { 'topmost': self.top_btn.isChecked(), 'dark': self.chk_dark.isChecked(), 'tray': self.chk_tray.isChecked(), 'tabs': { 'serial': self.serial_panel.collect_config(), 'tcp': self.tcp_panel.collect_config(), 'udp': self.udp_panel.collect_config(), }, } def _apply_panels(self, data): tabs_cfg = data.get('tabs') if isinstance(data, dict) else None if not isinstance(tabs_cfg, dict): return for panel, key in ((self.serial_panel, 'serial'), (self.tcp_panel, 'tcp'), (self.udp_panel, 'udp')): cfg = tabs_cfg.get(key) if isinstance(cfg, dict): panel.apply_config(cfg) def _save_app_config(self): try: with open(app_config_path(), 'w', encoding='utf-8') as f: json.dump(self._collect_all_config(), f, ensure_ascii=False, indent=2) except Exception: pass def _load_app_config(self): try: with open(app_config_path(), 'r', encoding='utf-8') as f: data = json.load(f) except Exception: return if not isinstance(data, dict): return if data.get('topmost'): self.top_btn.setChecked(True) if data.get('dark'): self.chk_dark.setChecked(True) if data.get('tray'): self.chk_tray.setChecked(True) self._apply_panels(data) def _profile_path(self): return os.path.splitext(app_config_path())[0] + '_profiles.json' def _load_profiles(self): try: with open(self._profile_path(), 'r', encoding='utf-8') as f: data = json.load(f) self._profiles = data if isinstance(data, dict) else {} except Exception: self._profiles = {} self._reload_profiles_combo() def _write_profiles(self): try: with open(self._profile_path(), 'w', encoding='utf-8') as f: json.dump(self._profiles, f, ensure_ascii=False, indent=2) except OSError: pass def _reload_profiles_combo(self, select=None): self.combo_profile.blockSignals(True) self.combo_profile.clear() self.combo_profile.addItem('← 选择预设 →') for name in self._profiles: self.combo_profile.addItem(name) select = self.combo_profile.currentText() if select is None else select idx = self.combo_profile.findText(select) if idx >= 0 and idx != 0: self.combo_profile.setCurrentIndex(idx) self.combo_profile.blockSignals(False) def _on_profile_selected(self, index): if index <= 0: return name = self.combo_profile.currentText() cfg = self._profiles.get(name) if not isinstance(cfg, dict): return if cfg.get('topmost'): self.top_btn.setChecked(True) else: self.top_btn.setChecked(False) if cfg.get('dark'): self.chk_dark.setChecked(True) else: self.chk_dark.setChecked(False) if cfg.get('tray'): self.chk_tray.setChecked(True) else: self.chk_tray.setChecked(False) self._apply_panels(cfg) self._save_app_config() self.info('预设已加载: ' + name) def _save_profile(self): text = self.combo_profile.currentText() if text == '← 选择预设 →': text = '' name, ok = QInputDialog.getText(self, '保存预设', '预设名称:', QLineEdit.Normal, text) if not ok or not name.strip(): return self._profiles[name.strip()] = self._collect_all_config() self._write_profiles() self._reload_profiles_combo(name.strip()) self.info('预设已保存: ' + name.strip()) def _del_profile(self): name = self.combo_profile.currentText() if name == '← 选择预设 →' or name not in self._profiles: return if QMessageBox.question(self, '删除预设', f'确定删除预设「{name}」吗？') \ != QMessageBox.Yes: return del self._profiles[name] self._write_profiles() self._reload_profiles_combo() self.info('预设已删除: ' + name) def closeEvent(self, event): self._save_app_config() if self._tray_enabled and self.tray_icon and self.tray_icon.isVisible(): event.ignore() self.hide() self.tray_icon.showMessage(APP_NAME, '程序已最小化到托盘', QSystemTrayIcon.Information, 2000) return if self.tray_icon: self.tray_icon.hide() super().closeEvent(event) if __name__ == '__main__': QApplication.setHighDpiScaleFactorRoundingPolicy( Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) app = QApplication(sys.argv) app.setApplicationName(APP_NAME) app.setFont(QFont('Microsoft YaHei', 9)) win = MainWindow() win.show() sys.exit(app.exec())