''' @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_timer" name = _("hsjl_timer") 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, 0x0A, 0x00, 0x00, 0x00, 0x05, 0x08, 0x06, 0x00, 0x00, 0x00, 0x7C, 0x64, 0x7D, 0x68, 0x00, 0x00, 0x00, 0x01, 0x73, 0x52, 0x47, 0x42, 0x00, 0xAE, 0xCE, 0x1C, 0xE9, 0x00, 0x00, 0x00, 0x04, 0x67, 0x41, 0x4D, 0x41, 0x00, 0x00, 0xB1, 0x8F, 0x0B, 0xFC, 0x61, 0x05, 0x00, 0x00, 0x00, 0x09, 0x70, 0x48, 0x59, 0x73, 0x00, 0x00, 0x12, 0x74, 0x00, 0x00, 0x12, 0x74, 0x01, 0xDE, 0x66, 0x1F, 0x78, 0x00, 0x00, 0x00, 0x14, 0x49, 0x44, 0x41, 0x54, 0x18, 0x57, 0x63, 0x50, 0x54, 0x54, 0xF9, 0x4F, 0x0C, 0x1E, 0xFC, 0x0A, 0x55, 0xFE, 0x03, 0x00, 0xAC, 0x06, 0x45, 0xBB, 0xD3, 0x7A, 0x66, 0x86, 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 E1 EE 00 CF 02 5A FE" #"data": "5A 54 E1 EE 01 00 00 00 01 15 00 55 00 05 55 57 58 80 00 00 00 00 00 00 00\n\ #00 00 00 01 00 00 00 00 75 17 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] self.config["customSendItems"] = default["customSendItems"] print("初始化的customSendItems:", self.config["customSendItems"]) 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): # 定义ID映射：原始ID -> 显示文本（带标识） id_display_map = { "01": "01（主）", "02": "02（从）", "03": "03（从）" } default_id = "01" # 原始ID默认值 # 按钮显示带标识的文本 toggleBtn = QPushButton(f"id: {id_display_map[default_id]}") # 存储原始ID（关键：组帧时用这个值） toggleBtn.setProperty("originalId", default_id) menu = QMenu(toggleBtn) def setId(original_id): # 更新显示文本（带标识） toggleBtn.setText(f"id: {id_display_map[original_id]}") # 更新存储的原始ID toggleBtn.setProperty("originalId", original_id) print(f"id设为 {original_id}") # 确认原始ID正确 # 菜单显示带标识的文本，但传递原始ID for original_id, display_text in id_display_map.items(): menu.addAction(display_text).triggered.connect( lambda checked, id=original_id: setId(id) ) toggleBtn.setMenu(menu) return toggleBtn def createtoIdButton(self): # 目的节点ID同理处理 id_display_map = { "01": "01（主）", "02": "02（从）", "03": "03（从）" } default_id = "01" toggleBtn = QPushButton(f"目的节点id: {id_display_map[default_id]}") toggleBtn.setProperty("originalId", default_id) # 存储原始ID menu = QMenu(toggleBtn) def setId(original_id): toggleBtn.setText(f"目的节点id: {id_display_map[original_id]}") toggleBtn.setProperty("originalId", original_id) print(f"目的节点id设为 {original_id}") for original_id, display_text in id_display_map.items(): menu.addAction(display_text).triggered.connect( lambda checked, id=original_id: setId(id) ) 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", "512kpbs", "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（如"01"） original_id = self.idButton.property("originalId") id_val = int(original_id, 16) # 转为16进制整数 except: id_val = 0x00 # 异常时默认值 frame.append(id_val) # 速率字段对应映射（参考原来速率的0x22等） rate_map = {"128kpbs": 0x22, "512kpbs": 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, toid_text, rate_text, slot_text, image_index=0): # 头部固定 frame = [0x5A, 0x54, 0xE4, 0xEE] if image_index == 0:#两个图片的序号不同 frame.append(0x01) else: frame.append(0x02) frame.extend([0x00, 0x00, 0x00]) # ID字段，转为整数 try: original_toid = self.toidButton.property("originalId") id_val = int(original_toid, 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 # 高8位 low_byte = data_length & 0xFF # 低8位 frame.extend([low_byte, high_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) # 计算数据段长度：从第三个字节（索引2）到dummy byte（当前frame的最后一个字节） # 长度 = 总字节数 - 前两个字节（因为前两个是帧头，不算数据段） data_segment_length = len(frame) -4 # len(frame)此时是：2（头） + 数据段内容长度 + 1（dummy） # 替换固定值0xC4为计算出的长度，再添加后续固定字节 frame.extend([data_segment_length, 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() toid_text = self.toidButton.text() cmd_str = self.buildimageCommand( switch_text, toid_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("电压状态: 未知") self.networktemLabel.setAlignment(Qt.AlignLeft) self.networktemLabel.setFont(QFont("Arial", 10)) self.networkvoltageLabel = QLabel("fgpa温度: 未知") 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.clearLayout(self.customSendItemsLayout) # 创建“开关”、“ID”、“时隙表”、“速率”按钮，供“总控”使用 self.toggleButton = self.createToggleButton() self.customSendItemsLayout.addWidget(self.toggleButton, 1, 0, 1, 2) self.idButton = self.createIdButton() self.customSendItemsLayout.addWidget(self.idButton, 0, 1) self.slotButton = self.createSlotButton() self.customSendItemsLayout.addWidget(self.slotButton, 0, 2) self.rateButton = self.createRateButton() self.customSendItemsLayout.addWidget(self.rateButton, 0, 0) self.toidButton = self.createtoIdButton() self.customSendItemsLayout.addWidget(self.toidButton, 0, 3) # 第一行固定插入“自检”、“图片传输”、“总控” for item in self.config["customSendItems"]: if item["caption"] == "自检项": print("找到自检项，准备添加到(0,0)") # 新增 w = self.createSendItemWidget(item) self.customSendItemsLayout.addWidget(w, 3, 0, 1, 2) 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秒 "512kpbs": 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() toid_text = self.toidButton.text() # 构建当前图片指令 cmd_str = self.buildimageCommand( switch_text, toid_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, 3, 2, 1, 2 ) 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, 1, 2, 1, 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) 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 == 0xE5: # 图像测试回令 result = self.evalimageTestRsp(binary_data) self.showReceiveDataSignal.emit(result) return elif cmd_id == 0xE1: # 自检回令 result = self.evalSelfTestRsp(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 evalSelfTestRsp(self, data: bytes): print('“”“”“”“”“”“”“”“”“”“”“”“') try: # 1. 先解析头部获取数据长度和图片类型标识 header_struct = Struct("<2s B B") # 前11字节结构 header_data = header_struct.unpack(data[:4]) frame_head, cmd_id, node_type= header_data print(data) print(header_data) img_flag = data[7:8] # 第八个字节 print('m') # 3. 动态构建完整结构体 struct_format = f"<2s B B BBBBbbb 2B 2B 2B BBbBB 3s B B B 2s" full_struct = Struct(struct_format) print('km') # 计算结构体要求的总长度 required_len = 30 # 头部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### evalselfTestRsp:: Failed to unpack data (data length {len(data)} byte).\n错误信息: {str(e)}" ret = "----------- 自检回令 -----------\n" ret += f"节点编号: 0x{payload[-2]:02X}\n" ret += "节点状态: " if payload[3] == 0xAA: ret += f"正常 (0x{payload[3]:02X})\n" network_status='正常' elif payload[6] == 0x55: ret += f"异常 (0x{payload[3]:02X})\n" network_status='异常' else: ret += f"【未知错误】(0x{payload[3]:02X})\n" network_status='未知错误' hwOffset = 4 ret += 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" ret += 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" print(payload[3]) ret += f"天线-2 工作 PLL: {(payload[hwOffset+7] >> 15) & 0x01:>1d}\n" ret += f"天线-2 工作频率: {(payload[hwOffset+7] & 0x7FFF)+3700:>4d}\n" print(payload[3]) ret += f"天线-3 工作 PLL: {(payload[hwOffset+8] >> 15) & 0x01:>1d}\n" ret += f"天线-3 工作频率: {(payload[hwOffset+8] & 0x7FFF)+3700:>4d}\n" print(payload[3]) ret += f"天线 PLL 锁定状态: {bin(payload[hwOffset+9])[2:].zfill(8)}\n" print(payload[3]) ret += f"天线工作模式: {bin(payload[hwOffset+10])[2:].zfill(8)}\n" ret += f"天线温度: {payload[hwOffset+11]:>4d}\n" print(payload[3]) ret += "--------------------------------\n" print(payload[3]) print(network_status) phase=payload[14] temp=payload[9] voltage=payload[4] self.networkStatusLabel.setText(f"节点状态: {network_status}") self.networkPhaseLabel.setText(f"天线温度: {phase}") self.networkvoltageLabel.setText(f"电压状态: {voltage}") self.networktemLabel.setText(f"fgpa温度: {temp}") 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('&&&&&') # 图像显示逻辑（添加缩放倍数控制） # 1. 手动设置缩放倍数（可根据需要修改，例如放大2倍） # 图像显示逻辑（修正后） if from_id == 0x01 or from_id == 0x02: # 图像通道 A → 左侧图像控件 if len(self.widgets) >= 1 and isinstance(self.widgets[0], graphWidgets["image"]): # 仅传递图像数据，缩放倍数已在控件初始化时通过config设置 self.widgets[0].onData(image_data) else: self.showReceiveDataSignal.emit("\n### No valid image widget found (A1)\n") 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) 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)