#!/usr/bin/env python # -*- coding: utf-8 -*- from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QLabel, QPushButton, QFormLayout, QMessageBox, QComboBox, QDoubleSpinBox, QCompleter) from PyQt5.QtCore import Qt, QThread, pyqtSignal from protocol.HART import common class ReadInfoThread(QThread): """读取或写入信息的线程类""" infoReceived = pyqtSignal(dict, str) errorOccurred = pyqtSignal(str, str) def __init__(self, hartComm, command_name, params=None): super().__init__() self.hartComm = hartComm self.command_name = command_name self.params = params if params is not None else {} self._is_running = True def stop(self): self._is_running = False def run(self): try: if not self.hartComm: raise Exception("HART通信对象未初始化") func = getattr(self.hartComm, self.command_name, None) if not func: raise Exception(f"未找到名为 {self.command_name} 的通信方法") response = func(**self.params) if self._is_running: msg = response[0] if isinstance(response, list) and response else response if msg and isinstance(msg, dict): if msg.get("status") != "fail": self.infoReceived.emit(msg, self.command_name) else: error_msg = msg.get("error", "未知错误") self.errorOccurred.emit(f"命令 {self.command_name} 执行失败: {error_msg}", self.command_name) else: self.errorOccurred.emit(f"收到无效响应: {response}", self.command_name) except Exception as e: self.errorOccurred.emit(str(e), self.command_name) class HartSensorConfigWidget(QWidget): def __init__(self, parent=None): super(HartSensorConfigWidget, self).__init__(parent) self.hartComm = None self.readThreads = {} self.units_map = common.UNITS_CODE self.reverse_units_map = common.REVERSE_UNITS_CODE self.transfer_fn_map = common.TRANSFER_FUNCTION_CODE self.reverse_transfer_fn_map = common.REVERSE_TRANSFER_FUNCTION_CODE self.initUI() def initUI(self): mainLayout = QVBoxLayout(self) displayLayout = QHBoxLayout() pvInfoGroup = QGroupBox("传感器信息 (Command 14)") pvInfoForm = QFormLayout() self.serialNoLabel = QLabel("--") self.sensorUpperLimitLabel = QLabel("--") self.sensorLowerLimitLabel = QLabel("--") self.sensorUnitLabel = QLabel("--") self.minSpanLabel = QLabel("--") pvInfoForm.addRow("传感器序列号:", self.serialNoLabel) pvInfoForm.addRow("传感器上限:", self.sensorUpperLimitLabel) pvInfoForm.addRow("传感器下限:", self.sensorLowerLimitLabel) pvInfoForm.addRow("传感器单位:", self.sensorUnitLabel) pvInfoForm.addRow("最小量程:", self.minSpanLabel) self.readPvInfoButton = QPushButton("读取传感器信息") self.readPvInfoButton.clicked.connect(lambda: self.readInfo('readPrimaryVariableInformation')) pvInfoForm.addRow(self.readPvInfoButton) pvInfoGroup.setLayout(pvInfoForm) outputInfoGroup = QGroupBox("输出信息 (Command 15)") outputInfoForm = QFormLayout() self.alarmCodeLabel = QLabel("--") self.transferFnLabel = QLabel("--") self.pvRangeUnitLabel = QLabel("--") self.pvUpperRangeLabel = QLabel("--") self.pvLowerRangeLabel = QLabel("--") self.dampingLabel = QLabel("--") self.writeProtectLabel = QLabel("--") self.privateLabelLabel = QLabel("--") outputInfoForm.addRow("报警动作:", self.alarmCodeLabel) outputInfoForm.addRow("输出特性:", self.transferFnLabel) outputInfoForm.addRow("PV量程单位:", self.pvRangeUnitLabel) outputInfoForm.addRow("PV上限值:", self.pvUpperRangeLabel) outputInfoForm.addRow("PV下限值:", self.pvLowerRangeLabel) outputInfoForm.addRow("阻尼值(s):", self.dampingLabel) outputInfoForm.addRow("写保护:", self.writeProtectLabel) outputInfoForm.addRow("私有标签:", self.privateLabelLabel) self.readOutputInfoButton = QPushButton("读取输出信息") self.readOutputInfoButton.clicked.connect(lambda: self.readInfo('readOutputInformation')) outputInfoForm.addRow(self.readOutputInfoButton) outputInfoGroup.setLayout(outputInfoForm) displayLayout.addWidget(pvInfoGroup) displayLayout.addWidget(outputInfoGroup) mainLayout.addLayout(displayLayout) configLayout = QHBoxLayout() writeGroup = QGroupBox("写入配置") writeForm = QFormLayout() self.rangeUnitComboBox = QComboBox() self.rangeUnitComboBox.setEditable(True) self.rangeUnitComboBox.addItems(sorted(self.units_map.values())) completer = QCompleter(self.rangeUnitComboBox.model()) completer.setFilterMode(Qt.MatchContains) completer.setCaseSensitivity(Qt.CaseInsensitive) self.rangeUnitComboBox.setCompleter(completer) self.upperRangeSpinBox = QDoubleSpinBox() self.upperRangeSpinBox.setRange(-999999.0, 999999.0) self.upperRangeSpinBox.setDecimals(4) self.lowerRangeSpinBox = QDoubleSpinBox() self.lowerRangeSpinBox.setRange(-999999.0, 999999.0) self.lowerRangeSpinBox.setDecimals(4) self.dampingSpinBox = QDoubleSpinBox() self.dampingSpinBox.setRange(0, 100) self.dampingSpinBox.setDecimals(2) self.transferFnComboBox = QComboBox() self.transferFnComboBox.addItems(self.transfer_fn_map.values()) self.setRangeButton = QPushButton("写入量程 (Cmd 35)") self.setRangeButton.clicked.connect(self.onSetRange) self.setDampingButton = QPushButton("写入阻尼 (Cmd 34)") self.setDampingButton.clicked.connect(self.onSetDamping) self.setTransferFnButton = QPushButton("写入输出函数 (Cmd 54)") self.setTransferFnButton.clicked.connect(self.onSetTransferFunction) writeForm.addRow("量程单位:", self.rangeUnitComboBox) writeForm.addRow("量程上限:", self.upperRangeSpinBox) writeForm.addRow("量程下限:", self.lowerRangeSpinBox) writeForm.addRow(self.setRangeButton) writeForm.addRow("阻尼(s):", self.dampingSpinBox) writeForm.addRow(self.setDampingButton) writeForm.addRow("输出函数:", self.transferFnComboBox) writeForm.addRow(self.setTransferFnButton) writeGroup.setLayout(writeForm) configLayout.addWidget(writeGroup) mainLayout.addLayout(configLayout) self.setHartComm(None) def setHartComm(self, hartComm): self.hartComm = hartComm is_connected = hartComm is not None self.readPvInfoButton.setEnabled(is_connected) self.readOutputInfoButton.setEnabled(is_connected) self.setRangeButton.setEnabled(is_connected) self.setDampingButton.setEnabled(is_connected) self.setTransferFnButton.setEnabled(is_connected) def readInfo(self, command_name): if not self.hartComm: QMessageBox.warning(self, "警告", "未连接到HART设备！") return if command_name in self.readThreads and self.readThreads[command_name].isRunning(): return thread = ReadInfoThread(self.hartComm, command_name) thread.infoReceived.connect(self.updateInfo) thread.errorOccurred.connect(self.handleError) thread.finished.connect(lambda: self.onReadFinished(command_name)) self.readThreads[command_name] = thread button = self.getButtonForCommand(command_name) if button: button.setProperty("original_text", button.text()) button.setText("读取中...") button.setEnabled(False) thread.start() def updateInfo(self, msg, command_name): if command_name == 'readPrimaryVariableInformation': serial_no = msg.get('serial_no') self.serialNoLabel.setText(f"{serial_no.hex().upper()}" if serial_no else "--") self.sensorUpperLimitLabel.setText(f"{msg.get('upper_limit', '--'):.4f}") self.sensorLowerLimitLabel.setText(f"{msg.get('lower_limit', '--'):.4f}") self.sensorUnitLabel.setText(common.get_unit_description(msg.get('sensor_limits_code'))) self.minSpanLabel.setText(f"{msg.get('min_span', '--'):.4f}") elif command_name == 'readOutputInformation': self.alarmCodeLabel.setText(self.getAlarmCodeDescription(msg.get('alarm_code'))) transfer_code = msg.get('transfer_fn_code') transfer_text = self.getTransferFunctionDescription(transfer_code) self.transferFnLabel.setText(transfer_text) if transfer_text in self.reverse_transfer_fn_map: self.transferFnComboBox.setCurrentText(transfer_text) unit_code = msg.get('primary_variable_range_code') unit_text = common.get_unit_description(unit_code) self.pvRangeUnitLabel.setText(unit_text) if unit_text in self.reverse_units_map: self.rangeUnitComboBox.setCurrentText(unit_text) upper_val = msg.get('upper_range_value') lower_val = msg.get('lower_range_value') self.pvUpperRangeLabel.setText(f"{upper_val:.4f}" if upper_val is not None else "--") self.pvLowerRangeLabel.setText(f"{lower_val:.4f}" if lower_val is not None else "--") if upper_val is not None: self.upperRangeSpinBox.setValue(upper_val) if lower_val is not None: self.lowerRangeSpinBox.setValue(lower_val) damping_val = msg.get('damping_value') self.dampingLabel.setText(f"{damping_val:.2f}" if damping_val is not None else "--") if damping_val is not None: self.dampingSpinBox.setValue(damping_val) self.writeProtectLabel.setText(self.getWriteProtectDescription(msg.get('write_protect'))) private_label = msg.get('private_label') self.privateLabelLabel.setText(f"{private_label}" if private_label is not None else "--") def onSetRange(self): upper = self.upperRangeSpinBox.value() lower = self.lowerRangeSpinBox.value() unit_text = self.rangeUnitComboBox.currentText() unit_code = self.reverse_units_map.get(unit_text) if unit_code is None: QMessageBox.warning(self, "输入错误", "无效的单位。请从列表中选择或正确输入。") return if upper <= lower: QMessageBox.warning(self, "输入错误", "量程上限必须大于下限。") return self.startWriteThread('writePrimaryVariableRange', {'unitsCode': unit_code, 'upperRange': upper, 'lowerRange': lower}) def onSetDamping(self): damping = self.dampingSpinBox.value() self.startWriteThread('writePrimaryVariableDamping', {'dampingTime': damping}) def onSetTransferFunction(self): fn_text = self.transferFnComboBox.currentText() fn_code = self.reverse_transfer_fn_map.get(fn_text) if fn_code is None: QMessageBox.warning(self, "输入错误", "无效的输出函数。") return self.startWriteThread('writePrimaryVariableOutputFunction', {'transferFunctionCode': fn_code}) def startWriteThread(self, command_name, params): if not self.hartComm: QMessageBox.warning(self, "警告", "未连接到HART设备！") return if command_name in self.readThreads and self.readThreads[command_name].isRunning(): return thread = ReadInfoThread(self.hartComm, command_name, params) thread.infoReceived.connect(self.onWriteSuccess) thread.errorOccurred.connect(self.handleError) thread.finished.connect(lambda: self.onReadFinished(command_name)) self.readThreads[command_name] = thread button = self.getButtonForCommand(command_name) if button: button.setProperty("original_text", button.text()) button.setText("写入中...") button.setEnabled(False) thread.start() def onWriteSuccess(self, msg, command_name): QMessageBox.information(self, "成功", f"命令 {command_name} 执行成功。") self.readInfo('readOutputInformation') def getAlarmCodeDescription(self, code): return {0: "高报", 1: "低报", 2: "保持最后输出"}.get(code, f"未知代码({code})") def getTransferFunctionDescription(self, code): return self.transfer_fn_map.get(code, f"未知代码({code})") def getWriteProtectDescription(self, code): return {0: "未保护", 1: "已保护"}.get(code, f"未知代码({code})") def handleError(self, errorMsg, command_name): QMessageBox.critical(self, "错误", f"执行 {command_name} 时出错:\\n{errorMsg}") def onReadFinished(self, command_name): button = self.getButtonForCommand(command_name) if button: original_text = button.property("original_text") if original_text: button.setText(original_text) button.setEnabled(self.hartComm is not None) def getButtonForCommand(self, command_name): if command_name == 'readPrimaryVariableInformation': return self.readPvInfoButton elif command_name == 'readOutputInformation': return self.readOutputInfoButton elif command_name == 'writePrimaryVariableRange': return self.setRangeButton elif command_name == 'writePrimaryVariableDamping': return self.setDampingButton elif command_name == 'writePrimaryVariableOutputFunction': return self.setTransferFnButton return None