74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
#!/usr/bin/env python
|
|
|
|
'''
|
|
MySQL Tag Server
|
|
Created on April 7, 2016
|
|
@author: Patrick McDonagh
|
|
@description: Continuously loops through a list of tags to store values from a PLC into a MySQL database
|
|
'''
|
|
|
|
from tag.tag import Tag
|
|
import traceback
|
|
import time
|
|
import requests
|
|
import json
|
|
|
|
# DEFAULTS
|
|
web_address = "http://localhost:3000"
|
|
scan_rate = 30 # seconds
|
|
save_all = "test" # use True, False, or any string
|
|
|
|
tag_store = {}
|
|
|
|
|
|
def main():
|
|
global web_address, scan_rate, save_all, tag_store
|
|
try:
|
|
# Get tags stored in database
|
|
get_tag_request_data = {'where': '{"tag_class": 5}'}
|
|
get_tag_request = requests.get('{}/tag'.format(web_address), params=get_tag_request_data)
|
|
tags = json.loads(get_tag_request.text)
|
|
except Exception, e:
|
|
print("Error getting tags: {}".format(e))
|
|
time.sleep(10)
|
|
main()
|
|
|
|
try:
|
|
sr_req_data = 'where={"parameter": "scan_rate"}'
|
|
sr_req = requests.get('{}/config?{}'.format(web_address, sr_req_data))
|
|
sr_try = json.loads(sr_req.text)
|
|
if len(sr_try) > 0:
|
|
scan_rate = int(sr_try[0]['val'])
|
|
except Exception, e:
|
|
print("Error getting scan rate: {}".format(e))
|
|
print("I'll just use {} seconds as the scan rate...".format(scan_rate))
|
|
|
|
try:
|
|
sa_req_data = {"where": {"parameter": "save_all"}}
|
|
sa_req = requests.get('{}/config'.format(web_address), params=sa_req_data)
|
|
sa_try = json.loads(sa_req.text)
|
|
if len(sa_try) > 0:
|
|
if sa_try[0]['val'].lower() == "true":
|
|
save_all = True
|
|
elif sa_try[0]['val'].lower() == "false":
|
|
save_all = False
|
|
except Exception, e:
|
|
print("Error getting save-all: {}".format(e))
|
|
print("I'll just use {} as the save-all parameter...".format(save_all))
|
|
|
|
for t in tags:
|
|
# name, tag, db_id, data_type, change_threshold, guarantee_sec, mapFn=None, device_type='CLX', ip_address='192.168.1.10'):
|
|
tag_store[t['name']] = Tag(t['name'], t['tag'], t['id'], t['data_type'], t['change_threshold'], t['guarantee_sec'], mapFn=t['map_function'], ip_address=t['deviceID']['address'])
|
|
|
|
while True:
|
|
for tag in tag_store:
|
|
try:
|
|
tag_store[tag].read('test')
|
|
except:
|
|
print("ERROR EVALUATING {}".format(tag))
|
|
traceback.print_exc()
|
|
time.sleep(scan_rate)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|