2080 lines
74 KiB
Python
2080 lines
74 KiB
Python
#!/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,
|
||
QToolBar,
|
||
)
|
||
|
||
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 RecvEdit(QPlainTextEdit):
|
||
"""接收区编辑框：Ctrl+滚轮 调整字体大小"""
|
||
|
||
MIN_FONT = 8
|
||
MAX_FONT = 40
|
||
DEFAULT_FONT = 11
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self._font_size = self.DEFAULT_FONT
|
||
|
||
def wheelEvent(self, event):
|
||
if event.modifiers() & Qt.ControlModifier:
|
||
delta = event.angleDelta().y()
|
||
if delta:
|
||
self.set_font_size(self._font_size + (1 if delta > 0 else -1))
|
||
event.accept()
|
||
return
|
||
super().wheelEvent(event)
|
||
|
||
def set_font_size(self, size):
|
||
self._font_size = max(self.MIN_FONT, min(self.MAX_FONT, int(size)))
|
||
font = self.font()
|
||
font.setPointSize(self._font_size)
|
||
self.setFont(font)
|
||
|
||
def font_size(self):
|
||
return self._font_size
|
||
|
||
|
||
class NoWheelSpinBox(QSpinBox):
|
||
def wheelEvent(self, event):
|
||
event.ignore()
|
||
|
||
|
||
class NoWheelComboBox(QComboBox):
|
||
def wheelEvent(self, event):
|
||
event.ignore()
|
||
|
||
|
||
def pack_int(val: int, width: int, endian: str) -> bytes:
|
||
width = int(width)
|
||
if width == 1:
|
||
return bytes([val & 0xFF])
|
||
prefix = '<' if endian == '小端' else '>'
|
||
return struct.pack(prefix + ('H' if width == 2 else 'I'), val & 0xFFFF if width == 2 else val & 0xFFFFFFFF)
|
||
|
||
|
||
def parse_match(hex_text: str):
|
||
"""解析匹配模板，?? 表示任意字节；返回 None 列表为通配"""
|
||
s = ''.join(hex_text.split())
|
||
s = s.replace(',', '').replace(':', '')
|
||
if len(s) % 2 != 0:
|
||
raise ValueError('匹配模板十六进制字符个数必须为偶数')
|
||
out = []
|
||
for i in range(0, len(s), 2):
|
||
pair = s[i:i + 2].upper()
|
||
if pair == '??':
|
||
out.append(None)
|
||
else:
|
||
out.append(int(pair, 16))
|
||
return out
|
||
|
||
|
||
def find_match(hex_text: str, data: bytes) -> int:
|
||
"""子串匹配：返回模板在 data 中的起始下标，未找到返回 -1。?? 为任意字节"""
|
||
try:
|
||
tmpl = parse_match(hex_text)
|
||
except ValueError:
|
||
return -1
|
||
n = len(tmpl)
|
||
if n == 0 or len(data) < n:
|
||
return -1
|
||
for i in range(len(data) - n + 1):
|
||
ok = True
|
||
for j, t in enumerate(tmpl):
|
||
if t is not None and t != data[i + j]:
|
||
ok = False
|
||
break
|
||
if ok:
|
||
return i
|
||
return -1
|
||
|
||
|
||
def match_rule(hex_text: str, data: bytes) -> bool:
|
||
"""子串匹配：接收数据中任一部分与模板一致即命中，?? 为任意字节"""
|
||
return find_match(hex_text, data) != -1
|
||
|
||
APP_NAME = "网络调试助手"
|
||
|
||
|
||
def app_config_path():
|
||
"""返回自动保存配置的文件路径（与主程序同目录）"""
|
||
if getattr(sys, 'frozen', False):
|
||
base = os.path.dirname(sys.executable)
|
||
else:
|
||
base = os.path.dirname(os.path.abspath(__file__))
|
||
return os.path.join(base, 'app_config.json')
|
||
|
||
|
||
BAUDS = ['1200', '2400', '4800', '9600', '19200', '38400', '57600', '115200']
|
||
PARITIES = {'无校验': serial.PARITY_NONE, '奇校验': serial.PARITY_ODD,
|
||
'偶校验': serial.PARITY_EVEN}
|
||
STOPBITS = {'1': serial.STOPBITS_ONE, '1.5': serial.STOPBITS_ONE_POINT_FIVE,
|
||
'2': serial.STOPBITS_TWO}
|
||
|
||
|
||
def ts():
|
||
return datetime.now().strftime('%H:%M:%S.%f')[:-3]
|
||
|
||
|
||
def bytes_to_hex(data: bytes) -> str:
|
||
return ' '.join(f'{b:02X}' for b in data)
|
||
|
||
|
||
def hex_to_bytes(text: str) -> bytes:
|
||
s = ''.join(text.split())
|
||
s = s.replace(',', '').replace(':', '')
|
||
if len(s) % 2 != 0:
|
||
raise ValueError('十六进制字符个数必须为偶数')
|
||
return bytes.fromhex(s)
|
||
|
||
|
||
def get_local_ip() -> str:
|
||
"""返回本机有效的非环回 IPv4 地址"""
|
||
addrs = []
|
||
try:
|
||
addrs = socket.gethostbyname_ex(socket.gethostname())[2]
|
||
except Exception:
|
||
pass
|
||
if not addrs:
|
||
try:
|
||
addrs = [info[4][0] for info in socket.getaddrinfo(
|
||
socket.gethostname(), None) if info[0] == socket.AF_INET]
|
||
except Exception:
|
||
addrs = []
|
||
for a in addrs:
|
||
if a and not a.startswith('127.'):
|
||
return a
|
||
if addrs:
|
||
return addrs[0]
|
||
return '127.0.0.1'
|
||
|
||
|
||
def make_stamp(opt: str) -> bytes:
|
||
now = datetime.now()
|
||
if opt == 'ASCII(YYYY-MM-DD HH:MM:SS)':
|
||
return now.strftime('%Y-%m-%d %H:%M:%S').encode('ascii')
|
||
if opt == 'ASCII(HH:MM:SS)':
|
||
return now.strftime('%H:%M:%S').encode('ascii')
|
||
if opt.startswith('Unix秒'):
|
||
fmt = '<I' 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 = RecvEdit()
|
||
self.recv.setReadOnly(True)
|
||
self.recv.setMaximumBlockCount(2000)
|
||
self.recv.setFont(QFont('Consolas', RecvEdit.DEFAULT_FONT))
|
||
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 collect_config(self):
|
||
return {
|
||
'recv': {
|
||
'display': 'HEX' if self.rb_recv_hex.isChecked() else 'ASCII',
|
||
'ts': self.chk_ts.isChecked(),
|
||
'info': self.chk_info.isChecked(),
|
||
},
|
||
'send': {
|
||
'format': 'HEX' if self.rb_send_hex.isChecked() else '文本',
|
||
'enter_send': self.chk_enter_send.isChecked(),
|
||
'crlf': self.chk_crlf.isChecked(),
|
||
'auto': self.chk_auto.isChecked(),
|
||
'interval': self.spin_interval.value(),
|
||
},
|
||
'reply': {
|
||
'enabled': self.chk_reply.isChecked(),
|
||
'parse': 'HEX' if self.rb_parse_hex.isChecked() else 'ASCII',
|
||
'respond': 'HEX' if self.rb_reply_hex.isChecked() else 'ASCII',
|
||
},
|
||
'font_size': self.recv.font_size(),
|
||
'reply_rules': copy.deepcopy(self.reply_rules),
|
||
'batch_rows': copy.deepcopy(self.batch_rows),
|
||
}
|
||
|
||
def apply_config(self, cfg):
|
||
if not isinstance(cfg, dict):
|
||
return []
|
||
msg = []
|
||
recv = cfg.get('recv')
|
||
if isinstance(recv, dict):
|
||
display = recv.get('display', 'HEX')
|
||
self.rb_recv_hex.setChecked(display == 'HEX')
|
||
self.rb_recv_ascii.setChecked(display != 'HEX')
|
||
if 'ts' in recv:
|
||
self.chk_ts.setChecked(bool(recv['ts']))
|
||
if 'info' in recv:
|
||
self.chk_info.setChecked(bool(recv['info']))
|
||
msg.append('接收设置')
|
||
send = cfg.get('send')
|
||
if isinstance(send, dict):
|
||
fmt = send.get('format', '文本')
|
||
self.rb_send_hex.setChecked(fmt == 'HEX')
|
||
self.rb_send_text.setChecked(fmt != 'HEX')
|
||
if 'enter_send' in send:
|
||
self.chk_enter_send.setChecked(bool(send['enter_send']))
|
||
if 'crlf' in send:
|
||
self.chk_crlf.setChecked(bool(send['crlf']))
|
||
if 'auto' in send:
|
||
self.chk_auto.setChecked(bool(send['auto']))
|
||
if 'interval' in send:
|
||
self.spin_interval.setValue(int(send['interval']))
|
||
msg.append('发送设置')
|
||
reply = cfg.get('reply')
|
||
if isinstance(reply, dict):
|
||
if 'enabled' in reply:
|
||
self.chk_reply.setChecked(bool(reply['enabled']))
|
||
if 'parse' in reply:
|
||
self.rb_parse_hex.setChecked(reply['parse'] == 'HEX')
|
||
self.rb_parse_ascii.setChecked(reply['parse'] != 'HEX')
|
||
if 'respond' in reply:
|
||
self.rb_reply_hex.setChecked(reply['respond'] == 'HEX')
|
||
self.rb_reply_ascii.setChecked(reply['respond'] != 'HEX')
|
||
msg.append('应答设置')
|
||
if 'font_size' in cfg:
|
||
try:
|
||
self.recv.set_font_size(int(cfg['font_size']))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
msg.append('字体大小')
|
||
rr = cfg.get('reply_rules')
|
||
if isinstance(rr, list) and all(isinstance(x, dict) for x in rr):
|
||
for r in rr:
|
||
r.setdefault('enabled', True)
|
||
self.reply_rules = rr
|
||
msg.append(f'应答配置 {len(rr)} 条')
|
||
br = cfg.get('batch_rows')
|
||
if isinstance(br, list) and all(isinstance(x, dict) for x in br):
|
||
self.batch_rows = [{'enabled': bool(x.get('enabled', True)),
|
||
'delay': int(x.get('delay', 0)),
|
||
'text': str(x.get('text', ''))} for x in br]
|
||
msg.append(f'批量发送 {len(br)} 条')
|
||
return msg
|
||
|
||
def export_config(self):
|
||
data = self.collect_config()
|
||
default = f'config_{datetime.now():%Y%m%d_%H%M%S}.json'
|
||
path, _ = QFileDialog.getSaveFileName(self, '导出配置', default,
|
||
'JSON 文件 (*.json);;所有文件 (*)')
|
||
if not path:
|
||
return
|
||
try:
|
||
with open(path, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
self.info(f'配置已导出: {path}')
|
||
except OSError as e:
|
||
self.info(f'导出失败: {e}')
|
||
|
||
def import_config(self):
|
||
path, _ = QFileDialog.getOpenFileName(self, '导入配置', '',
|
||
'JSON 文件 (*.json);;所有文件 (*)')
|
||
if not path:
|
||
return
|
||
try:
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
except Exception as e:
|
||
QMessageBox.warning(self, '导入失败', f'无法读取配置文件:\n{e}')
|
||
return
|
||
if not isinstance(data, dict):
|
||
QMessageBox.warning(self, '导入失败', '配置文件格式无效')
|
||
return
|
||
msg = self.apply_config(data)
|
||
if not msg:
|
||
QMessageBox.warning(self, '导入失败', '配置文件中没有可导入的数据')
|
||
return
|
||
self.info('配置已导入: ' + '，'.join(msg))
|
||
|
||
def toggle_save_log(self, checked: bool):
|
||
if checked:
|
||
d = QFileDialog.getExistingDirectory(self, '选择日志保存目录', '')
|
||
if not d:
|
||
self.chk_save_log.setChecked(False)
|
||
return
|
||
self._log_dir = d
|
||
self._log_fh = None
|
||
self._log_hour = None
|
||
self._rotate_log()
|
||
if self._log_fh:
|
||
self.log_rotate_timer.start(60000)
|
||
self.info(f'日志自动保存已开启 -> {d}（每小时一个文件，保留一个月循环覆盖）')
|
||
else:
|
||
self.chk_save_log.setChecked(False)
|
||
else:
|
||
self.log_rotate_timer.stop()
|
||
self._close_log()
|
||
if self._log_dir:
|
||
self.info('日志自动保存已停止')
|
||
self._log_dir = ''
|
||
|
||
def _log_path_for(self, dt) -> str:
|
||
return os.path.join(self._log_dir, f'log_{dt:%Y%m%d_%H}.txt')
|
||
|
||
def _close_log(self):
|
||
if self._log_fh:
|
||
try:
|
||
self._log_fh.close()
|
||
except OSError:
|
||
pass
|
||
self._log_fh = None
|
||
self._log_hour = None
|
||
|
||
def _rotate_log(self):
|
||
if not self._log_dir:
|
||
return
|
||
now = datetime.now()
|
||
key = now.strftime('%Y%m%d%H')
|
||
if self._log_fh and self._log_hour == key:
|
||
return
|
||
self._close_log()
|
||
try:
|
||
self._log_fh = open(self._log_path_for(now), 'a', encoding='utf-8')
|
||
self._log_hour = key
|
||
except OSError as e:
|
||
self.info(f'无法创建日志文件: {e}')
|
||
self._log_fh = None
|
||
self._cleanup_old_logs()
|
||
|
||
def _cleanup_old_logs(self):
|
||
try:
|
||
files = [f for f in os.listdir(self._log_dir)
|
||
if f.startswith('log_') and f.endswith('.txt')]
|
||
files.sort()
|
||
while len(files) > 720:
|
||
old = files.pop(0)
|
||
try:
|
||
os.remove(os.path.join(self._log_dir, old))
|
||
except OSError:
|
||
pass
|
||
except OSError:
|
||
pass
|
||
|
||
def info(self, text: str):
|
||
self.sigInfo.emit(text)
|
||
|
||
def _append_info(self, text: str):
|
||
if not self.chk_info.isChecked():
|
||
return
|
||
text = f'[{ts()}] INFO: {text}' if self.chk_ts.isChecked() else f'INFO: {text}'
|
||
self._append_line(text, 'gray')
|
||
|
||
|
||
class SerialPanel(BasePanel):
|
||
def __init__(self):
|
||
super().__init__('串口')
|
||
self.ser = None
|
||
self.reader = None
|
||
|
||
def build_top(self) -> QWidget:
|
||
w = QWidget()
|
||
g = QGridLayout(w)
|
||
self.combox_port = QComboBox()
|
||
self.combox_baud = QComboBox()
|
||
self.combox_baud.addItems(BAUDS)
|
||
self.combox_baud.setCurrentText('115200')
|
||
self.combox_data = QComboBox()
|
||
self.combox_data.addItems(['5', '6', '7', '8'])
|
||
self.combox_data.setCurrentText('8')
|
||
self.combox_parity = QComboBox()
|
||
self.combox_parity.addItems(list(PARITIES.keys()))
|
||
self.combox_stop = QComboBox()
|
||
self.combox_stop.addItems(list(STOPBITS.keys()))
|
||
self.combox_stop.setCurrentText('1')
|
||
self.btn_refresh = QPushButton('刷新')
|
||
self.btn_refresh.clicked.connect(self.refresh_ports)
|
||
self.btn_open = QPushButton('打开串口')
|
||
self.btn_open.clicked.connect(self.toggle_serial)
|
||
|
||
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 服务器
|
||
with self.lock:
|
||
targets = list(self.clients.values())
|
||
if not targets:
|
||
QMessageBox.warning(self, '提示', '没有已连接的客户端')
|
||
return False
|
||
ok = False
|
||
for c in targets:
|
||
try:
|
||
c.sendall(data)
|
||
ok = True
|
||
except Exception:
|
||
pass
|
||
return ok
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
@staticmethod
|
||
def _icon_path() -> str:
|
||
if getattr(sys, 'frozen', False):
|
||
base = sys._MEIPASS
|
||
else:
|
||
base = os.path.dirname(os.path.abspath(__file__))
|
||
p = os.path.join(base, '网络调试.ico')
|
||
return p if os.path.exists(p) else ''
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowTitle(APP_NAME)
|
||
self.resize(900, 680)
|
||
self.setMinimumSize(600, 460)
|
||
icon = self._icon_path()
|
||
if icon:
|
||
self.setWindowIcon(QIcon(icon))
|
||
tabs = QTabWidget()
|
||
self.serial_panel = SerialPanel()
|
||
self.tcp_panel = NetPanel('tcp')
|
||
self.udp_panel = NetPanel('udp')
|
||
tabs.addTab(self.serial_panel, '串口')
|
||
tabs.addTab(self.tcp_panel, 'TCP')
|
||
tabs.addTab(self.udp_panel, 'UDP')
|
||
self.setCentralWidget(tabs)
|
||
self.top_btn = QCheckBox('窗口置顶')
|
||
self.top_btn.setToolTip('选中后窗口保持在最前端')
|
||
self.top_btn.toggled.connect(self._toggle_topmost)
|
||
tb = QToolBar('选项')
|
||
tb.setMovable(False)
|
||
tb.addWidget(self.top_btn)
|
||
self.addToolBar(tb)
|
||
self._load_app_config()
|
||
|
||
def _toggle_topmost(self, on: bool):
|
||
try:
|
||
import ctypes
|
||
user32 = ctypes.windll.user32
|
||
user32.SetWindowPos.argtypes = [
|
||
ctypes.c_void_p, ctypes.c_void_p,
|
||
ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int,
|
||
ctypes.c_uint,
|
||
]
|
||
user32.SetWindowPos.restype = ctypes.c_int
|
||
HWND_TOPMOST = -1
|
||
HWND_NOTOPMOST = -2
|
||
SWP_NOMOVE = 0x0002
|
||
SWP_NOSIZE = 0x0001
|
||
SWP_NOACTIVATE = 0x0010
|
||
ok = user32.SetWindowPos(
|
||
ctypes.c_void_p(int(self.winId())),
|
||
ctypes.c_void_p(HWND_TOPMOST if on else HWND_NOTOPMOST),
|
||
0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE) != 0
|
||
if ok:
|
||
return
|
||
except Exception:
|
||
pass
|
||
self._fallback_topmost(on)
|
||
|
||
def _fallback_topmost(self, on: bool):
|
||
QTimer.singleShot(0, lambda: (self.setWindowFlag(Qt.WindowStaysOnTopHint, on),
|
||
self.show()))
|
||
|
||
def _save_app_config(self):
|
||
data = {'topmost': self.top_btn.isChecked(),
|
||
'tabs': {
|
||
'serial': self.serial_panel.collect_config(),
|
||
'tcp': self.tcp_panel.collect_config(),
|
||
'udp': self.udp_panel.collect_config(),
|
||
}}
|
||
try:
|
||
with open(app_config_path(), 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except OSError:
|
||
pass
|
||
|
||
def _load_app_config(self):
|
||
try:
|
||
with open(app_config_path(), 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
except Exception:
|
||
return
|
||
tabs_cfg = data.get('tabs') if isinstance(data, dict) else None
|
||
if isinstance(data, dict) and data.get('topmost'):
|
||
self.top_btn.setChecked(True)
|
||
if not isinstance(tabs_cfg, dict):
|
||
return
|
||
for panel, key in ((self.serial_panel, 'serial'),
|
||
(self.tcp_panel, 'tcp'),
|
||
(self.udp_panel, 'udp')):
|
||
cfg = tabs_cfg.get(key)
|
||
if isinstance(cfg, dict):
|
||
panel.apply_config(cfg)
|
||
|
||
def closeEvent(self, event):
|
||
self._save_app_config()
|
||
super().closeEvent(event)
|
||
|
||
|
||
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()) |