diff --git a/src/akkudoktoreos/class_akku.py b/src/akkudoktoreos/class_akku.py index 7c5e13a..5778c51 100644 --- a/src/akkudoktoreos/class_akku.py +++ b/src/akkudoktoreos/class_akku.py @@ -88,7 +88,7 @@ def ladezustand_in_prozent(self): return (self.soc_wh / self.kapazitaet_wh) * 100 def energie_abgeben(self, wh, hour): - if self.discharge_array[hour] == 0 : + if self.discharge_array[hour] == 0: return 0.0, 0.0 # No energy discharge and no losses # Calculate the maximum energy that can be discharged considering min_soc and efficiency @@ -122,7 +122,7 @@ def energie_laden(self, wh, hour, relative_power=0.0): if hour is not None and self.charge_array[hour] == 0: return 0, 0 # Charging not allowed in this hour if relative_power > 0.0: - wh=self.max_ladeleistung_w*relative_power + wh = self.max_ladeleistung_w * relative_power # If no value for wh is given, use the maximum charging power wh = wh if wh is not None else self.max_ladeleistung_w @@ -134,8 +134,8 @@ def energie_laden(self, wh, hour, relative_power=0.0): max_possible_charge_wh = max(max_possible_charge_wh, 0.0) # Ensure non-negative # The actually charged energy cannot exceed requested energy, charging power, or maximum possible charge - effektive_lademenge = min(wh, max_possible_charge_wh) - + effektive_lademenge = min(wh, max_possible_charge_wh) + # Energy actually stored in the battery geladene_menge = effektive_lademenge * self.lade_effizienz @@ -143,7 +143,7 @@ def energie_laden(self, wh, hour, relative_power=0.0): self.soc_wh += geladene_menge # Ensure soc_wh does not exceed max_soc_wh self.soc_wh = min(self.soc_wh, self.max_soc_wh) - + # Calculate losses verluste_wh = effektive_lademenge - geladene_menge return geladene_menge, verluste_wh diff --git a/src/akkudoktoreos/class_ems.py b/src/akkudoktoreos/class_ems.py index dfa6e67..b8b8b0a 100644 --- a/src/akkudoktoreos/class_ems.py +++ b/src/akkudoktoreos/class_ems.py @@ -1,8 +1,10 @@ from datetime import datetime from typing import Dict, List, Optional, Union -from akkudoktoreos.config import * + import numpy as np +from akkudoktoreos.config import prediction_hours + class EnergieManagementSystem: def __init__( @@ -23,16 +25,16 @@ def __init__( self.eauto = eauto self.haushaltsgeraet = haushaltsgeraet self.wechselrichter = wechselrichter - self.ac_charge_hours = np.full(prediction_hours,0) - self.dc_charge_hours = np.full(prediction_hours,1) - self.ev_charge_hours = np.full(prediction_hours,0) + self.ac_charge_hours = np.full(prediction_hours, 0) + self.dc_charge_hours = np.full(prediction_hours, 1) + self.ev_charge_hours = np.full(prediction_hours, 0) def set_akku_discharge_hours(self, ds: List[int]) -> None: self.akku.set_discharge_per_hour(ds) def set_akku_ac_charge_hours(self, ds: np.ndarray) -> None: self.ac_charge_hours = ds - + def set_akku_dc_charge_hours(self, ds: np.ndarray) -> None: self.dc_charge_hours = ds @@ -52,11 +54,11 @@ def simuliere_ab_jetzt(self) -> dict: return self.simuliere(start_stunde) def simuliere(self, start_stunde: int) -> dict: - ''' + """ hour: akku_soc_pro_stunde begin of the hour, initial hour state! last_wh_pro_stunde integral of last hour (end state) - ''' + """ lastkurve_wh = self.gesamtlast assert ( @@ -82,10 +84,10 @@ def simuliere(self, start_stunde: int) -> dict: akku_soc_pro_stunde[0] = self.akku.ladezustand_in_prozent() if self.eauto: eauto_soc_pro_stunde[0] = self.eauto.ladezustand_in_prozent() - - for stunde in range(start_stunde , ende): + + for stunde in range(start_stunde, ende): stunde_since_now = stunde - start_stunde - + # Accumulate loads and PV generation verbrauch = self.gesamtlast[stunde] verluste_wh_pro_stunde[stunde_since_now] = 0.0 @@ -95,28 +97,32 @@ def simuliere(self, start_stunde: int) -> dict: haushaltsgeraet_wh_pro_stunde[stunde_since_now] = ha_load # E-Auto handling - if self.eauto and self.ev_charge_hours[stunde]>0: - geladene_menge_eauto, verluste_eauto = self.eauto.energie_laden(None, stunde, relative_power=self.ev_charge_hours[stunde]) + if self.eauto and self.ev_charge_hours[stunde] > 0: + geladene_menge_eauto, verluste_eauto = self.eauto.energie_laden( + None, stunde, relative_power=self.ev_charge_hours[stunde] + ) verbrauch += geladene_menge_eauto verluste_wh_pro_stunde[stunde_since_now] += verluste_eauto - + if self.eauto: eauto_soc_pro_stunde[stunde_since_now] = self.eauto.ladezustand_in_prozent() # Process inverter logic erzeugung = self.pv_prognose_wh[stunde] - self.akku.set_charge_allowed_for_hour(self.dc_charge_hours[stunde],stunde) + self.akku.set_charge_allowed_for_hour(self.dc_charge_hours[stunde], stunde) netzeinspeisung, netzbezug, verluste, eigenverbrauch = ( self.wechselrichter.energie_verarbeiten(erzeugung, verbrauch, stunde) ) # AC PV Battery Charge if self.ac_charge_hours[stunde] > 0.0: - self.akku.set_charge_allowed_for_hour(1,stunde) - geladene_menge, verluste_wh = self.akku.energie_laden(None,stunde,relative_power=self.ac_charge_hours[stunde]) - #print(stunde, " ", geladene_menge, " ",self.ac_charge_hours[stunde]," ",self.akku.ladezustand_in_prozent()) + self.akku.set_charge_allowed_for_hour(1, stunde) + geladene_menge, verluste_wh = self.akku.energie_laden( + None, stunde, relative_power=self.ac_charge_hours[stunde] + ) + # print(stunde, " ", geladene_menge, " ",self.ac_charge_hours[stunde]," ",self.akku.ladezustand_in_prozent()) verbrauch += geladene_menge - netzbezug +=geladene_menge - verluste_wh_pro_stunde[stunde_since_now] += verluste_wh + netzbezug += geladene_menge + verluste_wh_pro_stunde[stunde_since_now] += verluste_wh netzeinspeisung_wh_pro_stunde[stunde_since_now] = netzeinspeisung netzbezug_wh_pro_stunde[stunde_since_now] = netzbezug diff --git a/src/akkudoktoreos/class_numpy_encoder.py b/src/akkudoktoreos/class_numpy_encoder.py index 3c9a2ac..3c42ad4 100644 --- a/src/akkudoktoreos/class_numpy_encoder.py +++ b/src/akkudoktoreos/class_numpy_encoder.py @@ -1,6 +1,8 @@ import json + import numpy as np + class NumpyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, np.ndarray): @@ -8,17 +10,16 @@ def default(self, obj): if isinstance(obj, np.generic): return obj.item() # Convert NumPy scalars to native Python types return super(NumpyEncoder, self).default(obj) - + @staticmethod def dumps(data): """ Static method to serialize a Python object into a JSON string using NumpyEncoder. - + Args: data: The Python object to serialize. - + Returns: str: A JSON string representation of the object. """ return json.dumps(data, cls=NumpyEncoder) - diff --git a/src/akkudoktoreos/class_optimize.py b/src/akkudoktoreos/class_optimize.py index 6f5791c..05d1902 100644 --- a/src/akkudoktoreos/class_optimize.py +++ b/src/akkudoktoreos/class_optimize.py @@ -12,7 +12,6 @@ from akkudoktoreos.visualize import visualisiere_ergebnisse - class optimization_problem: def __init__( self, @@ -37,8 +36,9 @@ def __init__( if fixed_seed is not None: random.seed(fixed_seed) - - def decode_charge_discharge(self, discharge_hours_bin: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: + def decode_charge_discharge( + self, discharge_hours_bin: np.ndarray + ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """ Decode the input array `discharge_hours_bin` into three separate arrays for AC charging, DC charging, and discharge. The function maps AC and DC charging values to relative power levels (0 to 1), while the discharge remains binary (0 or 1). @@ -50,7 +50,7 @@ def decode_charge_discharge(self, discharge_hours_bin: np.ndarray) -> Tuple[np.n 1: Discharge ("discharge") 2-6: AC charging with different power levels ("ac_charge") 7-8: DC charging Dissallowed/allowed ("dc_charge") - + Returns: - ac_charge (np.ndarray): Array with AC charging values as relative power (0-1), other values set to 0. - dc_charge (np.ndarray): Array with DC charging values as relative power (0-1), other values set to 0. @@ -60,7 +60,9 @@ def decode_charge_discharge(self, discharge_hours_bin: np.ndarray) -> Tuple[np.n discharge_hours_bin = np.array(discharge_hours_bin) # Create ac_charge array: Only consider values between 2 and 6 (AC charging power levels), set the rest to 0 - ac_charge = np.where((discharge_hours_bin >= 2) & (discharge_hours_bin <= 6), discharge_hours_bin - 1, 0) + ac_charge = np.where( + (discharge_hours_bin >= 2) & (discharge_hours_bin <= 6), discharge_hours_bin - 1, 0 + ) ac_charge = ac_charge / 5.0 # Normalize AC charge to range between 0 and 1 # Create dc_charge array: 7 = Not allowed (mapped to 0), 8 = Allowed (mapped to 1) @@ -68,16 +70,15 @@ def decode_charge_discharge(self, discharge_hours_bin: np.ndarray) -> Tuple[np.n if self.optimize_dc_charge: dc_charge = np.where(discharge_hours_bin == 8, 1, 0) else: - dc_charge = np.ones_like(discharge_hours_bin) # Set DC charge to 0 if optimization is disabled - + dc_charge = np.ones_like( + discharge_hours_bin + ) # Set DC charge to 0 if optimization is disabled # Create discharge array: Only consider value 1 (Discharge), set the rest to 0 (binary output) discharge = np.where(discharge_hours_bin == 1, 1, 0) return ac_charge, dc_charge, discharge - - # Custom mutation function that applies type-specific mutations def mutate(self, individual): """ @@ -92,40 +93,42 @@ def mutate(self, individual): Returns: - (tuple): The mutated individual as a tuple (required by DEAP). """ - + # Step 1: Mutate the charge/discharge states (idle, discharge, AC charge, DC charge) # Extract the relevant part of the individual for prediction hours, which represents the charge/discharge behavior. - charge_discharge_part = individual[:self.prediction_hours] - + charge_discharge_part = individual[: self.prediction_hours] + # Apply the mutation to the charge/discharge part - charge_discharge_mutated, = self.toolbox.mutate_charge_discharge(charge_discharge_part) - + (charge_discharge_mutated,) = self.toolbox.mutate_charge_discharge(charge_discharge_part) + # Ensure that no invalid states are introduced during mutation (valid values: 0-8) if self.optimize_dc_charge: charge_discharge_mutated = np.clip(charge_discharge_mutated, 0, 8) else: charge_discharge_mutated = np.clip(charge_discharge_mutated, 0, 6) - + # Use split_charge_discharge to split the mutated array into AC charge, DC charge, and discharge components - #ac_charge, dc_charge, discharge = self.split_charge_discharge(charge_discharge_mutated) - + # ac_charge, dc_charge, discharge = self.split_charge_discharge(charge_discharge_mutated) + # Optionally: You can process the split arrays further if needed, for example, # applying additional constraints or penalties, or keeping track of charging limits. # Reassign the mutated values back to the individual - individual[:self.prediction_hours] = charge_discharge_mutated + individual[: self.prediction_hours] = charge_discharge_mutated # Step 2: Mutate EV charging schedule if enabled if self.optimize_ev: # Extract the relevant part for EV charging schedule ev_charge_part = individual[self.prediction_hours : self.prediction_hours * 2] - + # Apply mutation on the EV charging schedule - ev_charge_part_mutated, = self.toolbox.mutate_ev_charge_index(ev_charge_part) - + (ev_charge_part_mutated,) = self.toolbox.mutate_ev_charge_index(ev_charge_part) + # Ensure the EV does not charge during fixed hours (set those hours to 0) - ev_charge_part_mutated[self.prediction_hours - self.fixed_eauto_hours :] = [0] * self.fixed_eauto_hours - + ev_charge_part_mutated[self.prediction_hours - self.fixed_eauto_hours :] = [ + 0 + ] * self.fixed_eauto_hours + # Reassign the mutated EV charging part back to the individual individual[self.prediction_hours : self.prediction_hours * 2] = ev_charge_part_mutated @@ -133,24 +136,27 @@ def mutate(self, individual): if self.opti_param["haushaltsgeraete"] > 0: # Extract the appliance part (typically a single value for the start hour) appliance_part = [individual[-1]] - + # Apply mutation on the appliance start hour - appliance_part_mutated, = self.toolbox.mutate_hour(appliance_part) - + (appliance_part_mutated,) = self.toolbox.mutate_hour(appliance_part) + # Reassign the mutated appliance part back to the individual individual[-1] = appliance_part_mutated[0] return (individual,) - # Method to create an individual based on the conditions def create_individual(self): # Start with discharge states for the individual - individual_components = [self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours)] + individual_components = [ + self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours) + ] # Add EV charge index values if optimize_ev is True if self.optimize_ev: - individual_components += [self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours)] + individual_components += [ + self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours) + ] # Add the start time of the household appliance if it's being optimized if self.opti_param["haushaltsgeraete"] > 0: @@ -171,7 +177,7 @@ def split_individual( discharge_hours_bin = individual[: self.prediction_hours] eautocharge_hours_float = ( individual[self.prediction_hours : self.prediction_hours * 2] - if self.optimize_ev + if self.optimize_ev else None ) @@ -200,30 +206,41 @@ def setup_deap_environment(self, opti_param: Dict[str, Any], start_hour: int) -> # Initialize toolbox with attributes and operations self.toolbox = base.Toolbox() if self.optimize_dc_charge: - self.toolbox.register("attr_discharge_state", random.randint, 0,8) + self.toolbox.register("attr_discharge_state", random.randint, 0, 8) else: - self.toolbox.register("attr_discharge_state", random.randint, 0,6) + self.toolbox.register("attr_discharge_state", random.randint, 0, 6) if self.optimize_ev: - self.toolbox.register("attr_ev_charge_index", random.randint, 0, len(possible_ev_charge_currents) - 1) + self.toolbox.register( + "attr_ev_charge_index", random.randint, 0, len(possible_ev_charge_currents) - 1 + ) self.toolbox.register("attr_int", random.randint, start_hour, 23) - # Register individual creation function self.toolbox.register("individual", self.create_individual) # Register population, mating, mutation, and selection functions self.toolbox.register("population", tools.initRepeat, list, self.toolbox.individual) self.toolbox.register("mate", tools.cxTwoPoint) - #self.toolbox.register("mutate", tools.mutFlipBit, indpb=0.1) + # self.toolbox.register("mutate", tools.mutFlipBit, indpb=0.1) # Register separate mutation functions for each type of value: # - Discharge state mutation (-5, 0, 1) if self.optimize_dc_charge: - self.toolbox.register("mutate_charge_discharge", tools.mutUniformInt, low=0, up=8, indpb=0.2) + self.toolbox.register( + "mutate_charge_discharge", tools.mutUniformInt, low=0, up=8, indpb=0.2 + ) else: - self.toolbox.register("mutate_charge_discharge", tools.mutUniformInt, low=0, up=6, indpb=0.2) + self.toolbox.register( + "mutate_charge_discharge", tools.mutUniformInt, low=0, up=6, indpb=0.2 + ) # - Float mutation for EV charging values - self.toolbox.register("mutate_ev_charge_index", tools.mutUniformInt, low=0, up=len(possible_ev_charge_currents) - 1, indpb=0.2) + self.toolbox.register( + "mutate_ev_charge_index", + tools.mutUniformInt, + low=0, + up=len(possible_ev_charge_currents) - 1, + indpb=0.2, + ) # - Start hour mutation for household devices self.toolbox.register("mutate_hour", tools.mutUniformInt, low=start_hour, up=23, indpb=0.2) @@ -246,8 +263,7 @@ def evaluate_inner( if self.opti_param.get("haushaltsgeraete", 0) > 0: ems.set_haushaltsgeraet_start(spuelstart_int, global_start_hour=start_hour) - ac,dc,discharge = self.decode_charge_discharge(discharge_hours_bin) - + ac, dc, discharge = self.decode_charge_discharge(discharge_hours_bin) ems.set_akku_discharge_hours(discharge) # Set DC charge hours only if DC optimization is enabled @@ -258,10 +274,10 @@ def evaluate_inner( if self.optimize_ev: eautocharge_hours_float = [ possible_ev_charge_currents[i] for i in eautocharge_hours_index - ] + ] ems.set_ev_charge_hours(eautocharge_hours_float) else: - ems.set_ev_charge_hours(np.full(self.prediction_hours, 0 )) + ems.set_ev_charge_hours(np.full(self.prediction_hours, 0)) return ems.simuliere(start_hour) def evaluate( @@ -279,16 +295,16 @@ def evaluate( o = self.evaluate_inner(individual, ems, start_hour) except Exception as e: return (100000.0,) # Return a high penalty in case of an exception - + gesamtbilanz = o["Gesamtbilanz_Euro"] * (-1.0 if worst_case else 1.0) - + discharge_hours_bin, eautocharge_hours_float, _ = self.split_individual(individual) # Small Penalty for not discharging gesamtbilanz += sum( 0.01 for i in range(self.prediction_hours) if discharge_hours_bin[i] == 0.0 ) - + # 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: # gesamtbilanz += sum( @@ -302,17 +318,16 @@ def evaluate( ) # 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) + # print(ems.akku.aktueller_energieinhalt()," * ", parameter["preis_euro_pro_wh_akku"] , " ", restwert_akku, " ", gesamtbilanz) gesamtbilanz += -restwert_akku - #print(gesamtbilanz) + # print(gesamtbilanz) if self.optimize_ev: - gesamtbilanz += max( - 0, - (parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()) * self.strafe, - ) - + gesamtbilanz += max( + 0, + (parameter["eauto_min_soc"] - ems.eauto.ladezustand_in_prozent()) * self.strafe, + ) return (gesamtbilanz,) @@ -333,7 +348,7 @@ def optimize( for _ in range(3): population.insert(0, creator.Individual(start_solution)) - #Run the evolutionary algorithm + # Run the evolutionary algorithm algorithms.eaMuPlusLambda( population, self.toolbox, @@ -384,7 +399,7 @@ def optimierung_ems( 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 parameter["eauto_min_soc"] - parameter["eauto_soc"] < 0: self.optimize_ev = False eauto = PVAkku( @@ -426,7 +441,7 @@ def optimierung_ems( "evaluate", lambda ind: self.evaluate(ind, ems, parameter, start_hour, worst_case), ) - start_solution, extra_data = self.optimize(parameter["start_solution"], ngen=ngen) # + start_solution, extra_data = self.optimize(parameter["start_solution"], ngen=ngen) # # Perform final evaluation on the best solution o = self.evaluate_inner(start_solution, ems, start_hour) @@ -434,7 +449,9 @@ def optimierung_ems( start_solution ) if self.optimize_ev: - eautocharge_hours_float = [possible_ev_charge_currents[i] for i in eautocharge_hours_float] + eautocharge_hours_float = [ + possible_ev_charge_currents[i] for i in eautocharge_hours_float + ] ac_charge, dc_charge, discharge = self.decode_charge_discharge(discharge_hours_bin) # Visualize the results @@ -472,7 +489,7 @@ def optimierung_ems( element_list = o[key].tolist() # Change the first value to None - #element_list[0] = None + # element_list[0] = None # Change the NaN to None (JSON) element_list = [ None if isinstance(x, (int, float)) and np.isnan(x) else x for x in element_list @@ -484,8 +501,8 @@ def optimierung_ems( # Return final results as a dictionary return { "ac_charge": ac_charge.tolist(), - "dc_charge":dc_charge.tolist(), - "discharge_allowed":discharge.tolist(), + "dc_charge": dc_charge.tolist(), + "discharge_allowed": discharge.tolist(), "eautocharge_hours_float": eautocharge_hours_float, "result": o, "eauto_obj": ems.eauto.to_dict(), @@ -493,4 +510,3 @@ def optimierung_ems( "spuelstart": spuelstart_int, "simulation_data": o, } - diff --git a/src/akkudoktoreos/class_soc_calc.py b/src/akkudoktoreos/class_soc_calc.py index f1d90ac..4b75f55 100644 --- a/src/akkudoktoreos/class_soc_calc.py +++ b/src/akkudoktoreos/class_soc_calc.py @@ -5,6 +5,7 @@ import numpy as np import pandas as pd + class BatteryDataProcessor: def __init__( self, @@ -84,7 +85,7 @@ def find_soc_points(self): condition_soc_0 = (self.data["battery_voltage"] <= self.voltage_low_threshold) & ( self.data["battery_current"].abs() <= self.current_low_threshold ) - + times_soc_100_all = self.data[condition_soc_100][ ["timestamp", "battery_voltage", "battery_current"] ] @@ -115,7 +116,10 @@ def calculate_resetting_soc(self, last_points_100_df, last_points_0_df): else: end_point = self.data.iloc[-1] # Verwenden des letzten Datensatzes als Endpunkt - if not last_points_100_df.empty and start_point["timestamp"] in last_points_100_df["timestamp"].values: + if ( + not last_points_100_df.empty + and start_point["timestamp"] in last_points_100_df["timestamp"].values + ): initial_soc = 100 elif start_point["timestamp"] in last_points_0_df["timestamp"].values: initial_soc = 0 @@ -235,7 +239,13 @@ def plot_data(self, last_points_100_df, last_points_0_df, soc_df): marker="o", label="100% SoC Points", ) - plt.scatter(last_points_0_df['timestamp'], last_points_0_df['battery_voltage'], color='red', marker='x', label='0% SoC Points') + plt.scatter( + last_points_0_df["timestamp"], + last_points_0_df["battery_voltage"], + color="red", + marker="x", + label="0% SoC Points", + ) plt.xlabel("Timestamp") plt.ylabel("Voltage (V)") plt.legend() @@ -256,7 +266,13 @@ def plot_data(self, last_points_100_df, last_points_0_df, soc_df): marker="o", label="100% SoC Points", ) - plt.scatter(last_points_0_df['timestamp'], last_points_0_df['battery_current'], color='red', marker='x', label='0% SoC Points') + plt.scatter( + last_points_0_df["timestamp"], + last_points_0_df["battery_current"], + color="red", + marker="x", + label="0% SoC Points", + ) plt.xlabel("Timestamp") plt.ylabel("Current (A)") plt.legend() @@ -284,10 +300,10 @@ def plot_data(self, last_points_100_df, last_points_0_df, soc_df): # MariaDB Verbindungsdetails config = { - 'user': 'soc', - 'password': 'Rayoflight123!', - 'host': '192.168.1.135', - 'database': 'sensor' + "user": "soc", + "password": "Rayoflight123!", + "host": "192.168.1.135", + "database": "sensor", } # Parameter festlegen @@ -295,7 +311,7 @@ def plot_data(self, last_points_100_df, last_points_0_df, soc_df): voltage_low_threshold = 48 # 0% SoC current_low_threshold = 2 # Niedriger Strom für beide Zustände gap = 30 # Zeitlücke in Minuten zum Gruppieren von Maxima/Minima - bat_capacity = 0.8*33 * 1000 / 48 + bat_capacity = 0.8 * 33 * 1000 / 48 # Zeitpunkt X definieren zeitpunkt_x = (datetime.now() - timedelta(weeks=4)).strftime("%Y-%m-%d %H:%M:%S") @@ -317,7 +333,7 @@ def plot_data(self, last_points_100_df, last_points_0_df, soc_df): last_points_100_df, last_points_0_df ) # soh_df = processor.calculate_soh(integration_results) - #processor.update_database_with_soc(soc_df) + # processor.update_database_with_soc(soc_df) processor.plot_data(last_points_100_df, last_points_0_df, soc_df) diff --git a/src/akkudoktoreos/class_strompreis.py b/src/akkudoktoreos/class_strompreis.py index 727fbe3..b951aba 100644 --- a/src/akkudoktoreos/class_strompreis.py +++ b/src/akkudoktoreos/class_strompreis.py @@ -22,20 +22,21 @@ def repeat_to_shape(array, target_shape): class HourlyElectricityPriceForecast: - def __init__(self, source, cache_dir="cache", charges=0.000228, prediction_hours=24, cache=True): # 228 + def __init__( + self, source, cache_dir="cache", charges=0.000228, prediction_hours=24, cache=True + ): # 228 self.cache_dir = cache_dir - self.cache=cache + self.cache = cache os.makedirs(self.cache_dir, exist_ok=True) self.cache_time_file = os.path.join(self.cache_dir, "cache_timestamp.txt") self.prices = self.load_data(source) self.charges = charges self.prediction_hours = prediction_hours - def load_data(self, source): cache_filename = self.get_cache_filename(source) if source.startswith("http"): - if os.path.exists(cache_filename) and not self.is_cache_expired() and self.cache==True: + if os.path.exists(cache_filename) and not self.is_cache_expired() and self.cache: print("Loading data from cache...") with open(cache_filename, "r") as file: json_data = json.load(file) diff --git a/src/akkudoktoreos/visualize.py b/src/akkudoktoreos/visualize.py index eab5feb..e656446 100644 --- a/src/akkudoktoreos/visualize.py +++ b/src/akkudoktoreos/visualize.py @@ -59,8 +59,6 @@ def visualisiere_ergebnisse( plt.grid(True) plt.legend() - - # PV forecast plt.subplot(3, 2, 3) plt.plot(hours, pv_forecast, label="PV Generation (Wh)", marker="x") @@ -111,19 +109,44 @@ def visualisiere_ergebnisse( plt.subplot(3, 2, 1) # Plot with transparency (alpha) and different linestyles plt.plot( - hours, ergebnisse["Last_Wh_pro_Stunde"], label="Load (Wh)", marker="o", linestyle="-", alpha=0.8 + hours, + ergebnisse["Last_Wh_pro_Stunde"], + label="Load (Wh)", + marker="o", + linestyle="-", + alpha=0.8, ) plt.plot( - hours, ergebnisse["Haushaltsgeraet_wh_pro_stunde"], label="Household Device (Wh)", marker="o", linestyle="--", alpha=0.8 + hours, + ergebnisse["Haushaltsgeraet_wh_pro_stunde"], + label="Household Device (Wh)", + marker="o", + linestyle="--", + alpha=0.8, ) plt.plot( - hours, ergebnisse["Netzeinspeisung_Wh_pro_Stunde"], label="Grid Feed-in (Wh)", marker="x", linestyle=":", alpha=0.8 + hours, + ergebnisse["Netzeinspeisung_Wh_pro_Stunde"], + label="Grid Feed-in (Wh)", + marker="x", + linestyle=":", + alpha=0.8, ) plt.plot( - hours, ergebnisse["Netzbezug_Wh_pro_Stunde"], label="Grid Consumption (Wh)", marker="^", linestyle="-.", alpha=0.8 + hours, + ergebnisse["Netzbezug_Wh_pro_Stunde"], + label="Grid Consumption (Wh)", + marker="^", + linestyle="-.", + alpha=0.8, ) plt.plot( - hours, ergebnisse["Verluste_Pro_Stunde"], label="Losses (Wh)", marker="^", linestyle="-", alpha=0.8 + hours, + ergebnisse["Verluste_Pro_Stunde"], + label="Losses (Wh)", + marker="^", + linestyle="-", + alpha=0.8, ) # Title and labels @@ -134,8 +157,6 @@ def visualisiere_ergebnisse( # Show legend with a higher number of columns to avoid overlap plt.legend(ncol=2) - - # Electricity prices hours_p = np.arange(0, len(strompreise)) plt.subplot(3, 2, 3) @@ -164,8 +185,6 @@ def visualisiere_ergebnisse( plt.legend(loc="upper left", bbox_to_anchor=(1, 1)) # Place legend outside the plot plt.grid(True, which="both", axis="x") # Grid for every hour - - # Plot for AC, DC charging, and Discharge status using bar charts ax1 = plt.subplot(3, 2, 5) hours = np.arange(0, prediction_hours) @@ -173,10 +192,20 @@ def visualisiere_ergebnisse( plt.bar(hours, ac, width=0.4, label="AC Charging (relative)", color="blue", alpha=0.6) # Plot DC charging as bars (relative values between 0 and 1) - plt.bar(hours + 0.4, dc, width=0.4, label="DC Charging (relative)", color="green", alpha=0.6) + plt.bar( + hours + 0.4, dc, width=0.4, label="DC Charging (relative)", color="green", alpha=0.6 + ) # Plot Discharge as bars (0 or 1, binary values) - plt.bar(hours, discharge, width=0.4, label="Discharge Allowed", color="red", alpha=0.6, bottom=np.maximum(ac, dc)) + plt.bar( + hours, + discharge, + width=0.4, + label="Discharge Allowed", + color="red", + alpha=0.6, + bottom=np.maximum(ac, dc), + ) # Configure the plot ax1.legend(loc="upper left") @@ -186,13 +215,11 @@ def visualisiere_ergebnisse( ax1.set_title("AC/DC Charging and Discharge Overview") ax1.grid(True) - if ist_dst_wechsel(datetime.datetime.now()): hours = np.arange(start_hour, prediction_hours - 1) else: hours = np.arange(start_hour, prediction_hours) - pdf.savefig() # Save the current figure state to the PDF plt.close() # Close the current figure to free up memory @@ -217,9 +244,17 @@ def visualisiere_ergebnisse( ) # Annotate costs for hour, value in enumerate(costs): - if value == None or np.isnan(value): - value=0 - axs[0].annotate(f'{value:.2f}', (hour, value), textcoords="offset points", xytext=(0,5), ha='center', fontsize=8, color='red') + if value is None or np.isnan(value): + value = 0 + axs[0].annotate( + f"{value:.2f}", + (hour, value), + textcoords="offset points", + xytext=(0, 5), + ha="center", + fontsize=8, + color="red", + ) # Plot revenues axs[0].plot( @@ -231,9 +266,17 @@ def visualisiere_ergebnisse( ) # Annotate revenues for hour, value in enumerate(revenues): - if value == None or np.isnan(value): - value=0 - axs[0].annotate(f'{value:.2f}', (hour, value), textcoords="offset points", xytext=(0,5), ha='center', fontsize=8, color='green') + if value is None or np.isnan(value): + value = 0 + axs[0].annotate( + f"{value:.2f}", + (hour, value), + textcoords="offset points", + xytext=(0, 5), + ha="center", + fontsize=8, + color="green", + ) # Title and labels axs[0].set_title("Financial Balance per Hour") @@ -242,8 +285,6 @@ def visualisiere_ergebnisse( axs[0].legend() axs[0].grid(True) - - # Summary of finances on the second axis (axs[1]) labels = ["Total Costs [€]", "Total Revenue [€]", "Total Balance [€]"] values = [total_costs, total_revenue, total_balance] diff --git a/src/akkudoktoreosserver/flask_server.py b/src/akkudoktoreosserver/flask_server.py index d8bd433..eeec9d7 100755 --- a/src/akkudoktoreosserver/flask_server.py +++ b/src/akkudoktoreosserver/flask_server.py @@ -15,9 +15,9 @@ 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_numpy_encoder import * from akkudoktoreos.class_strompreis import HourlyElectricityPriceForecast from akkudoktoreos.config import ( get_start_enddate, @@ -29,7 +29,10 @@ app = Flask(__name__) opt_class = optimization_problem( - prediction_hours=prediction_hours, strafe=10, optimization_hours=optimization_hours, verbose=True + prediction_hours=prediction_hours, + strafe=10, + optimization_hours=optimization_hours, + verbose=True, ) @@ -62,7 +65,7 @@ def flask_strompreis(): price_forecast = HourlyElectricityPriceForecast( source=f"https://api.akkudoktor.net/prices?start={date_now}&end={date}", prediction_hours=prediction_hours, - cache=False + cache=False, ) specific_date_prices = price_forecast.get_price_for_daterange( date_now, date @@ -250,7 +253,7 @@ def flask_optimize(): # Perform optimization simulation result = opt_class.optimierung_ems(parameter=parameter, start_hour=datetime.now().hour) - #print(result) + # print(result) # convert to JSON (None accepted by dumps) return NumpyEncoder.dumps(result)