Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-06-27 16:36:53 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

EV Charge Parameters optional + AC Charge first try (Parameter Reduction)

Browse Source
This commit is contained in:
Andreas 2024-10-14 10:10:12 +02:00 committed by Andreas
parent 6881710295
commit 2b5f0ee53c
5 changed files with 187 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/akkudoktoreos/class_akku.py
View File

@ -70,21 +70,32 @@ class PVAkku:
self.soc_wh = min(max(self.soc_wh, self.min_soc_wh), self.max_soc_wh)
self.discharge_array = np.full(self.hours, 1)
self.charge_array = np.full(self.hours, 1)
self.charge_array = np.full(self.hours, 0)
def set_discharge_per_hour(self, discharge_array):
assert len(discharge_array) == self.hours
self.discharge_array = np.array(discharge_array)
# Ensure no simultaneous charging and discharging in the same hour using NumPy mask
conflict_mask = (self.charge_array > 0) & (self.discharge_array > 0)
# Prioritize discharge by setting charge to 0 where both are > 0
self.charge_array[conflict_mask] = 0
def set_charge_per_hour(self, charge_array):
assert len(charge_array) == self.hours
self.charge_array = np.array(charge_array)
# Ensure no simultaneous charging and discharging in the same hour using NumPy mask
conflict_mask = (self.charge_array > 0) & (self.discharge_array > 0)
# Prioritize discharge by setting charge to 0 where both are > 0
self.charge_array[conflict_mask] = 0
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 and self.discharge_array[hour] == -1:
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

14
src/akkudoktoreos/class_ems.py
View File

