from utils.DBModels.ProtocolModel import ( ModbusTcpMasterVar, ModbusTcpSlaveVar, ModbusRtuMasterVar, ModbusRtuSlaveVar, HartVar, TcRtdVar, AnalogVar, HartSimulateVar ) from model.ProjectModel.GlobalConfigManager import GlobalConfigManager from model.ProjectModel.DeviceManage import DevicesManange from protocol.HartRtuSlaveManager import HartRtuSlaveManager from protocol.TCP.TCPVarManage import * from protocol.TCP.TemToMv import temToMv from protocol.RPC.RpcClient import RpcClient from protocol.RPC.RpcServer import RpcServer from protocol.ModBus.ModbusManager import ModbusManager # from protocol.ProfibusManager import ProfibusManager from utils import Globals import threading import time import json from typing import Dict, Any, Optional class ProtocolManage(object): """通讯变量查找类，用于根据变量名在数据库模型中检索变量信息""" def __init__(self): # 根据全局配置动态获取启用的模型类 self.MODEL_CLASSES = GlobalConfigManager.getEnabledModelClasses() # 根据配置决定是否初始化TCPVarManager（仅当启用IO或TCRTD模块时） if GlobalConfigManager.needsTcpVarManager(): # self.tcpVarManager = TCPVarManager('192.168.1.50', 5055) self.tcpVarManager = TCPVarManager('127.0.0.1', 8000) self.writeTC = [0] * 8 self.writeRTD = [0] * 8 else: self.tcpVarManager = None self.writeTC = None self.writeRTD = None self.RpcClient = None self.RpcServer = None self.varInfoCache = {} # 保持驼峰命名 self.historyDBManage = Globals.getValue('historyDBManage') self.variableValueCache = {} # {varName: value} self.cacheLock = threading.Lock() # Modbus 管理器 self.modbusManager = ModbusManager() # 设置 Modbus 管理器的缓存锁 self.modbusManager.setVariableCache(self.variableValueCache, self.cacheLock, self.varInfoCache) # PROFIBUS管理器 (已删除，准备重新构建) # self.profibusManager = DevicesManange() # self.profibusEnabled = False # HART模拟RTU从站管理器 self.hartRtuSlaveManager = None if GlobalConfigManager.isModuleEnabled('hartSimulateModule'): try: self.hartRtuSlaveManager = HartRtuSlaveManager() print("HART模拟RTU从站管理器已初始化") except Exception as e: print(f"初始化HART模拟RTU从站管理器失败: {e}") self.hartRtuSlaveManager = None self.refreshVarCache() self.readThreadStop = threading.Event() self.readThread = threading.Thread(target=self._backgroundReadAllVariables, daemon=True) self.readThread.start() def clearVarCache(self): """清空变量信息缓存""" self.varInfoCache.clear() def refreshVarCache(self): """重新加载所有变量信息到缓存（可选实现）""" self.varInfoCache.clear() # 重新获取最新的启用模型类列表 self.MODEL_CLASSES = GlobalConfigManager.getEnabledModelClasses() for modelClass in self.MODEL_CLASSES: try: for varInstance in modelClass.select(): varName = getattr(varInstance, 'varName', None) if varName: varData = {} for field in varInstance._meta.sorted_fields: fieldName = field.name varData[fieldName] = getattr(varInstance, fieldName) self.varInfoCache[varName] = { 'modelType': modelClass.__name__, 'variableData': varData } except Exception as e: print(f"刷新缓存时出错: {modelClass.__name__}: {e}") # self.getAllProfibusDevices() # PROFIBUS变量注册 (已删除，准备重新构建) def lookupVariable(self, variableName): """ 根据变量名检索变量信息（优先查缓存） :param variableName: 要查询的变量名 :return: 包含变量信息和模型类型的字典，如果未找到返回None """ if variableName in self.varInfoCache: return self.varInfoCache[variableName] for modelClass in self.MODEL_CLASSES: varInstance = modelClass.getByName(variableName) if varInstance: varData = {} for field in varInstance._meta.sorted_fields: fieldName = field.name varData[fieldName] = getattr(varInstance, fieldName) result = { 'modelType': modelClass.__name__, 'variableData': varData } self.varInfoCache[variableName] = result # 写入缓存 return result # PROFIBUS变量查找 (已删除，准备重新构建) return None def setClentMode(self, clentName, rabbitmqHost='localhost'): if self.RpcClient: self.RpcClient.close() self.RpcClient = RpcClient(clentName, rabbitmqHost, self) # 使用非阻塞方式启动RPC客户端 self.RpcClient.startNonBlocking() def closeClient(self): if self.RpcClient: self.RpcClient.close() self.RpcClient = None def setServerMode(self, rabbitmqHost='localhost'): if self.RpcServer: self.RpcServer.close() self.RpcServer = RpcServer(rabbitmqHost) def addClient(self, clientName): if self.RpcServer: self.RpcServer.addClient(clientName=clientName) else: return def closeServer(self): if self.RpcServer: self.RpcServer.close() self.RpcServer = None def writeVariableValue(self, variableName, value, trigger=None, timeoutMS=2000): """ 根据变量名写入变量值，根据变量类型进行不同处理（不保存到数据库） :param variableName: 变量名 :param value: 要写入的值 :return: 写入成功返回True，否则返回False """ varInfo = self.lookupVariable(variableName) if not varInfo: if self.RpcServer: try: existsVar, clientNames = self.RpcServer.existsVar(variableName) if existsVar: result = self.RpcServer.writeVar(variableName, value) return result.get("result") == "success" if result else False else: return False except Exception as e: print(f"RPC写入变量 {variableName} 失败: {e}") return False return False modelType = varInfo['modelType'] info = varInfo['variableData'] try: # 拆分为四个独立的Modbus协议条件判断 if modelType == 'ModbusTcpMasterVar': res = self.modbusManager.writeModbusTcpMasterValue(info, value) # print(res) if res == 'error': return elif modelType == 'ModbusTcpSlaveVar': res = self.modbusManager.writeModbusTcpSlaveValue(info, value) if res == 'error': return elif modelType == 'ModbusRtuMasterVar': res = self.modbusManager.writeModbusRtuMasterValue(info, value) if res == 'error': return elif modelType == 'ModbusRtuSlaveVar': res = self.modbusManager.writeModbusRtuSlaveValue(info, value) if res == 'error': return # HART协议变量处理 elif modelType == 'HartVar': pass # 温度/RTD变量处理 elif modelType == 'TcRtdVar': if not self.tcpVarManager: print("TCPVarManager未初始化，无法处理TcRtdVar变量") return False channel = int(info['channelNumber']) - 1 varType = info['varType'] compensationVar = float(info['compensationVar']) varModel = info['varModel'] model = self.getModelType(varModel) if self.getModelType(varModel) else localModel if model == localModel: if varType == 'PT100': self.writeRTD[channel] = value else: self.writeTC[channel] = value if varType in ['R', 'S', 'B', 'J', 'T', 'E', 'K', 'N', 'C', 'A', 'PT100'] and model != SimModel: value = temToMv(varType, value + compensationVar) if trigger and varType in ['R', 'S', 'B', 'J', 'T', 'E', 'K', 'N', 'C', 'A']: trigger = TCTrigger if trigger and varType in ['PT100']: trigger = RTDTrigger self.tcpVarManager.writeValue(varType, channel, value, trigger=trigger, model=model, timeoutMS=timeoutMS) # 模拟量变量处理 elif modelType == 'AnalogVar': if not self.tcpVarManager: print("TCPVarManager未初始化，无法处理AnalogVar变量") return False channel = int(info['channelNumber']) - 1 varType = info['varType'] varModel = info['varModel'] model = self.getModelType(varModel) if self.getModelType(varModel) else localModel if info['varType'] in ['AI', 'AO']: value = self.getRealAO(value, info['max'], info['min']) trigger = FPGATrigger if trigger else trigger self.tcpVarManager.writeValue(varType, channel, value, trigger=trigger, model=model, timeoutMS=timeoutMS) # PROFIBUS变量处理 (已删除，准备重新构建) elif modelType == 'ProfibusVar': # PROFIBUS变量写入功能 (已删除，准备重新构建) return False # HART模拟变量处理 elif modelType == 'HartSimulateVar': if not self.hartRtuSlaveManager: print("HART RTU从站管理器未初始化，无法处理HartSimulateVar变量") return False # 根据变量名确定是主变量还是动态变量 varName = variableName.lower() if 'primary' in varName or '主变量' in variableName: success = self.hartRtuSlaveManager.writeVariable('primaryVariable', float(value)) elif 'dynamic1' in varName or '动态变量1' in variableName: success = self.hartRtuSlaveManager.writeVariable('dynamicVariable1', float(value)) elif 'dynamic2' in varName or '动态变量2' in variableName: success = self.hartRtuSlaveManager.writeVariable('dynamicVariable2', float(value)) elif 'dynamic3' in varName or '动态变量3' in variableName: success = self.hartRtuSlaveManager.writeVariable('dynamicVariable3', float(value)) else: # 默认写入主变量 success = self.hartRtuSlaveManager.writeVariable('primaryVariable', float(value)) if not success: print(f"HART模拟变量 {variableName} 写入失败") return False rpcValue = self._prepareExternalValue(value) if self.RpcClient: self.RpcClient.setVarContent( variableName, rpcValue, info.get('min'), info.get('max'), info.get('varType') ) return True except Exception as e: print(f"写入变量值失败: {str(e)}") return False def _backgroundReadAllVariables(self, interval=1.0): """后台读取所有变量，增加间隔减少RPC调用频率""" while not self.readThreadStop.is_set(): try: allVarNames = list(self.getAllVariableNames()) # PROFIBUS区域更新 (已删除，准备重新构建) for varName in allVarNames: if self.readThreadStop.is_set(): break try: value = self._readVariableValueOriginal(varName) with self.cacheLock: self.variableValueCache[varName] = value except Exception as e: # 单个变量读取失败不影响其他变量 print(f"读取变量 {varName} 失败: {e}") continue # 增加间隔，减少RPC调用频率 time.sleep(interval) except Exception as e: print(f"后台读取线程异常: {e}") time.sleep(5) # 异常时等待更长时间 # ==================== 对外共享的 PROFIBUS 接口 (已删除，准备重新构建) ==================== def getAllVariableNames(self): return list(self.varInfoCache.keys()) def _readVariableValueOriginal(self, variableName): varInfo = self.lookupVariable(variableName) value = None if not varInfo: if self.RpcServer: try: # 使用safeRpcCall减少阻塞 existsVar, clientNames = self.safeRpcCall( self.RpcServer.existsVar, variableName ) or (False, None) if existsVar and clientNames: varData = self.safeRpcCall( self.RpcServer.getVarValue, clientNames[0], variableName ) if varData and 'value' in varData: # 安全地转换值 try: value = float(varData['value']) except (ValueError, TypeError): print(f"无法转换变量 {variableName} 的值: {varData['value']}") return None else: return None else: return None except Exception as e: print(f"RPC读取变量 {variableName} 失败: {e}") return None return None modelType = varInfo['modelType'] info = varInfo['variableData'] try: # 拆分为独立的协议条件判断 if modelType == 'ModbusTcpMasterVar': value = self.modbusManager.readModbusTcpMasterValue(info) # print(value) elif modelType == 'ModbusTcpSlaveVar': value = self.modbusManager.readModbusTcpSlaveValue(info) elif modelType == 'ModbusRtuMasterVar': value = self.modbusManager.readModbusRtuMasterValue(info) elif modelType == 'ModbusRtuSlaveVar': value = self.modbusManager.readModbusRtuSlaveValue(info) elif modelType == 'HartVar': pass elif modelType == 'TcRtdVar': if not self.tcpVarManager: return None channel = int(info['channelNumber']) - 1 varType = info['varType'] varModel = info['varModel'] model = self.getModelType(varModel) if model == SimModel: if varType == 'PT100': value = self.tcpVarManager.simRTDData[channel] elif varType in ['R', 'S', 'B', 'J', 'T', 'E', 'K', 'N', 'C', 'A']: value = self.tcpVarManager.simTCData[channel] else: if varType == 'PT100': value = self.writeRTD[channel] if self.writeRTD else 0 elif varType in ['R', 'S', 'B', 'J', 'T', 'E', 'K', 'N', 'C', 'A']: value = self.writeTC[channel] if self.writeTC else 0 elif modelType == 'AnalogVar': if not self.tcpVarManager: return None channel = int(info['channelNumber']) - 1 varType = info['varType'] varModel = info['varModel'] model = self.getModelType(varModel) value = self.tcpVarManager.readValue(varType, channel, model=model) if varType in ['AI', 'AO']: value = self.getRealAI(value, info['max'], info['min']) elif modelType == 'ProfibusVar': # PROFIBUS变量读取功能 (已删除，准备重新构建) # value = None # print(info) # device = self.profibusManager.getDevice(info['deviceName']) # values, qualityValueList = device.getAreaValues(info['id']) # print(values, qualityValueList, info['valueName']) pass elif modelType == 'HartSimulateVar': if not self.hartRtuSlaveManager: return None # 根据变量名确定是主变量还是动态变量 varName = variableName.lower() if 'primary' in varName or '主变量' in variableName: value = self.hartRtuSlaveManager.readVariable('primaryVariable') elif 'dynamic1' in varName or '动态变量1' in variableName: value = self.hartRtuSlaveManager.readVariable('dynamicVariable1') elif 'dynamic2' in varName or '动态变量2' in variableName: value = self.hartRtuSlaveManager.readVariable('dynamicVariable2') elif 'dynamic3' in varName or '动态变量3' in variableName: value = self.hartRtuSlaveManager.readVariable('dynamicVariable3') else: # 默认读取主变量 value = self.hartRtuSlaveManager.readVariable('primaryVariable') if value is not None and value != 'error': externalValue = self._prepareExternalValue(value) if self.RpcClient: self.RpcClient.setVarContent( variableName, externalValue, info.get('min'), info.get('max'), info.get('varType') ) if self.historyDBManage: self.historyDBManage.writeVarValue(variableName, externalValue) return value else: return None except Exception as e: print(f"读取变量值失败: {str(e)}") return None def _prepareExternalValue(self, value): if isinstance(value, (list, dict)): try: return json.dumps(value, ensure_ascii=False) except Exception: return str(value) return value def readVariableValue(self, variableName): with self.cacheLock: # print(self.variableValueCache) # print(variableName,111) if variableName in self.variableValueCache: # print(variableName) return self.variableValueCache[variableName] return self._readVariableValueOriginal(variableName) def recvDeltaT(self): if self.tcpVarManager: return self.tcpVarManager.recvDeltaT() return None def shutdown(self): if self.tcpVarManager: self.tcpVarManager.shutdown() # 关闭所有Modbus通讯 self.modbusManager.stopAllModbus() # 关闭HART RTU从站 if self.hartRtuSlaveManager: self.hartRtuSlaveManager.stopSlave() self.closeClient() self.closeServer() # 关闭后台读取线程 if hasattr(self, 'readThreadStop') and hasattr(self, 'readThread'): self.readThreadStop.set() self.readThread.join(timeout=1) # 关闭PROFIBUS管理器 (已删除，准备重新构建) def getRealAO(self, value, highValue, lowValue): if highValue: lowValue = float(lowValue) highValue = float(highValue) return (16 * (value - lowValue) + 4 * (highValue - lowValue)) / (1000 * (highValue - lowValue)) else: return value / 1000 def getRealAI(self, mA, highValue, lowValue): """将毫安值转换为实际工程值（getRealAO的反向计算）""" try: if highValue: lowValue = float(lowValue) highValue = float(highValue) return (1000 * mA * (highValue - lowValue) - 4 * (highValue - lowValue)) / 16.0 + lowValue else: return mA * 1000 except Exception as e: print(f"工程值转换失败: {str(e)}") return 0.0 def getModelType(self, varModel): if varModel == '本地值': return localModel elif varModel == '远程值': return NetModel elif varModel == '模拟值': return SimModel def disconnectClient(self, clientName): if self.RpcServer: self.RpcServer.removeClient(clientName) def safeRpcCall(self, method, *args, **kwargs): """安全的RPC调用，带超时和异常处理""" if not self.RpcServer: return None try: # 直接调用，但使用较短的超时时间 result = method(*args, **kwargs) return result except Exception as e: # 静默处理异常，避免日志过多 return None # ==================== Modbus 通讯管理方法（委托给 ModbusManager） ==================== def startModbusTcpMaster(self): """启动 Modbus TCP 主站""" return self.modbusManager.startModbusTcpMaster() def stopModbusTcpMaster(self): """停止 Modbus TCP 主站""" return self.modbusManager.stopModbusTcpMaster() def startModbusRtuMaster(self): """启动 Modbus RTU 主站""" return self.modbusManager.startModbusRtuMaster() def stopModbusRtuMaster(self): """停止 Modbus RTU 主站""" return self.modbusManager.stopModbusRtuMaster() def startModbusTcpSlave(self): """启动 Modbus TCP 从站""" return self.modbusManager.startModbusTcpSlave() def stopModbusTcpSlave(self): """停止 Modbus TCP 从站""" return self.modbusManager.stopModbusTcpSlave() def startModbusRtuSlave(self): """启动 Modbus RTU 从站""" return self.modbusManager.startModbusRtuSlave() def stopModbusRtuSlave(self): """停止 Modbus RTU 从站""" return self.modbusManager.stopModbusRtuSlave() def stopAllModbus(self): """停止所有 Modbus 通讯""" return self.modbusManager.stopAllModbus() def getModbusStatus(self): """获取所有 Modbus 通讯状态""" return self.modbusManager.getModbusStatus() def getModbusMessages(self): """获取 Modbus 报文信息""" return self.modbusManager.getMessages() def clearModbusMessages(self): """清空 Modbus 报文记录""" self.modbusManager.clearMessages() # ==================== HART模拟RTU从站管理方法 ==================== def startHartRtuSlave(self): """启动HART RTU从站""" if not self.hartRtuSlaveManager: return False return self.hartRtuSlaveManager.startSlave() def stopHartRtuSlave(self): """停止HART RTU从站""" if self.hartRtuSlaveManager: self.hartRtuSlaveManager.stopSlave() def writeHartVariable(self, variableName: str, value: float) -> bool: """写入HART变量值""" if not self.hartRtuSlaveManager: return False return self.hartRtuSlaveManager.writeVariable(variableName, value) def readHartVariable(self, variableName: str): """读取HART变量值""" if not self.hartRtuSlaveManager: return None return self.hartRtuSlaveManager.readVariable(variableName) # ==================== PROFIBUS设备管理方法 ==================== def getAllProfibusDevices(self): """获取所有PROFIBUS设备和区域信息，只返回变量名不为空的AREA""" try: # 使用DeviceManage的getAllDevice方法获取所有设备信息 devices = DevicesManange.getAllDevice() if not devices: return [] profibusDevices = [] for device in devices: deviceName, proType, masterSlaveModel, areaJson, pvLowerLimit, pvUpperLimit, pvUnit = device # 只处理PROFIBUS相关的设备（DP或PA类型） if proType in ['DP', 'PA']: deviceInfo = { 'deviceName': deviceName, 'proType': proType, 'masterSlaveModel': masterSlaveModel, 'pvLowerLimit': pvLowerLimit, 'pvUpperLimit': pvUpperLimit, 'pvUnit': pvUnit, 'areas': [] } # 解析区域信息，只添加变量名不为空的AREA if areaJson and areaJson != '[]': try: areas = json.loads(areaJson) for area in areas: valueName = area.get('valueName', '') # 只有当变量名不为空时才添加该AREA if valueName and valueName.strip() != '': areaInfo = { 'deviceName': deviceName, 'id': area.get('id', 0), 'type': area.get('type', ''), 'order': area.get('order', 'ABCD'), 'bytes': area.get('bytes', 4), 'valueName': valueName, 'nums': area.get('nums', 1), 'proType': proType, 'masterSlaveModel': masterSlaveModel, } # deviceInfo['areas'].append(areaInfo) self.varInfoCache[valueName] = { 'modelType': 'ProfibusVar', 'variableData': areaInfo } # print(f"缓存变量: {valueName}", areaInfo) except json.JSONDecodeError: # 如果JSON解析失败，跳过该设备 continue # 只有当设备有变量名不为空的AREA时才添加到结果中 # if deviceInfo['areas']: # profibusDevices.append(deviceInfo) # return profibusDevices except Exception as e: print(f"获取PROFIBUS设备信息失败: {str(e)}") return []