Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-08-25 06:52:23 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Migrate from Flask to FastAPI (#163)

Browse Source 
* Migrate from Flask to FastAPI

 * FastAPI migration:
    - Use pydantic model classes as input parameters to the
      data/calculation classes.
    - Interface field names changed to constructor parameter names (for
      simplicity only during transition, should be updated in a followup
      PR).
    - Add basic interface requirements (e.g. some values > 0, etc.).
 * Update tests for new data format.
 * Python requirement down to 3.9 (TypeGuard no longer needed)
 * Makefile: Add helpful targets (e.g. development server with reload)

* Move API doc from README to pydantic model classes (swagger)

 * Link to swagger.io with own openapi.yml.
 * Commit openapi.json and check with pytest for changes so the
   documentation is always up-to-date.

* Streamline docker

* FastAPI: Run startup action on dev server

 * Fix config for /strompreis, endpoint still broken however.

* test_openapi: Compare against docs/.../openapi.json

* Move fastapi to server/ submodule

 * See #187 for new repository structure.
This commit is contained in:
Dominique Lasserre
2024-11-15 22:27:25 +01:00
committed by GitHub
parent ed3226e522
commit f61665669f
38 changed files with 997 additions and 1331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
src/akkudoktoreos/class_akku.py
View File

@@ -1,32 +1,74 @@
from typing import Optional
import numpy as np
from pydantic import BaseModel, Field
def max_ladeleistung_w_field(default=None):
return Field(
default,
gt=0,
description="An integer representing the charging power of the battery in watts.",
)
def start_soc_prozent_field(description: str):
return Field(0, ge=0, le=100, description=description)
class BaseAkkuParameters(BaseModel):
kapazitaet_wh: int = Field(
gt=0, description="An integer representing the capacity of the battery in watt-hours."
)
lade_effizienz: float = Field(
0.88, gt=0, le=1, description="A float representing the charging efficiency of the battery."
)
entlade_effizienz: float = Field(0.88, gt=0, le=1)
max_ladeleistung_w: Optional[float] = max_ladeleistung_w_field()
start_soc_prozent: int = start_soc_prozent_field(
"An integer representing the state of charge of the battery at the **start** of the current hour (not the current state)."
)
min_soc_prozent: int = Field(
0,
ge=0,
le=100,
description="An integer representing the minimum state of charge (SOC) of the battery in percentage.",
)
max_soc_prozent: int = Field(100, ge=0, le=100)
class PVAkkuParameters(BaseAkkuParameters):
max_ladeleistung_w: Optional[float] = max_ladeleistung_w_field(5000)
class EAutoParameters(BaseAkkuParameters):
entlade_effizienz: float = 1.0
start_soc_prozent: int = start_soc_prozent_field(
"An integer representing the current state of charge (SOC) of the battery in percentage."
)
class PVAkku:
def __init__(
self,
kapazitaet_wh=None,
hours=None,
lade_effizienz=0.88,
entlade_effizienz=0.88,
max_ladeleistung_w=None,
start_soc_prozent=0,
min_soc_prozent=0,
max_soc_prozent=100,
):
def __init__(self, parameters: BaseAkkuParameters, hours: int = 24):
# Battery capacity in Wh
self.kapazitaet_wh = kapazitaet_wh
self.kapazitaet_wh = parameters.kapazitaet_wh
# Initial state of charge in Wh
self.start_soc_prozent = start_soc_prozent
self.soc_wh = (start_soc_prozent / 100) * kapazitaet_wh
self.hours = hours if hours is not None else 24 # Default to 24 hours if not specified
self.start_soc_prozent = parameters.start_soc_prozent
self.soc_wh = (parameters.start_soc_prozent / 100) * parameters.kapazitaet_wh
self.hours = hours
self.discharge_array = np.full(self.hours, 1)
self.charge_array = np.full(self.hours, 1)
# Charge and discharge efficiency
self.lade_effizienz = lade_effizienz
self.entlade_effizienz = entlade_effizienz
self.max_ladeleistung_w = max_ladeleistung_w if max_ladeleistung_w else self.kapazitaet_wh
self.min_soc_prozent = min_soc_prozent
self.max_soc_prozent = max_soc_prozent
self.lade_effizienz = parameters.lade_effizienz
self.entlade_effizienz = parameters.entlade_effizienz
self.max_ladeleistung_w = (
parameters.max_ladeleistung_w if parameters.max_ladeleistung_w else self.kapazitaet_wh
)
# Only assign for storage battery
self.min_soc_prozent = (
parameters.min_soc_prozent if isinstance(parameters, PVAkkuParameters) else 0
)
self.max_soc_prozent = parameters.max_soc_prozent
# Calculate min and max SoC in Wh
self.min_soc_wh = (self.min_soc_prozent / 100) * self.kapazitaet_wh
self.max_soc_wh = (self.max_soc_prozent / 100) * self.kapazitaet_wh

62
src/akkudoktoreos/class_ems.py
View File

@@ -2,27 +2,63 @@ from datetime import datetime
from typing import Dict, List, Optional, Union
import numpy as np
from pydantic import BaseModel, Field, model_validator
from typing_extensions import Self
from akkudoktoreos.class_akku import PVAkku
from akkudoktoreos.class_haushaltsgeraet import Haushaltsgeraet
from akkudoktoreos.class_inverter import Wechselrichter
from akkudoktoreos.config import EOSConfig
class EnergieManagementSystemParameters(BaseModel):
pv_prognose_wh: list[float] = Field(
description="An array of floats representing the forecasted photovoltaic output in watts for different time intervals."
)
strompreis_euro_pro_wh: list[float] = Field(
description="An array of floats representing the electricity price in euros per watt-hour for different time intervals."
)
einspeiseverguetung_euro_pro_wh: list[float] | float = Field(
description="A float or array of floats representing the feed-in compensation in euros per watt-hour."
)
preis_euro_pro_wh_akku: float
gesamtlast: list[float] = Field(
description="An array of floats representing the total load (consumption) in watts for different time intervals."
)
@model_validator(mode="after")
def validate_list_length(self) -> Self:
pv_prognose_length = len(self.pv_prognose_wh)
if (
pv_prognose_length != len(self.strompreis_euro_pro_wh)
or pv_prognose_length != len(self.gesamtlast)
or (
isinstance(self.einspeiseverguetung_euro_pro_wh, list)
and pv_prognose_length != len(self.einspeiseverguetung_euro_pro_wh)
)
):
raise ValueError("Input lists have different lengths")
return self
class EnergieManagementSystem:
def __init__(
self,
config: EOSConfig,
pv_prognose_wh: Optional[np.ndarray] = None,
strompreis_euro_pro_wh: Optional[np.ndarray] = None,
einspeiseverguetung_euro_pro_wh: Optional[np.ndarray] = None,
eauto: Optional[object] = None,
gesamtlast: Optional[np.ndarray] = None,
haushaltsgeraet: Optional[object] = None,
wechselrichter: Optional[object] = None,
parameters: EnergieManagementSystemParameters,
eauto: Optional[PVAkku] = None,
haushaltsgeraet: Optional[Haushaltsgeraet] = None,
wechselrichter: Optional[Wechselrichter] = None,
):
self.akku = wechselrichter.akku
self.gesamtlast = gesamtlast
self.pv_prognose_wh = pv_prognose_wh
self.strompreis_euro_pro_wh = strompreis_euro_pro_wh
self.einspeiseverguetung_euro_pro_wh = einspeiseverguetung_euro_pro_wh
self.gesamtlast = np.array(parameters.gesamtlast, float)
self.pv_prognose_wh = np.array(parameters.pv_prognose_wh, float)
self.strompreis_euro_pro_wh = np.array(parameters.strompreis_euro_pro_wh, float)
self.einspeiseverguetung_euro_pro_wh_arr = (
parameters.einspeiseverguetung_euro_pro_wh
if isinstance(parameters.einspeiseverguetung_euro_pro_wh, list)
else np.full(len(self.gesamtlast), parameters.einspeiseverguetung_euro_pro_wh, float)
)
self.eauto = eauto
self.haushaltsgeraet = haushaltsgeraet
self.wechselrichter = wechselrichter
@@ -134,7 +170,7 @@ class EnergieManagementSystem:
netzbezug * self.strompreis_euro_pro_wh[stunde]
)
einnahmen_euro_pro_stunde[stunde_since_now] = (
netzeinspeisung * self.einspeiseverguetung_euro_pro_wh[stunde]
netzeinspeisung * self.einspeiseverguetung_euro_pro_wh_arr[stunde]
)
# Akku SOC tracking
@@ -152,7 +188,7 @@ class EnergieManagementSystem:
"akku_soc_pro_stunde": akku_soc_pro_stunde,
"Einnahmen_Euro_pro_Stunde": einnahmen_euro_pro_stunde,
"Gesamtbilanz_Euro": gesamtkosten_euro,
"E-Auto_SoC_pro_Stunde": eauto_soc_pro_stunde,
"EAuto_SoC_pro_Stunde": eauto_soc_pro_stunde,
"Gesamteinnahmen_Euro": np.nansum(einnahmen_euro_pro_stunde),
"Gesamtkosten_Euro": np.nansum(kosten_euro_pro_stunde),
"Verluste_Pro_Stunde": verluste_wh_pro_stunde,

