Files
POC-VFD/POCloud/w_sqlite/poc.py
2016-03-26 17:09:54 -05:00

532 lines
23 KiB
Python

#!/usr/bin/python
import types
import traceback
import binascii
import threading
import time
import thread
import os
import struct
import sys
import serial
import minimalmodbus
import pickle
import re
from device_base import deviceBase
from datetime import datetime
import traceback
import requests
try:
import json
except:
import simplejson as json
import calendar
class Channel():
def __init__(self, mesh_name, data_type, chg_threshold, guarantee_sec):
self.mesh_name = mesh_name
self.data_type = data_type
self.last_value = None
self.value = None
self.last_send_time = 0
self.chg_threshold = chg_threshold
self.guarantee_sec = guarantee_sec
def checkSend(self, newVal, force):
v = self.data_type(newVal)
if self.data_type == bool or self.data_type == str:
if (self.last_send_time == 0) or (self.value is None) or not (self.value == v) or ((self.guarantee_sec > 0) and ((time.time() - self.last_send_time) > self.guarantee_sec)) or (force):
self.last_value = self.value
self.value = v
self.last_send_time = time.time()
return True
else:
return False
else:
if (self.last_send_time == 0) or (self.value is None) or (abs(self.value - v) > self.chg_threshold) or ((self.guarantee_sec > 0) and ((time.time() - self.last_send_time) > self.guarantee_sec)) or (force):
self.last_value = self.value
self.value = v
self.last_send_time = time.time()
return True
else:
return False
go_channels = {
'electricity_cost': Channel('go_electricity_cost', float, 0.0, 0),
'percent_run': Channel('go_percent_run', float, 0.0, 0),
'average_SPM': Channel('go_average_spm', float, 0.0, 0),
'peak_load': Channel('go_peak_load', float, 0.0, 0),
'polished_rod_HP': Channel('go_polished_rod_hp', float, 0.0, 0),
'lifting_cost': Channel('go_lifting_cost', float, 0.0, 0),
'full_card_production': Channel('go_full_card_production', float, 0.0, 0),
'fluid_above_pump': Channel('go_fluid_above_pump', float, 0.0, 0),
'production_calculated': Channel('go_production_calculated', float, 0.0, 0),
'kWh': Channel('go_kwh', float, 0.0, 3600),
'inflow_rate': Channel('go_inflow_rate', float, 0.0, 0),
'kWh_regen': Channel('go_kwh_regen', float, 0.0, 0),
'pump_intake_pressure': Channel('go_pump_intake_pressure', float, 0.0, 0),
'min_load': Channel('go_min_load', float, 0.0, 0),
}
card_channels = {
"card_id": Channel("card_history", int, 0, 0),
"card_type": Channel("cardtype", str, 0, 0),
"card": Channel(None, str, 0, 600)
}
status = Channel('status', str, 0, 0)
tag_channels = {
'polished_rod_hp': Channel('polished_rod_hp', float, 1.0, 3600),
'drive_torque_mode': Channel('drive_torque_mode', int, 1.0, 3600),
'downhole_gross_stroke': Channel('downhole_gross_stroke', float, 1.0, 3600),
'fluid_gradient': Channel('fluid_gradient', float, 0.01, 3600),
'tubing_head_pressure': Channel('tubing_head_pressure', float, 10.0, 3600),
'surface_min_load': Channel('surface_min_load', float, 500.0, 3600),
'downhole_fluid_load': Channel('downhole_fluid_load', float, 500.0, 3600),
'downhole_max_position': Channel('downhole_max_position', float, 1.0, 3600),
'downhole_net_stroke': Channel('downhole_net_stroke', float, 1.0, 3600),
'fillage_percent': Channel('fillage_percent', float, 5.0, 3600),
'pump_hp': Channel('pump_hp', float, 1.0, 3600),
'spm': Channel('SPM', float, 0.5, 3600),
'pump_intake_pressure': Channel('pump_intake_pressure', float, 200.0, 3600),
'speed_reference': Channel('speed_reference', float, 50.0, 3600),
'downhole_min_position': Channel('downhole_min_position', float, 1.0, 3600),
'tubing_movement': Channel('tubing_movement', float, 1.0, 3600),
'surface_max_load': Channel('surface_max_load', float, 500.0, 3600),
'stuffing_box_friction': Channel('stuffing_box_friction', float, 50.0, 3600),
'dt': Channel('dt', float, 0.001, 3600),
'fluid_level': Channel('fluid_above_pump', float, 100.0, 3600),
'torque_reference': Channel('torque_reference', float, 5.0, 3600),
'surface_stroke_length': Channel('surface_stroke_length', float, 1.0, 3600),
'downhole_adjusted_gross_stroke': Channel('downhole_adjusted_gross_stroke', float, 1.0, 3600),
'stroke_production': Channel('stroke_production', float, 0.001, 3600),
}
dt_channels = { # Current Daily Totals
'Electricity_Cost': Channel('dt_electricity_cost', float, 1.0, 3600),
'Downhole_Net_Stroke': Channel('dt_downhole_net_stroke', float, 2.0, 3600),
'Tubing_Movement': Channel('dt_tubing_movement', float, 1.0, 3600),
'Average_SPM': Channel('dt_average_spm', float, 0.50, 3600),
'Peak_Load': Channel('dt_peak_load', float, 500.0, 3600),
'kWh': Channel('dt_kWh', float, 5.0, 3600),
'Pump_HP': Channel('dt_pump_hp', float, 0.5, 3600),
'Percent_Run': Channel('dt_percent_run', float, 5.0, 3600),
'Projected_Production': Channel('dt_projected_production', float, 5.0, 3600),
'Pump_Intake_Presure': Channel('dt_pump_intake_pressure', float, 100.0, 3600),
'Inflow_Rate': Channel('dt_inflow_rate', float, 1.0, 3600),
'Calculated_Production': Channel('dt_calculated_production', float, 5.0, 3600),
'Fluid_Level': Channel('dt_fluid_level', float, 100.0, 3600),
'Lifting_Cost': Channel('dt_lifting_cost', float, 1.0, 3600),
'Polished_Rod_HP': Channel('dt_polished_rod_hp', float, 1.0, 3600),
'kWh_Regen': Channel('dt_kWh_regen', float, 1.0, 3600),
'Surface_Stroke_Length': Channel('dt_surface_stroke_length', float, 1.0, 3600),
'Full_Card_Production': Channel('dt_full_card_production', float, 10.0, 3600),
'Min_Load': Channel('dt_min_load', float, 500.0, 3600),
}
class start(threading.Thread, deviceBase):
def __init__(self, name=None, number=None, mac=None, Q=None, mcu=None, companyId=None, offset=None, mqtt=None, Nodes=None):
threading.Thread.__init__(self)
deviceBase.__init__(self, name=name, number=number, mac=mac, Q=Q, mcu=mcu, companyId=companyId, offset=offset, mqtt=mqtt, Nodes=Nodes)
self.daemon = True
self.forceSend = True
self.version = "2"
# self.device_address = "http://192.168.1.30/"
self.device_address = "http://localhost/"
self.cardLoopTimer = 600
self.finished = threading.Event()
threading.Thread.start(self)
self.statusChanged = False
self.al_status_last = False
self.dl_status_last = False
# load stored event ID's
try:
with open('eventIds.p', 'rb') as handle:
self.eventIds = pickle.load(handle)
print "found pickled eventID dictionary: {0}".format(self.eventIds)
except:
print "couldn't load enent ID's from pickle"
self.eventIds = []
# load stored Well Test ID's
try:
with open('welltestIDs.p', 'rb') as handle:
self.welltestIDs = pickle.load(handle)
print "found pickled welltestIDs dictionary: {0}".format(self.welltestIDs)
except:
print "couldn't load well test ID's from pickle"
self.welltestIDs = []
# load stored Fluid Shot ID's
try:
with open('fluidshotIDs.p', 'rb') as handle:
self.fluidshotIDs = pickle.load(handle)
print "found pickled fluidshotIDs dictionary: {0}".format(self.fluidshotIDs)
except:
print "couldn't load fluid shot ID's from pickle"
self.fluidshotIDs = []
# load stored note ID's
try:
with open('noteIDs.p', 'rb') as handle:
self.noteIDs = pickle.load(handle)
print "found pickled noteID dictionary: {0}".format(self.noteIDs)
except:
print "couldn't load note ID's from pickle"
self.noteIDs = []
# load stored wellconfig's
try:
with open('wellSetup.p', 'rb') as handle:
self.wellSetup = pickle.load(handle)
print "Found pickled Well Setup (but it's going to be too long to print)"
# print self.wellConfig
except:
print "couldn't load Well Setup from pickle"
self.wellSetup = []
self.sendtodbJSON("device_address", self.device_address, 0)
# this is a required function for all drivers, its goal is to upload some piece of data
# about your device so it can be seen on the web
def register(self):
channels["status"]["last_value"] = ""
def run(self):
self.runLoopStatus = ""
checkBackupEvery = 100
checkBackupSkipped = 1
while True:
try:
if self.forceSend:
print("!!!!!!!!!!!!!!! FORCE SEND !!!!!!!!!!!!!!!")
runLoopStatus = "checkStatus"
self.checkStatus()
runLoopStatus = "checkEvents"
self.checkEvents()
runLoopStatus = "checkNotes"
self.checkNotes()
runLoopStatus = "checkWellTests"
self.checkWellTests()
runLoopStatus = "checkFluidShots"
self.checkFluidShots()
runLoopStatus = "checkDailyTotals"
self.checkDailyTotals()
runLoopStatus = "checkGaugeOffData"
self.checkGaugeOffData()
runLoopStatus = "checkStoredValues"
self.checkStoredValues(self.forceSend)
runLoopStatus = "getDataLoggerStatus()"
self.getDataLoggerStatus()
if self.statusChanged:
runLoopStatus = "getLatestXCards"
self.forceSend = True
self.checkLatestCard(5)
else:
runLoopStatus = "checkLatestCard"
self.checkLatestCard()
# if self.forceSend or (checkBackupSkipped > checkBackupEvery):
# runLoopStatus = "checkBackup"
# self.checkBackup()
# checkBackupSkipped = 0
# checkBackupSkipped = checkBackupSkipped + 1
runLoopStatus = "Complete"
time.sleep(3)
self.forceSend = False
except Exception, e:
sleep_timer = 20
print "Error during {0} of run loop: {1}\nWill try again in {2} seconds...".format(runLoopStatus, e, sleep_timer)
traceback.print_exc()
time.sleep(sleep_timer)
def checkBackup(self):
backupList = json.loads(requests.get(self.device_address + "/json/backups").text)
file = backupList["backups"][0]
data = json.loads(requests.get(self.device_address + "/json/backups/" + file).text)
timestamp = time.time()
if data != self.wellSetup or self.forceSend:
self.sendtodbJSON("well_setup", json.dumps(data), timestamp)
self.wellSetup = data
with open('wellSetup.p', 'wb') as handle:
pickle.dump(self.wellSetup, handle)
def checkEvents(self):
data = json.loads(requests.get(self.device_address + "/json/event_list").text)
events = data["events"]
for event in events:
if int(event["id"]) not in self.eventIds:
timestamp = calendar.timegm(time.strptime(event["datetime"], '%Y-%m-%dT%H:%M:%S.%fZ'))
# we have a new event
self.sendtodbJSON("events", json.dumps(event), timestamp)
self.eventIds.append(int(event["id"]))
if len(self.eventIds) > 50:
del self.eventIds[0]
with open('eventIds.p', 'wb') as handle:
pickle.dump(self.eventIds, handle)
def checkNotes(self):
data = json.loads(requests.get(self.device_address + "/json/notes/get").text)
notes = data["notes"]
for note in notes:
if int(note["id"]) not in self.noteIDs:
timestamp = calendar.timegm(time.strptime(note["date_time"], '%Y-%m-%d %H:%M:%S'))
# we have a new note
self.sendtodbJSON("notes", json.dumps(note), timestamp)
self.noteIDs.append(int(note["id"]))
if len(self.noteIDs) > 50:
del self.noteIDs[0]
with open('noteIDs.p', 'wb') as handle:
pickle.dump(self.noteIDs, handle)
def checkFluidShots(self):
data = json.loads(requests.get(self.device_address + "/json/fluid_shot/get").text)
fluid_shots = data["fluid_shots"]
for shot in fluid_shots:
if int(shot["id"]) not in self.fluidshotIDs:
timestamp = calendar.timegm(time.strptime(shot["shot_datetime"], '%Y-%m-%d %H:%M:%S'))
# we have a new note
self.sendtodbJSON("fluidshots", json.dumps(shot), timestamp)
self.fluidshotIDs.append(int(shot["id"]))
if len(self.fluidshotIDs) > 50:
del self.fluidshotIDs[0]
with open('fluidshotIDs.p', 'wb') as handle:
pickle.dump(self.fluidshotIDs, handle)
def checkWellTests(self):
data = json.loads(requests.get(self.device_address + "/json/well_test/get").text)
well_tests = data["well_tests"]
for test in well_tests:
if int(test["id"]) not in self.welltestIDs:
timestamp = calendar.timegm(time.strptime(test["test_date"], '%Y-%m-%d %H:%M:%S'))
# we have a new note
self.sendtodbJSON("welltests", json.dumps(test), timestamp)
self.welltestIDs.append(int(test["id"]))
if len(self.welltestIDs) > 50:
del self.welltestIDs[0]
with open('welltestIDs.p', 'wb') as handle:
pickle.dump(self.welltestIDs, handle)
def checkStatus(self):
global status
statusMap = {
0: 'Stopped',
1: 'Running',
2: 'Pumped Off',
3: 'Faulted',
4: 'Starting',
5: 'Recovering',
100: 'Read Error',
1000: 'PLC Error',
9999: 'No Response'
}
st_response = requests.get(self.device_address + "/json/status")
if st_response.status_code == 200:
data = json.loads(st_response.text)
# date = data["ISOdate"]
status_read = data["run_status"]
if status.last_value != status_read:
self.statusChanged = True
print "Status has changed from {0} to {1} @ {2}".format(status.last_value, status_read, time.time())
else:
self.statusChanged = False
if self.statusChanged or self.forceSend:
self.status = status_read
# reg = "(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}).(\d{3})Z"
# fd = re.search(reg, date)
# dt = datetime(int(fd.group(1)), int(fd.group(2)), int(fd.group(3)), int(fd.group(4)), int(fd.group(5)), int(fd.group(6)), int(fd.group(7)))
# # timestamp = int(time.mktime(time.strptime(date, '%Y-%m-%dT%H:%M:%S.%fZ')))
# timestamp = calendar.timegm(dt.timetuple())
self.sendtodb("status", status_read, 0)
status.last_value = status_read
def checkDailyTotals(self):
data = json.loads(requests.get(self.device_address + "/json/totals").text)
if data['status'] == "OK":
totals = data["totals"]
timestamp = 0
for val in totals:
if val['name'] in dt_channels:
if dt_channels[val['name']].checkSend(val['value'], False):
self.sendtodb(dt_channels[val['name']].mesh_name, dt_channels[val['name']].value, timestamp)
else:
print("checkDailyTotalsError: {0}".format(data.message))
def checkGaugeOffData(self):
data = json.loads(requests.get(self.device_address + "/json/history").text)
day = data["hist"]
date = day['gauge_date']
reg = "(\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})"
fd = re.search(reg, date)
dt = datetime(int(fd.group(1)), int(fd.group(2)), int(fd.group(3)), int(fd.group(4)), int(fd.group(5)), int(fd.group(6)))
# timestamp = int(time.mktime(time.strptime(date, '%Y-%m-%dT%H:%M:%S.%fZ')))
timestamp = calendar.timegm(dt.timetuple())
for entry in day:
if entry in go_channels:
if go_channels[entry].checkSend(day[entry], False):
self.sendtodb(go_channels[entry].mesh_name, day[entry], timestamp)
def checkLatestCard(self, numCards = 1):
latest = ""
if numCards == 1:
latest = json.loads(requests.get(self.device_address + "/json/latestcard").text)
else:
latest = json.loads(requests.get(self.device_address + "/json/latestcard/{}".format(numCards)).text)
# check the card to see if its new
# 1. if its new send the folder/file_name to the card_history channel
# 2. if its new and its been 10 minutes since you last sent an entire card, then send up all of the data
for i in range(0, len(latest['card_data'])):
card = latest['card_data'][i]
if card_channels['card_id'].checkSend(card['Card_ID'], self.forceSend):
dateTime = str(card["Stroke_Time"])
reg = "(\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})"
fd = re.search(reg, dateTime)
dt = datetime(int(fd.group(1)), int(fd.group(2)), int(fd.group(3)), int(fd.group(4)), int(fd.group(5)), int(fd.group(6)))
timestamp = calendar.timegm(dt.timetuple())
card_timestamp = int(time.mktime(dt.timetuple()))
print "New card detected @ {0}".format(datetime.strftime(datetime.fromtimestamp(timestamp), "%Y-%m-%d %H:%M:%S.%f"))
# set the last value = to current value and upload your data
self.sendtodb("card_history", card_channels['card_id'].value, timestamp)
# check the last time the card was updated
if (time.time() - card_channels['card'].last_send_time) > self.cardLoopTimer or self.statusChanged or self.forceSend:
# its been 10 minutes, send the full upload
print "Either status has changed or last stored card is too old."
card_channels["card"].last_send_time = time.time()
self.process_card(card, timestamp, card_timestamp, sendCards=True)
return
else:
self.process_card(card, timestamp, card_timestamp, sendCards=False)
def process_card(self, data, data_timestamp, card_timestamp, sendCards=False):
# if sendCards = True then we upload all data no matter what, including cards
# check what type of data it is
# check if its changed, if it has, how long has it been since it changed
# NOTE: the initial vaue of "" is given to all channels in the channels object,
# so to avoid comparing a string to a float, and to make sure on startup we send all of the values, the first time through we send everything that has a "" as its last value
if sendCards:
self.sendtodb("cardtype", str(data['Card_Type']), int(data_timestamp))
s_p = data["Surface_Position"]
s_l = data["Surface_Load"]
d_p = data["Downhole_Position"]
d_l = data["Downhole_Load"]
newSc = "["
newDc = "["
for i in range(len(s_p)):
try:
if s_p[i] is None:
continue
if s_p[i] != 0.0 and s_l[i] != 0.0:
newSc += "[" + str(s_p[i]) + ", " + str(s_l[i]) + "], "
except:
pass
newSc += "[" + str(s_p[0]) + ", " + str(s_l[0]) + "]"
newSc += "]"
for i in range(len(d_p)):
try:
if d_p[i] is None:
continue
if d_p[i] != 0.0 and d_l[i] != 0.0:
newDc += "[" + str(d_p[i]) + ", " + str(d_l[i]) + "], "
except:
pass
newDc += "[" + str(d_p[0]) + ", " + str(d_l[0]) + "]"
newDc += "]"
self.sendtodb("sc", newSc, card_timestamp)
self.sendtodb("dc", newDc, card_timestamp)
def checkStoredValues(self, forceSend):
data = json.loads(requests.get(self.device_address + "/json/tagvalues").text)
if data['status'] == "OK":
vals = data['vals']
for val in vals:
if val['name'] in tag_channels:
if tag_channels[val['name']].checkSend(val['val'], forceSend):
self.sendtodbJSON(tag_channels[val['name']].mesh_name, tag_channels[val['name']].value, 0)
def getLatestXCards(self, numCards):
data = json.loads(requests.get(self.device_address + "/json/latest/" + str(int(numCards))).text)
for card in data['cards']:
card_data = json.loads(requests.get(self.device_address + "/json/" + data['folder'] + "/" + card).text)
dateTime = str(card_data["card_data"]["Stroke_Time"])
# timestamp = time.mktime(time.strptime(dateTime,'%Y-%m-%dT%H:%M:%S.%fZ'))
reg = "(\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})"
fd = re.search(reg, dateTime)
dt = datetime(int(fd.group(1)), int(fd.group(2)), int(fd.group(3)), int(fd.group(4)), int(fd.group(5)), int(fd.group(6)))
timestamp = calendar.timegm(dt.timetuple())
card_timestamp = int(time.mktime(dt.timetuple()))
channels["card_history"]["last_value"] = (data['folder'] + "/" + card)
self.sendtodb("card_history", (data['folder'] + "/" + card), card_timestamp)
self.process_card(card_data, timestamp, card_timestamp, sendCards=True)
def getDataLoggerStatus(self):
data = json.loads(requests.get(self.device_address + "/json/pythonstatus/").text)
al_status = "Not OK"
if data['status']['alarmLogger']:
al_status = "OK"
if al_status != self.al_status_last:
self.sendtodb("alarmlogger_status", al_status, 0)
self.al_status_last = al_status
dl_status = "Not OK"
if data['status']['dataLogger']:
dl_status = "OK"
if al_status != self.dl_status_last:
self.sendtodb("datalogger_status", dl_status, 0)
self.dl_status_last = dl_status
def poc_get_card(self, name, value):
self.getcard(value)
def poc_sync(self, name, value):
self.sendtodb("connected", "true", 0)
return True
def poc_set_address(self, name, value):
self.device_address = value
return True
def poc_refresh_data(self, name, value):
self.forceSend = True
return True