Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-04-19 00:45:22 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Price Prediction as JSON simulation output, config fixed electricty fees configurable + MyPy & Ruff

Browse Source
This commit is contained in:
Andreas 2024-12-21 13:26:41 +01:00 committed by Andreas
parent 6aa8838e5b
commit d2a83f6ea4
9 changed files with 78 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/akkudoktoreos/config.py
View File

@ -56,6 +56,7 @@ class EOSConfig(BaseModel):
penalty: int
available_charging_rates_in_percentage: list[float]
feed_in_tariff_eur_per_wh: int
electricty_price_fixed_fee: float
class BaseConfig(BaseModel):

3
src/akkudoktoreos/default.config.json
View File

@ -10,6 +10,7 @@
"available_charging_rates_in_percentage": [
0.0, 0.375, 0.5, 0.625, 0.75, 0.875, 1.0
],
"feed_in_tariff_eur_per_wh": 48
"feed_in_tariff_eur_per_wh": 48,
"electricty_price_fixed_fee": 0.00021
}
}

49
src/akkudoktoreos/optimization/genetic.py
View File

@ -1,13 +1,13 @@
import random
from tabnanny import verbose
from typing import Any, Optional, Tuple
import time
from pathlib import Path
from typing import Any, Optional, Tuple
import numpy as np
from deap import algorithms, base, creator, tools
from pydantic import BaseModel, Field, field_validator, model_validator
from typing_extensions import Self
from pathlib import Path
import sys
from akkudoktoreos.config import AppConfig
from akkudoktoreos.devices.battery import (
EAutoParameters,
@ -15,9 +15,6 @@ from akkudoktoreos.devices.battery import (
PVAkku,
PVAkkuParameters,
)
from akkudoktoreos.prediction.self_consumption_probability import (
self_consumption_probability_interpolator,
)
from akkudoktoreos.devices.generic import HomeAppliance, HomeApplianceParameters
from akkudoktoreos.devices.inverter import Wechselrichter, WechselrichterParameters
from akkudoktoreos.prediction.ems import (
@ -25,6 +22,9 @@ from akkudoktoreos.prediction.ems import (
EnergieManagementSystemParameters,
SimulationResult,
)
from akkudoktoreos.prediction.self_consumption_probability import (
self_consumption_probability_interpolator,
)
from akkudoktoreos.utils.utils import NumpyEncoder
from akkudoktoreos.visualize import visualisiere_ergebnisse
@ -126,7 +126,7 @@ class optimization_problem:
random.seed(fixed_seed)
def decode_charge_discharge(
self, discharge_hours_bin: list[int]
self, discharge_hours_bin: np.ndarray
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Decode the input array into ac_charge, dc_charge, and discharge arrays."""
discharge_hours_bin_np = np.array(discharge_hours_bin)
@ -229,8 +229,7 @@ class optimization_problem:
eautocharge_hours_index: Optional[np.ndarray],
washingstart_int: Optional[int],
) -> list[int]:
"""
Merge the individual components back into a single solution list.
"""Merge the individual components back into a single solution list.
Parameters:
discharge_hours_bin (np.ndarray): Binary discharge hours.
@ -262,8 +261,7 @@ class optimization_problem:
def split_individual(
self, individual: list[int]
) -> Tuple[np.ndarray, Optional[np.ndarray], Optional[int]]:
"""
Split the individual solution into its components.
"""Split the individual solution into its components.
Components:
1. Discharge hours (binary as int NumPy array),
@ -398,7 +396,6 @@ class optimization_problem:
worst_case: bool,
) -> Tuple[float]:
"""Evaluate the fitness of an individual solution based on the simulation results."""
try:
o = self.evaluate_inner(individual, ems, start_hour)
except Exception as e:
@ -414,36 +411,29 @@ class optimization_problem:
if self.optimize_ev:
eauto_soc_per_hour = np.array(o.get("EAuto_SoC_pro_Stunde", [])) # Beispielkey
# Angleichung von hinten
min_length = min(len(eauto_soc_per_hour), len(eautocharge_hours_index))
if eauto_soc_per_hour is None or eautocharge_hours_index is None:
raise ValueError("eauto_soc_per_hour or eautocharge_hours_index is None")
min_length = min(eauto_soc_per_hour.size, eautocharge_hours_index.size)
eauto_soc_per_hour_tail = eauto_soc_per_hour[-min_length:]
eautocharge_hours_index_tail = np.array(eautocharge_hours_index[-min_length:])
eautocharge_hours_index_tail = eautocharge_hours_index[-min_length:]
# Erstelle die Maske für die relevanten Abschnitte
# Mask
invalid_charge_mask = (eauto_soc_per_hour_tail == 100) & (
eautocharge_hours_index_tail > 0
)
# Überprüfen und anpassen der Ladezeiten
if np.any(invalid_charge_mask):
# Ignoriere den ersten ungültigen Eintrag
invalid_indices = np.where(invalid_charge_mask)[0]
if (
len(invalid_indices) > 1
): # Nur anpassen, wenn mehr als ein ungültiger Eintrag vorliegt
if len(invalid_indices) > 1:
eautocharge_hours_index_tail[invalid_indices[1:]] = 0
# Aktualisiere die letzten min_length-Einträge von eautocharge_hours_index
eautocharge_hours_index[-min_length:] = eautocharge_hours_index_tail.tolist()
# Rückschreiben der Anpassungen in `individual`
adjusted_individual = self.merge_individual(
discharge_hours_bin, eautocharge_hours_index, washingstart_int
)
# print("Vor:", individual)
individual[:] = adjusted_individual # Aktualisiere das ursprüngliche individual
# print("Nach:", individual)#
# Berechnung weiterer Metriken
individual.extra_data = ( # type: ignore[attr-defined]
@ -472,7 +462,7 @@ class optimization_problem:
return (gesamtbilanz,)
def optimize(
self, start_solution: Optional[list[float]] = None, ngen: int = 400
self, start_solution: Optional[list[float]] = None, ngen: int = 200
) -> Tuple[Any, dict[str, list[Any]]]:
"""Run the optimization process using a genetic algorithm."""
population = self.toolbox.population(n=300)
@ -588,13 +578,11 @@ class optimization_problem:
elapsed_time = time.time() - start_time
print(f"Time evaluate inner: {elapsed_time:.4f} sec.")
# Perform final evaluation on the best solution
print(start_solution[start_hour:])
print(start_hour)
o = self.evaluate_inner(start_solution, ems, start_hour)
discharge_hours_bin, eautocharge_hours_index, washingstart_int = self.split_individual(
start_solution
)
eautocharge_hours_float = (
[
self._config.eos.available_charging_rates_in_percentage[i]
@ -620,7 +608,6 @@ class optimization_problem:
config=self._config,
extra_data=extra_data,
)
return OptimizeResponse(
**{
"ac_charge": ac_charge,

9
src/akkudoktoreos/prediction/ems.py
View File

@ -78,6 +78,9 @@ class SimulationResult(BaseModel):
akku_soc_pro_stunde: list[Optional[float]] = Field(
description="The state of charge of the battery (not the EV) in percentage per hour."
)
Electricity_price: list[Optional[float]] = Field(
description="Used Electricity Price, including predictions"
)
@field_validator(
"Last_Wh_pro_Stunde",
@ -89,6 +92,7 @@ class SimulationResult(BaseModel):
"EAuto_SoC_pro_Stunde",
"Verluste_Pro_Stunde",
"Home_appliance_wh_per_hour",
"Electricity_price",
mode="before",
)
def convert_numpy(cls, field: Any) -> Any:
@ -171,6 +175,7 @@ class EnergieManagementSystem:
eauto_soc_pro_stunde = np.full((total_hours), np.nan)
verluste_wh_pro_stunde = np.full((total_hours), np.nan)
home_appliance_wh_per_hour = np.full((total_hours), np.nan)
electricity_price_per_hour = np.full((total_hours), np.nan)
# Set initial state
akku_soc_pro_stunde[0] = self.akku.ladezustand_in_prozent()
@ -222,7 +227,7 @@ class EnergieManagementSystem:
netzbezug_wh_pro_stunde[stunde_since_now] = netzbezug
verluste_wh_pro_stunde[stunde_since_now] += verluste
last_wh_pro_stunde[stunde_since_now] = verbrauch
electricity_price_per_hour[stunde_since_now] = self.strompreis_euro_pro_wh[stunde]
# Financial calculations
kosten_euro_pro_stunde[stunde_since_now] = (
netzbezug * self.strompreis_euro_pro_wh[stunde]
@ -252,6 +257,6 @@ class EnergieManagementSystem:
"Verluste_Pro_Stunde": verluste_wh_pro_stunde,
"Gesamt_Verluste": np.nansum(verluste_wh_pro_stunde),
"Home_appliance_wh_per_hour": home_appliance_wh_per_hour,
"Electricity_price": electricity_price_per_hour,
}
return out

31
src/akkudoktoreos/prediction/price_forecast.py
View File

@ -3,7 +3,8 @@ import json
import zoneinfo
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Sequence, Optional
from typing import Any, Optional, Sequence
import numpy as np
import requests
@ -28,7 +29,7 @@ class HourlyElectricityPriceForecast:
self,
source: str | Path,
config: AppConfig,
charges: float = 0.000228,
charges: float = 0.00021,
use_cache: bool = True,
): # 228
self.cache_dir = config.working_dir / config.directories.cache
@ -36,7 +37,7 @@ class HourlyElectricityPriceForecast:
if not self.cache_dir.is_dir():
raise SetupIncomplete(f"Output path does not exist: {self.cache_dir}.")
self.seven_day_mean = None
self.seven_day_mean = np.array([])
self.cache_time_file = self.cache_dir / "cache_timestamp.txt"
self.prices = self.load_data(source)
self.charges = charges
@ -97,7 +98,7 @@ class HourlyElectricityPriceForecast:
# Extract the price from 00:00 of the previous day
previous_day_prices = [
entry["marketpriceEurocentPerKWh"] + self.charges
entry["marketpriceEurocentPerKWh"] # + self.charges
for entry in self.prices
if previous_day_str in entry["end"]
]
@ -105,21 +106,22 @@ class HourlyElectricityPriceForecast:
# Extract all prices for the specified date
date_prices = [
entry["marketpriceEurocentPerKWh"] + self.charges
entry["marketpriceEurocentPerKWh"] # + self.charges
for entry in self.prices
if date_str in entry["end"]
]
print(f"getPrice: {len(date_prices)}")
# print(f"getPrice: {len(date_prices)}")
# Add the last price of the previous day at the start of the list
if len(date_prices) == 23:
date_prices.insert(0, last_price_of_previous_day)
# print(np.array(date_prices) / (1000.0 * 100.0))
# print("PRICE:")
# print(np.array(date_prices) / (1000.0 * 100.0) + self.charges)
return np.array(date_prices) / (1000.0 * 100.0) + self.charges
def get_average_price_last_7_days(self, end_date_str: Optional[str] = None) -> np.ndarray:
"""
Calculate the hourly average electricity price for the last 7 days.
"""Calculate the hourly average electricity price for the last 7 days.
Parameters:
end_date_str (Optional[str]): End date in the format "YYYY-MM-DD".
@ -138,8 +140,8 @@ class HourlyElectricityPriceForecast:
else:
end_date = datetime.strptime(end_date_str, "%Y-%m-%d").date()
if self.seven_day_mean != None:
return self.seven_day_mean
if self.seven_day_mean.size > 0:
return np.array([self.seven_day_mean])
# Calculate the start date (7 days before the end date)
start_date = end_date - timedelta(days=7)
@ -176,7 +178,6 @@ class HourlyElectricityPriceForecast:
self, start_date_str: str, end_date_str: str, repeat: bool = False
) -> np.ndarray:
"""Returns all prices between the start and end dates."""
start_date_utc = datetime.strptime(start_date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
end_date_utc = datetime.strptime(end_date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
start_date = start_date_utc.astimezone(zoneinfo.ZoneInfo("Europe/Berlin"))
@ -194,9 +195,9 @@ class HourlyElectricityPriceForecast:
# print(date_str, ":", daily_prices)
price_list_np = np.array(price_list)
print(price_list_np)
# print(price_list_np.shape, " ", self.prediction_hours)
# If prediction hours are greater than 0 and repeat is True
# print(price_list_np)
if self.prediction_hours > 0 and repeat:
# Check if price_list_np is shorter than prediction_hours
if price_list_np.size < self.prediction_hours:
@ -208,5 +209,5 @@ class HourlyElectricityPriceForecast:
# Concatenate existing values with the repeated values
price_list_np = np.concatenate((price_list_np, additional_values))
# print(price_list_np)
return price_list_np

8
src/akkudoktoreos/prediction/self_consumption_probability.py
View File

@ -1,22 +1,24 @@
#!/usr/bin/env python
import numpy as np
import pickle
from functools import lru_cache
# from scipy.interpolate import RegularGridInterpolator
from pathlib import Path
from typing import Tuple
import numpy as np
class self_consumption_probability_interpolator:
def __init__(self, filepath: str | Path):
self.filepath = filepath
self.interpolator = None
# self.interpolator = None
# Load the RegularGridInterpolator
with open(self.filepath, "rb") as file:
self.interpolator = pickle.load(file)
@lru_cache(maxsize=128)
def generate_points(self, load_1h_power: float, pv_power: float):
def generate_points(self, load_1h_power: float, pv_power: float) -> Tuple:
"""Generate the grid points for interpolation."""
partial_loads = np.arange(0, pv_power + 50, 50)
points = np.array([np.full_like(partial_loads, load_1h_power), partial_loads]).T

1
src/akkudoktoreos/server/fastapi_server.py
View File

@ -67,6 +67,7 @@ def fastapi_strompreis() -> list[float]:
source=f"https://api.akkudoktor.net/prices?start={date_start}&end={date_end}",
config=config,
use_cache=False,
charges=config.eos.electricty_price_fixed_fee,
)
# seven Day mean
specific_date_prices = price_forecast.get_price_for_daterange(

15
tests/test_class_ems.py
View File

@ -1,3 +1,5 @@
from pathlib import Path
import numpy as np
import pytest
@ -10,6 +12,9 @@ from akkudoktoreos.prediction.ems import (
EnergieManagementSystemParameters,
SimulationResult,
)
from akkudoktoreos.prediction.self_consumption_probability import (
self_consumption_probability_interpolator,
)
prediction_hours = 48
optimization_hours = 24
@ -25,8 +30,16 @@ def create_ems_instance(tmp_config: AppConfig) -> EnergieManagementSystem:
PVAkkuParameters(kapazitaet_wh=5000, start_soc_prozent=80, min_soc_prozent=10),
hours=prediction_hours,
)
# 1h Load to Sub 1h Load Distribution -> SelfConsumptionRate
sc = self_consumption_probability_interpolator(
Path(__file__).parent.resolve() / ".." / "data" / "regular_grid_interpolator.pkl"
)
akku.reset()
wechselrichter = Wechselrichter(WechselrichterParameters(max_leistung_wh=10000), akku)
wechselrichter = Wechselrichter(
WechselrichterParameters(max_leistung_wh=10000), akku, self_consumption_predictor=sc
)
# Household device (currently not used, set to None)
home_appliance = HomeAppliance(

15
tests/test_class_ems_2.py
View File

@ -1,3 +1,5 @@
from pathlib import Path
import numpy as np
import pytest
@ -9,6 +11,9 @@ from akkudoktoreos.prediction.ems import (
EnergieManagementSystem,
EnergieManagementSystemParameters,
)
from akkudoktoreos.prediction.self_consumption_probability import (
self_consumption_probability_interpolator,
)
prediction_hours = 48
optimization_hours = 24
@ -24,8 +29,16 @@ def create_ems_instance(tmp_config: AppConfig) -> EnergieManagementSystem:
PVAkkuParameters(kapazitaet_wh=5000, start_soc_prozent=80, min_soc_prozent=10),
hours=prediction_hours,
)
# 1h Load to Sub 1h Load Distribution -> SelfConsumptionRate
sc = self_consumption_probability_interpolator(
Path(__file__).parent.resolve() / ".." / "data" / "regular_grid_interpolator.pkl"
)
akku.reset()
wechselrichter = Wechselrichter(WechselrichterParameters(max_leistung_wh=10000), akku)
wechselrichter = Wechselrichter(
WechselrichterParameters(max_leistung_wh=10000), akku, self_consumption_predictor=sc
)
# Household device (currently not used, set to None)
home_appliance = HomeAppliance(