@ -27,8 +27,10 @@ class EnergieManagementSystem:
def set_akku_discharge_hours(self, ds: List[int]) -> None:
self.akku.set_discharge_per_hour(ds)
def set_akku_charge_hours(self, ds: List[int]) -> None:
self.akku.set_charge_per_hour(ds)
def set_eauto_charge_hours(self, ds: List[int]) -> None:
self.eauto.set_charge_per_hour(ds)
def set_haushaltsgeraet_start(self, ds: List[int], global_start_hour: int = 0) -> None:
@ -69,7 +71,7 @@ class EnergieManagementSystem:
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 + 1, ende):
stunde_since_now = stunde - start_stunde
@ -88,6 +90,14 @@ class EnergieManagementSystem:
verluste_wh_pro_stunde[stunde_since_now] += verluste_eauto
eauto_soc_pro_stunde[stunde_since_now] = self.eauto.ladezustand_in_prozent()
# AC PV Battery Charge
if self.akku.charge_array[stunde] > 0.0:
#soc_pre = self.akku.ladezustand_in_prozent()
geladene_menge, verluste_wh = self.akku.energie_laden(None,stunde)
#print(self.akku.charge_array[stunde], " ",geladene_menge," ",soc_pre," ",self.akku.ladezustand_in_prozent())
verbrauch += geladene_menge
verluste_wh_pro_stunde[stunde_since_now] += verluste_wh
# Process inverter logic
erzeugung = self.pv_prognose_wh[stunde]
netzeinspeisung, netzbezug, verluste, eigenverbrauch = (

154
src/akkudoktoreos/class_optimize.py
View File

@ -8,7 +8,7 @@ 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.config import moegliche_ladestroeme_in_prozent
from akkudoktoreos.config import possible_ev_charge_currents
from akkudoktoreos.visualize import visualisiere_ergebnisse
@ -26,21 +26,47 @@ class optimization_problem:
self.strafe = strafe
self.opti_param = None
self.fixed_eauto_hours = prediction_hours - optimization_hours
self.possible_charge_values = moegliche_ladestroeme_in_prozent
self.possible_charge_values = possible_ev_charge_currents
self.verbose = verbose
self.fix_seed = fixed_seed
self.optimize_ev = True
# Set a fixed seed for random operations if provided
if fixed_seed is not None:
random.seed(fixed_seed)
def split_charge_discharge(self, discharge_hours_bin: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
"""
Split the input array `discharge_hours_bin` into two separate arrays:
- `charge`: Contains only the negative values from `discharge_hours_bin` (charging values).
- `discharge`: Contains only the positive values from `discharge_hours_bin` (discharging values).
Parameters:
- discharge_hours_bin (np.ndarray): Input array with both positive and negative values.
Returns:
- charge (np.ndarray): Array with negative values from `discharge_hours_bin`, other values set to 0.
- discharge (np.ndarray): Array with positive values from `discharge_hours_bin`, other values set to 0.
"""
# Convert the input list to a NumPy array, if it's not already
discharge_hours_bin = np.array(discharge_hours_bin)
# Create charge array: Keep only negative values, set the rest to 0
charge = -np.where(discharge_hours_bin < 0, discharge_hours_bin, 0)
charge = charge / np.max(charge)
# Create discharge array: Keep only positive values, set the rest to 0
discharge = np.where(discharge_hours_bin > 0, discharge_hours_bin, 0)
return charge, discharge
def split_individual(
self, individual: List[float]
) -> Tuple[List[int], List[float], Optional[int]]:
"""
Split the individual solution into its components:
1. Discharge hours (-1 (Charge),0 (Nothing),1 (Discharge)),
2. Electric vehicle charge hours (float),
2. Electric vehicle charge hours (possible_charge_values),
3. Dishwasher start time (integer if applicable).
"""
discharge_hours_bin = individual[: self.prediction_hours]
@ -69,40 +95,60 @@ class optimization_problem:
# Initialize toolbox with attributes and operations
self.toolbox = base.Toolbox()
self.toolbox.register("attr_discharge_state", random.randint, -1, 1)
self.toolbox.register("attr_ev_charge_index", random.randint, 0, len(moegliche_ladestroeme_in_prozent) - 1)
self.toolbox.register("attr_discharge_state", random.randint, -5, 1)
if self.optimize_ev:
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 method based on household appliance parameter
if opti_param["haushaltsgeraete"] > 0:
self.toolbox.register(
"individual",
lambda: creator.Individual(
[self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours)]
+ [self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours)]
+ [self.toolbox.attr_int()]
),
)
else:
self.toolbox.register(
"individual",
lambda: creator.Individual(
[self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours)]
+ [self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours)]
),
)
# Function to create an individual based on the conditions
def create_individual():
# Start with discharge states for the individual
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)]
# Add the start time of the household appliance if it's being optimized
if self.opti_param["haushaltsgeraete"] > 0:
individual_components += [self.toolbox.attr_int()]
return creator.Individual(individual_components)
# Register individual creation function
self.toolbox.register("individual", create_individual)
# # Register individual creation method based on household appliance parameter
# if opti_param["haushaltsgeraete"] > 0:
# self.toolbox.register(
# "individual",
# lambda: creator.Individual(
# [self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours)]
# + [self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours)]
# + [self.toolbox.attr_int()]
# ),
# )
# else:
# self.toolbox.register(
# "individual",
# lambda: creator.Individual(
# [self.toolbox.attr_discharge_state() for _ in range(self.prediction_hours)]
# + [self.toolbox.attr_ev_charge_index() for _ in range(self.prediction_hours)]
# ),
# )
# 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)
# Register separate mutation functions for each type of value:
# - Discharge state mutation (-1, 0, 1)
self.toolbox.register("mutate_discharge", tools.mutUniformInt, low=0, up=1, indpb=0.1)
# - Discharge state mutation (-5, 0, 1)
self.toolbox.register("mutate_discharge", tools.mutUniformInt, low=-5, up=1, indpb=0.1)
# - Float mutation for EV charging values
self.toolbox.register("mutate_ev_charge_index", tools.mutUniformInt, low=0, up=len(moegliche_ladestroeme_in_prozent) - 1, indpb=0.1)
self.toolbox.register("mutate_ev_charge_index", tools.mutUniformInt, low=0, up=len(possible_ev_charge_currents) - 1, indpb=0.1)
# - Start hour mutation for household devices
self.toolbox.register("mutate_hour", tools.mutUniformInt, low=start_hour, up=23, indpb=0.3)
self.toolbox.register("mutate_hour", tools.mutUniformInt, low=start_hour, up=23, indpb=0.1)
# Custom mutation function that applies type-specific mutations
def mutate(individual):
@ -111,13 +157,15 @@ class optimization_problem:
individual[:self.prediction_hours]
)
# Mutate the EV charging indices
ev_charge_part = individual[self.prediction_hours : self.prediction_hours * 2]
ev_charge_part_mutated, = self.toolbox.mutate_ev_charge_index(ev_charge_part)
individual[self.prediction_hours : self.prediction_hours * 2] = ev_charge_part_mutated
if self.optimize_ev:
# Mutate the EV charging indices
ev_charge_part = individual[self.prediction_hours : self.prediction_hours * 2]
ev_charge_part_mutated, = self.toolbox.mutate_ev_charge_index(ev_charge_part)
ev_charge_part_mutated[self.prediction_hours - self.fixed_eauto_hours :] = [0] * self.fixed_eauto_hours
individual[self.prediction_hours : self.prediction_hours * 2] = ev_charge_part_mutated
# Mutate the appliance start hour if present
if len(individual) > self.prediction_hours * 2:
if self.opti_param["haushaltsgeraete"] > 0:
appliance_part = [individual[-1]]
appliance_part_mutated, = self.toolbox.mutate_hour(appliance_part)
individual[-1] = appliance_part_mutated[0]
@ -145,17 +193,18 @@ class optimization_problem:
if self.opti_param.get("haushaltsgeraete", 0) > 0:
ems.set_haushaltsgeraet_start(spuelstart_int, global_start_hour=start_hour)
ems.set_akku_discharge_hours(discharge_hours_bin)
eautocharge_hours_index[self.prediction_hours - self.fixed_eauto_hours :] = [
0
] * self.fixed_eauto_hours
charge, discharge = self.split_charge_discharge(discharge_hours_bin)
ems.set_akku_discharge_hours(discharge)
ems.set_akku_charge_hours(charge)
#print(charge)
eautocharge_hours_float = [
moegliche_ladestroeme_in_prozent[i] for i in eautocharge_hours_index
possible_ev_charge_currents[i] for i in eautocharge_hours_index
]
ems.set_eauto_charge_hours(eautocharge_hours_float)
if self.optimize_ev:
ems.set_eauto_charge_hours(eautocharge_hours_float)
return ems.simuliere(start_hour)
def evaluate(
@ -177,7 +226,7 @@ class optimization_problem:
gesamtbilanz = o["Gesamtbilanz_Euro"] * (-1.0 if worst_case else 1.0)
discharge_hours_bin, eautocharge_hours_float, _ = self.split_individual(individual)
max_ladeleistung = np.max(moegliche_ladestroeme_in_prozent)
#max_ladeleistung = np.max(possible_ev_charge_currents)
# Small Penalty for not discharging
gesamtbilanz += sum(
@ -185,11 +234,11 @@ class optimization_problem:
)
# Penalty for charging the electric vehicle during restricted hours
gesamtbilanz += sum(
self.strafe
for i in range(self.prediction_hours - self.fixed_eauto_hours, self.prediction_hours)
if eautocharge_hours_float[i] != 0.0
)
# gesamtbilanz += sum(
# self.strafe
# for i in range(self.prediction_hours - self.fixed_eauto_hours, self.prediction_hours)
# if eautocharge_hours_float[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:
@ -232,14 +281,14 @@ class optimization_problem:
for _ in range(3):
population.insert(0, creator.Individual(start_solution))
# Run the evolutionary algorithm
#Run the evolutionary algorithm
algorithms.eaMuPlusLambda(
population,
self.toolbox,
mu=100,
lambda_=200,
cxpb=0.7,
mutpb=0.3,
lambda_=150,
cxpb=0.5,
mutpb=0.5,
ngen=ngen,
stats=stats,
halloffame=hof,
@ -282,6 +331,10 @@ class optimization_problem:
)
akku.set_charge_per_hour(np.full(self.prediction_hours, 1))
self.optimize_ev = True
if parameter["eauto_min_soc"] - parameter["eauto_soc"] <0:
self.optimize_ev = False
eauto = PVAkku(
kapazitaet_wh=parameter["eauto_cap"],
hours=self.prediction_hours,
@ -382,3 +435,4 @@ class optimization_problem:
"spuelstart": spuelstart_int,
"simulation_data": o,
}

12
src/akkudoktoreos/config.py
View File

@ -5,18 +5,18 @@ output_dir = "output"
prediction_hours = 48
optimization_hours = 24
strafe = 10
moegliche_ladestroeme_in_prozent = [
possible_ev_charge_currents = [
0.0,
6.0 / 16.0,
7.0 / 16.0,
#7.0 / 16.0,
8.0 / 16.0,
9.0 / 16.0,
#9.0 / 16.0,
10.0 / 16.0,
11.0 / 16.0,
#11.0 / 16.0,
12.0 / 16.0,
13.0 / 16.0,
#13.0 / 16.0,
14.0 / 16.0,
15.0 / 16.0,
#15.0 / 16.0,
1.0,
]

69
src/akkudoktoreos/visualize.py
View File

@ -160,26 +160,40 @@ def visualisiere_ergebnisse(
plt.grid(True, which="both", axis="x") # Grid for every hour
ax1 = plt.subplot(3, 2, 3)
# Plot für die discharge_hours-Werte
for hour, value in enumerate(discharge_hours):
# Festlegen der Farbe und des Labels basierend auf dem Wert
if value > 0: # Positive Werte (Entladung)
color = "red"
label = "Discharge" if hour == 0 else "" # Label nur beim ersten Eintrag hinzufügen
elif value < 0: # Negative Werte (Ladung)
color = "blue"
label = "Charge" if hour == 0 else ""
else:
continue # Überspringe 0-Werte
# Erstellen der Farbbereiche mit `axvspan`
ax1.axvspan(
hour,
hour + 1,
color="red",
ymax=value,
hour, # Start der Stunde
hour + 1, # Ende der Stunde
ymin=0, # Untere Grenze
ymax=abs(value), # Obere Grenze: abs(value), um die Höhe richtig darzustellen
color=color,
alpha=0.3,
label="Discharge Possibility" if hour == 0 else "",
label=label
)
for hour, value in enumerate(laden_moeglich):
ax1.axvspan(
hour,
hour + 1,
color="green",
ymax=value,
alpha=0.3,
label="Charging Possibility" if hour == 0 else "",
# Annotieren der Werte in der Mitte des Farbbereichs
ax1.text(
hour + 0.5, # In der Mitte des Bereichs
abs(value) / 2, # In der Mitte der Höhe
f'{value:.2f}', # Wert mit zwei Dezimalstellen
ha='center',
va='center',
fontsize=8,
color='black'
)
ax1.legend(loc="upper left")
ax1.set_xlim(0, prediction_hours)
pdf.savefig() # Save the current figure state to the PDF
plt.close() # Close the current figure to free up memory
@ -192,26 +206,47 @@ def visualisiere_ergebnisse(
losses = ergebnisse["Gesamt_Verluste"]
# Costs and revenues per hour on the first axis (axs[0])
costs = ergebnisse["Kosten_Euro_pro_Stunde"]
revenues = ergebnisse["Einnahmen_Euro_pro_Stunde"]
# Plot costs
axs[0].plot(
hours,
ergebnisse["Kosten_Euro_pro_Stunde"],
costs,
label="Costs (Euro)",
marker="o",
color="red",
)
# Annotate costs
for hour, value in enumerate(costs):
print(hour, " ", value)
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')
# Plot revenues
axs[0].plot(
hours,
ergebnisse["Einnahmen_Euro_pro_Stunde"],
revenues,
label="Revenue (Euro)",
marker="x",
color="green",
)
# 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')
# Title and labels
axs[0].set_title("Financial Balance per Hour")
axs[0].set_xlabel("Hour")
axs[0].set_ylabel("Euro")
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]