20
src/akkudoktoreos/class_haushaltsgeraet.py
View File

@@ -1,11 +1,25 @@
import numpy as np
from pydantic import BaseModel, Field
class HaushaltsgeraetParameters(BaseModel):
verbrauch_wh: int = Field(
gt=0,
description="An integer representing the energy consumption of a household device in watt-hours.",
)
dauer_h: int = Field(
gt=0,
description="An integer representing the usage duration of a household device in hours.",
)
class Haushaltsgeraet:
def __init__(self, hours=None, verbrauch_wh=None, dauer_h=None):
def __init__(self, parameters: HaushaltsgeraetParameters, hours=24):
self.hours = hours # Total duration for which the planning is done
self.verbrauch_wh = verbrauch_wh # Total energy consumption of the device in kWh
self.dauer_h = dauer_h # Duration of use in hours
self.verbrauch_wh = (
parameters.verbrauch_wh # Total energy consumption of the device in kWh
)
self.dauer_h = parameters.dauer_h # Duration of use in hours
self.lastkurve = np.zeros(self.hours) # Initialize the load curve with zeros
def set_startzeitpunkt(self, start_hour, global_start_hour=0):

15
src/akkudoktoreos/class_inverter.py
View File

@@ -1,6 +1,17 @@
from pydantic import BaseModel, Field
from akkudoktoreos.class_akku import PVAkku
class WechselrichterParameters(BaseModel):
max_leistung_wh: float = Field(10000, gt=0)
class Wechselrichter:
def __init__(self, max_leistung_wh, akku):
self.max_leistung_wh = max_leistung_wh # Maximum power that the inverter can handle
def __init__(self, parameters: WechselrichterParameters, akku: PVAkku):
self.max_leistung_wh = (
parameters.max_leistung_wh # Maximum power that the inverter can handle
)
self.akku = akku # Connection to a battery object
def energie_verarbeiten(self, erzeugung, verbrauch, hour):

248
src/akkudoktoreos/class_optimize.py
View File

