nmelone/MaxWaterSystem-Lite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0fe4826e39d9c96d998184679577483e414b089f
MaxWaterSystem-Lite/flow-monitor.py
Patrick McDonagh ca1aa5e0f6 Increments to version 12
2018-05-15 15:23:34 -05:00

571 lines
24 KiB
Python
Raw Blame History

"""Driver for connecting Flow Monitor to Meshify."""
import threading
import time
from datetime import datetime
import sqlite3
import logging
import sys
from device_base import deviceBase
import persistence
from utilities import get_public_ip_address
# LOGGING SETUP
from logging.handlers import RotatingFileHandler
log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(funcName)s(%(lineno)d) %(message)s')
logFile = './flowmonitor.log'
my_handler = RotatingFileHandler(logFile, mode='a', maxBytes=500*1024, backupCount=2, encoding=None, delay=0)
my_handler.setFormatter(log_formatter)
my_handler.setLevel(logging.INFO)
logger = logging.getLogger('flowmonitor')
logger.setLevel(logging.INFO)
logger.addHandler(my_handler)
console_out = logging.StreamHandler(sys.stdout)
console_out.setFormatter(log_formatter)
logger.addHandler(console_out)
logger.info("flowmonitor startup")
CREATE_FLOWDATA_TABLE = """CREATE TABLE flow_data (
id integer PRIMARY KEY,
gal_totalizer_value float,
bbl_totalizer_value float,
gal_monthly_totalizer float,
bbl_monthly_totalizer float,
last_measured_timestamp integer
);"""
INSERT_BLANK_FLOWDATA = """INSERT INTO flow_data (
id,
gal_totalizer_value,
bbl_totalizer_value,
gal_monthly_totalizer,
bbl_monthly_totalizer,
last_measured_timestamp)
VALUES (1, 0.0, 0.0, 0.0, 0.0, 0);"""
CLEAR_FLOWDATA = """UPDATE flow_data SET
gal_totalizer_value=0.0,
bbl_totalizer_value=0.0,
gal_monthly_totalizer=0.0,
bbl_monthly_totalizer=0.0,
last_measured_timestamp=0 WHERE id=1;"""
UPDATE_FLOWDATA = """UPDATE flow_data SET
gal_totalizer_value=?,
bbl_totalizer_value=?,
gal_monthly_totalizer=?,
bbl_monthly_totalizer=?,
last_measured_timestamp=? WHERE id=1"""
PERSIST = persistence.load()
def migrate_from_sqlite_to_json_scaling():
"""Migrate data from SQLite db to JSON, if needed"""
global PERSIST
try:
PERSIST['flow_raw_min']
logger.info("Don't need to migrate from SQLite to JSON.")
return False
except (KeyError, TypeError):
logger.info("Migrating from SQLite to JSON.")
if not PERSIST:
PERSIST = {}
conn = sqlite3.connect('/root/python_firmware/drivers/flow-monitor.db')
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM scaling_data WHERE id = 1') # dummy query for checking database
stored_data = cursor.fetchone()
flow_raw_min = stored_data[1]
flow_raw_max = stored_data[2]
flow_gpm_min = stored_data[3]
flow_gpm_max = stored_data[4]
except (sqlite3.OperationalError, TypeError):
logger.info("No stored data in SQLite for scaling, we'll just use default values")
return False
PERSIST["flow_raw_min"] = flow_raw_min
PERSIST["flow_raw_max"] = flow_raw_max
PERSIST["flow_gpm_min"] = flow_gpm_min
PERSIST["flow_gpm_max"] = flow_gpm_max
persistence.store(PERSIST)
class ChannelSimple(object):
"""Simple Meshify channel structure."""
def __init__(self, meshify_name, senddelta_value, senddelta_time):
"""Initialize the channel with variables."""
self.meshify_name = meshify_name
self.senddelta_time = senddelta_time
self.senddelta_value = senddelta_value
self.last_sent_value = None
self.last_sent_timestamp = 0
def check_if_send_needed(self, value, timestamp):
"""Check to see if the value needs to be pushed."""
if self.last_sent_value is None or self.last_sent_timestamp == 0:
return True
if abs(value - self.last_sent_value) > self.senddelta_value:
return True
if (timestamp - self.last_sent_timestamp) > self.senddelta_time:
return True
return False
def update(self, last_sent_value, last_sent_timestamp):
"""Update values after a push."""
self.last_sent_value = last_sent_value
self.last_sent_timestamp = last_sent_timestamp
def scale(raw_val, raw_min, raw_max, eu_min, eu_max):
"""Scale a raw value."""
slope = (eu_max - eu_min) / (raw_max - raw_min)
intercept = eu_max - (slope * raw_max)
return slope * raw_val + intercept
def is_today(tstamp):
"""Check if a given timestamp belongs to the current date."""
midnight_today = datetime.today().replace(hour=0, minute=0, second=0, microsecond=0)
midnight_ts = (midnight_today - datetime(1970, 1, 1)).total_seconds()
return tstamp >= midnight_ts
def is_thismonth(tstamp):
"""Check if a given timestamp belongs to the current month."""
today = datetime.today()
tstamp_date = datetime.fromtimestamp(tstamp)
return today.month == tstamp_date.month
class start(threading.Thread, deviceBase):
"""Start class required for driver."""
def __init__(self, name=None, number=None, mac=None, Q=None, mcu=None,
companyId=None, offset=None, mqtt=None, Nodes=None):
"""Initalize the driver."""
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)
# Default Scaling Values for Flowmeter
self.flow_raw_min = 3.89
self.flow_raw_max = 19.54
self.flow_gpm_min = 0.0
self.flow_gpm_max = 100.0
self.pressure_raw_min = 0.00
self.pressure_raw_max = 10.00
self.pressure_psi_min = 0.00
self.pressure_psi_max = 600.0
self.GPM_IGNORE_LIMIT = 1.0
self.daemon = True
self.version = "12"
self.finished = threading.Event()
threading.Thread.start(self)
# 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):
"""Register the driver."""
self.channels["status"]["last_value"] = ""
def run(self):
"""Run the driver."""
# Configuration Parameters
total_time_store_delta = 600 # seconds
flow_time_store_delta = 600 # seconds
pressure_time_store_delta = 600 # seconds
gal_per_bbl = 42.0
startup_wait_seconds = 30
ip_check_after = 3600 # check public IP address after an hour
# Initialization
gpm_val = 0.0
date_reset = False
month_reset = True # True because False was causing problems when starting up on the first of the month
gal_totalizer_value = 0.0
bbl_totalizer_value = 0.0
gal_monthly_totalizer = 0.0
bbl_monthly_totalizer = 0.0
# Channels
galtotal_ch = ChannelSimple('gal_total', 100.0, total_time_store_delta)
bbltotal_ch = ChannelSimple('bbl_total', galtotal_ch.senddelta_value/gal_per_bbl, total_time_store_delta)
galtotalthismonth_ch = ChannelSimple('gal_total_thismonth', galtotal_ch.senddelta_value, total_time_store_delta)
bbltotalthismonth_ch = ChannelSimple('bbl_total_thismonth', galtotalthismonth_ch.senddelta_value/gal_per_bbl, total_time_store_delta)
gpmflow_ch = ChannelSimple('gpm_flow', 10.0, flow_time_store_delta)
bpdflow_ch = ChannelSimple('bpd_flow', gpmflow_ch.senddelta_value * 34.2857, flow_time_store_delta)
psipressure_ch = ChannelSimple('psi_pressure', 10.0, pressure_time_store_delta)
runstatus_ch = ChannelSimple('run_status', 0.5, 600)
# Startup timer.
# Waits for connection to Meshify before attempting to send data
wait_loops = 0
while wait_loops < startup_wait_seconds:
logger.info("Waiting to start flowmonitor driver in {} seconds".format(startup_wait_seconds - wait_loops))
wait_loops += 1
time.sleep(1)
##############################################
# THIS IS THE ACTUAL DRIVER CODE
# (executes after waiting for startup timer to complete)
##############################################
# Determine public IP address and send to Meshify
public_ip_address = get_public_ip_address()
self.sendtodb('public_ip_address', public_ip_address, 0)
ip_checked_time = time.time()
# Attempt to retrieve data stored in the database
last_measured_timestamp = time.time()
conn = sqlite3.connect('/root/python_firmware/drivers/flow-monitor.db')
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM flow_data WHERE id = 1') # dummy query for checking database
stored_data = cursor.fetchone()
gal_totalizer_value = stored_data[1]
bbl_totalizer_value = stored_data[2]
gal_monthly_totalizer = stored_data[3]
bbl_monthly_totalizer = stored_data[4]
last_measured_timestamp = stored_data[5]
except sqlite3.OperationalError:
# Caught if the table does not exist in the database.
logger.warning("No table flow_data in the database. I'll create it now.")
cursor.execute(CREATE_FLOWDATA_TABLE)
cursor.execute(INSERT_BLANK_FLOWDATA)
conn.commit()
except IndexError:
# Reset the database if the correct size data is not in the database.
self.flowmonitor_resetdatabase(None, None)
# Load scaling data
migrate_from_sqlite_to_json_scaling()
try:
self.flow_raw_min = PERSIST["flow_raw_min"]
self.flow_raw_max = PERSIST["flow_raw_max"]
self.flow_gpm_min = PERSIST["flow_gpm_min"]
self.flow_gpm_max = PERSIST["flow_gpm_max"]
except KeyError:
logger.warning("No flow scaling data exists in persist file. We'll add some now")
PERSIST["flow_raw_min"] = self.flow_raw_min
PERSIST["flow_raw_max"] = self.flow_raw_max
PERSIST["flow_gpm_min"] = self.flow_gpm_min
PERSIST["flow_gpm_max"] = self.flow_gpm_max
persistence.store(PERSIST)
self.sendtodb("setrawmin", self.flow_raw_min, 0)
self.sendtodb("setrawmax", self.flow_raw_max, 0)
self.sendtodb("setgpmmin", self.flow_gpm_min, 0)
self.sendtodb("setgpmmax", self.flow_gpm_max, 0)
try:
self.pressure_raw_min = PERSIST["pressure_raw_min"]
self.pressure_raw_max = PERSIST["pressure_raw_max"]
self.pressure_psi_min = PERSIST["pressure_psi_min"]
self.pressure_psi_max = PERSIST["pressure_psi_max"]
except KeyError:
logger.warning("No flow scaling data exists in persist file. We'll add some now")
PERSIST["pressure_raw_min"] = self.pressure_raw_min
PERSIST["pressure_raw_max"] = self.pressure_raw_max
PERSIST["pressure_psi_min"] = self.pressure_psi_min
PERSIST["pressure_psi_max"] = self.pressure_psi_max
persistence.store(PERSIST)
self.sendtodb("setpressurerawmin", self.pressure_raw_min, 0)
self.sendtodb("setpressurerawmax", self.pressure_raw_max, 0)
self.sendtodb("setpressurepsimin", self.pressure_psi_min, 0)
self.sendtodb("setpressurepsimax", self.pressure_psi_max, 0)
try:
self.GPM_IGNORE_LIMIT = PERSIST['gpm_ignore_limit']
except Exception:
logger.warning("no persisted GPM Ignore Limit. Using default of {}".format(self.GPM_IGNORE_LIMIT))
self.flowmonitor_setgpmignorelimit("gpm_ignore_limit", self.GPM_IGNORE_LIMIT)
# on bootup, if the day has changed, clear the totalizers.
# this would happen if the device is off when the time changes to Midnight.
if not is_today(last_measured_timestamp):
gal_totalizer_value = 0.0
bbl_totalizer_value = 0.0
last_measured_timestamp = time.time()
# DRIVER LOOP
while True:
try:
mcu_status = self.mcu.getDict() # Gets a dictionary of the IO states
# {
# 'bat': u'23.10',
# 'ver': u'Mar 16 2016 21:29:31',
# 'dout3': 'Off',
# 'temp': u'40.37',
# 'vin': u'24.6',
# 'pulse': u'0',
# 'dout4': 'Off',
# 'dout1': 'Off',
# 'din2': 'Off',
# 'din1': 'Off',
# 'dout2': 'On',
# 'cloop': u'0.0',
# 'analog4': u'0.0',
# 'analog3': u'0.0',
# 'analog2': u'0.0',
# 'analog1': u'0.0',
# 'relay1': 'Off'
# }
cloop_val = float(mcu_status['cloop'])
analog1_val = float(mcu_status['analog1'])
din1_val = 1 if mcu_status['din1'] == 'On' else 0 # Check DIGITAL INPUT 1 for run status
gpm_val = scale(cloop_val, self.flow_raw_min, self.flow_raw_max, self.flow_gpm_min, self.flow_gpm_max)
psi_val = scale(analog1_val, self.pressure_raw_min, self.pressure_raw_max, self.pressure_psi_min, self.pressure_psi_max)
if gpm_val < self.GPM_IGNORE_LIMIT:
gpm_val = 0
bpd_val = (gpm_val / gal_per_bbl) * 60.0 * 24.0 # Computes BPD from GPM
now = time.time()
time_diff = now - last_measured_timestamp
if time_diff > 0:
# Volume flowed since last measuring
gal_flow_delta = (time_diff / 60.0) * gpm_val
bbl_flow_delta = (time_diff / 60.0) * (1.0 / 60.0) * (1.0 / 24.0) * bpd_val
# Increment totalizers
gal_totalizer_value += gal_flow_delta
bbl_totalizer_value += bbl_flow_delta
gal_monthly_totalizer += gal_flow_delta
bbl_monthly_totalizer += bbl_flow_delta
last_measured_timestamp = now
# Update the database with the most recent totalizer values.
cursor.execute(UPDATE_FLOWDATA, (gal_totalizer_value, bbl_totalizer_value, gal_monthly_totalizer, bbl_monthly_totalizer, last_measured_timestamp))
conn.commit()
logger.info('gpm: {}, bpd: {}, psi: {}, gal: {}, bbl:{}, month_gal:{}, month_bbl:{}'.format(
gpm_val, bpd_val, psi_val, gal_totalizer_value, bbl_totalizer_value, gal_monthly_totalizer, bbl_monthly_totalizer))
# Channel Checks:
# check to see if the value needs to be sent to Meshify
if galtotal_ch.check_if_send_needed(gal_totalizer_value, now):
self.sendtodb(galtotal_ch.meshify_name, gal_totalizer_value, 0)
galtotal_ch.update(gal_totalizer_value, now)
if bbltotal_ch.check_if_send_needed(bbl_totalizer_value, now):
self.sendtodb(bbltotal_ch.meshify_name, bbl_totalizer_value, 0)
bbltotal_ch.update(bbl_totalizer_value, now)
if galtotalthismonth_ch.check_if_send_needed(gal_monthly_totalizer, now):
self.sendtodb(galtotalthismonth_ch.meshify_name, gal_monthly_totalizer, 0)
galtotalthismonth_ch.update(gal_monthly_totalizer, now)
if bbltotalthismonth_ch.check_if_send_needed(bbl_monthly_totalizer, now):
self.sendtodb(bbltotalthismonth_ch.meshify_name, bbl_monthly_totalizer, 0)
bbltotalthismonth_ch.update(bbl_monthly_totalizer, now)
if gpmflow_ch.check_if_send_needed(gpm_val, now):
self.sendtodb(gpmflow_ch.meshify_name, gpm_val, 0)
gpmflow_ch.update(gpm_val, now)
if bpdflow_ch.check_if_send_needed(bpd_val, now):
self.sendtodb(bpdflow_ch.meshify_name, bpd_val, 0)
bpdflow_ch.update(bpd_val, now)
if psipressure_ch.check_if_send_needed(psi_val, now):
self.sendtodb(psipressure_ch.meshify_name, psi_val, 0)
psipressure_ch.update(psi_val, now)
if runstatus_ch.check_if_send_needed(din1_val, now):
self.sendtodb(runstatus_ch.meshify_name, din1_val, 0)
runstatus_ch.update(din1_val, now)
# Check for the clock hitting midnight for resetting the daily totalizer value
if time.localtime(now)[3] == 0 and not date_reset:
self.sendtodb('gal_total_yesterday', gal_totalizer_value, 0)
self.sendtodb('bbl_total_yesterday', bbl_totalizer_value, 0)
gal_totalizer_value = 0.0
bbl_totalizer_value = 0.0
# Update the database with cleared values
cursor.execute(UPDATE_FLOWDATA, (gal_totalizer_value, bbl_totalizer_value, gal_monthly_totalizer, bbl_monthly_totalizer, last_measured_timestamp))
conn.commit()
date_reset = True
# Once the hour goes to anything other than 0 (midnight), unset the Date Reset status
if time.localtime(now)[3] != 0 and date_reset:
date_reset = False
# Check for a new month for resetting the monthly totalizers
if time.localtime(now)[2] == 1 and not month_reset:
self.sendtodb('gal_total_lastmonth', gal_monthly_totalizer, 0)
self.sendtodb('bbl_total_lastmonth', bbl_monthly_totalizer, 0)
gal_monthly_totalizer = 0.0
bbl_monthly_totalizer = 0.0
# Update the database with cleared values
cursor.execute(UPDATE_FLOWDATA, (gal_totalizer_value, bbl_totalizer_value, gal_monthly_totalizer, bbl_monthly_totalizer, last_measured_timestamp))
conn.commit()
month_reset = True
# once it's no longer the 1st of the month, unset the Month Reset status
if time.localtime(now)[2] != 1 and month_reset:
month_reset = False
# Periodically check the public IP address to see if it has changed.
if (now - ip_checked_time) > ip_check_after:
test_public_ip = get_public_ip_address()
if not test_public_ip == public_ip_address:
self.sendtodb('public_ip_address', test_public_ip, 0)
public_ip_address = test_public_ip
ip_checked_time = now
if mcu_status['din2'] == 'On':
self.flowmonitor_startcmd(None, None)
if mcu_status['din1'] == 'Off' and mcu_status['relay1'] == "On":
self.flowmonitor_stopcmd(None, None)
except Exception as e:
logger.error("problem in the driver: {}".format(e))
time.sleep(2)
def flowmonitor_startcmd(self, name, value):
"""Start the well."""
self.mcu.relay1(str(1))
return True
def flowmonitor_stopcmd(self, name, value):
"""Stop the well."""
self.mcu.relay1(str(0))
return True
def flowmonitor_resetdatabase(self, name, value):
"""Reset the database back to blank."""
conn = sqlite3.connect('/root/python_firmware/drivers/flow-monitor.db')
cursor = conn.cursor()
try:
cursor.execute('SELECT * FROM flow_data WHERE id = 1')
cursor.execute('DROP TABLE flow_data')
cursor.execute(CREATE_FLOWDATA_TABLE)
cursor.execute(INSERT_BLANK_FLOWDATA)
conn.commit()
logger.info("DATABASE HAS BEEN RESET!")
except sqlite3.OperationalError:
logger.warning("No table flow_data in the database. I'll create it now.")
cursor.execute(CREATE_FLOWDATA_TABLE)
cursor.execute(INSERT_BLANK_FLOWDATA)
conn.commit()
return(True)
def flowmonitor_setrawmin(self, name, value):
"""Set the raw min scaling value."""
try:
self.flow_raw_min = float(value)
self.sendtodb("setrawmin", self.flow_raw_min, 0)
PERSIST['flow_raw_min'] = self.flow_raw_min
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.flow_raw_min: {}".format(e))
return(True)
def flowmonitor_setrawmax(self, name, value):
"""Set the raw max scaling value."""
try:
self.flow_raw_max = float(value)
self.sendtodb("setrawmax", self.flow_raw_max, 0)
PERSIST['flow_raw_max'] = self.flow_raw_max
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.flow_raw_max: {}".format(e))
return(True)
def flowmonitor_setgpmmin(self, name, value):
"""Set the gpm min scaling value."""
try:
self.flow_gpm_min = float(value)
self.sendtodb("setgpmmin", self.flow_gpm_min, 0)
PERSIST['flow_gpm_min'] = self.flow_gpm_min
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.flow_gpm_min: {}".format(e))
return(True)
def flowmonitor_setgpmmax(self, name, value):
"""Set the gpm max scaling value."""
try:
self.flow_gpm_max = float(value)
self.sendtodb("setgpmmax", self.flow_gpm_max, 0)
PERSIST['flow_gpm_max'] = self.flow_gpm_max
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.flow_gpm_max: {}".format(e))
return(True)
def flowmonitor_setpressurerawmin(self, name, value):
"""Set the pressure raw min scaling value."""
try:
self.pressure_raw_min = float(value)
self.sendtodb("setpressurerawmin", self.pressure_raw_min, 0)
PERSIST['pressure_raw_min'] = self.pressure_raw_min
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.pressure_raw_min: {}".format(e))
return(True)
def flowmonitor_setpressurerawmax(self, name, value):
"""Set the pressure raw max scaling value."""
try:
self.pressure_raw_max = float(value)
self.sendtodb("setpressurerawmax", self.pressure_raw_max, 0)
PERSIST['pressure_raw_max'] = self.pressure_raw_max
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.pressure_raw_max: {}".format(e))
return(True)
def flowmonitor_setpressurepsimin(self, name, value):
"""Set the pressure psi min scaling value."""
try:
self.pressure_psi_min = float(value)
self.sendtodb("setpressurepsimin", self.pressure_psi_min, 0)
PERSIST['pressure_psi_min'] = self.pressure_psi_min
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.pressure_psi_min: {}".format(e))
return(True)
def flowmonitor_setpressurepsimax(self, name, value):
"""Set the pressure psi max scaling value."""
try:
self.pressure_psi_max = float(value)
self.sendtodb("setpressurepsimax", self.pressure_psi_max, 0)
PERSIST['pressure_psi_max'] = self.pressure_psi_max
persistence.store(PERSIST)
except Exception as e:
logger.error("Could not set self.pressure_psi_max: {}".format(e))
return(True)
def flowmonitor_setgpmignorelimit(self, name, value):
"""Set the GPM Ignore Limit."""
try:
self.GPM_IGNORE_LIMIT = float(value)
self.sendtodb("setgpmignorelimit", self.GPM_IGNORE_LIMIT, 0)
return True
except Exception as e:
logger.error("Error during flowmonitor_setgpmignorelimit: {}".format(e))
return False
Reference in New Issue View Git Blame Copy Permalink