from PyQt5 import QtCore, QtWidgets from PyQt5.QtCore import QSize, Qt, QTimer from PyQt5.QtGui import QIcon from PyQt5.QtWidgets import QSplitter, QPushButton, QFileDialog, QMessageBox, QVBoxLayout from UI.VarManages.VarTable import VarTableView, HartTableView, TcRtdTableView, AnalogTableView, \ HartSimulateTableView, RpcVarTableView from UI.VarManages.MessageWidget import MessageWidget from model.ProjectModel.ProjectManage import ProjectManage from model.ProjectModel.VarManage import ModbusVarManage from utils import Globals # 移除 Celery 相关导入，现在使用 ProtocolManage 和 ModbusManager from protocol.TCP.TemToMv import temToMv from protocol.TCP.Analog import getRealAO import re class VarWidgets(QtWidgets.QWidget): _isPopenOpen = False def __init__(self, modbusType, parent=None): super(VarWidgets, self).__init__(parent) self.modbusType = modbusType self.setupUI() def setupUI(self): self.setAttribute(Qt.WA_StyledBackground, True) self.createBtn = QPushButton(QIcon('./Static/add.png'), '新建变量') self.createBtn.setObjectName('createBtn') self.createBtn.setIconSize(QSize(22, 22)) self.createBtn.clicked.connect(self.createVar) self.forceBtn = QPushButton(QIcon('./Static/start.png'), '批量强制') self.forceBtn.setObjectName('forceBtn') self.forceBtn.setIconSize(QSize(22, 22)) self.forceBtn.clicked.connect(self.forceVar) self.clearBtn = QPushButton(QIcon('./Static/clear.png'), '清除颜色') self.clearBtn.setObjectName('clearBtn') self.clearBtn.setIconSize(QSize(22, 22)) self.clearBtn.clicked.connect(self.clearColour) # self.importBtn = QPushButton(QIcon(':/Static/import.png'), '导入变量') # self.importBtn.setObjectName('importBtn') # self.importBtn.setIconSize(QSize(22, 22)) # self.importBtn.clicked.connect(self.importVar) # self.exportBtn = QPushButton(QIcon(':/Static/export.png'), '导出变量') # self.exportBtn.setObjectName('exportBtn') # self.exportBtn.setIconSize(QSize(22, 22)) # self.exportBtn.clicked.connect(self.exportVar) self.messageBtn = QPushButton(QIcon('./Static/message.png'), '查看报文') self.messageBtn.setObjectName('messageBtn') self.messageBtn.setIconSize(QSize(22, 22)) self.messageBtn.clicked.connect(self.showMessage) self.startProtocolBtn = QPushButton(QIcon('./Static/startProtocol.png'), '开始通讯') self.startProtocolBtn.setObjectName('startProtocolBtn') self.startProtocolBtn.setIconSize(QSize(22, 22)) self.startProtocolBtn.clicked.connect(self.startProtocol) self.varView = VarTableView(self.modbusType, self) self.varView.setObjectName('varView') self.proxy = QtCore.QSortFilterProxyModel(self) self.proxy.setSourceModel(self.varView.model) self.varView.setModel(self.proxy) self.varView.proxy = self.proxy self.timer = QTimer(self) # 将定时器超时信号与槽函数showTime()连接 self.timer.timeout.connect(self.proxy.invalidate) self.timer.start(50) # 启动timer Globals.setValue(str(self.modbusType) + 'Table', self.varView) self.gridLayout = QtWidgets.QGridLayout(self) self.gridLayout.addWidget(self.createBtn,0, 0, 1, 1) self.gridLayout.addWidget(self.forceBtn, 0, 1, 1, 1) self.gridLayout.addWidget(self.messageBtn, 0, 2, 1, 1) self.gridLayout.addWidget(self.startProtocolBtn, 0, 3, 1, 1) self.gridLayout.addWidget(QSplitter(), 0, 4, 1, 20) self.gridLayout.addWidget(self.clearBtn, 0, 4, 1, 1) # self.gridLayout.addWidget(self.importBtn, 0, 24, 1, 1) # self.gridLayout.addWidget(self.exportBtn, 0, 25, 1, 1) self.gridLayout.addWidget(self.varView, 1, 0, 10, 26) self.gridLayout.setSpacing(20) self.gridLayout.setContentsMargins(20, 30, 20, 20) # self.comboBox.currentIndexChanged.connect(self.on_comboBox_currentIndexChanged) self.horizontalHeader = self.varView.horizontalHeader() self.horizontalHeader.sectionClicked.connect(self.on_view_horizontalHeader_sectionClicked) def clearColour(self): Globals.clearValue('forceVars') @QtCore.pyqtSlot(int) def on_view_horizontalHeader_sectionClicked(self, logicalIndex): self.logicalIndex = logicalIndex self.on_comboBox_currentIndexChanged(self.logicalIndex) menuName = '_' + str(logicalIndex) + 'Menu' if not hasattr(self, menuName): setattr(self, menuName, QtWidgets.QMenu(self)) self.menuValues = getattr(self, menuName) menuEdit = QtWidgets.QLineEdit() inputAction = QtWidgets.QWidgetAction(self.menuValues) inputAction.setDefaultWidget(menuEdit) self.menuValues.addAction(inputAction) menuEdit.textChanged.connect(self.on_lineEdit_textChanged) self.menuValues = getattr(self, menuName) self.romoveAction(self.menuValues) self.signalMapper = QtCore.QSignalMapper(self) self.menuValues.mouseReleaseEvent = self._menu_mouseReleaseEvent actionAll = QtWidgets.QAction("All", self) actionAll.triggered.connect(self.on_actionAll_triggered) actionAll.setProperty('canHide', True) actionAll.setCheckable(True) self.menuValues.addAction(actionAll) self.menuValues.addSeparator() valuesUnique = [self.proxy.data(self.proxy.index(row, self.logicalIndex)) for row in range(self.proxy.rowCount()) ] for actionNumber, actionName in enumerate((list(set(valuesUnique)))): action = QtWidgets.QAction(str(actionName), self) self.signalMapper.setMapping(action, actionNumber) action.triggered.connect(self.signalMapper.map) action.setCheckable(True) self.menuValues.addAction(action) self.signalMapper.mapped.connect(self.on_signalMapper_mapped) headerPos =self.varView.mapToGlobal(self.horizontalHeader.pos()) posY = headerPos.y() + self.horizontalHeader.height() posX = headerPos.x() + self.horizontalHeader.sectionPosition(self.logicalIndex) getattr(self, menuName).exec_(QtCore.QPoint(posX, posY)) @QtCore.pyqtSlot() def on_actionAll_triggered(self): # 显示全部 filterColumn = self.logicalIndex filterString = QtCore.QRegExp( "", QtCore.Qt.CaseInsensitive, QtCore.QRegExp.RegExp ) self.proxy.setFilterRegExp(filterString) self.proxy.setFilterKeyColumn(filterColumn) @QtCore.pyqtSlot(int) def on_signalMapper_mapped(self, i): # stringAction = self.signalMapper.mapping(i).text() stringActions = '|'.join([x.text() for x in getattr(self, '_' + str(self.logicalIndex) + 'Menu').actions() if x.isChecked()]) filterColumn = self.logicalIndex filterString = QtCore.QRegExp( stringActions, QtCore.Qt.CaseSensitive, # QtCore.QRegExp.FixedString ) self.proxy.setFilterRegExp(filterString) self.proxy.setFilterKeyColumn(filterColumn) @QtCore.pyqtSlot(str) def on_lineEdit_textChanged(self, text): # 搜索框文字变化函数 search = QtCore.QRegExp( text, QtCore.Qt.CaseInsensitive, QtCore.QRegExp.RegExp ) self.proxy.setFilterRegExp(search) @QtCore.pyqtSlot(int) def on_comboBox_currentIndexChanged(self, index): self.proxy.setFilterKeyColumn(index) def _menu_mouseReleaseEvent(self, event): action = self.menuValues.actionAt(event.pos()) if not action: # 没有找到action就交给QMenu自己处理 return QtWidgets.QMenu.mouseReleaseEvent(self.menuValues, event) if action.property('canHide'): # 如果有该属性则给菜单自己处理 return QtWidgets.QMenu.mouseReleaseEvent(self.menuValues, event) # 找到了QAction则只触发Action action.activate(action.Trigger) def romoveAction(self, menu): # 删除输入框之外的按钮1 for action in menu.actions(): if type(action) != QtWidgets.QWidgetAction: menu.removeAction(action) def _checkAction(self): # 三个action都响应该函数 self.labelInfo.setText('\n'.join(['{}\t选中：{}'.format( action.text(), action.isChecked()) for action in self.menuValues.actions()])) def createVar(self): self.varView.model.append_data(['', '', '', '', '', '', '', '', '', '', '','本地值','int']) def forceVar(self): check = [i for i, x in enumerate(self.varView.model.checkList) if x == 'Checked'] if not check: QMessageBox.information(self, '提示', '请先勾选要强制的变量', QMessageBox.Yes) return protocolManage = Globals.getValue('protocolManage') if not protocolManage: QMessageBox.warning(self, '错误', '协议管理器未初始化', QMessageBox.Yes) return forceVars = Globals.getValue('forceVars') if forceVars is None: forceVars = set() for i in check: varMes = self.varView.model.datas[i] value, name, des, varType, slaveID, address, min, max, order = str(varMes[1]), str(varMes[3]), str(varMes[4]), str(varMes[5]), str(varMes[6]), str(varMes[7]), str(varMes[8]), str(varMes[9]), str(varMes[10]) pattern = re.compile(r'[^0-9\.-]+') if not value or re.findall(pattern, str(value)): QMessageBox.warning(self, '警告', "请输入强制值或数字", QMessageBox.Yes) return if min and max: try: if float(value) < float(min) or float(value) > float(max): QMessageBox.warning(self, '警告', "超出量程范围", QMessageBox.Yes) return except Exception: QMessageBox.warning(self, '警告', "量程范围格式错误", QMessageBox.Yes) return elif min and not max: try: if float(value) < float(min): QMessageBox.warning(self, '警告', "超出量程范围", QMessageBox.Yes) return except Exception: QMessageBox.warning(self, '警告', "量程范围格式错误", QMessageBox.Yes) return elif max and not min: try: if float(value) > float(max): QMessageBox.warning(self, '警告', "超出量程范围", QMessageBox.Yes) return except Exception: QMessageBox.warning(self, '警告', "量程范围格式错误", QMessageBox.Yes) return # 使用新的 ProtocolManage 进行变量写入 try: result = protocolManage.writeVariableValue(name, value) if result: forceVars.add(name) else: QMessageBox.information(self, '提示', f'变量 {name} 写入失败', QMessageBox.Yes) except Exception as e: QMessageBox.warning(self, '错误', f'写入变量 {name} 时发生错误: {str(e)}', QMessageBox.Yes) Globals.setValue('forceVars', forceVars) def importVar(self): if Globals.getValue('currentPro'): path = QFileDialog.getOpenFileName(self, "Choose File", "./","VarExcel (*.xlsx);;All Files (*)")[0] if not path: return result = ModbusVarManage.importModbusVar(path) QMessageBox.question(self.parent(), '提示', result, QMessageBox.Yes) self.varView.model.initTable() def exportVar(self): path = QFileDialog.getSaveFileName(self, "Choose File", "./","VarExcel (*.xlsx);;All Files (*)")[0] if not path: return ModbusVarManage.exportModbusVar(path) def showMessage(self): self.messageWidget = MessageWidget() self.messageWidget.show() def startProtocol(self): protocolManage = Globals.getValue('protocolManage') if not protocolManage: QMessageBox.warning(self, '错误', '协议管理器未初始化', QMessageBox.Yes) return if not self._isPopenOpen: # 根据 modbusType 启动对应的 Modbus 通讯 print(self.modbusType) try: success = False if self.modbusType == 'ModbusTcpMaster': # TCP Master success = protocolManage.startModbusTcpMaster() elif self.modbusType == 'ModbusTcpSlave': # TCP Slave success = protocolManage.startModbusTcpSlave() elif self.modbusType == 'ModbusRtuMaster': # RTU Master success = protocolManage.startModbusRtuMaster() elif self.modbusType == 'ModbusRtuSlave': # RTU Slave success = protocolManage.startModbusRtuSlave() if success: self._isPopenOpen = True self.startProtocolBtn.setText('停止通讯') self.startProtocolBtn.setIcon(QIcon('./Static/pause.png')) else: QMessageBox.warning(self, '错误', 'Modbus 通讯启动失败', QMessageBox.Yes) except Exception as e: QMessageBox.warning(self, '错误', f'启动通讯时发生错误: {str(e)}', QMessageBox.Yes) else: # 停止对应的 Modbus 通讯 try: success = False if self.modbusType == 'ModbusTcpMaster': # TCP Master success = protocolManage.stopModbusTcpMaster() elif self.modbusType == 'ModbusTcpSlave': # TCP Slave success = protocolManage.stopModbusTcpSlave() elif self.modbusType == 'ModbusRtuMaster': # RTU Master success = protocolManage.stopModbusRtuMaster() elif self.modbusType == 'ModbusRtuSlave': # RTU Slave success = protocolManage.stopModbusRtuSlave() if success: self._isPopenOpen = False self.startProtocolBtn.setText('开始通讯') self.startProtocolBtn.setIcon(QIcon('./Static/startProtocol.png')) else: QMessageBox.warning(self, '错误', 'Modbus 通讯停止失败', QMessageBox.Yes) except Exception as e: QMessageBox.warning(self, '错误', f'停止通讯时发生错误: {str(e)}', QMessageBox.Yes) def initIcon(self): self._isPopenOpen = False self.startProtocolBtn.setText('开始通讯') self.startProtocolBtn.setIcon(QIcon('./Static/startProtocol.png')) class HartWidgets(VarWidgets): def __init__(self, parent=None): super(HartWidgets, self).__init__(parent) def setupUI(self): self.setAttribute(Qt.WA_StyledBackground, True) self.startProtocolBtn = QPushButton(QIcon('./Static/startProtocol.png'), '开始通讯') self.startProtocolBtn.setObjectName('startProtocolBtn') self.startProtocolBtn.setIconSize(QSize(22, 22)) self.startProtocolBtn.clicked.connect(self.startProtocol) self.varView = HartTableView() self.varView.setObjectName('HartTable') self.proxy = QtCore.QSortFilterProxyModel(self) self.proxy.setSourceModel(self.varView.model) self.varView.setModel(self.proxy) self.varView.proxy = self.proxy self.timer = QTimer(self) # 将定时器超时信号与槽函数showTime()连接 self.timer.timeout.connect(self.proxy.invalidate) self.timer.start(50) # 启动timer Globals.setValue('HartTable', self.varView) self.gridLayout = QtWidgets.QGridLayout(self) self.gridLayout.addWidget(self.startProtocolBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.varView, 1, 0, 10, 26) self.gridLayout.setSpacing(20) self.gridLayout.setContentsMargins(20, 30, 20, 20) # self.comboBox.currentIndexChanged.connect(self.on_comboBox_currentIndexChanged) self.horizontalHeader = self.varView.horizontalHeader() self.horizontalHeader.sectionClicked.connect(self.on_view_horizontalHeader_sectionClicked) def startProtocol(self): pass class TcRtdWidgets(VarWidgets): def __init__(self, parent=None): super(TcRtdWidgets, self).__init__(parent) def setupUI(self): self.setAttribute(Qt.WA_StyledBackground, True) self.forceBtn = QPushButton(QIcon('./Static/start.png'), '批量强制') self.forceBtn.setObjectName('forceBtn') self.forceBtn.setIconSize(QSize(22, 22)) self.forceBtn.clicked.connect(self.forceVar) self.startProtocolBtn = QPushButton(QIcon('./Static/startProtocol.png'), '开始通讯') self.startProtocolBtn.setObjectName('startProtocolBtn') self.startProtocolBtn.setIconSize(QSize(22, 22)) self.startProtocolBtn.clicked.connect(self.startProtocol) self.clearBtn = QPushButton(QIcon('./Static/clear.png'), '清除颜色') self.clearBtn.setObjectName('clearBtn') self.clearBtn.setIconSize(QSize(22, 22)) self.clearBtn.clicked.connect(self.clearColour) self.varView = TcRtdTableView() self.varView.setObjectName('TcRtdTable') self.proxy = QtCore.QSortFilterProxyModel(self) self.proxy.setSourceModel(self.varView.model) self.varView.setModel(self.proxy) self.varView.proxy = self.proxy self.timer = QTimer(self) # 将定时器超时信号与槽函数showTime()连接 self.timer.timeout.connect(self.proxy.invalidate) self.timer.start(50) # 启动timer Globals.setValue('TcRtdTable', self.varView) self.gridLayout = QtWidgets.QGridLayout(self) # self.gridLayout.addWidget(self.createBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.forceBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.startProtocolBtn, 0, 1, 1, 1) self.gridLayout.addWidget(self.clearBtn, 0, 2, 1, 1) self.gridLayout.addWidget(self.varView, 1, 0, 10, 26) self.gridLayout.setSpacing(20) self.gridLayout.setContentsMargins(20, 30, 20, 20) self.horizontalHeader = self.varView.horizontalHeader() self.horizontalHeader.sectionClicked.connect(self.on_view_horizontalHeader_sectionClicked) def createVar(self): self.varView.model.append_data(['', '', '', '', '', '', '', '', '','']) def forceVar(self): check = [i for i, x in enumerate(self.varView.model.checkList) if x == 'Checked'] if not check: QMessageBox.information(self, '提示', '请先勾选要强制的变量', QMessageBox.Yes) return forceVars = Globals.getValue('forceVars') if forceVars is None: forceVars = set() for i in check: value = self.varView.model.datas[i][1] varName = self.varView.model.datas[i][3] min_ = self.varView.model.datas[i][7] max_ = self.varView.model.datas[i][8] pattern = re.compile(r'[^0-9\.-]+') if not value or re.findall(pattern, str(value)): QMessageBox.warning(self, '警告', "请输入强制值或数字", QMessageBox.Yes) return if min_ and max_: try: if float(value) < float(min_) or float(value) > float(max_): QMessageBox.warning(self, '警告', "超出量程范围", QMessageBox.Yes) return except Exception: QMessageBox.warning(self, '警告', "量程范围格式错误", QMessageBox.Yes) return res = Globals.getValue('protocolManage').writeVariableValue(varName, value) if res: forceVars.add(varName) else: QMessageBox.information(self, '提示', f'变量 {varName} 写入失败', QMessageBox.Yes) Globals.setValue('forceVars', forceVars) # self.varView.model.refreshValueCache() # self.varView.model.layoutChanged.emit() def startProtocol(self): pass class AnalogWidgets(VarWidgets): def __init__(self, parent=None): super(AnalogWidgets, self).__init__(parent) def setupUI(self): self.setAttribute(Qt.WA_StyledBackground, True) self.forceBtn = QPushButton(QIcon('./Static/start.png'), '批量强制') self.forceBtn.setObjectName('forceBtn') self.forceBtn.setIconSize(QSize(22, 22)) self.forceBtn.clicked.connect(self.forceVar) self.startProtocolBtn = QPushButton(QIcon('./Static/startProtocol.png'), '开始通讯') self.startProtocolBtn.setObjectName('startProtocolBtn') self.startProtocolBtn.setIconSize(QSize(22, 22)) self.startProtocolBtn.clicked.connect(self.startProtocol) self.clearBtn = QPushButton(QIcon('./Static/clear.png'), '清除颜色') self.clearBtn.setObjectName('clearBtn') self.clearBtn.setIconSize(QSize(22, 22)) self.clearBtn.clicked.connect(self.clearColour) self.varView = AnalogTableView() self.varView.setObjectName('AnalogTable') self.proxy = QtCore.QSortFilterProxyModel(self) self.proxy.setSourceModel(self.varView.model) self.varView.setModel(self.proxy) self.varView.proxy = self.proxy self.timer = QTimer(self) # 将定时器超时信号与槽函数showTime()连接 self.timer.timeout.connect(self.proxy.invalidate) self.timer.start(50) # 启动timer Globals.setValue('AnalogTable', self.varView) self.gridLayout = QtWidgets.QGridLayout(self) # self.gridLayout.addWidget(self.createBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.forceBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.startProtocolBtn, 0, 1, 1, 1) self.gridLayout.addWidget(self.clearBtn, 0, 2, 1, 1) self.gridLayout.addWidget(self.varView, 1, 0, 10, 26) self.gridLayout.setSpacing(20) self.gridLayout.setContentsMargins(20, 30, 20, 20) self.horizontalHeader = self.varView.horizontalHeader() self.horizontalHeader.sectionClicked.connect(self.on_view_horizontalHeader_sectionClicked) def createVar(self): self.varView.model.append_data(['', '', '', '', '', '', '', '', '']) def forceVar(self): check = [i for i, x in enumerate(self.varView.model.checkList) if x == 'Checked'] if not check: QMessageBox.information(self, '提示', '请先勾选要强制的变量', QMessageBox.Yes) return forceVars = Globals.getValue('forceVars') if forceVars is None: forceVars = set() for i in check: value = self.varView.model.datas[i][1] varName = self.varView.model.datas[i][3] min_ = self.varView.model.datas[i][7] max_ = self.varView.model.datas[i][8] pattern = re.compile(r'[^0-9\.-]+') if not value or re.findall(pattern, str(value)): QMessageBox.warning(self, '警告', "请输入强制值或数字", QMessageBox.Yes) return if min_ and max_: try: if float(value) < float(min_) or float(value) > float(max_): QMessageBox.warning(self, '警告', "超出量程范围", QMessageBox.Yes) return except Exception: QMessageBox.warning(self, '警告', "量程范围格式错误", QMessageBox.Yes) return res = Globals.getValue('protocolManage').writeVariableValue(varName, float(value)) if res: forceVars.add(varName) else: QMessageBox.information(self, '提示', f'变量 {varName} 写入失败', QMessageBox.Yes) Globals.setValue('forceVars', forceVars) # self.varView.model.refreshValueCache() # self.varView.model.layoutChanged.emit() def startProtocol(self): pass class HartSimulateWidgets(VarWidgets): def __init__(self, parent=None): super(HartSimulateWidgets, self).__init__(parent) def setupUI(self): self.setAttribute(Qt.WA_StyledBackground, True) self.startProtocolBtn = QPushButton(QIcon('./Static/startProtocol.png'), '开始通讯') self.startProtocolBtn.setObjectName('startProtocolBtn') self.startProtocolBtn.setIconSize(QSize(22, 22)) self.startProtocolBtn.clicked.connect(self.startProtocol) self.varView = HartSimulateTableView() self.varView.setObjectName('HartSimulateTable') self.proxy = QtCore.QSortFilterProxyModel(self) self.proxy.setSourceModel(self.varView.model) self.varView.setModel(self.proxy) self.varView.proxy = self.proxy self.timer = QTimer(self) # 将定时器超时信号与槽函数showTime()连接 self.timer.timeout.connect(self.proxy.invalidate) self.timer.start(50) # 启动timer Globals.setValue('HartSimulateTable', self.varView) self.gridLayout = QtWidgets.QGridLayout(self) self.gridLayout.addWidget(self.startProtocolBtn, 0, 0, 1, 1) self.gridLayout.addWidget(self.varView, 1, 0, 10, 26) self.gridLayout.setSpacing(20) self.gridLayout.setContentsMargins(20, 30, 20, 20) self.horizontalHeader = self.varView.horizontalHeader() self.horizontalHeader.sectionClicked.connect(self.on_view_horizontalHeader_sectionClicked) def startProtocol(self): pass