SolarControl/solarcontrol.py

#!/usr/bin/python3

import paho.mqtt.client as mqtt
import os
from dotenv import load_dotenv
from time import time, sleep
import json
from threading import Thread
import requests
import math

# get env vars
load_dotenv()
mq_broker = os.getenv('MQBROKER', None)
mq_port = int(os.getenv('MQPORT', None))
mq_user = os.getenv('MQUSER', None)
mq_pwd = os.getenv('MQPWD', None)
mq_topic_p_house = os.getenv('MQTOPIC_P_HOUSE', None)
mq_topic_p_solar = os.getenv('MQTOPIC_P_SOLAR', None)
maximum_data_ts_deviation = int(os.getenv('MAXIMUM_DATA_TS_DEVIATION', 5))
opendtu_address = os.getenv('OPENDTU_ADDR', None)
opendtu_user = os.getenv('OPENDTU_USER', None)
opendtu_pwd = os.getenv('OPENDTU_PWD', None)
opendtu_inverter_sn = os.getenv('OPENDTU_INVERTER_SN', None)
limit_correction_factor = float(os.getenv('LIMIT_CORRECTION_FACTOR', 1.0))
limit_update_interval = float(os.getenv('LIMIT_UPDATE_INTERVAL', 5))
power_target = int(os.getenv('POWER_TARGET', 50))
power_target_min = float(os.getenv('POWER_TARGET_MIN', 0))
power_target_max = float(os.getenv('POWER_TARGET_MAX', 100))
power_damping_factor = float(os.getenv('POWER_DAMPING_FACTOR', 0.3))
power_limit_change_treshold = float(os.getenv('POWER_LIMIT_CHANGE_TRESHOLD', 0.5))
power_limit_type = int(os.getenv('POWER_LIMIT_TYPE', 1))
dry_run = bool(int(os.getenv('DRY_RUN', 1)))

# some checks for the correctness of supplied data
if power_target_min < 0: power_target_min = 0
if power_target_max > 100: power_target_max = 100
if power_damping_factor < 0: power_damping_factor = 0.0
if power_damping_factor > 1: power_damping_factor = 1.0


# create the powers dict (containing the current use) and data variables (for thread sharing)
powers_raw = {"solar": 0, "solar_ts": 0, "house": 0, "house_ts": 0}
powers = {"total": None, "total_house": None, "total_solar": None, "timestamp": 0}

# define mqtt callbacks
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Connected with result code {reason_code}")
    client.subscribe("lge320/#")
    client.subscribe("solar/ac/#")
def on_message(client, userdata, msg):
    #print(f"{msg.topic}: {msg.payload}")
    if msg.topic == mq_topic_p_house:
        powers_raw["house_ts"] = time()
        powers_raw["house"] = math.floor(((powers_raw["house"] + float(json.loads(msg.payload)["total_act_power"])) / 2)*10)/10
    elif msg.topic == mq_topic_p_solar:
        powers_raw["solar_ts"] = time()
        powers_raw["solar"] = math.floor(((powers_raw["solar"] + float(msg.payload)) / 2)*10)/10

# initialize the mqtt client
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.on_connect = on_connect
mqttc.on_message = on_message

mqttc.username_pw_set(mq_user, mq_pwd)
mqttc.connect(mq_broker, mq_port, 60)

mqttc.loop_start()

def threaded_current_power_calculation():
    while True:
        last_powers = powers_raw.copy()
        while last_powers == powers_raw:
            sleep(0.2)
        if abs(powers_raw["solar_ts"] - powers_raw["house_ts"]) < maximum_data_ts_deviation and powers_raw["solar"] != None and powers_raw["house"] != None:
            print(f"Current total P: {powers_raw['solar'] + powers_raw['house']} | Solar P: {powers_raw['solar']} | House P: {powers_raw['house']}")
            powers["total"] = powers_raw["solar"] + powers_raw["house"]
            powers["total_house"] = powers_raw["house"]
            powers["total_solar"] = powers_raw["solar"]
            powers["timestamp"] = time()

power_calc_thread = Thread(target=threaded_current_power_calculation)
power_calc_thread.start()


def threaded_solar_power_limit_setting():
    last_time = time()
    while True:
        # Get current openDTU current limit status
        status = requests.get(opendtu_address.strip("/") + "/api/limit/status", auth=(opendtu_user, opendtu_pwd)).json().copy()
        
        while(time() - powers["timestamp"] > 1): # wait until recent data is available
            sleep(0.2)

        if status[opendtu_inverter_sn]["limit_set_status"] == "Ok" and powers["total"] != None:
            # calculate percentage to set the limit to
            new_ideal_limit = math.floor(((powers["total"]-power_target) / status[opendtu_inverter_sn]["max_power"]) * limit_correction_factor * 1000)/10  # * 100 because its a percentage
            if new_ideal_limit > power_target_max: new_ideal_limit = power_target_max
            if new_ideal_limit < power_target_min: new_ideal_limit = power_target_min

            cur_limit = status[opendtu_inverter_sn]["limit_relative"]

            diff_cur_new_limit = cur_limit - new_ideal_limit

            if abs(diff_cur_new_limit) >= power_limit_change_treshold:
                # dampen only when not giving away energy to the provider
                if powers["total_house"] < 0:
                    new_dampened_limit = new_ideal_limit
                else:
                    new_dampened_limit = cur_limit - (diff_cur_new_limit * (1-power_damping_factor))
                new_dampened_limit = math.floor(new_dampened_limit*10)/10
                if new_dampened_limit > power_target_max: new_dampened_limit = power_target_max
                if new_dampened_limit < power_target_min: new_dampened_limit = power_target_min
                if not dry_run:
                    data_to_send = 'data={"serial":"'+opendtu_inverter_sn+'","limit_type":'+str(power_limit_type)+',"limit_value":'+str(new_dampened_limit)+'}'
                    r = requests.post(opendtu_address.strip("/") + "/api/limit/config", data=data_to_send, headers={'Content-Type':'text/plain'}, auth=(opendtu_user, opendtu_pwd))
                    print("Setting new limit over the API: " + str(new_dampened_limit) + f"% (previously: {cur_limit}% | currently targeting: {new_ideal_limit}%)... ", end="")
                    #print(f"Current limit: {cur_limit}, new ideal limit: {new_ideal_limit}, new dampened limit: {new_dampened_limit}")
                    if json.loads(r.text)["type"] == "success":
                       print("Success!")
                    else: 
                       print(f"Unsuccessful :( Code: {json.loads(r.text)['code']} ({json.loads(r.text)['type']})")
                       print(r.text)  # keep and uncomment for debug reasons
                else:
                    print("Now the new limit would be set over the API (but DRY_RUN is either not specified or True): " + str(new_limit) + "%")


        while (time() - last_time) < limit_update_interval:
            sleep(0.5)
        last_time = time()

solar_power_limit_set_thread = Thread(target=threaded_solar_power_limit_setting)
solar_power_limit_set_thread.start()