commit e9b3100322c4cc1b243cfdff15edc844afe51fb0 Author: YuanHongbin <975559679@qq.com> Date: Fri Aug 7 10:01:41 2026 +0800 初始提交：网络调试助手（串口/TCP/UDP） diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0c2d952 --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +# Python +__pycache__/ +*.pyc +*.pyo + +# PyInstaller build output +build/ +dist/ +*.spec.bak + +# Config/backup artifacts (not part of source) +*.corrupt.bak +corrupt_backup.py +decompiled_network_debugger.py +network_debugger_repaired.py +clean_consts.txt +mod_doc.txt +docs.json \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..50e48b6 --- /dev/null +++ b/README.md @@ -0,0 +1,29 @@ +# 网络调试助手 + +串口 / TCP / UDP 网络调试客户端，基于 Python + PySide6。 + +## 主要功能 + +- **串口**：波特率、数据位、校验位、停止位可配置，自动刷新端口 +- **TCP / UDP**：客户端/服务器模式，支持多客户端 +- **收发**：文本 / 十六进制格式，时间戳、回车发送、定时发送 +- **自动应答**：应答规则多选，按命中模板拼包回复（固定字节/时间戳/长度/CRC/序号/接收数据） +- **批量发送**：多条定时发送，可循环，条目可增删与上下移动 +- **日志**：按小时自动保存，保留一个月循环覆盖 +- **配置**：应答配置与批量发送配置可导出 / 导入（JSON） +- **图标**：应用图标 `网络调试.ico` + +## 运行 + +```bash +pip install PySide6 pyserial +python network_debugger.py +``` + +## 生成 exe + +```bash +python -m PyInstaller "网络调试助手.spec" +``` + +输出在 `dist/` 目录。 \ No newline at end of file diff --git a/network_debugger.py b/network_debugger.py new file mode 100644 index 0000000..72cd220 --- /dev/null +++ b/network_debugger.py @@ -0,0 +1,1900 @@ +#!/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 +from PySide6.QtGui import QFont, QColor, QTextCharFormat, QTextCursor, QIcon +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, +) + +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 = ['小端', '大端'] + + +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): + """回车即发送的发送框，支持↑↓键浏览发送历史""" + + enterPressed = Signal() + histNav = Signal(int) + + 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 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 = "网络调试助手" +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) + 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) + 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) + 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.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 = QPlainTextEdit() + self.recv.setReadOnly(True) + self.recv.setMaximumBlockCount(2000) + self.recv.setStyleSheet('QPlainTextEdit{font-family:Consolas;font-size:11pt}') + 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.setStyleSheet('QPlainTextEdit{font-family:Consolas;font-size:11pt}') + 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_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_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_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_clear_recv) + btn_row.addWidget(self.btn_clear_send) + btn_row.addStretch() + 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) + 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') + 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 _append_colored(self, text: str, color: str): + cur = self.recv.textCursor() + cur.movePosition(QTextCursor.End) + if color: + fmt = QTextCharFormat() + fmt.setForeground(QColor(color)) + cur.setCharFormat(fmt) + cur.insertText(text + '\n') + self.recv.setTextCursor(cur) + bar = self.recv.verticalScrollBar() + bar.setValue(bar.maximum()) + + 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_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 export_config(self): + data = {'reply_rules': copy.deepcopy(self.reply_rules), + 'batch_rows': copy.deepcopy(self.batch_rows)} + 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 = [] + rr = data.get('reply_rules') + if isinstance(rr, list) and rr 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 = data.get('batch_rows') + if isinstance(br, list): + if not all(isinstance(x, dict) for x in br): + QMessageBox.warning(self, '导入失败', '批量发送配置格式无效') + return + 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)} 条') + 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) + + 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.btn_open, 2, 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, + ) + 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 + + +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 服务器 + sel = self.combox_client.currentText() + with self.lock: + if sel and sel in self.clients: + targets = [self.clients[sel]] + else: + 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() + tabs.addTab(SerialPanel(), '串口') + tabs.addTab(NetPanel('tcp'), 'TCP') + tabs.addTab(NetPanel('udp'), 'UDP') + self.setCentralWidget(tabs) + + +if __name__ == '__main__': + app = QApplication(sys.argv) + app.setApplicationName(APP_NAME) + app.setFont(QFont('Microsoft YaHei', 9)) + win = MainWindow() + win.show() + sys.exit(app.exec()) \ No newline at end of file diff --git a/网络调试.ico b/网络调试.ico new file mode 100644 index 0000000..be77dc5 Binary files /dev/null and b/网络调试.ico differ diff --git a/网络调试助手.spec b/网络调试助手.spec new file mode 100644 index 0000000..199274c --- /dev/null +++ b/网络调试助手.spec @@ -0,0 +1,39 @@ +# -*- mode: python ; coding: utf-8 -*- + + +a = Analysis( + ['network_debugger.py'], + pathex=[], + binaries=[], + datas=[('网络调试.ico', '.')], + hiddenimports=[], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + noarchive=False, + optimize=0, +) +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.datas, + [], + name='网络调试助手', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=False, + icon='网络调试.ico', + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +)