7 Commits

5 changed files with 95 additions and 13 deletions

View File

8
Dockerfile Normal file
View File

@@ -0,0 +1,8 @@
FROM python:3.13-slim
RUN useradd --create-home --shell /bin/bash solarcontrol
WORKDIR /script
COPY solarcontrol.py ./
RUN pip install --upgrade pip
USER solarcontrol
RUN pip install --user --no-cache-dir paho-mqtt python-dotenv requests
CMD ["python", "solarcontrol.py"]

View File

@@ -1,3 +1,50 @@
# SolarControl # SolarControl
Enforce a zero export (or whatever consumption you like) policy with an OpenDTU-controlled inverter and energy data from MQTT Enforce a zero export (or whatever consumption you like) policy with an OpenDTU-controlled inverter and energy data from MQTT
## Configuring and usage
### Docker
To run it using docker, try the `docker-compose.yaml` file present in this repository. You will need the .env file too for that reasons, so the easiest thing is to just run the following commands:
```bash
git clone https://git.privacynerd.de/BlueFox/SolarControl.git && cd SolarControl
vi .env # adjust the script to your needs
docker-compose up -d && docker-compose logs -f
```
### Bare-bone
The script can be configured using the .env file where you can adjust it (hopefully perfect) to your needs. After that, just run the script:
```bash
python3 lge320reader.py
```
Please note: the .env file needs to be in the same folder or any other folder higher up in the directory structure as the script (more specifically, the WORKDIR). It is just a help, actually, the script searches for specific variables in its environment variables. It only loads the .env file so that you do not need to `export` all the files before running (see https://pypi.org/project/python-dotenv/ for more details). This also means that when using it in docker, you can set the docker containers environment file to that .env file and it will be accepted too.
## Updating
To update, simply use `git pull` to pull the latest changes. Afterwards, you need to restart your script (with docker, just use `docker-compose up -d --force-recreate`).
## Building docker
To build the image for docker, simply use the following commands:
```bash
docker login # login to docker hub
docker buildx create --name buildx-multi-arch
docker buildx use buildx-multi-arch
docker buildx build --no-cache --platform linux/amd64,linux/386,linux/arm/v5,linux/arm/v7,linux/arm64/v8,linux/ppc64le,linux/s390x -t bluefox42/solarcontrol:<VERSION> -t bluefox42/solarcontrol:latest . --push
```
## License
This project is licensed under the terms of the GNU General Public License v3.0 or later, see [COPYING](COPYING).

8
docker-compose.yaml Normal file
View File

@@ -0,0 +1,8 @@
services:
solarcontrol:
image: bluefox42/solarcontrol:latest
hostname: solarcontrol
container_name: solarcontrol
env_file: ".env"
restart: unless-stopped
stop_grace_period: 15s

View File

@@ -1,5 +1,6 @@
#!/usr/bin/python3 #!/usr/bin/python3
import paho.mqtt.client as mqtt import paho.mqtt.client as mqtt
import os import os
from dotenv import load_dotenv from dotenv import load_dotenv
@@ -7,6 +8,7 @@ from time import time, sleep
import json import json
from threading import Thread from threading import Thread
import requests import requests
import urllib3 # only for exception handling
import math import math
@@ -101,12 +103,17 @@ def threaded_solar_power_limit_setting():
last_time = time() last_time = time()
while True: while True:
# Get current openDTU current limit status # Get current openDTU current limit status
try:
status = requests.get(opendtu_address.strip("/") + "/api/limit/status", auth=(opendtu_user, opendtu_pwd)).json().copy() status = requests.get(opendtu_address.strip("/") + "/api/limit/status", auth=(opendtu_user, opendtu_pwd)).json().copy()
except (requests.exceptions.RequestException, urllib3.exceptions.HTTPError) as e:
print(f"{bcolors.ERROR}Some error occured while trying to reach out to OpenDTU to get latest data about the inverter limit status. Skipping for now.{bcolors.ENDC}\n==== START OF EXCEPTION ====\n{e}\n==== END OF EXCEPTION ====")
sleep(0.2) # wait some time (to avaid cpu overload on continuous unavailability of the service)
continue # skip this loop pass
while(time() - powers["timestamp"] > 1): # wait until recent data is available while(time() - powers["timestamp"] > 1): # wait until recent data is available
sleep(0.2) sleep(0.2)
if status[opendtu_inverter_sn]["limit_set_status"] == "Ok" and powers["total"] != None: if status[opendtu_inverter_sn]["limit_set_status"] == "Ok" and powers["total"] != None and status[opendtu_inverter_sn]["max_power"] != 0:
# calculate percentage to set the limit to # 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 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_max: new_ideal_limit = power_target_max
@@ -126,17 +133,29 @@ def threaded_solar_power_limit_setting():
if new_dampened_limit > power_target_max: new_dampened_limit = power_target_max 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 new_dampened_limit < power_target_min: new_dampened_limit = power_target_min
if not dry_run: if not dry_run:
request_failed = False
data_to_send = 'data={"serial":"'+opendtu_inverter_sn+'","limit_type":'+str(power_limit_type)+',"limit_value":'+str(new_dampened_limit)+'}' data_to_send = 'data={"serial":"'+opendtu_inverter_sn+'","limit_type":'+str(power_limit_type)+',"limit_value":'+str(new_dampened_limit)+'}'
print(f"Setting new limit via the API: {bcolors.OKBLUE}{bcolors.BOLD}{str(new_dampened_limit)}%{bcolors.ENDC} ({bcolors.OKGREEN}previously: {bcolors.BOLD}{cur_limit}%{bcolors.ENDC} | {bcolors.WARNING}currently targeting: {bcolors.BOLD}{new_ideal_limit}%{bcolors.ENDC})... ", end="")
try:
r = requests.post(opendtu_address.strip("/") + "/api/limit/config", data=data_to_send, headers={'Content-Type':'text/plain'}, auth=(opendtu_user, opendtu_pwd)) r = requests.post(opendtu_address.strip("/") + "/api/limit/config", data=data_to_send, headers={'Content-Type':'text/plain'}, auth=(opendtu_user, opendtu_pwd))
print(f"Setting new limit over the API: {bcolors.OKBLUE}{bcolors.BOLD}{str(new_dampened_limit)}%{bcolors.ENDC} ({bcolors.OKGREEN}previously: {bcolors.BOLD}{cur_limit}%{bcolors.ENDC} | {bcolors.WARNING}currently targeting: {bcolors.BOLD}{new_ideal_limit}%{bcolors.ENDC})... ", end="") except BaseException as e:
#print(f"Current limit: {cur_limit}, new ideal limit: {new_ideal_limit}, new dampened limit: {new_dampened_limit}") request_failed = True
if request_failed:
print(f"{bcolors.FAIL}Unsuccessful :( The http request failed somehow. Printing traceback.{bcolors.ENDC}\n{e}")
else:
if "type" in json.loads(r.text).keys() and "code" in json.loads(r.text).keys():
if json.loads(r.text)["type"] == "success": if json.loads(r.text)["type"] == "success":
print(bcolors.OKGREEN+"Success!"+bcolors.ENDC) print(f"{bcolors.OKGREEN}Success!{bcolors.ENDC}")
else: else:
print(f"{bcolors.FAIL}Unsuccessful :( Code: {bcolors.BOLD}{json.loads(r.text)['code']} ({json.loads(r.text)['type']}){bcolors.ENDC}") print(f"{bcolors.FAIL}Unsuccessful :( Code: {bcolors.BOLD}{json.loads(r.text)['code']} ({json.loads(r.text)['type']}){bcolors.ENDC}")
print(r.text) # keep and uncomment for debug reasons print(r.text) # keep and uncomment for debug reasons
else: else:
print(f"{bcolors.WARNING}Now the new limit would be set over the API (but DRY_RUN is either not specified or True): {str(new_limit)}%") print(f"{bcolors.FAIL}Failed. Got some weird response from OpenDTU while trying to set the new limit. Maybe the OpenDTU API has changed?{bcolors.ENDC}\nOpenDTUs response: (needs to be in JSON and contain the keys 'code' and 'type')\n{r.text}")
else:
print(f"{bcolors.WARNING}Now the new limit would be set via the API (but DRY_RUN is either not specified or True): {str(new_limit)}%")
elif status[opendtu_inverter_sn]["max_power"] == 0:
print(f"{bcolors.ERROR}OpenDTU is reporting strange values for the inverter's maximum power output. Skipping it for now.{bcolors.ENDC}")
while (time() - last_time) < limit_update_interval: while (time() - last_time) < limit_update_interval: