YuanHongbin/network-debugger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
network-debugger/network_debugger.py
T

1900 lines
68 KiB
Python
Raw Permalink Normal View History

初始提交：网络调试助手（串口/TCP/UDP）
2026-08-07 10:01:41 +08:00
#!/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' if '小端' in opt else '>I'
return struct.pack(fmt, int(now.timestamp()))
if opt.startswith('Unix毫秒'):
fmt = '<Q' if '小端' in opt else '>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())
Reference in New Issue Copy Permalink