|
|
from pythonnet import load
|
|
|
load('coreclr')
|
|
|
import clr # 与其他模块不同此处不直接导入pythonnet 而是导入clr（python与c#交互核心）
|
|
|
dll_path = 'protocol\\FF\\NationalInstruments.Fieldbus' # 为了方便理解这里给到dll所在的绝对路径且包含dll文件本身
|
|
|
clr.AddReference(dll_path) #加载c#库
|
|
|
clr.AddReference('System.Collections')
|
|
|
clr.AddReference('System')
|
|
|
from NationalInstruments.FieldBus import Session, FBDate
|
|
|
from NationalInstruments.FieldBus.ParameterHandler import Parameter, ParameterType
|
|
|
from NationalInstruments.FieldBus.ElementHandler import *
|
|
|
from System.Collections.Generic import *
|
|
|
from System import *
|
|
|
# import numpy as np
|
|
|
|
|
|
|
|
|
# session = Session()
|
|
|
# session.Open()
|
|
|
# link = session.Links[0]
|
|
|
# link.Open()
|
|
|
# print([x.Tag for x in link.Devices])
|
|
|
# dev = link.Devices[1]
|
|
|
# VFD = dev.GetVfdByOrdinalNumber(0)
|
|
|
# VFD.Open()
|
|
|
# blocks = VFD.Blocks
|
|
|
# print([x.Tag for x in blocks])
|
|
|
# block = blocks[1]
|
|
|
# block.Open()
|
|
|
# Params = block.Parameters
|
|
|
# print([x.Type.ToString() + ' ' + x.Name for x in Params])
|
|
|
# print(byte(10))
|
|
|
class ParamItem(object):
|
|
|
def __init__(self, param):
|
|
|
self.param = param
|
|
|
self.name = param.Name
|
|
|
self.type = param.Type.ToString()
|
|
|
self.elemDict = {}
|
|
|
match(param.Type):
|
|
|
case ParameterType.ODT_SIMPLEVAR:
|
|
|
self.elements = [param.Element]
|
|
|
case ParameterType.ODT_ARRAY:
|
|
|
self.elements = param.Elements
|
|
|
case ParameterType.ODT_RECORD:
|
|
|
self.elements = param.Elements
|
|
|
|
|
|
def getElements(self):
|
|
|
for elem in self.elements:
|
|
|
self.elemDict[elem.Name] = [elem.Type.ToString(), elem.ValueString(), elem]
|
|
|
return self.elemDict
|
|
|
|
|
|
|
|
|
class FFProtocol(object):
|
|
|
"""docstring for FF"""
|
|
|
paramDict = {}
|
|
|
def __init__(self):
|
|
|
pass
|
|
|
|
|
|
def open(self):
|
|
|
self.session = Session()
|
|
|
self.session.Open()
|
|
|
self.link = self.session.Links[0]
|
|
|
self.link.Open()
|
|
|
|
|
|
|
|
|
def getDevices(self):
|
|
|
self.devices = [x for x in self.link.Devices][1:]
|
|
|
for device in self.devices:
|
|
|
device.Open()
|
|
|
VFD = device.GetVfdByOrdinalNumber(0)
|
|
|
VFD.Open()
|
|
|
return [x.Tag for x in self.devices]
|
|
|
|
|
|
def getBlocks(self, devTag):
|
|
|
device = self.link.GetDeviceByTag(devTag)
|
|
|
VFD = device.GetVfdByOrdinalNumber(0)
|
|
|
blocks = [x for x in VFD.Blocks]
|
|
|
for block in blocks:
|
|
|
block.Open()
|
|
|
return [x.Tag for x in blocks]
|
|
|
|
|
|
def getParams(self, devTag, blockTag):
|
|
|
device = self.link.GetDeviceByTag(devTag)
|
|
|
VFD = device.GetVfdByOrdinalNumber(0)
|
|
|
block = VFD.GetBlockByTag(blockTag)
|
|
|
Params = block.Parameters
|
|
|
paramList = []
|
|
|
for param in Params:
|
|
|
paramList.append(ParamItem(param = param))
|
|
|
if devTag not in self.paramDict:
|
|
|
self.paramDict[devTag] = {blockTag : paramList}
|
|
|
else:
|
|
|
self.paramDict[devTag][blockTag] = paramList
|
|
|
# print(self.paramDict)
|
|
|
return paramList
|
|
|
|
|
|
def refreshParams(self, reParams = None):
|
|
|
if reParams:
|
|
|
for param in reParams:
|
|
|
param.Read()
|
|
|
# print(reParams)
|
|
|
# for device, blockDict in self.paramDict.items():
|
|
|
# for block, paramList in blockDict.items():
|
|
|
# else:
|
|
|
# param.param.Read()
|
|
|
# for elem in param.elements:
|
|
|
# return elem.get_Value()
|
|
|
|
|
|
def writeValue(self, param, elem, value):
|
|
|
param.Read()
|
|
|
match(elem.Type):
|
|
|
case ElementType.FF_BOOLEAN:
|
|
|
if value in [True, False]:
|
|
|
elem.Value = value
|
|
|
elif value in ['True', 'true']:
|
|
|
elem.Value = True
|
|
|
elif value in ['False', 'false']:
|
|
|
elem.Value = False
|
|
|
case ElementType.FF_INTEGER8 | ElementType.FF_INTEGER16:
|
|
|
value = int(value)
|
|
|
elem.Value = Int16(value)
|
|
|
case ElementType.FF_INTEGER32 | ElementType.FF_UNSIGNED16:
|
|
|
value = int(value)
|
|
|
elem.Value = Int32(value)
|
|
|
case ElementType.FF_UNSIGNED8:
|
|
|
value = int(value)
|
|
|
elem.Value = Byte(value)
|
|
|
case ElementType.FF_UNSIGNED32:
|
|
|
value = int(value)
|
|
|
elem.Value = Int64(value)
|
|
|
case ElementType.FF_FLOAT:
|
|
|
value = float(value)
|
|
|
elem.Value = Single(value)
|
|
|
case ElementType.FF_VISIBLE_STRING:
|
|
|
value = str(value)
|
|
|
elem.Value = value
|
|
|
case ElementType.FF_OCTET_STRING:
|
|
|
value = str(value)
|
|
|
if len(value) % 2 != 0:
|
|
|
return
|
|
|
startIndex = 0
|
|
|
# byteArray = [x for x in range(int(len(value) / 2))]
|
|
|
byteArray = Array.CreateInstance(Byte, int(len(value) / 2))
|
|
|
for i in range(int(len(value) / 2)):
|
|
|
varStr = value[startIndex:startIndex + 2]
|
|
|
startIndex += 2
|
|
|
byteArray[i] = Byte(int(varStr))
|
|
|
elem.Value = byteArray
|
|
|
case ElementType.FF_DATE:
|
|
|
value = str(value)
|
|
|
valueList = [int(x) for x in value.split(':')]
|
|
|
elem.Value.Year = Byte(valueList[0])
|
|
|
elem.Value.Month = Byte(valueList[1])
|
|
|
elem.Value.DayOfMonth =Byte(valueList[2])
|
|
|
elem.Value.Hour = Byte(valueList[3])
|
|
|
elem.Value.Minute = Byte(valueList[4])
|
|
|
elem.Value.Millisecond = int(valueList[5])
|
|
|
case ElementType.FF_TIMEOFDAY | ElementType.FF_TIME_DIFF:
|
|
|
value = str(value)
|
|
|
valueList = [int(x) for x in value.split(':')]
|
|
|
elem.Value.Day = int(valueList[0])
|
|
|
elem.Value.Millisecond = Int64(valueList[1])
|
|
|
case ElementType.FF_BIT_STRING:
|
|
|
value = str(value)
|
|
|
for s in value:
|
|
|
if s not in ['0', '1']:
|
|
|
return
|
|
|
if len(value) % 8 != 0:
|
|
|
return
|
|
|
j = -1
|
|
|
byteArray = Array.CreateInstance(Byte, int(len(value) / 8))
|
|
|
for i in range(len(value)):
|
|
|
if i % 8 == 0:
|
|
|
j += 1
|
|
|
if value[i] == '1':
|
|
|
# print(type(byteArray[j]))
|
|
|
byteArray[j] += 1<<(7-(i % 8))
|
|
|
byteArray[j] = Byte(byteArray[j])
|
|
|
elem.Value = byteArray
|
|
|
case ElementType.FF_TIME_VALUE:
|
|
|
value = str(value)
|
|
|
valueList = [int(x) for x in value.split(':')]
|
|
|
elem.Value.Lower = Int64(valueList[0])
|
|
|
elem.Value.Upper = Int64(valueList[1])
|
|
|
|
|
|
param.Write()
|
|
|
|
|
|
|
|
|
|
|
|
def close(self):
|
|
|
for device in self.devices:
|
|
|
device.Close()
|
|
|
self.link.Close()
|
|
|
self.session.Close()
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
a = FFProtocol()
|
|
|
a.open()
|
|
|
devs = a.getDevices()
|
|
|
blocks = a.getBlocks(devs[0])
|
|
|
|
|
|
par = a.getParams(devs[0], blocks[0])[7]
|
|
|
|
|
|
ele1 = par.getElements()['VALUE_1'][2]
|
|
|
ele2 = par.getElements()['VALUE_2'][2]
|
|
|
ele3 = par.getElements()['VALUE_3'][2]
|
|
|
ele4 = par.getElements()['VALUE_4'][2]
|
|
|
ele5 = par.getElements()['VALUE_5'][2]
|
|
|
ele6 = par.getElements()['VALUE_6'][2]
|
|
|
ele7 = par.getElements()['VALUE_7'][2]
|
|
|
ele8 = par.getElements()['VALUE_8'][2]
|
|
|
ele9 = par.getElements()['VALUE_9'][2]
|
|
|
ele10 = par.getElements()['VALUE_10'][2]
|
|
|
ele11 = par.getElements()['VALUE_11'][2]
|
|
|
ele12 = par.getElements()['VALUE_12'][2]
|
|
|
ele13 = par.getElements()['VALUE_13'][2]
|
|
|
ele14 = par.getElements()['VALUE_14'][2]
|
|
|
ele15 = par.getElements()['VALUE_15'][2]
|
|
|
|
|
|
a.writeValue(par.param, ele1, True)
|
|
|
a.writeValue(par.param, ele2, 1)
|
|
|
a.writeValue(par.param, ele3, 2)
|
|
|
a.writeValue(par.param, ele4, 3)
|
|
|
a.writeValue(par.param, ele5, 4)
|
|
|
a.writeValue(par.param, ele6, 5)
|
|
|
a.writeValue(par.param, ele7, 6)
|
|
|
a.writeValue(par.param, ele8, 1.1)
|
|
|
a.writeValue(par.param, ele9, 'test')
|
|
|
a.writeValue(par.param, ele10, '9999494949494949494949494949494949494949494949494949494949494949')
|
|
|
a.writeValue(par.param, ele11, '23:02:18:14:49:50')
|
|
|
a.writeValue(par.param, ele12, '18:59')
|
|
|
a.writeValue(par.param, ele13, '25:60')
|
|
|
a.writeValue(par.param, ele14, '1111000000001111')
|
|
|
a.writeValue(par.param, ele15, '55:55')
|
|
|
|
|
|
|
|
|
a.close() |