@@ -1,17 +1,165 @@
import random
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Optional, Tuple
import numpy as np
from deap import algorithms, base, creator, tools
from pydantic import BaseModel, Field, model_validator
from typing_extensions import Self
from akkudoktoreos.class_akku import PVAkku
from akkudoktoreos.class_ems import EnergieManagementSystem
from akkudoktoreos.class_haushaltsgeraet import Haushaltsgeraet
from akkudoktoreos.class_inverter import Wechselrichter
from akkudoktoreos.class_akku import EAutoParameters, PVAkku, PVAkkuParameters
from akkudoktoreos.class_ems import (
EnergieManagementSystem,
EnergieManagementSystemParameters,
)
from akkudoktoreos.class_haushaltsgeraet import (
Haushaltsgeraet,
HaushaltsgeraetParameters,
)
from akkudoktoreos.class_inverter import Wechselrichter, WechselrichterParameters
from akkudoktoreos.config import AppConfig
from akkudoktoreos.visualize import visualisiere_ergebnisse
class OptimizationParameters(BaseModel):
ems: EnergieManagementSystemParameters
pv_akku: PVAkkuParameters
wechselrichter: WechselrichterParameters = WechselrichterParameters()
eauto: EAutoParameters
spuelmaschine: Optional[HaushaltsgeraetParameters] = None
temperature_forecast: list[float] = Field(
"An array of floats representing the temperature forecast in degrees Celsius for different time intervals."
)
start_solution: Optional[list[float]] = Field(
None, description="Can be `null` or contain a previous solution (if available)."
)
@model_validator(mode="after")
def validate_list_length(self) -> Self:
arr_length = len(self.ems.pv_prognose_wh)
if arr_length != len(self.temperature_forecast):
raise ValueError("Input lists have different lenghts")
return self
class EAutoResult(BaseModel):
"""This object contains information related to the electric vehicle and its charging and discharging behavior."""
charge_array: list[float] = Field(
description="Indicates for each hour whether the EV is charging (`0` for no charging, `1` for charging)."
)
discharge_array: list[int] = Field(
description="Indicates for each hour whether the EV is discharging (`0` for no discharging, `1` for discharging)."
)
entlade_effizienz: float = Field(description="The discharge efficiency as a float.")
hours: int = Field("Amount of hours the simulation is done for.")
kapazitaet_wh: int = Field("The capacity of the EV’s battery in watt-hours.")
lade_effizienz: float = Field("The charging efficiency as a float.")
max_ladeleistung_w: int = Field(description="The maximum charging power of the EV in watts.")
soc_wh: float = Field(
description="The state of charge of the battery in watt-hours at the start of the simulation."
)
start_soc_prozent: int = Field(
description="The state of charge of the battery in percentage at the start of the simulation."
)
class SimulationResult(BaseModel):
"""This object contains the results of the simulation and provides insights into various parameters over the entire forecast period."""
Last_Wh_pro_Stunde: list[Optional[float]] = Field(description="TBD")
EAuto_SoC_pro_Stunde: list[Optional[float]] = Field(
description="The state of charge of the EV for each hour."
)
Einnahmen_Euro_pro_Stunde: list[Optional[float]] = Field(
description="The revenue from grid feed-in or other sources in euros per hour."
)
Gesamt_Verluste: float = Field(
description="The total losses in watt-hours over the entire period."
)
Gesamtbilanz_Euro: float = Field(
description="The total balance of revenues minus costs in euros."
)
Gesamteinnahmen_Euro: float = Field(description="The total revenues in euros.")
Gesamtkosten_Euro: float = Field(description="The total costs in euros.")
Haushaltsgeraet_wh_pro_stunde: list[Optional[float]] = Field(
description="The energy consumption of a household appliance in watt-hours per hour."
)
Kosten_Euro_pro_Stunde: list[Optional[float]] = Field(
description="The costs in euros per hour."
)
Netzbezug_Wh_pro_Stunde: list[Optional[float]] = Field(
description="The grid energy drawn in watt-hours per hour."
)
Netzeinspeisung_Wh_pro_Stunde: list[Optional[float]] = Field(
description="The energy fed into the grid in watt-hours per hour."
)
Verluste_Pro_Stunde: list[Optional[float]] = Field(
description="The losses in watt-hours per hour."
)
akku_soc_pro_stunde: list[Optional[float]] = Field(
description="The state of charge of the battery (not the EV) in percentage per hour."
)
# class SimulationData(BaseModel):
# """An object containing the simulated data."""
#
# Last_Wh_pro_Stunde: list[Optional[float]] = Field(description="TBD")
# EAuto_SoC_pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated state of charge of the electric car per hour.",
# )
# Einnahmen_Euro_pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated income in euros per hour."
# )
# Gesamt_Verluste: float = Field(description="The total simulated losses in watt-hours.")
# Gesamtbilanz_Euro: float = Field(description="The total simulated balance in euros.")
# Gesamteinnahmen_Euro: float = Field(description="The total simulated income in euros.")
# Gesamtkosten_Euro: float = Field(description="The total simulated costs in euros.")
# Haushaltsgeraet_wh_pro_stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated energy consumption of a household appliance in watt-hours per hour."
# )
# Kosten_Euro_pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated costs in euros per hour."
# )
# Netzbezug_Wh_pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated grid consumption in watt-hours per hour."
# )
# Netzeinspeisung_Wh_pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated grid feed-in in watt-hours per hour."
# )
# Verluste_Pro_Stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated losses per hour."
# )
# akku_soc_pro_stunde: list[Optional[float]] = Field(
# description="An array of floats representing the simulated state of charge of the battery in percentage per hour."
# )
class OptimizeResponse(BaseModel):
"""**Note**: The first value of "Last_Wh_pro_Stunde", "Netzeinspeisung_Wh_pro_Stunde" and "Netzbezug_Wh_pro_Stunde", will be set to null in the JSON output and represented as NaN or None in the corresponding classes' data returns. This approach is adopted to ensure that the current hour's processing remains unchanged."""
ac_charge: list[float] = Field(
description="Array with AC charging values as relative power (0-1), other values set to 0."
)
dc_charge: list[float] = Field(
description="Array with DC charging values as relative power (0-1), other values set to 0."
)
discharge_allowed: list[int] = Field(
description="Array with discharge values (1 for discharge, 0 otherwise)."
)
result: SimulationResult
eauto_obj: EAutoResult
start_solution: Optional[list[float]] = Field(
None,
description="An array of binary values (0 or 1) representing a possible starting solution for the simulation.",
)
spuelstart: Optional[int] = Field(
None,
description="Can be `null` or contain an object representing the start of washing (if applicable).",
)
# simulation_data: Optional[SimulationData] = None
class optimization_problem:
def __init__(
self,
@@ -164,8 +312,8 @@ class optimization_problem:
return creator.Individual(individual_components)
def split_individual(
self, individual: List[float]
) -> Tuple[List[int], List[float], Optional[int]]:
self, individual: list[float]
) -> Tuple[list[int], list[float], Optional[int]]:
"""Split the individual solution into its components.
Components:
@@ -187,7 +335,7 @@ class optimization_problem:
)
return discharge_hours_bin, eautocharge_hours_float, spuelstart_int
def setup_deap_environment(self, opti_param: Dict[str, Any], start_hour: int) -> None:
def setup_deap_environment(self, opti_param: dict[str, Any], start_hour: int) -> None:
"""Set up the DEAP environment with fitness and individual creation rules."""
self.opti_param = opti_param
@@ -250,8 +398,8 @@ class optimization_problem:
self.toolbox.register("select", tools.selTournament, tournsize=3)
def evaluate_inner(
self, individual: List[float], ems: EnergieManagementSystem, start_hour: int
) -> Dict[str, Any]:
self, individual: list[float], ems: EnergieManagementSystem, start_hour: int
) -> dict[str, Any]:
"""Simulates the energy management system (EMS) using the provided individual solution.
This is an internal function.
@@ -283,9 +431,9 @@ class optimization_problem:
def evaluate(
self,
individual: List[float],
individual: list[float],
ems: EnergieManagementSystem,
parameter: Dict[str, Any],
parameters: OptimizationParameters,
start_hour: int,
worst_case: bool,
) -> Tuple[float]:
@@ -305,7 +453,7 @@ class optimization_problem:
)
# Penalty for not meeting the minimum SOC (State of Charge) requirement
# if parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent() <= 0.0 and self.optimize_ev:
# if parameters.eauto_min_soc_prozent - ems.eauto.ladezustand_in_prozent() <= 0.0 and self.optimize_ev:
# gesamtbilanz += sum(
# self.strafe for ladeleistung in eautocharge_hours_float if ladeleistung != 0.0
# )
@@ -313,26 +461,27 @@ class optimization_problem:
individual.extra_data = (
o["Gesamtbilanz_Euro"],
o["Gesamt_Verluste"],
parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent(),
parameters.eauto.min_soc_prozent - ems.eauto.ladezustand_in_prozent(),
)
# Adjust total balance with battery value and penalties for unmet SOC
restwert_akku = ems.akku.aktueller_energieinhalt() * parameter["preis_euro_pro_wh_akku"]
# print(ems.akku.aktueller_energieinhalt()," * ", parameter["preis_euro_pro_wh_akku"] , " ", restwert_akku, " ", gesamtbilanz)
restwert_akku = ems.akku.aktueller_energieinhalt() * parameters.ems.preis_euro_pro_wh_akku
# print(ems.akku.aktueller_energieinhalt()," * ", parameters.ems.preis_euro_pro_wh_akku , " ", restwert_akku, " ", gesamtbilanz)
gesamtbilanz += -restwert_akku
# print(gesamtbilanz)
if self.optimize_ev:
gesamtbilanz += max(
0,
(parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()) * self.strafe,
(parameters.eauto.min_soc_prozent - ems.eauto.ladezustand_in_prozent())
* self.strafe,
)
return (gesamtbilanz,)
def optimize(
self, start_solution: Optional[List[float]] = None, ngen: int = 400
) -> Tuple[Any, Dict[str, List[Any]]]:
self, start_solution: Optional[list[float]] = None, ngen: int = 400
) -> Tuple[Any, dict[str, list[Any]]]:
"""Run the optimization process using a genetic algorithm."""
population = self.toolbox.population(n=300)
hof = tools.HallOfFame(1)
@@ -373,61 +522,50 @@ class optimization_problem:
def optimierung_ems(
self,
parameter: Optional[Dict[str, Any]] = None,
parameters: OptimizationParameters,
start_hour: Optional[int] = None,
worst_case: bool = False,
startdate: Optional[Any] = None, # startdate is not used!
*,
ngen: int = 600,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Perform EMS (Energy Management System) optimization and visualize results."""
einspeiseverguetung_euro_pro_wh = np.full(
self.prediction_hours, parameter["einspeiseverguetung_euro_pro_wh"]
self.prediction_hours, parameters.ems.einspeiseverguetung_euro_pro_wh
)
# Initialize PV and EV batteries
akku = PVAkku(
kapazitaet_wh=parameter["pv_akku_cap"],
parameters.pv_akku,
hours=self.prediction_hours,
start_soc_prozent=parameter["pv_soc"],
min_soc_prozent=parameter["min_soc_prozent"],
max_ladeleistung_w=5000,
)
akku.set_charge_per_hour(np.full(self.prediction_hours, 1))
self.optimize_ev = True
if parameter["eauto_min_soc"] - parameter["eauto_soc"] < 0:
if parameters.eauto.min_soc_prozent - parameters.eauto.start_soc_prozent < 0:
self.optimize_ev = False
eauto = PVAkku(
kapazitaet_wh=parameter["eauto_cap"],
parameters.eauto,
hours=self.prediction_hours,
lade_effizienz=parameter["eauto_charge_efficiency"],
entlade_effizienz=1.0,
max_ladeleistung_w=parameter["eauto_charge_power"],
start_soc_prozent=parameter["eauto_soc"],
)
eauto.set_charge_per_hour(np.full(self.prediction_hours, 1))
# Initialize household appliance if applicable
spuelmaschine = (
Haushaltsgeraet(
parameters=parameters.spuelmaschine,
hours=self.prediction_hours,
verbrauch_wh=parameter["haushaltsgeraet_wh"],
dauer_h=parameter["haushaltsgeraet_dauer"],
)
if parameter["haushaltsgeraet_dauer"] > 0
if parameters.spuelmaschine is not None
else None
)
# Initialize the inverter and energy management system
wr = Wechselrichter(10000, akku)
wr = Wechselrichter(parameters.wechselrichter, akku)
ems = EnergieManagementSystem(
config=self._config.eos,
gesamtlast=parameter["gesamtlast"],
pv_prognose_wh=parameter["pv_forecast"],
strompreis_euro_pro_wh=parameter["strompreis_euro_pro_wh"],
einspeiseverguetung_euro_pro_wh=einspeiseverguetung_euro_pro_wh,
self._config.eos,
parameters.ems,
eauto=eauto,
haushaltsgeraet=spuelmaschine,
wechselrichter=wr,
@@ -437,9 +575,9 @@ class optimization_problem:
self.setup_deap_environment({"haushaltsgeraete": 1 if spuelmaschine else 0}, start_hour)
self.toolbox.register(
"evaluate",
lambda ind: self.evaluate(ind, ems, parameter, start_hour, worst_case),
lambda ind: self.evaluate(ind, ems, parameters, start_hour, worst_case),
)
start_solution, extra_data = self.optimize(parameter["start_solution"], ngen=ngen) #
start_solution, extra_data = self.optimize(parameters.start_solution, ngen=ngen)
# Perform final evaluation on the best solution
o = self.evaluate_inner(start_solution, ems, start_hour)
@@ -455,16 +593,16 @@ class optimization_problem:
ac_charge, dc_charge, discharge = self.decode_charge_discharge(discharge_hours_bin)
# Visualize the results
visualisiere_ergebnisse(
gesamtlast=parameter["gesamtlast"],
pv_forecast=parameter["pv_forecast"],
strompreise=parameter["strompreis_euro_pro_wh"],
ergebnisse=o,
ac=ac_charge,
dc=dc_charge,
discharge=discharge,
temperature=parameter["temperature_forecast"],
start_hour=start_hour,
einspeiseverguetung_euro_pro_wh=einspeiseverguetung_euro_pro_wh,
parameters.ems.gesamtlast,
parameters.ems.pv_prognose_wh,
parameters.ems.strompreis_euro_pro_wh,
o,
ac_charge,
dc_charge,
discharge,
parameters.temperature_forecast,
start_hour,
einspeiseverguetung_euro_pro_wh,
config=self._config,
extra_data=extra_data,
)
@@ -477,7 +615,7 @@ class optimization_problem:
"Netzbezug_Wh_pro_Stunde",
"Kosten_Euro_pro_Stunde",
"Einnahmen_Euro_pro_Stunde",
"E-Auto_SoC_pro_Stunde",
"EAuto_SoC_pro_Stunde",
"Verluste_Pro_Stunde",
"Haushaltsgeraet_wh_pro_stunde",
]
@@ -507,5 +645,5 @@ class optimization_problem:
"eauto_obj": ems.eauto.to_dict(),
"start_solution": start_solution,
"spuelstart": spuelstart_int,
"simulation_data": o,
# "simulation_data": o,
}

5
src/akkudoktoreos/class_pv_forecast.py
View File

@@ -108,6 +108,11 @@ def validate_pv_forecast_data(data) -> str:
return data_type
class ForecastResponse(BaseModel):
temperature: list[float]
pvpower: list[float]
class ForecastData:
"""Stores forecast data for PV power and weather parameters.

0
src/akkudoktoreosserver/data/load_profiles.npz → src/akkudoktoreos/data/load_profiles.npz
View File

0
src/akkudoktoreos/server/__init__.py Normal file
View File

252
src/akkudoktoreos/server/fastapi_server.py Executable file
View File

@@ -0,0 +1,252 @@
#!/usr/bin/env python3
import os
from datetime import datetime
from pathlib import Path
from typing import Annotated, Any, Optional
import matplotlib
import uvicorn
from fastapi.exceptions import HTTPException
# Sets the Matplotlib backend to 'Agg' for rendering plots in environments without a display
matplotlib.use("Agg")
import pandas as pd
from fastapi import FastAPI, Query
from fastapi.responses import FileResponse, RedirectResponse
from akkudoktoreos.class_load import LoadForecast
from akkudoktoreos.class_load_container import Gesamtlast
from akkudoktoreos.class_load_corrector import LoadPredictionAdjuster
from akkudoktoreos.class_optimize import (
OptimizationParameters,
OptimizeResponse,
optimization_problem,
)
from akkudoktoreos.class_pv_forecast import ForecastResponse, PVForecast
from akkudoktoreos.class_strompreis import HourlyElectricityPriceForecast
from akkudoktoreos.config import (
SetupIncomplete,
get_start_enddate,
get_working_dir,
load_config,
)
app = FastAPI(
title="Akkudoktor-EOS",
description="This project provides a comprehensive solution for simulating and optimizing an energy system based on renewable energy sources. With a focus on photovoltaic (PV) systems, battery storage (batteries), load management (consumer requirements), heat pumps, electric vehicles, and consideration of electricity price data, this system enables forecasting and optimization of energy flow and costs over a specified period.",
summary="Comprehensive solution for simulating and optimizing an energy system based on renewable energy sources",
version="0.0.1",
license_info={
"name": "Apache 2.0",
"url": "https://www.apache.org/licenses/LICENSE-2.0.html",
},
)
working_dir = get_working_dir()
# copy config to working directory. Make this a CLI option later
config = load_config(working_dir, True)
opt_class = optimization_problem(config)
server_dir = Path(__file__).parent.resolve()
class PdfResponse(FileResponse):
media_type = "application/pdf"
@app.get("/strompreis")
def fastapi_strompreis() -> list[float]:
# Get the current date and the end date based on prediction hours
date_now, date = get_start_enddate(config.eos.prediction_hours, startdate=datetime.now().date())
price_forecast = HourlyElectricityPriceForecast(
source=f"https://api.akkudoktor.net/prices?start={date_now}&end={date}",
config=config,
use_cache=False,
)
specific_date_prices = price_forecast.get_price_for_daterange(
date_now, date
) # Fetch prices for the specified date range
return specific_date_prices.tolist()
@app.post("/gesamtlast")
def fastapi_gesamtlast(
year_energy: float,
measured_data: list[dict[str, Any]],
hours: int = config.eos.prediction_hours,
) -> list[float]:
"""Endpoint to handle total load calculation based on the latest measured data."""
# Measured data in JSON format
measured_data_df = pd.DataFrame(measured_data)
measured_data_df["time"] = pd.to_datetime(measured_data_df["time"])
# Ensure datetime has timezone info for accurate calculations
if measured_data_df["time"].dt.tz is None:
measured_data_df["time"] = measured_data_df["time"].dt.tz_localize("Europe/Berlin")
else:
measured_data_df["time"] = measured_data_df["time"].dt.tz_convert("Europe/Berlin")
# Remove timezone info after conversion to simplify further processing
measured_data_df["time"] = measured_data_df["time"].dt.tz_localize(None)
# Instantiate LoadForecast and generate forecast data
lf = LoadForecast(
filepath=server_dir / ".." / "data" / "load_profiles.npz", year_energy=year_energy
)
forecast_list = []
# Generate daily forecasts for the date range based on measured data
for single_date in pd.date_range(
measured_data_df["time"].min().date(), measured_data_df["time"].max().date()
):
date_str = single_date.strftime("%Y-%m-%d")
daily_forecast = lf.get_daily_stats(date_str)
mean_values = daily_forecast[0]
fc_hours = [single_date + pd.Timedelta(hours=i) for i in range(24)]
daily_forecast_df = pd.DataFrame({"time": fc_hours, "Last Pred": mean_values})
forecast_list.append(daily_forecast_df)
# Concatenate all daily forecasts into a single DataFrame
predicted_data = pd.concat(forecast_list, ignore_index=True)
# Create LoadPredictionAdjuster instance to adjust the predictions based on measured data
adjuster = LoadPredictionAdjuster(measured_data_df, predicted_data, lf)
adjuster.calculate_weighted_mean() # Calculate weighted mean for adjustment
adjuster.adjust_predictions() # Adjust predictions based on measured data
future_predictions = adjuster.predict_next_hours(hours) # Predict future load
# Extract household power predictions
leistung_haushalt = future_predictions["Adjusted Pred"].values
gesamtlast = Gesamtlast(prediction_hours=hours)
gesamtlast.hinzufuegen(
"Haushalt", leistung_haushalt
) # Add household load to total load calculation
# Calculate the total load
last = gesamtlast.gesamtlast_berechnen() # Compute total load
return last.tolist()
@app.get("/gesamtlast_simple")
def fastapi_gesamtlast_simple(year_energy: float) -> list[float]:
date_now, date = get_start_enddate(
config.eos.prediction_hours, startdate=datetime.now().date()
) # Get the current date and prediction end date
###############
# Load Forecast
###############
lf = LoadForecast(
filepath=server_dir / ".." / "data" / "load_profiles.npz", year_energy=year_energy
) # Instantiate LoadForecast with specified parameters
leistung_haushalt = lf.get_stats_for_date_range(date_now, date)[
0
] # Get expected household load for the date range
gesamtlast = Gesamtlast(
prediction_hours=config.eos.prediction_hours
) # Create Gesamtlast instance
gesamtlast.hinzufuegen(
"Haushalt", leistung_haushalt
) # Add household load to total load calculation
# ###############
# # WP (Heat Pump)
# ##############
# leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours
# gesamtlast.hinzufuegen("Heatpump", leistung_wp) # Add heat pump load to total load calculation
last = gesamtlast.gesamtlast_berechnen() # Calculate total load
print(last) # Output total load
return last.tolist() # Return total load as JSON
@app.get("/pvforecast")
def fastapi_pvprognose(url: str, ac_power_measurement: Optional[float] = None) -> ForecastResponse:
date_now, date = get_start_enddate(config.eos.prediction_hours, startdate=datetime.now().date())
###############
# PV Forecast
###############
PVforecast = PVForecast(
prediction_hours=config.eos.prediction_hours, url=url
) # Instantiate PVForecast with given parameters
if ac_power_measurement is not None:
PVforecast.update_ac_power_measurement(
date_time=datetime.now(),
ac_power_measurement=ac_power_measurement,
) # Update measurement
# Get PV forecast and temperature forecast for the specified date range
pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now, date)
temperature_forecast = PVforecast.get_temperature_for_date_range(date_now, date)
# Return both forecasts as a JSON response
ret = {
"temperature": temperature_forecast.tolist(),
"pvpower": pv_forecast.tolist(),
}
return ret
@app.post("/optimize")
def fastapi_optimize(
parameters: OptimizationParameters,
start_hour: Annotated[
Optional[int], Query(description="Defaults to current hour of the day.")
] = None,
) -> OptimizeResponse:
if start_hour is None:
start_hour = datetime.now().hour
# Perform optimization simulation
result = opt_class.optimierung_ems(parameters=parameters, start_hour=start_hour)
# print(result)
# convert to JSON (None accepted by dumps)
return result
@app.get("/visualization_results.pdf", response_class=PdfResponse)
def get_pdf():
# Endpoint to serve the generated PDF with visualization results
output_path = config.working_dir / config.directories.output
if not output_path.is_dir():
raise SetupIncomplete(f"Output path does not exist: {output_path}.")
file_path = output_path / "visualization_results.pdf"
if not file_path.is_file():
raise HTTPException(status_code=404, detail="No visualization result available.")
return FileResponse(file_path)
@app.get("/site-map", include_in_schema=False)
def site_map():
return RedirectResponse(url="/docs")
@app.get("/", include_in_schema=False)
def root():
# Redirect the root URL to the site map
return RedirectResponse(url="/docs")
if __name__ == "__main__":
try:
config.run_setup()
except Exception as e:
print(f"Failed to initialize: {e}")
exit(1)
# Set host and port from environment variables or defaults
host = os.getenv("EOS_RUN_HOST", "0.0.0.0")
port = os.getenv("EOS_RUN_PORT", 8503)
try:
uvicorn.run(app, host=host, port=int(port)) # Run the FastAPI application
except Exception as e:
print(
f"Could not bind to host {host}:{port}. Error: {e}"
) # Error handling for binding issues
exit(1)
else:
# started from cli / dev server
config.run_setup()

