shed-loads.py

#!/usr/bin/env python

# This script reads battery power from Venus OS DBus
# and turns off smart sockets to shed heavy loads when 
# the battery is discharging

import dbus
import requests

system_bus = dbus.SystemBus()

# Pylontech module connected over CAN
BATTERY_ADDRESS = "com.victronenergy.battery.socketcan_can1"

# What happens if I have more than one battery module?
# Would this value report power of the 1st module or of
# the whole battery bank?
POWER_PATH = "/Dc/0/Power"

# switch hostname is announced with mDNS
BOILER_SWITCH_URL = "http://boiler.local"
BOILER_SWITCH_API_KEY = "C60XXXXXXXXXXXXX"

HTTP_TIMEOUT = 3

# Returns battery power in Watts
# Negative power means battery is discharging
def get_power():
  # Connect to the battery service on the dbus
  battery_service = system_bus.get_object(
    BATTERY_ADDRESS, 
    POWER_PATH
  )

  # Get the GetValue method from the battery service
  get_value_method = battery_service.get_dbus_method(
    'GetValue', 
    'com.victronenergy.BusItem'
  )

  return get_value_method()

def switch_boiler(mode):
  # espurna RESTful API request
  r = requests.put(
    url = "%s/api/relay/0" % BOILER_SWITCH_URL,
    data = {"value": mode},
    headers = {"Api-Key": BOILER_SWITCH_API_KEY},
    timeout = HTTP_TIMEOUT
  )

  if r.status_code != 200:
    print("HTTP Error, boiler switch responded with status code %s" % r.status_code)
    return False

  return True
  
def shed_load():
  #print("Requesting socket OFF")
  switch_boiler("off")

def enable_load():
  #print("Requesting socket ON")
  switch_boiler("on")

def main():
  power = get_power()
  if power < 0:
    #print("Battery is discharging at %sW" % power)
    shed_load()
  elif power > 0:
    #print("Battery is charging at %sW" % power)
    enable_load()
  else:
    #print("Battery is idle")
    enable_load()

main()