josh/hj_datalink
1
0
Fork 0
forked from 0xha0/df_datalink
Code Issues Pull Requests Releases Activity

feat: 新增图片传输上位机

Browse Source
This commit is contained in:
josh
2025-08-20 10:30:39 +08:00
parent e56afc4a78
commit cb9dc0d934
1 changed files with 915 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/hsjl_plugin_timer.py
+915
View File
@@ -0,0 +1,915 @@
'''
@brief This is an example plugin, can receive and send data
@author
@date
@license LGPL-3.0
'''
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QGridLayout,QHBoxLayout, QTextEdit, QPushButton, QLabel, QScrollArea, QSplitter
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtGui import QFont, QTextCursor
from struct import Struct
try:
from PyQt5.QtWidgets import QFrame
import utils_ui
from plugins.base import Plugin_Base
from conn import ConnectionStatus
from i18n import _
from widgets import statusBar
from plugins.graph_widgets import graphWidgets
from PyQt5.QtWidgets import QPushButton, QMenu
from PyQt5.QtWidgets import QComboBox
from PyQt5 import QtCore
except ImportError:
from COMTool import utils_ui
from COMTool.plugins.base import Plugin_Base
from COMTool.i18n import _
from COMTool.conn import ConnectionStatus
from COMTool.widgets import statusBar
from COMTool.plugins.graph_widgets import graphWidgets
class Plugin(Plugin_Base):
id = "hsjl_time"
name = _("hsjl_time")
showReceiveDataSignal = pyqtSignal(str)
# 图片1的字节数据（已存在，提取为常量）
IMAGE1_DATA = [
0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D, 0x49, 0x48, 0x44, 0x52,
0x00, 0x00, 0x00, 0x10, 0x00, 0x00, 0x00, 0x10, 0x08, 0x06, 0x00, 0x00, 0x00, 0x1F, 0xF3, 0xFF,
0x61, 0x00, 0x00, 0x00, 0x01, 0x73, 0x52, 0x47, 0x42, 0x00, 0xAE, 0xCE, 0x1C, 0xE9, 0x00, 0x00,
0x00, 0x75, 0x49, 0x44, 0x41, 0x54, 0x38, 0x4F, 0x63, 0x64, 0xA0, 0x10, 0x30, 0x52, 0xA8, 0x9F,
0x61, 0xD4, 0x00, 0x06, 0xEC, 0x61, 0x20, 0xD0, 0xFE, 0xDA, 0x81, 0x91, 0x85, 0xF9, 0xCD, 0xFB,
0x52, 0xA1, 0x2B, 0xA0, 0x40, 0x06, 0xF1, 0x41, 0xF4, 0x87, 0x4A, 0xD1, 0x03, 0xE8, 0x81, 0x8E,
0x35, 0x10, 0xF9, 0x3A, 0x5F, 0xD7, 0x33, 0xFE, 0xFF, 0x1F, 0xFF, 0x9F, 0xE1, 0x7F, 0x26, 0xE3,
0x7F, 0x46, 0x3E, 0x06, 0x46, 0xC6, 0x55, 0x8C, 0xFF, 0x19, 0x1C, 0x89, 0x36, 0x00, 0x64, 0x0B,
0xC4, 0x10, 0x86, 0x52, 0x06, 0x06, 0x06, 0x6E, 0x86, 0xFF, 0xFF, 0xC3, 0x3E, 0x56, 0x8A, 0xAD,
0xC6, 0x16, 0xE5, 0x78, 0xA3, 0x11, 0x6C, 0xC8, 0xBF, 0xFF, 0xD7, 0x70, 0x69, 0x06, 0x19, 0x38,
0x9A, 0x0E, 0x06, 0x43, 0x18, 0x00, 0x00, 0x4E, 0x62, 0x24, 0x11, 0xDE, 0x82, 0x35, 0xD8, 0x00,
0x00, 0x00, 0x00, 0x49, 0x45, 0x4E, 0x44, 0xAE, 0x42, 0x60, 0x82
]
# 图片2的字节数据（新添加）
IMAGE2_DATA = [
0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D, 0x49, 0x48, 0x44, 0x52,
0x00, 0x00, 0x00, 0x8B, 0x00, 0x00, 0x00, 0x22, 0x04, 0x03, 0x00, 0x00, 0x00, 0x9D, 0x24, 0x37,
0xEA, 0x00, 0x00, 0x00, 0x2D, 0x50, 0x4C, 0x54, 0x45, 0xFF, 0xFF, 0xFF, 0x27, 0x26, 0x36, 0x27,
0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26,
0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x27, 0x26, 0x36,
0x27, 0x26, 0x36, 0x27, 0x26, 0x36, 0x99, 0xBB, 0x2D, 0xFF, 0x00, 0x00, 0x00, 0x0E, 0x74, 0x52,
0x4E, 0x53, 0x00, 0x11, 0x44, 0x99, 0xEE, 0xAA, 0xBB, 0x22, 0xCC, 0x33, 0x66, 0x88, 0xDD, 0x77,
0x52, 0x6E, 0x24, 0x11, 0x00, 0x00, 0x00, 0x01, 0x62, 0x4B, 0x47, 0x44, 0x00, 0x88, 0x05, 0x1D,
0x48, 0x00, 0x00, 0x00, 0x09, 0x70, 0x48, 0x59, 0x73, 0x00, 0x00, 0x00, 0x48, 0x00, 0x00, 0x00,
0x48, 0x00, 0x46, 0xC9, 0x6B, 0x3E, 0x00, 0x00, 0x00, 0x94, 0x49, 0x44, 0x41, 0x54, 0x48, 0xC7,
0x63, 0x60, 0x00, 0x02, 0x21, 0x25, 0x0A, 0x80, 0x00, 0x03, 0x04, 0x30, 0x99, 0xBC, 0xA3, 0x04,
0x3C, 0x09, 0x02, 0x9B, 0xC2, 0x68, 0x47, 0x91, 0x29, 0x40, 0x90, 0x0E, 0x32, 0xA6, 0x82, 0x52,
0x53, 0xDE, 0x3D, 0x99, 0x00, 0x74, 0x4C, 0x1F, 0xC5, 0xC6, 0xBC, 0xF3, 0x64, 0x60, 0x60, 0xA6,
0xDC, 0x94, 0x77, 0xCF, 0x18, 0x18, 0xEA, 0xA8, 0x60, 0xCC, 0xBB, 0x05, 0x0C, 0xFB, 0xA8, 0x61,
0x4C, 0x00, 0xC3, 0x39, 0x6A, 0x18, 0xF3, 0x80, 0x81, 0x1A, 0xA6, 0x8C, 0x1A, 0x33, 0x6A, 0xCC,
0xA8, 0x31, 0xA3, 0xC6, 0xD0, 0xD1, 0x98, 0x3E, 0x6A, 0x18, 0xE3, 0xC0, 0x70, 0x8F, 0x1A, 0xC6,
0x04, 0x30, 0x94, 0x50, 0xC3, 0x18, 0x05, 0x06, 0x6E, 0x2A, 0x98, 0x02, 0xAC, 0xA7, 0x18, 0xF2,
0x28, 0x37, 0xE6, 0x20, 0xD0, 0x18, 0x31, 0x8A, 0x4D, 0x79, 0x31, 0x01, 0xD4, 0x16, 0xD8, 0x42,
0xA9, 0x31, 0x86, 0x90, 0x96, 0xC9, 0x0D, 0xCA, 0xDC, 0xB2, 0x1D, 0x68, 0x06, 0x00, 0x45, 0x8D,
0x41, 0x11, 0x72, 0x0F, 0xD2, 0x4E, 0x00, 0x00, 0x00, 0x25, 0x74, 0x45, 0x58, 0x74, 0x64, 0x61,
0x74, 0x65, 0x3A, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x00, 0x32, 0x30, 0x31, 0x36, 0x2D, 0x30,
0x35, 0x2D, 0x31, 0x38, 0x54, 0x31, 0x34, 0x3A, 0x33, 0x37, 0x3A, 0x35, 0x34, 0x2B, 0x30, 0x38,
0x3A, 0x30, 0x30, 0x9D, 0x24, 0x28, 0xF9, 0x00, 0x00, 0x00, 0x25, 0x74, 0x45, 0x58, 0x74, 0x64,
0x61, 0x74, 0x65, 0x3A, 0x6D, 0x6F, 0x64, 0x69, 0x66, 0x79, 0x00, 0x32, 0x30, 0x31, 0x36, 0x2D,
0x30, 0x35, 0x2D, 0x31, 0x38, 0x54, 0x31, 0x34, 0x3A, 0x33, 0x37, 0x3A, 0x35, 0x34, 0x2B, 0x30,
0x38, 0x3A, 0x30, 0x30, 0xEC, 0x79, 0x90, 0x45, 0x00, 0x00, 0x00, 0x1E, 0x74, 0x45, 0x58, 0x74,
0x70, 0x73, 0x3A, 0x48, 0x69, 0x52, 0x65, 0x73, 0x42, 0x6F, 0x75, 0x6E, 0x64, 0x69, 0x6E, 0x67,
0x42, 0x6F, 0x78, 0x00, 0x31, 0x33, 0x39, 0x78, 0x33, 0x34, 0x2B, 0x30, 0x2B, 0x30, 0x87, 0x35,
0x32, 0x1E, 0x00, 0x00, 0x00, 0x1C, 0x74, 0x45, 0x58, 0x74, 0x70, 0x73, 0x3A, 0x4C, 0x65, 0x76,
0x65, 0x6C, 0x00, 0x41, 0x64, 0x6F, 0x62, 0x65, 0x2D, 0x33, 0x2E, 0x30, 0x20, 0x45, 0x50, 0x53,
0x46, 0x2D, 0x33, 0x2E, 0x30, 0x0A, 0x9B, 0x70, 0xBB, 0xE3, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45,
0x4E, 0x44, 0xAE, 0x42, 0x60, 0x82
]
# 在类初始化时添加定时器相关变量（onInit方法中）
def onInit(self, config):
self.image_index = 0 # 图片索引：0-图片1，1-图片2
self.cycle_count = 0 # 循环次数计数
self.send_timer = QtCore.QTimer() # 新增：定时器
self.send_timer.setInterval(1000) # 间隔1秒
self.send_timer.timeout.connect(self.send_next_image) # 绑定定时发送函数
default = {
"version": 1,
"customSendItems": [
{
"caption": "遥测",
"icon": "fa5.play-circle",
"data": "5A 54 E6 EE 00 D4 02 5A FE"
},
{
"caption": "图片传输",
"icon": "fa5.play-circle",
"data": "5A 54 E4 EE 00 00 00 01 02 00 BB 89 50 4E 47 0D 0A 1A 0A 00 00 00 0D 49 48 44 52 00 00 00 10 00 00 \n\
00 10 08 06 00 00 00 1F F3 FF 61 00 00 00 01 73 52 47 42 00 AE CE 1C E9 00 00 00 75 49 \n\
44 41 54 38 4F 63 64 A0 10 30 52 A8 9F 61 D4 00 06 EC 61 20 D0 FE DA 81 91 85 F9 CD FB \n\
52 A1 2B A0 40 06 F1 41 F4 87 4A D1 03 E8 81 8E 35 10 F9 3A 5F D7 33 FE FF 1F FF 9F E1 \n\
7F 26 E3 7F 46 3E 06 46 C6 55 8C FF 19 1C 89 36 00 64 0B C4 10 86 52 06 06 06 6E 86 FF \n\
FF C3 3E 56 8A AD C6 16 E5 78 A3 11 6C C8 BF FF D7 70 69 06 19 38 9A 0E 06 43 18 00 00 \n\
4E 62 24 11 DE 82 35 D8 00 00 00 00 49 45 4E 44 AE 42 60 82 00 76 C4 5A FE"
},
{
"caption": "总控",
"icon": "fa5.play-circle",
"data": "5A 54 E3 EE AA 03 22 00 00 A0 06 5A FE"
},
]
}
self.config = config
for k in default:
if k not in self.config:
self.config[k] = default[k]
def createToggleButton(self):
toggleBtn = QPushButton("开关: Off")
menu = QMenu(toggleBtn)
actionOn = menu.addAction("On")
actionOff = menu.addAction("Off")
def setOn():
toggleBtn.setText("开关: On")
print("开关设为 On")
def setOff():
toggleBtn.setText("开关: Off")
print("开关设为 Off")
actionOn.triggered.connect(setOn)
actionOff.triggered.connect(setOff)
toggleBtn.setMenu(menu)
return toggleBtn
def createIdButton(self):
toggleBtn = QPushButton("id: 01")
menu = QMenu(toggleBtn)
ids = ["01", "02", "03"]
def setId(text):
toggleBtn.setText(f"id: {text}")
print(f"id设为 {text}")
for id_text in ids:
menu.addAction(id_text).triggered.connect(lambda checked, t=id_text: setId(t))
toggleBtn.setMenu(menu)
return toggleBtn
def createSlotButton(self):
slotBtn = QPushButton("时隙表: 01") # 默认选中01
menu = QMenu(slotBtn)
slotOptions = ["01", "02", "03"]
def setSlot(text):
slotBtn.setText(f"时隙表: {text}")
print(f"时隙表设为 {text}")
for opt in slotOptions:
menu.addAction(opt).triggered.connect(lambda checked, t=opt: setSlot(t))
slotBtn.setMenu(menu)
return slotBtn
def createRateButton(self):
rateBtn = QPushButton("速率: 128kpbs") # 默认选中01
menu = QMenu(rateBtn)
rateOptions = ["128kpbs", "256kpbs", "2Mbps"]
def setRate(text):
rateBtn.setText(f"速率: {text}")
print(f"速率设为 {text}")
for opt in rateOptions:
menu.addAction(opt).triggered.connect(lambda checked, t=opt: setRate(t))
rateBtn.setMenu(menu)
return rateBtn
def buildBindCommand(self, switch_text, id_text, rate_text, slot_text):
"""
switch_text: "开关: On" 或 "开关: Off"
id_text: 形如 "id: 00", 只取后两位字符串
rate_text: "速率: 01" 对应 1/2/3
slot_text: "时隙表: 01" 对应 1/2/3
返回拼接好的指令字符串，如 "5A 54 E3 EE AA 01 22 01 00 ?? ？？ 5A FE"
"""
# 头部固定
frame = [0x5A, 0x54, 0xE3, 0xEE]
# 开关字段：On -> 0xAA，Off -> 0x55
if switch_text.endswith("On"):
frame.append(0xAA)
else:
frame.append(0x55)
# ID字段，转为整数
try:
id_val = int(id_text.split(":")[1].strip(), 16)
except:
id_val = 0x00
frame.append(id_val)
# 速率字段对应映射（参考原来速率的0x22等）
rate_map = {"128kpbs": 0x22, "256kpbs": 0x33, "2Mbps": 0x44}
rate_key = rate_text.split(":")[1].strip()
rate_val = rate_map.get(rate_key, 0x22)
frame.append(rate_val)
# 时隙表，直接转为整数
try:
slot_val = int(slot_text.split(":")[1].strip())
except:
slot_val = 0x00
frame.append(slot_val)
# 后续固定填充和校验（填充字节+校验和+06 5A FE）
frame.append(0x00) # 填充字节
# 校验和计算（从第3字节开始到最后填充前）
checksum = sum(frame[2:]) & 0xFF
frame.append(checksum)
# 结尾
frame.extend([0x06, 0x5A, 0xFE])
# 转成空格分隔的大写16进制字符串
return " ".join(f"{b:02X}" for b in frame)
def buildimageCommand(self, switch_text, id_text, rate_text, slot_text, image_index=0):
# 头部固定
frame = [0x5A, 0x54, 0xE4, 0xEE, 0x00, 0x00, 0x00]
if image_index == 0:#两个图片的序号不同
frame.append(0x01)
else:
frame.append(0x02)
# ID字段，转为整数
try:
id_val = int(id_text.split(":")[1].strip(), 16)
except:
id_val = 0x01
frame.append(id_val)
# 根据索引选择图片数据
if image_index == 0:
image_data = self.IMAGE1_DATA
else:
image_data = self.IMAGE2_DATA
# 计算图片数据长度（2字节表示，这里使用大端模式）
data_length = len(image_data)
# 拆分长度为高8位和低8位
high_byte = (data_length >> 8) & 0xFF
low_byte = data_length & 0xFF
frame.extend([high_byte, low_byte]) # 替换固定字段为动态长度
# 根据索引选择图片数据
if image_index == 0:
frame.extend(self.IMAGE1_DATA)
else:
frame.extend(self.IMAGE2_DATA)
frame.append(0x00) # 填充字节
# 校验和计算（从第3字节开始到最后填充前）
checksum = sum(frame[2:]) & 0xFF
frame.append(checksum)
# 结尾
frame.extend([0xC4, 0x5A, 0xFE])
# 转成空格分隔的大写16进制字符串
return " ".join(f"{b:02X}" for b in frame)
def send_next_image(self):
if self.cycle_count < 5: # 循环5次
# 读取按钮状态
switch_text = self.toggleButton.text()
id_text = self.idButton.text()
slot_text = self.slotButton.text()
rate_text = self.rateButton.text()
# 发送当前索引对应的图片
cmd_str = self.buildimageCommand(
switch_text, id_text, rate_text, slot_text,
self.image_index # 传入当前图片索引
)
print(f"第{self.cycle_count + 1}次发送，图片{self.image_index + 1}: {cmd_str}")
# 实际发送数据
data_bytes = self.parseSendData(
cmd_str,
encoding=self.configGlobal["encoding"],
isHexStr=True
)
if data_bytes:
self.send(data_bytes, callback=self.onSent)
# 切换图片索引（0<->1）
self.image_index = 1 - self.image_index
self.cycle_count += 1
else:
# 循环结束，重置状态
self.send_timer.stop()
self.cycle_count = 0
self.image_index = 0
# 重新启用按钮
self.image_send_btn.setEnabled(True)
print("循环发送完成")
def onWidgetMain(self, parent):
self.mainWidget = QSplitter(Qt.Vertical)
imageContainer = QSplitter(Qt.Horizontal)
self.widgets = []
for i in range(2):
if "image" in graphWidgets:
imageWidgetClass = graphWidgets["image"]
imageWidget = imageWidgetClass(
hintSignal=self.hintSignal,
rmCallback=lambda widget: None,
send=self.sendData,
config={}
)
self.widgets.append(imageWidget)
frame = QFrame()
frame.setFrameShape(QFrame.StyledPanel)
frame.setFrameShadow(QFrame.Raised)
frame.setStyleSheet("QFrame { border: 1px solid #999; border-radius: 6px; margin: 4px; }")
layout = QVBoxLayout()
layout.setContentsMargins(6, 6, 6, 6)
layout.addWidget(imageWidget)
frame.setLayout(layout)
imageContainer.addWidget(frame)
else:
placeholder = QLabel("No graph widget configured")
placeholder.setAlignment(Qt.AlignCenter)
imageContainer.addWidget(placeholder)
self.mainWidget.addWidget(imageContainer)
self.customSendScroll = QScrollArea()
self.customSendScroll.setMinimumHeight(20)
self.customSendScroll.setWidgetResizable(True)
self.customSendScroll.setVerticalScrollBarPolicy(Qt.ScrollBarAsNeeded)
self.customSendScroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
sendItemsWrapper = QWidget()
self.customSendScroll.setWidget(sendItemsWrapper)
sendItemsOuterWrapper = QWidget()
sendItemsOuterWrapper.setProperty("class", "scrollbar2")
layout0 = QVBoxLayout()
layout0.setContentsMargins(0, 8, 0, 0)
sendItemsOuterWrapper.setLayout(layout0)
layout0.addWidget(self.customSendScroll)
sendItemsLayoutWrapper = QVBoxLayout()
sendItemsLayoutWrapper.setContentsMargins(0, 0, 0, 0)
sendItemsWrapper.setLayout(sendItemsLayoutWrapper)
self.customItems = QWidget()
self.customSendItemsLayout = QGridLayout()
self.customSendItemsLayout.setContentsMargins(8, 4, 8, 4)
self.customItems.setLayout(self.customSendItemsLayout)
sendItemsLayoutWrapper.addWidget(self.customItems)
sendItemsLayoutWrapper.addStretch(0)
self.mainWidget.addWidget(sendItemsOuterWrapper)
# 网络状态标签初始化
self.networkStatusLabel = QLabel("网络连接状态: 未连接")
self.networkStatusLabel.setAlignment(Qt.AlignLeft)
self.networkStatusLabel.setFont(QFont("Arial", 10))
self.networkPhaseLabel = QLabel("网络运行阶段: 未知")
self.networkPhaseLabel.setAlignment(Qt.AlignLeft)
self.networkPhaseLabel.setFont(QFont("Arial", 10))
self.networktemLabel = QLabel("fpga温度: 未知")
self.networktemLabel.setAlignment(Qt.AlignLeft)
self.networktemLabel.setFont(QFont("Arial", 10))
self.networkvoltageLabel = QLabel("电压监测: 未知")
self.networkvoltageLabel.setAlignment(Qt.AlignLeft)
self.networkvoltageLabel.setFont(QFont("Arial", 10))
# 设置主窗口比例
self.mainWidget.setStretchFactor(0, 6) # 图像区域
self.mainWidget.setStretchFactor(1, 4) # 自定义发送区域
return self.mainWidget
def clearLayout(self, layout):
while layout.count():
item = layout.takeAt(0)
widget = item.widget()
if widget:
widget.setParent(None)
def createSendItemWidget(self, item):
caption = item["caption"]
itemWidget = QWidget()
layout = QHBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
itemWidget.setLayout(layout)
send = QPushButton("")
label = QLabel(caption)
icon = item.get("icon", "fa.send")
utils_ui.setButtonIcon(send, icon)
send.setProperty("class", "smallBtn")
def sendCustomData():
self.sendCustomItem(item)
send.clicked.connect(sendCustomData)
layout.addWidget(label)
layout.addWidget(send)
return itemWidget
def onUiInitDone(self):
# 去掉自检项
self.config["customSendItems"] = [item for item in self.config["customSendItems"] if item["caption"] != "自检"]
# 清空旧布局
self.clearLayout(self.customSendItemsLayout)
# 创建“开关”、“ID”、“时隙表”、“速率”按钮，供“总控”使用
self.toggleButton = self.createToggleButton()
self.customSendItemsLayout.addWidget(self.toggleButton, 1, 0)
self.idButton = self.createIdButton()
self.customSendItemsLayout.addWidget(self.idButton, 1, 1)
self.slotButton = self.createSlotButton()
self.customSendItemsLayout.addWidget(self.slotButton, 1, 2)
self.rateButton = self.createRateButton()
self.customSendItemsLayout.addWidget(self.rateButton, 1, 3)
# 第一行固定插入“遥测”、“图片传输”、“总控”
for item in self.config["customSendItems"]:
if item["caption"] == "遥测":
w = self.createSendItemWidget(item)
self.customSendItemsLayout.addWidget(w, 0, 0)
elif item["caption"] == "图片传输":
totalControlWidget = QWidget()
layout = QHBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
totalControlWidget.setLayout(layout)
sendBtn = QPushButton("")
self.image_send_btn = sendBtn
label = QLabel("图片传输")
utils_ui.setButtonIcon(sendBtn, item.get("icon", "fa.send"))
sendBtn.setProperty("class", "smallBtn")
# 初始化定时器（确保只初始化一次）
self.send_timer = QtCore.QTimer()
def start_image_cycle():
# 禁用按钮防止重复触发
sendBtn.setEnabled(False)
# 重置计数和索引
self.cycle_count = 0 # 总循环次数（每次循环包含图片1+图片2）
self.image_index = 0 # 当前图片索引（0:图片1，1:图片2）
# 新增：根据当前速率设置定时器间隔
# 1. 定义速率与间隔的映射关系（单位：毫秒）
rate_interval_map = {
"128kpbs": 1500, # 128kpbs对应1.5秒
"256kpbs": 1000, # 256kpbs对应1秒
"2Mbps": 500 # 2Mbps对应0.5秒
}
# 2. 获取当前选中的速率
rate_text = self.rateButton.text() # 格式为"速率: xxx"
rate_key = rate_text.split(":")[1].strip() # 提取"128kpbs"等关键部分
# 3. 获取对应的间隔时间（默认1秒）
interval = rate_interval_map.get(rate_key, 1000)
# 4. 启动定时器（使用动态计算的间隔）
self.send_timer.start(interval)
# 定义循环发送逻辑
def send_image_cycle():
# 读取按钮状态
switch_text = self.toggleButton.text()
id_text = self.idButton.text()
slot_text = self.slotButton.text()
rate_text = self.rateButton.text()
# 构建当前图片指令
cmd_str = self.buildimageCommand(
switch_text, id_text, rate_text, slot_text, self.image_index
)
print(f"第{self.cycle_count + 1}次循环，图片{self.image_index + 1}: {cmd_str}")
# 发送数据
data_bytes = self.parseSendData(
cmd_str,
encoding=self.configGlobal["encoding"],
isHexStr=True
)
if data_bytes:
self.send(data_bytes, callback=self.onSent)
# 切换图片索引（0<->1）
self.image_index = 1 - self.image_index
# 当图片索引回到0时，代表完成一次完整循环（图片1+图片2）
if self.image_index == 0:
self.cycle_count += 1
# 达到5次循环则停止
if self.cycle_count >= 5:
self.send_timer.stop()
sendBtn.setEnabled(True) # 恢复按钮可用
print("循环发送完成")
# 绑定定时器触发事件
self.send_timer.timeout.connect(send_image_cycle)
# 绑定按钮点击事件（启动循环）
sendBtn.clicked.connect(start_image_cycle)
layout.addWidget(label)
layout.addWidget(sendBtn)
self.customSendItemsLayout.addWidget(totalControlWidget, 0, 1)
elif item["caption"] == "总控":
totalControlWidget = QWidget()
layout = QHBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
totalControlWidget.setLayout(layout)
sendBtn = QPushButton("")
label = QLabel("总控")
utils_ui.setButtonIcon(sendBtn, item.get("icon", "fa.send"))
sendBtn.setProperty("class", "smallBtn")
def sendTotalControl():
# 读取4个按钮当前文本
switch_text = self.toggleButton.text()
id_text = self.idButton.text()
slot_text = self.slotButton.text()
rate_text = self.rateButton.text()
# 动态构造指令字符串S
cmd_str = self.buildBindCommand(switch_text, id_text, rate_text, slot_text)
print(f"总控: {cmd_str}")
data_bytes = self.parseSendData(cmd_str,
encoding=self.configGlobal["encoding"],
isHexStr=True)
if data_bytes:
self.send(data_bytes, callback=self.onSent)
sendBtn.clicked.connect(sendTotalControl)
layout.addWidget(label)
layout.addWidget(sendBtn)
self.customSendItemsLayout.addWidget(totalControlWidget, 0, 2)
# 第三行放4个状态标签
self.customSendItemsLayout.addWidget(self.networkStatusLabel, 2, 0)
self.customSendItemsLayout.addWidget(self.networkPhaseLabel, 2, 1)
self.customSendItemsLayout.addWidget(self.networkvoltageLabel, 2, 2)
self.customSendItemsLayout.addWidget(self.networktemLabel, 2, 3)
# 初始化结构体
self.structStatusCheckRsp = Struct(
"<BB Q IIII IBBBBIB III III III III III i II 3s IIII d Q dBB dB H B 2s BBBbbb HHHBBbB B 3s")
def insertSendItem(self, item = None, load = False):
caption = item["caption"]
itemWidget = QWidget()
layout = QHBoxLayout()
layout.setContentsMargins(0,0,0,0)
itemWidget.setLayout(layout)
send = QPushButton("")
label = QLabel(caption)
if (not "icon" in item) or not item["icon"]:
item["icon"] = "fa.send"
utils_ui.setButtonIcon(send, item["icon"])
send.setProperty("class", "smallBtn")
def sendCustomData(idx):
self.sendCustomItem(self.config["customSendItems"][idx])
send.clicked.connect(lambda: sendCustomData(self.customSendItemsLayout.indexOf(itemWidget)))
layout.addWidget(label)
layout.addWidget(send)
count = self.customSendItemsLayout.count()
row = count // 4 # 每 4 个换行
col = count % 4
self.customSendItemsLayout.addWidget(itemWidget,row,col)
if not load:
self.config["customSendItems"].append(item)
return item
def onReceived(self, data: bytes):
print('>')
print(data)
self.statusBar.addRx(len(data)) # 确保self在此处有定义（类成员）
# 新增：判断数据类型并获取二进制帧数据
binary_data = None
try:
# 尝试解码为十六进制字符串（处理第一种格式）
hex_str = data.decode('utf-8')
# 清理空格和换行
cleaned_hex = hex_str.replace(' ', '').replace('\n', '').replace('\r', '')
# 验证是否为有效的十六进制字符
if all(c in '0123456789ABCDEFabcdef' for c in cleaned_hex):
binary_data = bytes.fromhex(cleaned_hex)
print(f"转换后的二进制数据长度: {len(binary_data)} 字节")
else:
# 不是纯十六进制字符串，视为原始二进制数据
binary_data = data
except UnicodeDecodeError:
# 解码失败，说明是原始二进制数据（处理第二种格式）
binary_data = data
print(f"直接使用原始二进制数据，长度: {len(binary_data)} 字节")
except ValueError as e:
print(f"十六进制转换失败: {e}")
return
if not binary_data:
print("无有效二进制数据")
return
# 第二步：对二进制帧数据解析指令
try:
# 帧头为0x5A54（即b'ZT'），前2字节为帧头，第3字节为cmd_id
if len(binary_data) >= 3 and binary_data[:2] == b'\x5A\x54':
cmd_id = binary_data[2]
print(f"收到指令: cmd_id=0x{cmd_id:02X}")
if cmd_id == 0xE4: # 图像测试回令
result = self.evalimageTestRsp(binary_data)
self.showReceiveDataSignal.emit(result)
return
elif cmd_id == 0xE6: # 遥测回令
result = self.evalStatusCheckRsp(binary_data)
self.showReceiveDataSignal.emit(result)
return
else:
print("未匹配帧头（5A54），进入PNG解析逻辑")
except Exception as e:
print(f"指令解析失败: {e}")
# 第三步：PNG解析逻辑
if not hasattr(self, "png_buffer"):
self.png_buffer = bytearray()
self.png_buffer.extend(binary_data) # 使用处理后的二进制数据累积
png_start = self.png_buffer.find(b'\x89PNG\r\n\x1a\n')
print(f"png_start: {png_start}")
if png_start != -1:
png_candidate = self.png_buffer[png_start:]
iend_pos = png_candidate.rfind(b'IEND\xae\x42\x60\x82')
if iend_pos != -1:
complete_png = png_candidate[:iend_pos + 8]
for widget in self.widgets:
widget.onData(complete_png)
self.png_buffer.clear()
else:
pass
else:
# 超过一定长度仍未找到PNG头则清空，避免内存溢出
if len(self.png_buffer) > 1024 * 1024: # 1MB阈值
self.png_buffer.clear()
def printFrame(self, data: bytes):
ret = ''
for i in range(len(data)):
ret += f'{data[i]:02X}'
ret += '\n' if i%16 == 15 else ' '
ret += '\n'
return ret
def evalStatusCheckRsp(self, data: bytes):
try:
payload = self.structStatusCheckRsp.unpack(data)
except :
return f"\n### evalStatusCheckRsp:: Failed to unpack data (data length {len(data)} byte).\n"
ret = "----------- 遥测回令 -----------\n"
ret += "节点类型: "
if payload[1] == 0xEE:
ret += f"C频段 (0x{payload[1]:02X})\n"
else:
ret += f"【未知错误】(0x{payload[1]:02X})\n"
ret += f"时间戳: {payload[2]/1e5:.2f} ms\n"
sysbusOffset = 3
linkOffset = 7
phyOffset = 36
hwOffset = 47
ret += "\n -- 总线状态\n"
ret += f"配置装订: {payload[sysbusOffset + 0]:08X}\n"
ret += f"接收数据计数: {payload[sysbusOffset + 1]:>6d}\n"
ret += f"发送数据计数: {payload[sysbusOffset + 2]:>6d}\n"
ret += f"遥测请求计数: {payload[sysbusOffset + 3]:>6d}\n"
ret += "\n -- 链路层状态\n"
ret += f"时元号: {payload[linkOffset + 0]:>6d}\n"
ret += f"时隙号: {payload[linkOffset + 1]:>3d}\n"
ret += "当前角色: "
if payload[linkOffset + 2] == 0xAA:
ret += f"簇首 (0x{payload[linkOffset + 2]:02X})\n"
elif payload[linkOffset + 2] == 0xCC:
ret += f"成员 (0x{payload[linkOffset + 2]:02X})\n"
else:
ret += f"【未知错误】(0x{payload[linkOffset + 2]:02X})\n"
# 网络连接状态更新
network_status = "未连接"
if payload[linkOffset + 3] == 0x11:
network_status = "已连接"
elif payload[linkOffset + 3] == 0x00:
network_status = "未连接"
else:
network_status = f"未知状态 (0x{payload[linkOffset + 3]:02X})"
# 网络运行阶段更新
phase = "未知"
if payload[linkOffset + 4] == 0x11:
phase = "建网阶段"
elif payload[linkOffset + 4] == 0x22:
phase = "业务通信"
else:
phase = f"未知阶段 (0x{payload[linkOffset + 4]:02X})"
self.networkStatusLabel.setText(f"网络连接状态: {network_status}")
self.networkPhaseLabel.setText(f"网络运行阶段: {phase}")
ret += f"重建/入网计数: {payload[linkOffset + 5]:>6d}\n"
ret += f"通信天线位号: {payload[linkOffset + 6]:>1d}\n"
ret += f"建网帧发送计数: {payload[linkOffset + 7]:>6d}\n"
ret += f"建网帧接收计数: {payload[linkOffset + 8]:>6d}\n"
ret += f"建网帧正确计数: {payload[linkOffset + 9]:>6d}\n"
ret += f"应答帧发送计数: {payload[linkOffset + 10]:>6d}\n"
ret += f"应答帧接收计数: {payload[linkOffset + 11]:>6d}\n"
ret += f"应答帧正确计数: {payload[linkOffset + 12]:>6d}\n"
ret += f"分发帧发送计数: {payload[linkOffset + 13]:>6d}\n"
ret += f"分发帧接收计数: {payload[linkOffset + 14]:>6d}\n"
ret += f"分发帧正确计数: {payload[linkOffset + 15]:>6d}\n"
ret += f"网管帧发送计数: {payload[linkOffset + 16]:>6d}\n"
ret += f"网管帧接收计数: {payload[linkOffset + 17]:>6d}\n"
ret += f"网管帧正确计数: {payload[linkOffset + 18]:>6d}\n"
ret += f"RTT 帧发送计数: {payload[linkOffset + 19]:>6d}\n"
ret += f"RTT 帧接收计数: {payload[linkOffset + 20]:>6d}\n"
ret += f"RTT 帧正确计数: {payload[linkOffset + 21]:>6d}\n"
ret += f"业务帧发送计数: {payload[linkOffset + 22]:>6d}\n"
ret += f"业务帧接收计数: {payload[linkOffset + 23]:>6d}\n"
ret += f"业务帧正确计数: {payload[linkOffset + 24]:>6d}\n"
ret += f"时间偏差记录: {payload[linkOffset + 25]/1e2:.2f} us\n"
ret += f"发送数据 ID: {payload[linkOffset + 26]:>6d}\n"
ret += f"接收数据 ID: {payload[linkOffset + 27]:>6d}\n"
ret += "\n -- 物理层状态\n"
ret += f"捕获计数: {payload[phyOffset + 0]:>6d}\n"
ret += f"帧同步计数: {payload[phyOffset + 1]:>6d}\n"
ret += f"译码计数: {payload[phyOffset + 2]:>6d}\n"
ret += f"CRC 正确计数:{payload[phyOffset + 3]:>6d}\n"
ret += f"多普勒估计: {payload[phyOffset + 4]:.2f} Hz\n"
ret += f"正确接收时刻: {payload[phyOffset + 5]/1e2:.2f} us\n"
ret += f"捕获峰均比: {payload[phyOffset + 6]:.2f}\n"
ret += f"捕获相位值: {payload[phyOffset + 7]:>3d}\n"
ret += f"捕获频偏通道: {payload[phyOffset + 8]:>3d}\n"
ret += f"帧同步峰均比: {payload[phyOffset + 9]:.2f}\n"
ret += f"帧同步频偏通道: {payload[phyOffset + 10]:>3d}\n"
ret += f"交织输入数据幅度统计:\t0x{payload[phyOffset + 11]:04X}\n"
ret += f"天线控制模式: {bin(payload[phyOffset + 10])[2:].zfill(8)}\n"
ret += "\n -- 硬件状态\n"
voltage = f"FPGA 电压监测: {bin(payload[hwOffset+0])[2:].zfill(8)}\n"
ret += f"时钟锁定状态: {bin(payload[hwOffset+1])[2:].zfill(8)}\n"
ret += f"AD 锁定状态: {bin(payload[hwOffset+2])[2:].zfill(8)}\n"
ret += f"AD 温度 1: {payload[hwOffset+3]:>4d}\n"
ret += f"AD 温度 2: {payload[hwOffset+4]:>4d}\n"
temp = f"FPGA 温度: {payload[hwOffset+5]:>4d}\n"
ret += f"天线-1 工作 PLL: {(payload[hwOffset+6] >> 15) & 0x01:>1d}\n"
ret += f"天线-1 工作频率: {(payload[hwOffset+6] & 0x7FFF)+3700:>4d}\n"
ret += f"天线-2 工作 PLL: {(payload[hwOffset+7] >> 15) & 0x01:>1d}\n"
ret += f"天线-2 工作频率: {(payload[hwOffset+7] & 0x7FFF)+3700:>4d}\n"
ret += f"天线-3 工作 PLL: {(payload[hwOffset+8] >> 15) & 0x01:>1d}\n"
ret += f"天线-3 工作频率: {(payload[hwOffset+8] & 0x7FFF)+3700:>4d}\n"
ret += f"天线 PLL 锁定状态: {bin(payload[hwOffset+9])[2:].zfill(8)}\n"
ret += f"天线工作模式: {bin(payload[hwOffset+10])[2:].zfill(8)}\n"
ret += f"天线温度: {payload[hwOffset+11]:>4d}\n"
print('+')
print(voltage)
print('_')
self.networkvoltageLabel.setText(f"电压监测: {voltage}")
self.networktemLabel.setText(f"fgpa温度: {temp}")
ret += "--------------------------------\n"
return ret + '\n' + self.printFrame(data)
def evalimageTestRsp(self, data: bytes):
print('“”“”“”“”“”“”“”“”“”“”“”“')
try:
# 1. 先解析头部获取数据长度和图片类型标识
header_struct = Struct(">2s B B I B H") # 前11字节结构
header_data = header_struct.unpack(data[:11])
frame_head, cmd_id, node_type, data_id, from_id, data_len = header_data
print(data)
print(header_data)
print(data_len)
# 2. 获取第八个字节的图片类型标识(01/02)
img_flag = data[7:8] # 第八个字节
print('m')
# 3. 动态构建完整结构体
struct_format = f"<2s B B I B H {data_len}s B B B 2s"
full_struct = Struct(struct_format)
print('km')
# 计算结构体要求的总长度
required_len = 11 + data_len + 5 # 头部11 + 数据段data_len + 尾部5
print(f"预期长度: {required_len}, 实际数据长度: {len(data)}")
payload = full_struct.unpack(data)
print('lm')
except Exception as e:
print('x')
return f"\n### evalimageTestRsp:: Failed to unpack data (data length {len(data)} byte).\n错误信息: {str(e)}"
ret = "----------- 图像测试回令 -----------\n"
# 字段索引对应（保留原始变量名）
frame_head = payload[0] # bytes: 2 字节帧头
cmd_id = payload[1] # int: 1 字节指令
node_type = payload[2] # int: 1 字节节点类型
data_id = payload[3] # int: 4 字节数据 ID
from_id = payload[4] # int: 1 字节源节点 ID
data_len = payload[5] # int: 2 字节数据长度
image_data = payload[6] # bytes: 动态长度
padding = payload[7] # int: 1 字节填充
checksum = payload[8] # int: 1 字节校验和
data_len2 = payload[9] # int: 1 字节数据段长度
frame_tail = payload[10] # bytes: 2 字节帧尾
# 格式化输出（保留原始输出格式）
ret += f"帧头: {frame_head.hex().upper()} (ascii: {frame_head.decode(errors='ignore')})\n"
ret += f"指令: 0x{cmd_id:02X}\n"
ret += f"节点类型: 0x{node_type:02X}\n"
ret += f"数据 ID: 0x{data_id:08X}\n"
ret += f"源节点 ID: {from_id}\n"
ret += f"数据长度字段: {data_len} 字节\n"
ret += f"数据段实际长度: {len(image_data)} 字节\n"
ret += f"填充字节: 0x{padding:02X}\n"
ret += f"校验和: 0x{checksum:02X}\n"
ret += f"数据段长度: {data_len2} (十六进制: 0x{data_len2:02X})\n"
ret += f"帧尾: {frame_tail.hex().upper()}\n"
print('&&&&&')
# 图像显示逻辑（保留原始控件调用方式）
# 截取有效图像数据
if from_id == 0x01 or from_id == 0x02:
# 图像通道 A → 左侧图像控件
if len(self.widgets) >= 1 and isinstance(self.widgets[0], graphWidgets["image"]):
self.widgets[0].onData(image_data) # 发送有效数据
else:
self.showReceiveDataSignal.emit("\n### No valid image widget found (A1)\n")
print(from_id)
elif from_id == 0x03:
# 图像通道 B → 右侧图像控件
if len(self.widgets) >= 2 and isinstance(self.widgets[1], graphWidgets["image"]):
self.widgets[1].onData(image_data) # 发送有效数据
else:
self.showReceiveDataSignal.emit("\n### No valid image widget found (A2)\n")
print(from_id)
print('+')
ret += "------------------------------------\n"
return ret + '\n' + self.printFrame(data)
def onSent(self, ok, msg, length, path):
if ok:
self.statusBar.addTx(length)
else:
self.hintSignal.emit("error", _("Error"), _("Send data failed!") + " " + msg)
def sendCustomItem(self, item):
data = item["data"]
dateBytes = self.parseSendData(data, encoding = self.configGlobal["encoding"], isHexStr = True)
if dateBytes:
self.send(dateBytes, callback=self.onSent)