2
src/akkudoktoreos/visualize.py
View File

@@ -179,7 +179,7 @@ def visualisiere_ergebnisse(
plt.plot(hours, ergebnisse["akku_soc_pro_stunde"], label="PV Battery (%)", marker="x")
plt.plot(
hours,
ergebnisse["E-Auto_SoC_pro_Stunde"],
ergebnisse["EAuto_SoC_pro_Stunde"],
label="E-Car Battery (%)",
marker="x",
)

305
src/akkudoktoreosserver/flask_server.py
View File

@@ -1,305 +0,0 @@
#!/usr/bin/env python3
import os
from datetime import datetime
from typing import Any, TypeGuard
import matplotlib
# Sets the Matplotlib backend to 'Agg' for rendering plots in environments without a display
matplotlib.use("Agg")
import pandas as pd
from flask import Flask, jsonify, redirect, request, send_from_directory, url_for
from akkudoktoreos.class_load import LoadForecast
from akkudoktoreos.class_load_container import Gesamtlast
from akkudoktoreos.class_load_corrector import LoadPredictionAdjuster
from akkudoktoreos.class_numpy_encoder import NumpyEncoder
from akkudoktoreos.class_optimize import optimization_problem
from akkudoktoreos.class_pv_forecast import PVForecast
from akkudoktoreos.class_strompreis import HourlyElectricityPriceForecast
from akkudoktoreos.config import (
SetupIncomplete,
get_start_enddate,
get_working_dir,
load_config,
)
app = Flask(__name__)
working_dir = get_working_dir()
# copy config to working directory. Make this a CLI option later
config = load_config(working_dir, True)
opt_class = optimization_problem(config)
def isfloat(num: Any) -> TypeGuard[float]:
"""Check if a given input can be converted to float."""
if num is None:
return False
if isinstance(num, str):
num = num.strip() # Strip any surrounding whitespace
try:
float_value = float(num)
return not (
float_value == float("inf")
or float_value == float("-inf")
or float_value != float_value
) # Excludes NaN or Infinity
except (ValueError, TypeError):
return False
@app.route("/strompreis", methods=["GET"])
def flask_strompreis():
# Get the current date and the end date based on prediction hours
date_now, date = get_start_enddate(config.eos.prediction_hours, startdate=datetime.now().date())
price_forecast = HourlyElectricityPriceForecast(
source=f"https://api.akkudoktor.net/prices?start={date_now}&end={date}",
config=config.eos.prediction_hours,
use_cache=False,
)
specific_date_prices = price_forecast.get_price_for_daterange(
date_now, date
) # Fetch prices for the specified date range
return jsonify(specific_date_prices.tolist())
# Endpoint to handle total load calculation based on the latest measured data
@app.route("/gesamtlast", methods=["POST"])
def flask_gesamtlast():
# Retrieve data from the JSON body
data = request.get_json()
# Extract year_energy and prediction_hours from the request JSON
year_energy = float(data.get("year_energy"))
prediction_hours = int(data.get("hours", 48)) # Default to 48 hours if not specified
# Measured data in JSON format
measured_data_json = data.get("measured_data")
measured_data = pd.DataFrame(measured_data_json)
measured_data["time"] = pd.to_datetime(measured_data["time"])
# Ensure datetime has timezone info for accurate calculations
if measured_data["time"].dt.tz is None:
measured_data["time"] = measured_data["time"].dt.tz_localize("Europe/Berlin")
else:
measured_data["time"] = measured_data["time"].dt.tz_convert("Europe/Berlin")
# Remove timezone info after conversion to simplify further processing
measured_data["time"] = measured_data["time"].dt.tz_localize(None)
# Instantiate LoadForecast and generate forecast data
file_path = os.path.join("data", "load_profiles.npz")
lf = LoadForecast(filepath=file_path, year_energy=year_energy)
forecast_list = []
# Generate daily forecasts for the date range based on measured data
for single_date in pd.date_range(
measured_data["time"].min().date(), measured_data["time"].max().date()
):
date_str = single_date.strftime("%Y-%m-%d")
daily_forecast = lf.get_daily_stats(date_str)
mean_values = daily_forecast[0]
hours = [single_date + pd.Timedelta(hours=i) for i in range(24)]
daily_forecast_df = pd.DataFrame({"time": hours, "Last Pred": mean_values})
forecast_list.append(daily_forecast_df)
# Concatenate all daily forecasts into a single DataFrame
predicted_data = pd.concat(forecast_list, ignore_index=True)
# Create LoadPredictionAdjuster instance to adjust the predictions based on measured data
adjuster = LoadPredictionAdjuster(measured_data, predicted_data, lf)
adjuster.calculate_weighted_mean() # Calculate weighted mean for adjustment
adjuster.adjust_predictions() # Adjust predictions based on measured data
future_predictions = adjuster.predict_next_hours(prediction_hours) # Predict future load
# Extract household power predictions
leistung_haushalt = future_predictions["Adjusted Pred"].values
gesamtlast = Gesamtlast(prediction_hours=prediction_hours)
gesamtlast.hinzufuegen(
"Haushalt", leistung_haushalt
) # Add household load to total load calculation
# Calculate the total load
last = gesamtlast.gesamtlast_berechnen() # Compute total load
return jsonify(last.tolist())
@app.route("/gesamtlast_simple", methods=["GET"])
def flask_gesamtlast_simple():
if request.method == "GET":
year_energy = float(
request.args.get("year_energy")
) # Get annual energy value from query parameters
date_now, date = get_start_enddate(
config.eos.prediction_hours, startdate=datetime.now().date()
) # Get the current date and prediction end date
###############
# Load Forecast
###############
server_dir = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(server_dir, "data", "load_profiles.npz")
print(file_path)
lf = LoadForecast(
filepath=file_path, year_energy=year_energy
) # Instantiate LoadForecast with specified parameters
leistung_haushalt = lf.get_stats_for_date_range(date_now, date)[
0
] # Get expected household load for the date range
gesamtlast = Gesamtlast(
prediction_hours=config.eos.prediction_hours
) # Create Gesamtlast instance
gesamtlast.hinzufuegen(
"Haushalt", leistung_haushalt
) # Add household load to total load calculation
# ###############
# # WP (Heat Pump)
# ##############
# leistung_wp = wp.simulate_24h(temperature_forecast) # Simulate heat pump load for 24 hours
# gesamtlast.hinzufuegen("Heatpump", leistung_wp) # Add heat pump load to total load calculation
last = gesamtlast.gesamtlast_berechnen() # Calculate total load
print(last) # Output total load
return jsonify(last.tolist()) # Return total load as JSON
@app.route("/pvforecast", methods=["GET"])
def flask_pvprognose():
if request.method == "GET":
# Retrieve URL and AC power measurement from query parameters
url = request.args.get("url")
ac_power_measurement = request.args.get("ac_power_measurement")
date_now, date = get_start_enddate(
config.eos.prediction_hours, startdate=datetime.now().date()
)
###############
# PV Forecast
###############
PVforecast = PVForecast(
prediction_hours=config.eos.prediction_hours, url=url
) # Instantiate PVForecast with given parameters
if isfloat(ac_power_measurement): # Check if the AC power measurement is a valid float
PVforecast.update_ac_power_measurement(
date_time=datetime.now(),
ac_power_measurement=float(ac_power_measurement),
) # Update measurement
# Get PV forecast and temperature forecast for the specified date range
pv_forecast = PVforecast.get_pv_forecast_for_date_range(date_now, date)
temperature_forecast = PVforecast.get_temperature_for_date_range(date_now, date)
# Return both forecasts as a JSON response
ret = {
"temperature": temperature_forecast.tolist(),
"pvpower": pv_forecast.tolist(),
}
return jsonify(ret)
@app.route("/optimize", methods=["POST"])
def flask_optimize():
if request.method == "POST":
from datetime import datetime
# Retrieve optimization parameters from the request JSON
parameter = request.json
# Check for required parameters
required_parameters = [
"preis_euro_pro_wh_akku",
"strompreis_euro_pro_wh",
"gesamtlast",
"pv_akku_cap",
"einspeiseverguetung_euro_pro_wh",
"pv_forecast",
"temperature_forecast",
"eauto_min_soc",
"eauto_cap",
"eauto_charge_efficiency",
"eauto_charge_power",
"eauto_soc",
"pv_soc",
"start_solution",
"haushaltsgeraet_dauer",
"haushaltsgeraet_wh",
]
# Identify any missing parameters
missing_params = [p for p in required_parameters if p not in parameter]
if missing_params:
return jsonify(
{"error": f"Missing parameter: {', '.join(missing_params)}"}
), 400 # Return error for missing parameters
# Optional min SoC PV Battery
if "min_soc_prozent" not in parameter:
parameter["min_soc_prozent"] = 0
# Perform optimization simulation
result = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour)
# print(result)
# convert to JSON (None accepted by dumps)
return NumpyEncoder.dumps(result)
@app.route("/visualization_results.pdf")
def get_pdf():
# Endpoint to serve the generated PDF with visualization results
output_path = config.working_dir / config.directories.output
if not output_path.is_dir():
raise SetupIncomplete(f"Output path does not exist: {output_path}.")
return send_from_directory(output_path, "visualization_results.pdf")
@app.route("/site-map")
def site_map():
# Function to generate a site map of valid routes in the application
def print_links(links):
content = "<h1>Valid routes</h1><ul>"
for link in links:
content += f"<li><a href='{link}'>{link}</a></li>"
content += "</ul>"
return content
# Check if the route has no empty parameters
def has_no_empty_params(rule):
defaults = rule.defaults if rule.defaults is not None else ()
arguments = rule.arguments if rule.arguments is not None else ()
return len(defaults) >= len(arguments)
# Collect all valid GET routes without empty parameters
links = []
for rule in app.url_map.iter_rules():
if "GET" in rule.methods and has_no_empty_params(rule):
url = url_for(rule.endpoint, **(rule.defaults or {}))
links.append(url)
return print_links(sorted(links)) # Return the sorted links as HTML
@app.route("/")
def root():
# Redirect the root URL to the site map
return redirect("/site-map", code=302)
if __name__ == "__main__":
try:
config.run_setup()
# Set host and port from environment variables or defaults
host = os.getenv("FLASK_RUN_HOST", "0.0.0.0")
port = os.getenv("FLASK_RUN_PORT", 8503)
app.run(debug=True, host=host, port=port) # Run the Flask application
except Exception as e:
print(
f"Could not bind to host {host}:{port}. Error: {e}"
) # Error handling for binding issues