Hi @dlmpryor - Here are the 3 scripts that I am using. First is to get the data from the device - in my case a selectronic inverter.
This inverter also checks for updates etc every midnight, and I found that it can cause a problem if I am trying to pull down data during this time, as it seems to randomly check sometime between midnight and 1:00 am. So I just put my loop to sleep for 6300 seconds from midnight (arbitary length - probably only needs 30 mins).
If it is not in this midnight period, then use python requests to pull down the data, but using a try loop to manage the occasional data corruption. If there is a problem (an exception) basically sleep for 10 s and try again. The downloaded data is in json format. I change the timestamp format and round down some of the 15 decimal places(!) to 2 and write to a file. This file is very small and is updated every 10 seconds.
import dateutil.parser
from datetime import datetime, time
import pytz # not sure if needed here
begin_time = time(0,1)
end_time = time(1,45)
while True:
dt = datetime.now().time()
if (begin_time<=dt<=end_time):
print('gone to sleep at ',dt)
sleep(6300)
print('woken and reconnected')
else:
pass
r = requests.get('http://192.168.20.xxx/cgi-bin/solarmonweb/devices/xyzetc/point')
# need to manage errors if http fails with poor json
try:
datajson = r.json() # get the data in json
extract = datajson['items']
extract['timestamp'] = (datetime.fromtimestamp(extract['timestamp']).strftime('%Y-%m-%d %H:%M:%S'))
extractcolumns = list(extract.keys())
extract['soc'] = round(extract['battery_soc'],2)
extract['load'] = round(extract['load_w'],2)
extract['pvwatts'] = round(extract['solarinverter_w'],2)
print(json.dumps(extract))
with open('selectronic.json','w') as outfile:
json.dump(extract,outfile)
except:
sleep(10)
sleep(10)
The second script reads this data file and pulls out the SoC value I want to monitor (data[‘soc’]) and appends it to an empty list (‘files’). This list will continue to get longer unless trimmed, so I ‘pop’ the first value when the list reaches 10 entries and test if the list values are strictly decreasing (if yes, result will be True). The list (‘files’) is converted to a string and appended to a string ‘call_email’ that is run with os.system(call_email). As currently setup, an email is sent every 30 mins regardless of the SoC, but this can be turned off. Further down the loop the last SoC value is tested, and if less than 50 (%) triggers another email that can also activate an SMS text message. As shown below, it is just the normal email, but can be changed to the SMS script.
The last 6 lines of code run the schedule - which is effectively just printing a line of dots while waiting for the 30 min schedule to trigger. This probably could more simply done using sleep, but I am trying out schedule as well.
#
# script to keep tabs on selectPro data and email when out of bounds.
# uses pymail2/3.py to send emails
# looks at selectronic.json which is written to file by soc_mqtt_mod1.py
# every 10 sec
# this json file can be used to check threshold values etc in almost real time (no data logged)
# lots of print statements to help in setting up - can be commented out
import time, datetime
import sys
import json
import schedule, time, datetime
import pathlib
import os
count = 0
files = []
fname = pathlib.Path('/home/tony/python/emon_select/selectronic.json')
assert fname.exists(), ('No such file: {fname}') # check that the file exists
# test if SoC continuously decreasing - will return True or False
def strictly_decreasing(files):
return all(files[i] > files[i+1] for i in range(len(files)-1))
# read json and put into dict
def read_json():
#print('def read_json called')
global count
with open('selectronic.json') as json_file:
data = json.load(json_file)
#check if soc getting low
print('json data: ',data)
files.append(data['soc'])
print('length of files = ',len(files))
# make string for calling email
call_email = 'python3 pymail3.py '+ str(files)
os.system(call_email)
if len(files) == 10:
files.pop(0)
result = strictly_decreasing(files)
print(files)
if len(files) == 9 and result and data['soc']<50:
count += 1
if count <= 1:
os.system('python3 pymail3.py passedSoC')
print('email alert sent!')
else:
print('all seems OK...')
count = 0
time.sleep(20)
if data['soc'] < 50:
print('SoC low alert! - send email...')
schedule.every(30).minutes.do(read_json)
while True:
schedule.run_pending()
sys.stdout.write('.')
sys.stdout.flush()
time.sleep(10)
The last script actually sends the email. All 3 scripts could probably be merged into 1, but I found it easier to troubleshoot as separate scripts - and the email script needed tuning.
import time, datetime
import smtplib, ssl
import sys
from email.message import EmailMessage
message = []
message = sys.argv
print('Argument List :', str(sys.argv))
print('Argument List #1:', message[1])
tstamp = datetime.datetime.now(tz=None)
date_time = tstamp.strftime('%d-%m-%Y %H:%M:%S')
sender_email = "[email protected]"
receiver_email = "[email protected]"
msg = EmailMessage()
msg.set_content('Stuff passed through ' + str(message) + ' timestamp = ' + str(date_time))
msg['Subject'] = 'Something like python test notification: '
msg['From'] = sender_email
msg['To'] = receiver_email
port = 465 # For SSL
password = 'my gmail pwd here' # or use manual input
#input("Type your password and press enter: ")
# Create a secure SSL context
context = ssl.create_default_context()
with smtplib.SMTP_SSL("smtp.gmail.com", port, context=context) as server:
server.login("[email protected]", password)
server.send_message(msg)
This works with gmail, but should work with other smtp mail systems with a bit of tuning. It’s pretty simple to setup a second or third gmail account - just google it.
