106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
from pycomm.ab_comm.clx import Driver as ClxDriver
|
|
import sys
|
|
from time import sleep
|
|
from readConfig import readConfig
|
|
|
|
def closeEnough(a,b):
|
|
return abs(a - b) <= 0.1
|
|
|
|
configProperties = readConfig()
|
|
|
|
def readTag(addr, tag):
|
|
time.sleep(0.01)
|
|
c = ClxDriver()
|
|
if c.open(configProperties['PLC_IP_ADDRESS']):
|
|
out = {}
|
|
try:
|
|
result = c.read_tag([tagName])
|
|
if result[0][2] == None:
|
|
raise ValueError('Tag not found')
|
|
out['status'] = "success"
|
|
out['value'] = result[0][1]
|
|
out['type'] = result[0][2]
|
|
except Exception, e:
|
|
out['status'] = "error"
|
|
out['message'] = "Tag Not Found"
|
|
c.close()
|
|
return out
|
|
|
|
|
|
def main(tag, value):
|
|
global configProperties
|
|
r = 0
|
|
|
|
readObj = readTag(tag)
|
|
if readObj['status'] == "error":
|
|
return readObj
|
|
elif readObj['status'] == 'success':
|
|
tagType = readObj['type']
|
|
|
|
if tagType[:-3] == "INT" or tagType == "BOOL":
|
|
value = int(value)
|
|
elif tagType == "REAL":
|
|
value = float(value)
|
|
c = ClxDriver()
|
|
if c.open(configProperties['PLC_IP_ADDRESS']):
|
|
r = c.write_tag(tag, value, tagType)
|
|
else:
|
|
return {"status": 'error', "message": "not connected to PLC"}
|
|
c.close()
|
|
|
|
sleep(2)
|
|
|
|
return readTag(tag)
|
|
|
|
def writeTagAndVerify(tag,value, sleepValue=2):
|
|
"""Writes the specified value to tag and confirms that the value has been set"""
|
|
global configProperties
|
|
r = 0
|
|
|
|
readObj = readTag(tag)
|
|
if readObj['status'] == "error":
|
|
return readObj
|
|
elif readObj['status'] == 'success':
|
|
tagType = readObj['type']
|
|
prevValue = readObj['value']
|
|
if tagType[:-3] == "INT" or tagType == "BOOL":
|
|
value = int(value)
|
|
elif tagType == "REAL":
|
|
value = float(value)
|
|
c = ClxDriver()
|
|
if c.open(configProperties['PLC_IP_ADDRESS']):
|
|
r = c.write_tag(tag, value, tagType)
|
|
sleep(float(sleepValue))
|
|
newObj = readTag(tag)
|
|
if newObj['status'] == "error":
|
|
return newObj
|
|
elif newObj['status'] == 'success':
|
|
newTagType = newObj['type']
|
|
newValue = newObj['value']
|
|
if (newTagType == tagType):
|
|
if (closeEnough(newValue, value)):
|
|
return {'status':'success', 'value':newValue, 'type':newTagType, 'verified':'true','prevValue':prevValue}
|
|
else:
|
|
return {"status": 'error', "message": "The tag value does not match the specified value", "value":newValue}
|
|
else:
|
|
return {"status": 'error', "message": "Somehow the tag type has changed"}
|
|
else:
|
|
return {"status": 'error', "message": "not connected to PLC"}
|
|
c.close()
|
|
|
|
def toggle(tag, toggleVal, timeToggled=2):
|
|
val = int(toggleVal)
|
|
set_tag = writeTagAndVerify(tag,toggleVal,sleepValue=1)
|
|
if set_tag['status']== 'success':
|
|
sleep(timeToggled)
|
|
reset_tag = writeTagAndVerify(tag, set_tag['prevValue'],sleepValue=1)
|
|
if reset_tag['status'] == "success":
|
|
return {'status':'success'}
|
|
else:
|
|
return {'status':'error', 'message':reset_tag['message']}
|
|
else:
|
|
return {'status':'error', 'message':set_tag['message']}
|
|
|
|
if __name__ == '__main__':
|
|
print main(sys.argv[1], sys.argv[2])
|