118 lines
3.1 KiB
Python
118 lines
3.1 KiB
Python
#!/usr/bin/env python
|
|
|
|
'''
|
|
Created on Oct 1, 2014
|
|
|
|
@author: PJMcdona
|
|
'''
|
|
|
|
|
|
#from prettytable import PrettyTable
|
|
import csv
|
|
from datetime import datetime
|
|
import os
|
|
import sys
|
|
from random import randint
|
|
from time import sleep
|
|
import MySQLdb
|
|
|
|
def main(fullfilepath):
|
|
|
|
db = MySQLdb.connect(host="127.0.0.1",user="website",passwd="henrypump",db="WellData")
|
|
cur = db.cursor()
|
|
query = "SELECT * FROM WellData.config ORDER BY dateChanged DESC LIMIT 1;"
|
|
cur.execute(query)
|
|
setup = cur.fetchall()
|
|
db.commit()
|
|
db.close()
|
|
|
|
PLC_IP_ADDRESS = setup[0][2]
|
|
PLC_TYPE = setup[0][1]
|
|
|
|
|
|
|
|
#TUXEIP Connection to PLC
|
|
from tuxeip import TuxEIP, LGX, LGX_REAL, LGX_BOOL, LGX_INT, LGX_DINT, LGX_SINT
|
|
tux = TuxEIP(libpath="/usr/lib/libtuxeip.so")
|
|
sess = tux.OpenSession(PLC_IP_ADDRESS)
|
|
reg = tux.RegisterSession(sess)
|
|
conn = tux.ConnectPLCOverCNET(sess, LGX, 1, 100, 123, randint(0,9999), 123, 321, 100, 5000, 1, '01')
|
|
|
|
def WriteToPLC(tag, ttype, value):
|
|
try:
|
|
r=0
|
|
if ttype == "INT":
|
|
r = tux.WriteLGXData(sess, conn, tag, LGX_INT, int(value), 1)
|
|
elif ttype == "DINT":
|
|
r = tux.WriteLGXData(sess, conn, tag, LGX_DINT, int(value), 1)
|
|
elif ttype == "SINT":
|
|
r = tux.WriteLGXData(sess, conn, tag, LGX_SINT, int(value), 1)
|
|
elif ttype == "REAL":
|
|
r = tux.WriteLGXData(sess, conn, tag, LGX_REAL, float(value), 1)
|
|
elif ttype == "BOOL":
|
|
r = tux.WriteLGXData(sess, conn, tag, LGX_BOOL, int(value), 1)
|
|
else:
|
|
print "invalid type", ttype
|
|
r = -1
|
|
if r == -1:
|
|
print "Error writing to", tag
|
|
except:
|
|
print "Error writing to", tag
|
|
|
|
def ReadFromPLC(tag, tag_type):
|
|
try:
|
|
if (tag_type == "REAL"):
|
|
a = tux.ReadLGXDataAsFloat(sess, conn, tag, 1)
|
|
result = round(float(a[0]),3)
|
|
#elif (tag_type == "INT" or tag_type == "DINT" or tag_type =="SINT"):
|
|
elif tag_type[-3:] == "INT":
|
|
a = tux.ReadLGXDataAsInteger(sess, conn, tag, 1)
|
|
result = int(a[0])
|
|
elif (tag_type == "BOOL"):
|
|
a = tux.ReadLGXDataAsInteger(sess, conn, tag, 1)
|
|
intermediate = int(a[0])
|
|
if intermediate == 0:
|
|
result = 0
|
|
else:
|
|
result = 1
|
|
elif (tag_type =="STRING"):
|
|
a = tux.ReadLGXDataAsInteger(sess, conn, tag, 82)
|
|
word = ""
|
|
for ch in range(len(a)):
|
|
word = word + unichr(int(a[ch]))
|
|
result = word
|
|
else:
|
|
result = "?"
|
|
return result
|
|
except:
|
|
print "Unable to read " + tag
|
|
return 911
|
|
|
|
errors = []
|
|
myfile = open(fullfilepath, 'rb')
|
|
wr = csv.reader(myfile)
|
|
for row in wr:
|
|
if len(row) == 3:
|
|
(tag, value, tagType) = row
|
|
WriteToPLC(tag, tagType, value)
|
|
actual = ReadFromPLC(tag, tagType)
|
|
verify = False
|
|
if tagType == "REAL":
|
|
verify = actual == float(value)
|
|
elif tagType[-3:] == "INT":
|
|
verify = actual == int(value)
|
|
elif tagType == "BOOL":
|
|
verify = actual == int(value)
|
|
if not verify:
|
|
errors.append( "Validation Error:", tag, "does not equal", value, "(actual value:", actual,")")
|
|
myfile.close()
|
|
print "Restore Complete with", len(errors), "errors."
|
|
if len(errors) > 0:
|
|
print "-------------------------"
|
|
print "-- ERRORS --"
|
|
print "-------------------------"
|
|
for i in range(0,len(errors)):
|
|
print i+1,"-",errors[i]
|
|
if __name__ == '__main__':
|
|
main(sys.argv[1])
|