diff --git a/single_test_optimization.py b/single_test_optimization.py index 83a99d7..ac3fd09 100644 --- a/single_test_optimization.py +++ b/single_test_optimization.py @@ -10,12 +10,15 @@ import numpy as np from akkudoktoreos.config.config import get_config from akkudoktoreos.core.ems import get_ems +from akkudoktoreos.core.logging import get_logger from akkudoktoreos.optimization.genetic import ( OptimizationParameters, optimization_problem, ) from akkudoktoreos.prediction.prediction import get_prediction +get_logger(__name__, logging_level="DEBUG") + def prepare_optimization_real_parameters() -> OptimizationParameters: """Prepare and return optimization parameters with real world data. diff --git a/src/akkudoktoreos/optimization/genetic.py b/src/akkudoktoreos/optimization/genetic.py index a05f849..8847f0f 100644 --- a/src/akkudoktoreos/optimization/genetic.py +++ b/src/akkudoktoreos/optimization/genetic.py @@ -1,3 +1,4 @@ +import logging import random import time from pathlib import Path @@ -14,6 +15,7 @@ from akkudoktoreos.core.coreabc import ( EnergyManagementSystemMixin, ) from akkudoktoreos.core.ems import EnergieManagementSystemParameters, SimulationResult +from akkudoktoreos.core.logging import get_logger from akkudoktoreos.devices.battery import ( Battery, ElectricVehicleParameters, @@ -25,6 +27,8 @@ from akkudoktoreos.devices.inverter import Inverter, InverterParameters from akkudoktoreos.prediction.interpolator import SelfConsumptionPropabilityInterpolator from akkudoktoreos.utils.utils import NumpyEncoder +logger = get_logger(__name__) + class OptimizationParameters(BaseModel): ems: EnergieManagementSystemParameters @@ -113,10 +117,14 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi self.fix_seed = fixed_seed self.optimize_ev = True self.optimize_dc_charge = False + self.fitness_history: dict[str, Any] = {} - # Set a fixed seed for random operations if provided - if fixed_seed is not None: - random.seed(fixed_seed) + # Set a fixed seed for random operations if provided or in debug mode + if self.fix_seed is not None: + random.seed(self.fix_seed) + elif logger.level == logging.DEBUG: + self.fix_seed = random.randint(1, 100000000000) + random.seed(self.fix_seed) def decode_charge_discharge( self, discharge_hours_bin: np.ndarray @@ -493,6 +501,8 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi hof = tools.HallOfFame(1) stats = tools.Statistics(lambda ind: ind.fitness.values) stats.register("min", np.min) + stats.register("avg", np.mean) + stats.register("max", np.max) if self.verbose: print("Start optimize:", start_solution) @@ -503,7 +513,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi population.insert(0, creator.Individual(start_solution)) # Run the evolutionary algorithm - algorithms.eaMuPlusLambda( + pop, log = algorithms.eaMuPlusLambda( population, self.toolbox, mu=100, @@ -516,6 +526,14 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi verbose=self.verbose, ) + # Store fitness history + self.fitness_history = { + "gen": log.select("gen"), # Generation numbers (X-axis) + "avg": log.select("avg"), # Average fitness for each generation (Y-axis) + "max": log.select("max"), # Maximum fitness for each generation (Y-axis) + "min": log.select("min"), # Minimum fitness for each generation (Y-axis) + } + member: dict[str, list[float]] = {"bilanz": [], "verluste": [], "nebenbedingung": []} for ind in population: if hasattr(ind, "extra_data"): @@ -627,6 +645,8 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi "start_solution": start_solution, "spuelstart": washingstart_int, "extra_data": extra_data, + "fitness_history": self.fitness_history, + "fixed_seed": self.fix_seed, } from akkudoktoreos.utils.visualize import prepare_visualize diff --git a/src/akkudoktoreos/utils/visualize.py b/src/akkudoktoreos/utils/visualize.py index 2ab17a6..d4b4768 100644 --- a/src/akkudoktoreos/utils/visualize.py +++ b/src/akkudoktoreos/utils/visualize.py @@ -1,4 +1,7 @@ +import json +import logging import os +import textwrap from collections.abc import Sequence from typing import Callable, Optional, Union @@ -7,8 +10,11 @@ import numpy as np from matplotlib.backends.backend_pdf import PdfPages from akkudoktoreos.core.coreabc import ConfigMixin +from akkudoktoreos.core.logging import get_logger from akkudoktoreos.optimization.genetic import OptimizationParameters +logger = get_logger(__name__) + class VisualizationReport(ConfigMixin): def __init__(self, filename: str = "visualization_results.pdf") -> None: @@ -50,40 +56,42 @@ class VisualizationReport(ConfigMixin): def _save_group_to_pdf(self, group: list[Callable[[], None]]) -> None: """Save a group of charts to the PDF.""" fig_count = len(group) # Number of charts in the group + if fig_count == 0: - print("Attempted to save an empty group to PDF!") # Warn if group is empty - return # Prevent saving an empty group + print("Attempted to save an empty group to PDF!") + return - # Create a figure layout based on the number of charts + # Check for special charts before creating layout + special_keywords = {"add_text_page", "add_json_page"} + for chart_func in group: + if any(keyword in chart_func.__qualname__ for keyword in special_keywords): + chart_func() # Special chart functions handle their own rendering + return + + # Create layout only if no special charts are detected if fig_count == 3: - # Layout for three charts: 1 full-width on top, 2 below - fig = plt.figure(figsize=(14, 10)) # Set a larger figure size - ax1 = fig.add_subplot(2, 1, 1) # Full-width subplot - ax2 = fig.add_subplot(2, 2, 3) # Bottom left subplot - ax3 = fig.add_subplot(2, 2, 4) # Bottom right subplot - - # Store axes in a list for easy access + fig = plt.figure(figsize=(14, 10)) + ax1 = fig.add_subplot(2, 1, 1) + ax2 = fig.add_subplot(2, 2, 3) + ax3 = fig.add_subplot(2, 2, 4) axs = [ax1, ax2, ax3] else: - # Dynamic layout for any other number of charts - cols = 2 if fig_count > 1 else 1 # Determine number of columns - rows = (fig_count // 2) + (fig_count % 2) # Calculate required rows - fig, axs = plt.subplots(rows, cols, figsize=(14, 7 * rows)) # Create subplots - # If axs is a 2D array of axes, flatten it into a 1D list - # if isinstance(axs, np.ndarray): + cols = 2 if fig_count > 1 else 1 + rows = (fig_count + 1) // 2 + fig, axs = plt.subplots(rows, cols, figsize=(14, 7 * rows)) axs = list(np.array(axs).reshape(-1)) - # Draw each chart in the corresponding axes + # Render each chart in its corresponding axis for idx, chart_func in enumerate(group): - plt.sca(axs[idx]) # Set current axes - chart_func() # Call the chart function to draw + plt.sca(axs[idx]) # Set current axis + chart_func() # Render the chart - # Hide any unused axes + # Save the figure to the PDF and clean up for idx in range(fig_count, len(axs)): - axs[idx].set_visible(False) # Hide unused axes - self.pdf_pages.savefig(fig) # Save the figure to the PDF + axs[idx].set_visible(False) - plt.close(fig) # Close the figure to free up memory + self.pdf_pages.savefig(fig) # Save the figure to the PDF + plt.close(fig) def create_line_chart( self, @@ -232,6 +240,63 @@ class VisualizationReport(ConfigMixin): self.add_chart_to_group(chart) # Add chart function to current group + def add_text_page(self, text: str, title: Optional[str] = None, fontsize: int = 12) -> None: + """Add a page with text content to the PDF.""" + + def chart() -> None: + fig = plt.figure(figsize=(8.5, 11)) # Create a standard page size + plt.axis("off") # Turn off axes for a clean page + wrapped_text = textwrap.fill(text, width=80) # Wrap text to fit the page width + y = 0.95 # Start at the top of the page + + if title: + plt.text(0.5, y, title, ha="center", va="top", fontsize=fontsize + 4, weight="bold") + y -= 0.05 # Add space after the title + + plt.text(0.5, y, wrapped_text, ha="center", va="top", fontsize=fontsize, wrap=True) + self.pdf_pages.savefig(fig) # Save the figure as a page in the PDF + plt.close(fig) # Close the figure to free up memory + + self.add_chart_to_group(chart) # Treat the text page as a "chart" in the group + + def add_json_page( + self, json_obj: dict, title: Optional[str] = None, fontsize: int = 12 + ) -> None: + """Add a page with a formatted JSON object to the PDF. + + Args: + json_obj (dict): The JSON object to display. + title (Optional[str]): An optional title for the page. + fontsize (int): The font size for the JSON text. + """ + + def chart() -> None: + # Convert JSON object to a formatted string + json_str = json.dumps(json_obj, indent=4) + + fig = plt.figure(figsize=(8.5, 11)) # Standard page size + plt.axis("off") # Turn off axes for a clean page + + y = 0.95 # Start at the top of the page + if title: + plt.text(0.5, y, title, ha="center", va="top", fontsize=fontsize + 4, weight="bold") + y -= 0.05 # Add space after the title + + # Split the JSON string into lines and render them + lines = json_str.splitlines() + for line in lines: + plt.text(0.05, y, line, ha="left", va="top", fontsize=fontsize, family="monospace") + y -= 0.02 # Move down for the next line + + # Stop if the text exceeds the page + if y < 0.05: + break + + self.pdf_pages.savefig(fig) # Save the figure as a page in the PDF + plt.close(fig) # Close the figure to free up memory + + self.add_chart_to_group(chart) # Treat the JSON page as a "chart" in the group + def generate_pdf(self) -> None: """Generate the PDF report with all the added chart groups.""" self._initialize_pdf() # Initialize the PDF @@ -366,7 +431,6 @@ def prepare_visualize( c=extra_data["nebenbedingung"], ) - # Example usage values_list = [ [ results["result"]["Gesamtkosten_Euro"], @@ -422,7 +486,25 @@ def prepare_visualize( if filtered_balance.size > 0 or filtered_losses.size > 0: report.finalize_group() - + if logger.level == logging.DEBUG or results["fixed_seed"]: + report.create_line_chart( + 0, + [ + results["fitness_history"]["avg"], + results["fitness_history"]["max"], + results["fitness_history"]["min"], + ], + title=f"DEBUG: Generation Fitness for seed {results['fixed_seed']}", + xlabel="Generation", + ylabel="Fitness", + labels=[ + "avg", + "max", + "min", + ], + markers=[".", ".", "."], + ) + report.finalize_group() # Generate the PDF report report.generate_pdf() @@ -500,6 +582,41 @@ def generate_example_report(filename: str = "example_report.pdf") -> None: report.finalize_group() # Finalize the third group of charts + logger.setLevel(logging.DEBUG) # set level for example report + + if logger.level == logging.DEBUG: + report.create_line_chart( + x_hours, + [np.array([0.2, 0.25, 0.3, 0.35])], + title="DEBUG", + xlabel="DEBUG", + ylabel="DEBUG", + ) + report.finalize_group() # Finalize the third group of charts + + report.add_text_page( + text=" Bisher passierte folgendes:" + "Am Anfang wurde das Universum erschaffen." + "Das machte viele Leute sehr wütend und wurde allent-" + "halben als Schritt in die falsche Richtung angesehen...", + title="Don't Panic!", + fontsize=14, + ) + report.finalize_group() + + sample_json = { + "name": "Visualization Report", + "version": 1.0, + "charts": [ + {"type": "line", "data_points": 50}, + {"type": "bar", "categories": 10}, + ], + "metadata": {"author": "AI Assistant", "date": "2025-01-11"}, + } + + report.add_json_page(json_obj=sample_json, title="Formatted JSON Data", fontsize=10) + report.finalize_group() + # Generate the PDF report report.generate_pdf()