Shelly 1PM auslesen mit Python
Das Shelly 1PM kann sowohl schalten als auch den Energiverbrauch der angeschlossenen Geräte messen. Es wird über ein REST-API kontrolliert.
Das API liefert JSON aus, mit ausführlichem Status (editiert):
{
"mqtt": {
"connected": false
},
"time": "",
"serial": 1,
"has_update": false,
"relays": [
{
"ison": true,
"has_timer": false,
"overpower": false
}
],
"meters": [
{
"power": 0.76,
"is_valid": true,
"timestamp": 0,
"counters": [
0.0,
0.0,
0.0
],
"total": 0
}
],
"inputs": [
{
"input": 0
}
],
"ext_temperature": {},
"temperature": 48.78,
"overtemperature": false,
"tmp": {
"tC": 48.78,
"tF": 119.8,
"is_valid": "true"
},
"ram_total": 50680,
"ram_free": 40608,
"fs_size": 233681,
"fs_free": 173441,
"uptime": 4391554
}
Mit dem requests-Modul von Python kann der Status bequem ausgelesen und analysiert werden.
Das Python-Script
Da das Script wahrscheinlich längere Zeit ohne direkte Ueberwachung läuft loggt es Probleme in einem File namens shellypower.log.
Die Ausgabe erfolgt auf die Konsole und kann durch mittels Redirection oder tee in ein File geschreiben werden.
Die Requests sind mit einem Timeout versehen und bei jedem Fehler wird die Zeit zwischen den Abfragen verlängert.
from datetime import datetime
import requests
import logging
import json
import time
logger = logging.getLogger( "shellypower" )
# URL for the shelly status
URL = "http://192.168.11.22/status"
# Username and password
USERNAME = "username"
PASSWORD = "password"
# Time between readings
INTERVAL = 10
# Delay after a failure, doubled for every failure
DELAY = 10
delay = DELAY
def handle_exception( ex ):
global delay
text = f"{type(ex).__name__}: {ex}"
logger.error( text )
time.sleep( delay )
if delay < 3600:
delay *= 2
def main():
while True:
try:
# Read the status from the shelly
response = requests.get( URL, timeout=1, auth=(USERNAME,PASSWORD) )
response.raise_for_status()
# Decode the status
status = response.json()
# Read the current power of the first meter
power = status["meters"][0]["power"]
# Print a timestamp and the power
timestamp = datetime.now().strftime( "%Y-%m-%dT%H:%M:%S" )
print( f"{timestamp},{power}", flush=True )
delay = DELAY
time.sleep( INTERVAL )
except requests.exceptions.RequestException as re:
handle_exception( re )
except json.JSONDecodeError as je:
handle_exception( je )
if __name__ == '__main__':
# Configure the logger
logging.basicConfig( filename="shellypower.log",
level=logging.INFO,
format="%(asctime)s,%(levelname)s,\"%(message)s\"",
datefmt="%d.%m.%Y,%H:%M:%S" )
logger.info("shellypower V1.0" )
try:
main()
except KeyboardInterrupt:
logger.info( "Aborted" )