You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
569 lines
19 KiB
Python
569 lines
19 KiB
Python
|
|
import openpyxl
|
|
|
|
from utils.DBModels.ProtocolModel import ModbusTcpMasterVar, ModbusRtuMasterVar, ModbusRtuSlaveVar,\
|
|
ModbusTcpSlaveVar, HartVar, TcRtdVar, AnalogVar, FFSimulateVar, HartSimulateVar
|
|
from utils import Globals
|
|
|
|
# ID 从站地址 变量名 变量描述 功能码 寄存器地址 工程量下限 工程量上限
|
|
|
|
|
|
class ModbusVarManage(object):
|
|
ModBusVarClass = {
|
|
'ModbusTcpMaster': ModbusTcpMasterVar,
|
|
'ModbusTcpSlave': ModbusRtuMasterVar,
|
|
'ModbusRtuMaster': ModbusRtuSlaveVar,
|
|
'ModbusRtuSlave': ModbusTcpSlaveVar,
|
|
}
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
|
|
@classmethod
|
|
def getVarClass(self, modbusType):
|
|
"""根据类型获取对应的变量类"""
|
|
return self.ModBusVarClass.get(modbusType)
|
|
|
|
@classmethod
|
|
def importModbusVar(self, path):
|
|
# 导入modbus变量表
|
|
wb = openpyxl.load_workbook(path)
|
|
ws = wb.active
|
|
IDIndex = None
|
|
slaveIndex = None
|
|
nameIndex = None
|
|
desIndex = None
|
|
typeIndex = None
|
|
addrIndex = None
|
|
minIndex = None
|
|
maxIndex = None
|
|
|
|
for index, cell in enumerate(list(ws.iter_rows())[0]):
|
|
if cell.value in ['id', 'ID']:
|
|
IDIndex = index
|
|
if cell.value == '从站地址':
|
|
slaveIndex = index
|
|
if cell.value == '变量名':
|
|
nameIndex = index
|
|
if cell.value == '变量描述':
|
|
desIndex = index
|
|
if cell.value == '变量类型':
|
|
typeIndex = index
|
|
if cell.value == '寄存器地址':
|
|
addrIndex = index
|
|
if cell.value == '工程量下限':
|
|
minIndex = index
|
|
if cell.value == '工程量上限':
|
|
maxIndex = index
|
|
# print(IDIndex, slaveIndex, nameIndex, desIndex, typeIndex, addrIndex, minIndex, maxIndex)
|
|
if IDIndex == None or slaveIndex == None or nameIndex == None or desIndex == None or typeIndex == None or addrIndex == None or minIndex == None or maxIndex == None:
|
|
# print('表头错误')
|
|
return '表头错误'
|
|
errorConList = []
|
|
for index, row in enumerate(list(ws.iter_rows())[1:]):
|
|
try:
|
|
l = [str(x.value) for x in row]
|
|
varName = l[nameIndex]
|
|
varType = l[typeIndex]
|
|
des = '' if l[desIndex] == 'None' else l[desIndex]
|
|
address = l[addrIndex]
|
|
slaveID = l[slaveIndex]
|
|
min = '' if l[minIndex] == 'None' else l[minIndex]
|
|
max = '' if l[maxIndex] == 'None' else l[maxIndex]
|
|
if {varName, varType, address, slaveID} & {'None'}:
|
|
errorConList.append('第{}行导入失败，有关键字段为空'.format(str(index + 1)))
|
|
continue
|
|
|
|
if varType not in ['0', '1', '3', '4']:
|
|
errorConList.append('第{}行导入失败，变量类型有误'.format(str(index + 1)))
|
|
continue
|
|
if ModBusVar.getByName(varName):
|
|
ModBusVar.update(varType = varType, description = des, address = address, slaveID = slaveID, min = min, max = max, order = 'int').where(ModBusVar.varName == varName).execute()
|
|
else:
|
|
ModBusVarModel = ModBusVar()
|
|
ModBusVarModel.createVar(varName = varName, varType = varType, des = des, address = address, slaveID = slaveID, min = min, max = max, order = 'int')
|
|
except Exception as e:
|
|
errorConList.append('第{}行导入失败，{}'.format(str(index + 1), str(e)))
|
|
continue
|
|
else:
|
|
if errorConList:
|
|
return '\r\n'.join(errorConList)
|
|
else:
|
|
return '导入成功'
|
|
|
|
@classmethod
|
|
def exportModbusVar(self, excelPath):
|
|
vars = ModBusVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
|
|
wb = openpyxl.Workbook()
|
|
# Create a new worksheet
|
|
ws = wb.active
|
|
ws.title = "Modbus Variables"
|
|
ws.append(['ID', '从站地址', '变量名', '变量描述', '变量类型', '寄存器地址', '工程量下限', '工程量上限'])
|
|
for var in vars:
|
|
ws.append([var.id, var.slaveID, var.varName, var.description, var.varType, var.address, var.min, var.max])
|
|
# Save the workbook
|
|
wb.save(excelPath)
|
|
|
|
|
|
|
|
@classmethod
|
|
def createVar(self, varName, varType, des, address, slaveID, min, max, order, modbusType):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
varType = str(varType)
|
|
address = str(address)
|
|
slaveID = str(slaveID)
|
|
min = str(min)
|
|
max = str(max)
|
|
order = order
|
|
if self.getVarClass(modbusType).getByName(name):
|
|
return 1
|
|
else:
|
|
|
|
varModel =self.getVarClass(modbusType)()
|
|
varModel.createVar(varName = varName, varType = varType, des = des, address = address, slaveID = slaveID, min = min, max = max, order = order)
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name, modbusType):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
self.getVarClass(modbusType).deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des, varType, slaveID, address, min, max, order, modbusType):
|
|
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
varType = str(varType)
|
|
slaveID = str(slaveID)
|
|
address = str(address)
|
|
min = str(min)
|
|
max = str(max)
|
|
if Nname == name:
|
|
self.getVarClass(modbusType).update(varName = Nname, description = des, varType = varType, address = address, slaveID = slaveID, min = min, max = max, order = order).where(self.getVarClass(modbusType).varName == name).execute()
|
|
elif self.getVarClass(modbusType).getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not self.getVarClass(modbusType).getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
self.getVarClass(modbusType).update(varName = Nname, description = des, varType = varType, address = address, slaveID = slaveID, min = min, max = max, order = order).where(self.getVarClass(modbusType).varName == name).execute()
|
|
|
|
@classmethod
|
|
def editOrder(self, name, order, modbusType):
|
|
name = str(name)
|
|
order = str(order)
|
|
self.getVarClass(modbusType).update(order = order).where(self.getVarClass(modbusType).varName == name).execute()
|
|
|
|
@classmethod
|
|
def getAllVar(self, modbusType):
|
|
# 查询所有变量
|
|
|
|
vars = self.getVarClass(modbusType).get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description, var.varType, var.slaveID, var.address, var.min, var.max, var.order])
|
|
return l
|
|
|
|
@classmethod
|
|
def getByName(self, name, modbusType):
|
|
# 查询指定变量信息
|
|
var = self.getVarClass(modbusType).getByName(name)
|
|
if var:
|
|
return [var.id, var.varName, var.description, var.varType, var.slaveID, var.address, var.min, var.max]
|
|
else:
|
|
return False
|
|
|
|
class HartVarManage(object):
|
|
def __init__(self):
|
|
super(HartVarManage, self).__init__()
|
|
|
|
@classmethod
|
|
def createVar(self, varName, des):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
if HartVar.getByName(name):
|
|
print('已有同名变量')
|
|
else:
|
|
varModel = HartVar()
|
|
varModel.createVar(varName = varName, des = des)
|
|
|
|
@classmethod
|
|
def getAllVar(self):
|
|
# 查询所有变量
|
|
vars = HartVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description])
|
|
return l
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
HartVar.deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des):
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
|
|
if Nname == name:
|
|
HartVar.update(varName = Nname, description = des).where(HartVar.varName == name).execute()
|
|
elif HartVar.getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not HartVar.getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
HartVar.update(varName = Nname, description = des).where(HartVar.varName == name).execute()
|
|
|
|
@classmethod
|
|
def getByName(self, name):
|
|
# 查询指定变量信息
|
|
var = HartVar.getByName(name)
|
|
if var:
|
|
varMes = [var.id, var.varName, var.description]
|
|
[varMes.append('') for x in range(10)]
|
|
return varMes
|
|
else:
|
|
return False
|
|
|
|
def initVar(self):
|
|
self.createVar(varName='name', des='TC')
|
|
|
|
|
|
class TcRtdManage(object):
|
|
def __init__(self):
|
|
super(TcRtdManage, self).__init__()
|
|
|
|
@classmethod
|
|
def createVar(self, varName, varType, des, min, max, compensationVar):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
if TcRtdVar.getByName(name):
|
|
print('已有同名变量')
|
|
else:
|
|
varModel = TcRtdVar()
|
|
varModel.createVar(varName = varName, des = des, varType = varType, min = min, max = max, compensationVar = compensationVar)
|
|
|
|
@classmethod
|
|
def getAllVar(self):
|
|
# 查询所有变量
|
|
vars = TcRtdVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description, var.varType, var.min, var.max, var.compensationVar])
|
|
return l
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
TcRtdVar.deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des, varType, min, max, compensationVar):
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
varType = str(varType)
|
|
min = str(min)
|
|
max = str(max)
|
|
compensationVar = str(compensationVar)
|
|
if Nname == name:
|
|
TcRtdVar.update(varName=Nname, description=des, varType=varType, min=min, max=max, compensationVar = compensationVar).where(TcRtdVar.varName == name).execute()
|
|
elif TcRtdVar.getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not TcRtdVar.getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
TcRtdVar.update(varName=Nname, description=des, varType=varType, min=min, max=max, compensationVar = compensationVar).where(TcRtdVar.varName == name).execute()
|
|
|
|
@classmethod
|
|
def getByName(self, name):
|
|
# 查询指定变量信息
|
|
var = TcRtdVar.getByName(name)
|
|
if var:
|
|
|
|
return [var.id, var.varName, var.description, var.varType, var.min, var.max, var.compensationVar]
|
|
else:
|
|
return False
|
|
|
|
def initVar(self):
|
|
# 创建变量
|
|
for i in range(1, 9):
|
|
name = 'Value' + str(i)
|
|
self.createVar(varName=name, varType='R', des='TC', min='100', max='200', compensationVar = '0')
|
|
for i in range(9, 17):
|
|
name = 'Value' + str(i)
|
|
self.createVar(varName=name, varType='PT100', des='RTD', min='100', max='200', compensationVar = '0')
|
|
|
|
@classmethod
|
|
def editvarType(self, name, varType):
|
|
name = str(name)
|
|
TcRtdVar.update(varType = str(varType)).where(TcRtdVar.varName == name).execute()
|
|
|
|
class AnalogManage(object):
|
|
def __init__(self):
|
|
super(AnalogManage, self).__init__()
|
|
|
|
@classmethod
|
|
def createVar(self, varName, varType, des, min, max):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
if AnalogVar.getByName(name):
|
|
print('已有同名变量')
|
|
else:
|
|
varModel = AnalogVar()
|
|
varModel.createVar(varName = varName, des = des, varType = varType, min = min, max = max)
|
|
|
|
@classmethod
|
|
def getAllVar(self):
|
|
# 查询所有变量
|
|
vars = AnalogVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description, var.varType, var.min, var.max])
|
|
return l
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
AnalogVar.deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des, varType, min, max):
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
varType = str(varType)
|
|
min = str(min)
|
|
max = str(max)
|
|
if Nname == name:
|
|
AnalogVar.update(varName=Nname, description=des, varType=varType, min=min, max=max).where(AnalogVar.varName == name).execute()
|
|
elif AnalogVar.getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not AnalogVar.getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
AnalogVar.update(varName=Nname, description=des, varType=varType, min=min, max=max).where(AnalogVar.varName == name).execute()
|
|
|
|
@classmethod
|
|
def getByName(self, name):
|
|
# 查询指定变量信息
|
|
var = AnalogVar.getByName(name)
|
|
if var:
|
|
|
|
return [var.id, var.varName, var.description, var.varType, var.min, var.max]
|
|
else:
|
|
return False
|
|
|
|
def initVar(self):
|
|
for i in range(1, 9):
|
|
name = 'Value' + str(i)
|
|
if i < 5:
|
|
des = '有源4-20mA输出' + str(i)
|
|
self.createVar(varName=name, varType='AO', des=des, min='100', max='200')
|
|
else:
|
|
des = '无源4-20mA输出' + str(i)
|
|
self.createVar(varName=name, varType='AO', des=des, min='100', max='200')
|
|
for i in range(1, 9):
|
|
name = 'Value' + str(i + 8)
|
|
if i < 5:
|
|
des = '有源频率输出' + str(i)
|
|
self.createVar(varName=name, varType='Hz', des=des, min='100', max='200')
|
|
elif i < 7:
|
|
des = '无源NPN频率输出' + str(i)
|
|
self.createVar(varName=name, varType='Hz', des=des, min='100', max='200')
|
|
else:
|
|
des = '无源PNP频率输出' + str(i)
|
|
self.createVar(varName=name, varType='Hz', des=des, min='100', max='200')
|
|
for i in range(1, 9):
|
|
name = 'Value' + str(i + 16)
|
|
self.createVar(varName=name, varType='DO', des='DO', min='0', max='1')
|
|
|
|
class FFSimulateManage(object):
|
|
def __init__(self):
|
|
super(FFSimulateManage, self).__init__()
|
|
|
|
@classmethod
|
|
def createVar(self, varName, varType, des):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
if FFSimulateVar.getByName(name):
|
|
print('已有同名变量')
|
|
else:
|
|
varModel = FFSimulateVar()
|
|
varModel.createVar(varName = varName, des = des, varType = varType)
|
|
|
|
@classmethod
|
|
def getAllVar(self):
|
|
# 查询所有变量
|
|
vars = FFSimulateVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description, var.varType])
|
|
return l
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
FFSimulateVar.deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des, varType):
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
varType = str(varType)
|
|
|
|
if Nname == name:
|
|
FFSimulateVar.update(varName=Nname, description=des, varType=varType).where(FFSimulateVar.varName == name).execute()
|
|
elif FFSimulateVar.getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not FFSimulateVar.getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
FFSimulateVar.update(varName=Nname, description=des, varType=varType).where(FFSimulateVar.varName == name).execute()
|
|
|
|
@classmethod
|
|
def getByName(self, name):
|
|
# 查询指定变量信息
|
|
var = FFSimulateVar.getByName(name)
|
|
if var:
|
|
|
|
return [var.id, var.varName, var.description, var.varType]
|
|
else:
|
|
return False
|
|
|
|
def initVar(self):
|
|
for i in range(1, 9):
|
|
name = 'Value' + str(i)
|
|
if i < 5:
|
|
self.createVar(varName=name, varType='AI', des='AI')
|
|
else:
|
|
self.createVar(varName=name, varType='DI', des='DI')
|
|
for i in range(9, 17):
|
|
name = 'Value' + str(i)
|
|
if i < 13:
|
|
self.createVar(varName=name, varType='AO', des='AO')
|
|
else:
|
|
self.createVar(varName=name, varType='DO', des='DO')
|
|
|
|
class HartSimulateVarManage(object):
|
|
def __init__(self):
|
|
super(HartSimulateVarManage, self).__init__()
|
|
|
|
@classmethod
|
|
def createVar(self, varName, des):
|
|
# 创建变量
|
|
name = str(varName)
|
|
des = str(des)
|
|
if HartSimulateVar.getByName(name):
|
|
print('已有同名变量')
|
|
else:
|
|
varModel = HartSimulateVar()
|
|
varModel.createVar(varName = varName, des = des)
|
|
|
|
@classmethod
|
|
def getAllVar(self):
|
|
# 查询所有变量
|
|
vars = HartSimulateVar.get_all()
|
|
if vars is 'error':
|
|
return
|
|
l = []
|
|
for var in vars:
|
|
l.append([var.id, var.varName, var.description])
|
|
return l
|
|
|
|
|
|
@classmethod
|
|
def deleteVar(self, name):
|
|
# 删除变量
|
|
name = str(name)
|
|
# print(name)
|
|
HartSimulateVar.deleteVar(name = name)
|
|
|
|
|
|
@classmethod
|
|
def editVar(self, name, Nname, des):
|
|
# 修改变量信息
|
|
name = str(name)
|
|
Nname = str(Nname)
|
|
des = str(des)
|
|
|
|
if Nname == name:
|
|
HartSimulateVar.update(varName = Nname, description = des).where(HartSimulateVar.varName == name).execute()
|
|
elif HartSimulateVar.getByName(Nname):
|
|
print('已有同名变量')
|
|
return
|
|
elif not HartSimulateVar.getByName(name):
|
|
print('不存在的变量')
|
|
return
|
|
else:
|
|
HartSimulateVar.update(varName = Nname, description = des).where(HartSimulateVar.varName == name).execute()
|
|
|
|
@classmethod
|
|
def getByName(self, name):
|
|
# 查询指定变量信息
|
|
var = HartSimulateVar.getByName(name)
|
|
if var:
|
|
varMes = [var.id, var.varName, var.description]
|
|
[varMes.append('') for x in range(10)]
|
|
return varMes
|
|
else:
|
|
return False
|
|
|
|
def initVar(self):
|
|
self.createVar(varName='name', des='TC') |