Refactored class_optimize.py

- Optimized Imports: Removed unused imports and organized them.
- Refactored Code: Introduced split_individual function for clarity.
- Improved Efficiency: Enhanced penalty calculation and streamlined loops.
- Updated Evaluation Logic: Better handling of penalties in evaluate.
- Type Hints added
- fixed seed option added for automated tests
- verbose comment added, default False

Notes:
- isfloat is only used in flask_server.py
- start_hour is not used in this class
This commit is contained in:
Normann 2024-10-04 03:11:24 +02:00 committed by Andreas
parent f10b64e7c6
commit e2bca5aba1

View File

@ -1,25 +1,23 @@
import os import os
import random
import sys import sys
from typing import Any, Dict, List, Optional, Tuple
import matplotlib
import numpy as np import numpy as np
from deap import algorithms, base, creator, tools
from modules.class_akku import PVAkku from modules.class_akku import PVAkku
from modules.class_ems import EnergieManagementSystem, Wechselrichter from modules.class_ems import EnergieManagementSystem
from modules.class_haushaltsgeraet import Haushaltsgeraet from modules.class_haushaltsgeraet import Haushaltsgeraet
from modules.class_inverter import Wechselrichter
from modules.visualize import visualisiere_ergebnisse from modules.visualize import visualisiere_ergebnisse
matplotlib.use("Agg") # Setzt das Backend auf Agg
import random
from datetime import datetime, timedelta
from deap import algorithms, base, creator, tools
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import moegliche_ladestroeme_in_prozent from config import moegliche_ladestroeme_in_prozent
def isfloat(num): def isfloat(num: Any) -> bool:
"""Check if a given input can be converted to float."""
try: try:
float(num) float(num)
return True return True
@ -27,281 +25,209 @@ def isfloat(num):
return False return False
def differential_evolution(
population,
toolbox,
cxpb,
mutpb,
ngen,
stats=None,
halloffame=None,
verbose=__debug__,
):
"""Differential Evolution Algorithm"""
# Evaluate the entire population
fitnesses = list(map(toolbox.evaluate, population))
for ind, fit in zip(population, fitnesses):
ind.fitness.values = fit
if halloffame is not None:
halloffame.update(population)
logbook = tools.Logbook()
logbook.header = ["gen", "nevals"] + (stats.fields if stats else [])
for gen in range(ngen):
# Generate the next generation by mutation and recombination
for i, target in enumerate(population):
a, b, c = random.sample([ind for ind in population if ind != target], 3)
mutant = toolbox.clone(a)
for k in range(len(mutant)):
mutant[k] = c[k] + mutpb * (a[k] - b[k]) # Mutation step
if random.random() < cxpb: # Recombination step
mutant[k] = target[k]
# Evaluate the mutant
mutant.fitness.values = toolbox.evaluate(mutant)
# Replace if mutant is better
if mutant.fitness > target.fitness:
population[i] = mutant
# Update hall of fame
if halloffame is not None:
halloffame.update(population)
# Gather all the fitnesses in one list and print the stats
record = stats.compile(population) if stats else {}
logbook.record(gen=gen, nevals=len(population), **record)
if verbose:
print(logbook.stream)
return population, logbook
class optimization_problem: class optimization_problem:
def __init__(self, prediction_hours=24, strafe=10, optimization_hours=24): def __init__(
self.prediction_hours = prediction_hours # self,
prediction_hours: int = 24,
strafe: float = 10,
optimization_hours: int = 24,
verbose: bool = False,
fixed_seed: Optional[int] = None,
):
"""Initialize the optimization problem with the required parameters."""
self.prediction_hours = prediction_hours
self.strafe = strafe self.strafe = strafe
self.opti_param = None self.opti_param = None
self.fixed_eauto_hours = prediction_hours - optimization_hours self.fixed_eauto_hours = prediction_hours - optimization_hours
self.possible_charge_values = moegliche_ladestroeme_in_prozent self.possible_charge_values = moegliche_ladestroeme_in_prozent
self.verbose = verbose
self.fix_seed = fixed_seed
def split_individual(self, individual): # Set a fixed seed for random operations if provided
if fixed_seed is not None:
random.seed(fixed_seed)
def split_individual(
self, individual: List[float]
) -> Tuple[List[int], List[float], Optional[int]]:
""" """
Teilt das gegebene Individuum in die verschiedenen Parameter auf: Split the individual solution into its components:
- Entladeparameter (discharge_hours_bin) 1. Discharge hours (binary),
- Ladeparameter (eautocharge_hours_float) 2. Electric vehicle charge hours (float),
- Haushaltsgeräte (spuelstart_int, falls vorhanden) 3. Dishwasher start time (integer if applicable).
""" """
# Extrahiere die Entlade- und Ladeparameter direkt aus dem Individuum discharge_hours_bin = individual[: self.prediction_hours]
discharge_hours_bin = individual[
: self.prediction_hours
] # Erste 24 Werte sind Bool (Entladen)
eautocharge_hours_float = individual[ eautocharge_hours_float = individual[
self.prediction_hours : self.prediction_hours * 2 self.prediction_hours : self.prediction_hours * 2
] # Nächste 24 Werte sind Float (Laden) ]
spuelstart_int = (
spuelstart_int = None individual[-1]
if self.opti_param and self.opti_param.get("haushaltsgeraete", 0) > 0: if self.opti_param and self.opti_param.get("haushaltsgeraete", 0) > 0
spuelstart_int = individual[ else None
-1 )
] # Letzter Wert ist Startzeit für Haushaltsgerät
return discharge_hours_bin, eautocharge_hours_float, spuelstart_int return discharge_hours_bin, eautocharge_hours_float, spuelstart_int
def setup_deap_environment(self, opti_param, start_hour): 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 self.opti_param = opti_param
if "FitnessMin" in creator.__dict__: # Remove existing FitnessMin and Individual classes from creator if present
del creator.FitnessMin for attr in ["FitnessMin", "Individual"]:
if "Individual" in creator.__dict__: if attr in creator.__dict__:
del creator.Individual del creator.__dict__[attr]
# Create new FitnessMin and Individual classes
creator.create("FitnessMin", base.Fitness, weights=(-1.0,)) creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
creator.create("Individual", list, fitness=creator.FitnessMin) creator.create("Individual", list, fitness=creator.FitnessMin)
# PARAMETER # Initialize toolbox with attributes and operations
self.toolbox = base.Toolbox() self.toolbox = base.Toolbox()
self.toolbox.register("attr_bool", random.randint, 0, 1) self.toolbox.register("attr_bool", random.randint, 0, 1)
self.toolbox.register( self.toolbox.register("attr_float", random.uniform, 0, 1)
"attr_float", random.uniform, 0, 1
) # Für kontinuierliche Werte zwischen 0 und 1 (z.B. für E-Auto-Ladeleistung)
# self.toolbox.register("attr_choice", random.choice, self.possible_charge_values) # Für diskrete Ladeströme
self.toolbox.register("attr_int", random.randint, start_hour, 23) self.toolbox.register("attr_int", random.randint, start_hour, 23)
################### # Register individual creation method based on household appliance parameter
# Haushaltsgeraete
# print("Haushalt:",opti_param["haushaltsgeraete"])
if opti_param["haushaltsgeraete"] > 0: if opti_param["haushaltsgeraete"] > 0:
self.toolbox.register(
def create_individual(): "individual",
attrs = [ lambda: creator.Individual(
self.toolbox.attr_bool() for _ in range(self.prediction_hours) [self.toolbox.attr_bool() for _ in range(self.prediction_hours)]
] # 24 Bool-Werte für Entladen + [self.toolbox.attr_float() for _ in range(self.prediction_hours)]
attrs += [ + [self.toolbox.attr_int()]
self.toolbox.attr_float() for _ in range(self.prediction_hours) ),
] # 24 Float-Werte für Laden )
attrs.append(self.toolbox.attr_int()) # Haushaltsgerät-Startzeit
return creator.Individual(attrs)
else: else:
self.toolbox.register(
"individual",
lambda: creator.Individual(
[self.toolbox.attr_bool() for _ in range(self.prediction_hours)]
+ [self.toolbox.attr_float() for _ in range(self.prediction_hours)]
),
)
def create_individual(): # Register population, mating, mutation, and selection functions
attrs = [
self.toolbox.attr_bool() for _ in range(self.prediction_hours)
] # 24 Bool-Werte für Entladen
attrs += [
self.toolbox.attr_float() for _ in range(self.prediction_hours)
] # 24 Float-Werte für Laden
return creator.Individual(attrs)
self.toolbox.register(
"individual", create_individual
) # tools.initCycle, creator.Individual, (self.toolbox.attr_bool,self.toolbox.attr_bool), n=self.prediction_hours+1)
self.toolbox.register( self.toolbox.register(
"population", tools.initRepeat, list, self.toolbox.individual "population", tools.initRepeat, list, self.toolbox.individual
) )
self.toolbox.register("mate", tools.cxTwoPoint) self.toolbox.register("mate", tools.cxTwoPoint)
self.toolbox.register("mutate", tools.mutFlipBit, indpb=0.1) self.toolbox.register("mutate", tools.mutFlipBit, indpb=0.1)
# self.toolbox.register("mutate", mutate_choice, self.possible_charge_values, indpb=0.1)
# self.toolbox.register("mutate", tools.mutUniformInt, low=0, up=len(self.possible_charge_values)-1, indpb=0.1)
self.toolbox.register("select", tools.selTournament, tournsize=3) self.toolbox.register("select", tools.selTournament, tournsize=3)
def evaluate_inner(self, individual, ems, start_hour): def evaluate_inner(
self, individual: List[float], ems: EnergieManagementSystem, start_hour: int
) -> Dict[str, Any]:
"""
Internal evaluation function that simulates the energy management system (EMS)
using the provided individual solution.
"""
ems.reset() ems.reset()
# print("Spuel:",self.opti_param)
discharge_hours_bin, eautocharge_hours_float, spuelstart_int = ( discharge_hours_bin, eautocharge_hours_float, spuelstart_int = (
self.split_individual(individual) self.split_individual(individual)
) )
if self.opti_param.get("haushaltsgeraete", 0) > 0:
# Haushaltsgeraete
if self.opti_param["haushaltsgeraete"] > 0:
ems.set_haushaltsgeraet_start(spuelstart_int, global_start_hour=start_hour) ems.set_haushaltsgeraet_start(spuelstart_int, global_start_hour=start_hour)
# discharge_hours_bin = np.full(self.prediction_hours,0)
ems.set_akku_discharge_hours(discharge_hours_bin) ems.set_akku_discharge_hours(discharge_hours_bin)
eautocharge_hours_float[self.prediction_hours - self.fixed_eauto_hours :] = [
# Setze die festen Werte für die letzten x Stunden 0.0
for i in range( ] * self.fixed_eauto_hours
self.prediction_hours - self.fixed_eauto_hours, self.prediction_hours
):
eautocharge_hours_float[i] = (
0.0 # Setze die letzten x Stunden auf einen festen Wert (oder vorgegebenen Wert)
)
# print(eautocharge_hours_float)
ems.set_eauto_charge_hours(eautocharge_hours_float) ems.set_eauto_charge_hours(eautocharge_hours_float)
return ems.simuliere(start_hour)
o = ems.simuliere(start_hour) def evaluate(
self,
return o individual: List[float],
ems: EnergieManagementSystem,
# Fitness-Funktion (muss Ihre EnergieManagementSystem-Logik integrieren) parameter: Dict[str, Any],
def evaluate(self, individual, ems, parameter, start_hour, worst_case): start_hour: int,
worst_case: bool,
) -> Tuple[float]:
"""
Evaluate the fitness of an individual solution based on the simulation results.
"""
try: try:
o = self.evaluate_inner(individual, ems, start_hour) o = self.evaluate_inner(individual, ems, start_hour)
except Exception: except Exception:
return (100000.0,) return (100000.0,) # Return a high penalty in case of an exception
gesamtbilanz = o["Gesamtbilanz_Euro"] gesamtbilanz = o["Gesamtbilanz_Euro"] * (-1.0 if worst_case else 1.0)
if worst_case: discharge_hours_bin, eautocharge_hours_float, _ = self.split_individual(
gesamtbilanz = gesamtbilanz * -1.0 individual
discharge_hours_bin, eautocharge_hours_float, spuelstart_int = (
self.split_individual(individual)
) )
max_ladeleistung = np.max(moegliche_ladestroeme_in_prozent) max_ladeleistung = np.max(moegliche_ladestroeme_in_prozent)
strafe_überschreitung = 0.0 # Penalty for not discharging
gesamtbilanz += sum(
0.01 for i in range(self.prediction_hours) if discharge_hours_bin[i] == 0.0
)
# Ladeleistung überschritten? # Penalty for charging the electric vehicle during restricted hours
for ladeleistung in eautocharge_hours_float: gesamtbilanz += sum(
if ladeleistung > max_ladeleistung: self.strafe
# Berechne die Überschreitung for i in range(
überschreitung = ladeleistung - max_ladeleistung self.prediction_hours - self.fixed_eauto_hours, self.prediction_hours
# Füge eine Strafe hinzu (z.B. 10 Einheiten Strafe pro Prozentpunkt Überschreitung) )
strafe_überschreitung += ( if eautocharge_hours_float[i] != 0.0
self.strafe * 10 )
) # Hier ist die Strafe proportional zur Überschreitung
# Für jeden Discharge 0, eine kleine Strafe von 1 Cent, da die Lastvertelung noch fehlt. Also wenn es egal ist, soll er den Akku entladen lassen # Penalty for exceeding maximum charge power
for i in range(0, self.prediction_hours): gesamtbilanz += sum(
if ( self.strafe * 10
discharge_hours_bin[i] == 0.0 for ladeleistung in eautocharge_hours_float
): # Wenn die letzten x Stunden von einem festen Wert abweichen if ladeleistung > max_ladeleistung
gesamtbilanz += 0.01 # Bestrafe den Optimierer )
# E-Auto nur die ersten self.fixed_eauto_hours # Penalty for not meeting the minimum SOC (State of Charge) requirement
for i in range( if parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent() <= 0.0:
self.prediction_hours - self.fixed_eauto_hours, self.prediction_hours gesamtbilanz += sum(
): self.strafe
if ( for ladeleistung in eautocharge_hours_float
eautocharge_hours_float[i] != 0.0 if ladeleistung != 0.0
): # Wenn die letzten x Stunden von einem festen Wert abweichen )
gesamtbilanz += self.strafe # Bestrafe den Optimierer
# Überprüfung, ob der Mindest-SoC erreicht wird
final_soc = (
ems.eauto.ladezustand_in_prozent()
) # Nimmt den SoC am Ende des Optimierungszeitraums
if (parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()) <= 0.0:
# print (parameter['eauto_min_soc']," " ,ems.eauto.ladezustand_in_prozent()," ",(parameter['eauto_min_soc']-ems.eauto.ladezustand_in_prozent()))
for i in range(0, self.prediction_hours):
if (
eautocharge_hours_float[i] != 0.0
): # Wenn die letzten x Stunden von einem festen Wert abweichen
gesamtbilanz += self.strafe # Bestrafe den Optimierer
eauto_roi = parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()
individual.extra_data = ( individual.extra_data = (
o["Gesamtbilanz_Euro"], o["Gesamtbilanz_Euro"],
o["Gesamt_Verluste"], o["Gesamt_Verluste"],
eauto_roi, parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent(),
) )
restenergie_akku = ems.akku.aktueller_energieinhalt() # Adjust total balance with battery value and penalties for unmet SOC
restwert_akku = restenergie_akku * parameter["preis_euro_pro_wh_akku"] restwert_akku = (
# print(restenergie_akku) ems.akku.aktueller_energieinhalt() * parameter["preis_euro_pro_wh_akku"]
# print(parameter["preis_euro_pro_wh_akku"]) )
# print(restwert_akku) gesamtbilanz += (
# print() max(
strafe = 0.0 0,
strafe = max( (parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent())
0, * self.strafe,
(parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()) )
* self.strafe, - restwert_akku
) )
gesamtbilanz += strafe - restwert_akku + strafe_überschreitung
# gesamtbilanz += o["Gesamt_Verluste"]/10000.0
return (gesamtbilanz,) return (gesamtbilanz,)
# Genetischer Algorithmus def optimize(
def optimize(self, start_solution=None): self, start_solution: Optional[List[float]] = None
) -> Tuple[Any, Dict[str, List[Any]]]:
"""Run the optimization process using a genetic algorithm."""
population = self.toolbox.population(n=300) population = self.toolbox.population(n=300)
hof = tools.HallOfFame(1) hof = tools.HallOfFame(1)
stats = tools.Statistics(lambda ind: ind.fitness.values) stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", np.mean)
stats.register("min", np.min) stats.register("min", np.min)
stats.register("max", np.max)
print("Start:", start_solution) if self.verbose:
print("Start optimize:", start_solution)
if start_solution is not None and start_solution != -1: # Insert the start solution into the population if provided
population.insert(0, creator.Individual(start_solution)) if start_solution not in [None, -1]:
population.insert(1, creator.Individual(start_solution)) for _ in range(3):
population.insert(2, creator.Individual(start_solution)) population.insert(0, creator.Individual(start_solution))
# Run the evolutionary algorithm
algorithms.eaMuPlusLambda( algorithms.eaMuPlusLambda(
population, population,
self.toolbox, self.toolbox,
@ -312,11 +238,8 @@ class optimization_problem:
ngen=400, ngen=400,
stats=stats, stats=stats,
halloffame=hof, halloffame=hof,
verbose=True, verbose=self.verbose,
) )
# algorithms.eaSimple(population, self.toolbox, cxpb=0.3, mutpb=0.3, ngen=200, stats=stats, halloffame=hof, verbose=True)
# algorithms.eaMuCommaLambda(population, self.toolbox, mu=100, lambda_=200, cxpb=0.2, mutpb=0.4, ngen=300, stats=stats, halloffame=hof, verbose=True)
# population, log = differential_evolution(population, self.toolbox, cxpb=0.2, mutpb=0.5, ngen=200, stats=stats, halloffame=hof, verbose=True)
member = {"bilanz": [], "verluste": [], "nebenbedingung": []} member = {"bilanz": [], "verluste": [], "nebenbedingung": []}
for ind in population: for ind in population:
@ -329,42 +252,28 @@ class optimization_problem:
return hof[0], member return hof[0], member
def optimierung_ems( def optimierung_ems(
self, parameter=None, start_hour=None, worst_case=False, startdate=None self,
): parameter: Optional[Dict[str, Any]] = None,
############ start_hour: Optional[int] = None,
# Parameter worst_case: bool = False,
############ startdate: Optional[Any] = None, # startdate is not used!
if startdate is None: ) -> Dict[str, Any]:
date = ( """
datetime.now().date() + timedelta(hours=self.prediction_hours) Perform EMS (Energy Management System) optimization and visualize results.
).strftime("%Y-%m-%d") """
date_now = datetime.now().strftime("%Y-%m-%d")
else:
date = (startdate + timedelta(hours=self.prediction_hours)).strftime(
"%Y-%m-%d"
)
date_now = startdate.strftime("%Y-%m-%d")
# print("Start_date:",date_now)
akku_size = parameter["pv_akku_cap"] # Wh
einspeiseverguetung_euro_pro_wh = np.full( einspeiseverguetung_euro_pro_wh = np.full(
self.prediction_hours, parameter["einspeiseverguetung_euro_pro_wh"] self.prediction_hours, parameter["einspeiseverguetung_euro_pro_wh"]
) # = # € / Wh 7/(1000.0*100.0) )
discharge_array = np.full(
self.prediction_hours, 1 # Initialize PV and EV batteries
) # np.array([1, 0, 1, 0, 1, 1, 1, 1, 1, 0, 1, 1, 0, 1, 1, 0, 1, 1, 1, 1, 1, 1, 1, 1, 0, 1, 0, 0, 0, 1, 1, 1, 1, 0, 1, 1, 0, 0, 0, 0, 1, 0, 0, 1, 0, 0, 1, 0]) #
akku = PVAkku( akku = PVAkku(
kapazitaet_wh=akku_size, kapazitaet_wh=parameter["pv_akku_cap"],
hours=self.prediction_hours, hours=self.prediction_hours,
start_soc_prozent=parameter["pv_soc"], start_soc_prozent=parameter["pv_soc"],
max_ladeleistung_w=5000, max_ladeleistung_w=5000,
) )
akku.set_charge_per_hour(discharge_array) akku.set_charge_per_hour(np.full(self.prediction_hours, 1))
laden_moeglich = np.full(
self.prediction_hours, 1
) # np.array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 1, 0, 1, 1, 0, 1, 1, 1, 0, 0, 0, 0, 1, 1, 1, 1, 0, 1, 1, 0, 0])
eauto = PVAkku( eauto = PVAkku(
kapazitaet_wh=parameter["eauto_cap"], kapazitaet_wh=parameter["eauto_cap"],
hours=self.prediction_hours, hours=self.prediction_hours,
@ -373,121 +282,71 @@ class optimization_problem:
max_ladeleistung_w=parameter["eauto_charge_power"], max_ladeleistung_w=parameter["eauto_charge_power"],
start_soc_prozent=parameter["eauto_soc"], start_soc_prozent=parameter["eauto_soc"],
) )
eauto.set_charge_per_hour(laden_moeglich) eauto.set_charge_per_hour(np.full(self.prediction_hours, 1))
min_soc_eauto = parameter["eauto_min_soc"]
start_params = parameter["start_solution"]
############### # Initialize household appliance if applicable
# spuelmaschine spuelmaschine = (
############## Haushaltsgeraet(
print(parameter)
if parameter["haushaltsgeraet_dauer"] > 0:
spuelmaschine = Haushaltsgeraet(
hours=self.prediction_hours, hours=self.prediction_hours,
verbrauch_kwh=parameter["haushaltsgeraet_wh"], verbrauch_kwh=parameter["haushaltsgeraet_wh"],
dauer_h=parameter["haushaltsgeraet_dauer"], dauer_h=parameter["haushaltsgeraet_dauer"],
) ).set_startzeitpunkt(start_hour)
spuelmaschine.set_startzeitpunkt(start_hour) # Startet jetzt if parameter["haushaltsgeraet_dauer"] > 0
else: else None
spuelmaschine = None )
###############
# PV Forecast
###############
# PVforecast = PVForecast(filepath=os.path.join(r'test_data', r'pvprognose.json'))
# PVforecast = PVForecast(prediction_hours = self.prediction_hours, url=pv_forecast_url)
# #print("PVPOWER",parameter['pvpowernow'])
# if isfloat(parameter['pvpowernow']):
# PVforecast.update_ac_power_measurement(date_time=datetime.now(), ac_power_measurement=float(parameter['pvpowernow']))
# #PVforecast.print_ac_power_and_measurement()
pv_forecast = parameter[
"pv_forecast"
] # PVforecast.get_pv_forecast_for_date_range(date_now,date) #get_forecast_for_date(date)
temperature_forecast = parameter[
"temperature_forecast"
] # PVforecast.get_temperature_for_date_range(date_now,date)
###############
# Strompreise
###############
specific_date_prices = parameter["strompreis_euro_pro_wh"]
print(specific_date_prices)
# print("https://api.akkudoktor.net/prices?start="+date_now+"&end="+date)
# Initialize the inverter and energy management system
wr = Wechselrichter(10000, akku) wr = Wechselrichter(10000, akku)
ems = EnergieManagementSystem( ems = EnergieManagementSystem(
gesamtlast=parameter["gesamtlast"], gesamtlast=parameter["gesamtlast"],
pv_prognose_wh=pv_forecast, pv_prognose_wh=parameter["pv_forecast"],
strompreis_euro_pro_wh=specific_date_prices, strompreis_euro_pro_wh=parameter["strompreis_euro_pro_wh"],
einspeiseverguetung_euro_pro_wh=einspeiseverguetung_euro_pro_wh, einspeiseverguetung_euro_pro_wh=einspeiseverguetung_euro_pro_wh,
eauto=eauto, eauto=eauto,
haushaltsgeraet=spuelmaschine, haushaltsgeraet=spuelmaschine,
wechselrichter=wr, wechselrichter=wr,
) )
o = ems.simuliere(start_hour)
############### # Setup the DEAP environment and optimization process
# Optimizer Init self.setup_deap_environment(
############## {"haushaltsgeraete": 1 if spuelmaschine else 0}, start_hour
opti_param = {} )
opti_param["haushaltsgeraete"] = 0 self.toolbox.register(
if spuelmaschine is not None: "evaluate",
opti_param["haushaltsgeraete"] = 1 lambda ind: self.evaluate(ind, ems, parameter, start_hour, worst_case),
)
start_solution, extra_data = self.optimize(parameter["start_solution"])
self.setup_deap_environment(opti_param, start_hour) # Perform final evaluation on the best solution
o = self.evaluate_inner(start_solution, ems, start_hour)
def evaluate_wrapper(individual):
return self.evaluate(individual, ems, parameter, start_hour, worst_case)
self.toolbox.register("evaluate", evaluate_wrapper)
start_solution, extra_data = self.optimize(start_params)
best_solution = start_solution
o = self.evaluate_inner(best_solution, ems, start_hour)
eauto = ems.eauto.to_dict()
spuelstart_int = None
discharge_hours_bin, eautocharge_hours_float, spuelstart_int = ( discharge_hours_bin, eautocharge_hours_float, spuelstart_int = (
self.split_individual(best_solution) self.split_individual(start_solution)
) )
print(parameter) # Visualize the results
print(best_solution)
visualisiere_ergebnisse( visualisiere_ergebnisse(
parameter["gesamtlast"], parameter["gesamtlast"],
pv_forecast, parameter["pv_forecast"],
specific_date_prices, parameter["strompreis_euro_pro_wh"],
o, o,
discharge_hours_bin, discharge_hours_bin,
eautocharge_hours_float, eautocharge_hours_float,
temperature_forecast, parameter["temperature_forecast"],
start_hour, start_hour,
self.prediction_hours, self.prediction_hours,
einspeiseverguetung_euro_pro_wh, einspeiseverguetung_euro_pro_wh,
extra_data=extra_data, extra_data=extra_data,
) )
os.system("cp visualisierungsergebnisse.pdf ~/") os.system("cp visualisierungsergebnisse.pdf ~/")
# 'Eigenverbrauch_Wh_pro_Stunde': eigenverbrauch_wh_pro_stunde, # Return final results as a dictionary
# 'Netzeinspeisung_Wh_pro_Stunde': netzeinspeisung_wh_pro_stunde,
# 'Netzbezug_Wh_pro_Stunde': netzbezug_wh_pro_stunde,
# 'Kosten_Euro_pro_Stunde': kosten_euro_pro_stunde,
# '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,
# 'Gesamteinnahmen_Euro': sum(einnahmen_euro_pro_stunde),
# 'Gesamtkosten_Euro': sum(kosten_euro_pro_stunde),
# "Verluste_Pro_Stunde":verluste_wh_pro_stunde,
# "Gesamt_Verluste":sum(verluste_wh_pro_stunde),
# "Haushaltsgeraet_wh_pro_stunde":haushaltsgeraet_wh_pro_stunde
# print(eauto)
return { return {
"discharge_hours_bin": discharge_hours_bin, "discharge_hours_bin": discharge_hours_bin,
"eautocharge_hours_float": eautocharge_hours_float, "eautocharge_hours_float": eautocharge_hours_float,
"result": o, "result": o,
"eauto_obj": eauto, "eauto_obj": ems.eauto.to_dict(),
"start_solution": best_solution, "start_solution": start_solution,
"spuelstart": spuelstart_int, "spuelstart": spuelstart_int,
"simulation_data": o, "simulation_data": o,
} }