mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-09-11 10:26:38 +00:00
The isolated pre-commit mypy hook previously omitted runtime type information that make mypy used, hiding errors involving dependencies such as Pydantic and Pendulum. Makefile, pre-commit and CI now run the same full-project typing policy in the development environment defined by uv.lock. - Use uv run --locked --exact --extra dev and the same mypy arguments for Makefile and the local hook. Check all of src and tests, including on configuration-only changes. - Pin Python 3.13 for local development and the pre-commit CI job, and install the locked pre-commit version in CI. - Disable incremental analysis because existing Pendulum cache state changes mypy 2.3.1 diagnostics. Document the policy, the performance tradeoff and the existing typing debt. - Add a regression test that exercises Makefile, the hook and the CI command in a temporary project, accepting valid dependency types and detecting deliberate Pydantic/Pendulum assignment errors. Resolve the newly detected mypy diagnostics. - Enable the numpydantic and Pydantic mypy plugins, retaining strict Pydantic constructor typing with init_typed = true. Validate raw/coercible payloads through model_validate. - Propagate concrete record, provider and time-window types through generic collections, factories and lookup methods. Preserve runtime field inspection and generated time-window documentation. - Align Pendulum annotations with actual factory/arithmetic results while retaining Pydantic validation adapters at runtime. Correct optional values, array boundaries, REST handlers and plotting interfaces. - Add pinned scipy-stubs and types-psutil, update uv.lock, and supply the plugins' dependencies. - Add runtime regression coverage for validated path defaults, normalized time-series metadata, generic field inspection, invalid timestamps and unsupported provider imports. Runtime and compatibility details: - Validate path defaults as Path objects while retaining raw string defaults needed by migration serialization with exclude_defaults. - Normalize feed-in tariff lists and default charge rates to NumPy arrays; reject missing timestamps/uninitialized values explicitly. Importing into a provider without import support returns HTTP 400. - Public JSON schemas and OpenAPI structure match main (excluding the generated version). Signed-off-by: dr-dimitry Signed-off-by: dr-dimitry Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com> Co-authored-by: dr-dimitri <87113560+dr-dimitri@users.noreply.github.com> Co-authored-by: Normann <github@koldrack.com>
821 lines
30 KiB
Python
821 lines
30 KiB
Python
import json
|
|
import textwrap
|
|
from collections.abc import Sequence
|
|
from io import BytesIO
|
|
from typing import Callable, Optional, Union
|
|
|
|
import matplotlib
|
|
import matplotlib.dates as mdates
|
|
import matplotlib.pyplot as plt
|
|
import numpy as np
|
|
import pendulum
|
|
from matplotlib.backends.backend_pdf import PdfPages
|
|
|
|
from akkudoktoreos.core.coreabc import ConfigMixin, get_ems
|
|
from akkudoktoreos.optimization.genetic.geneticsolution import GeneticSolution
|
|
from akkudoktoreos.utils.datetimeutil import DateTime, to_datetime
|
|
|
|
matplotlib.use(
|
|
"Agg"
|
|
) # non-interactive backend that can only write to files, backend needed to stay in main thread.
|
|
|
|
|
|
debug_visualize: bool = False
|
|
|
|
|
|
class GeneticVisualizationReport(ConfigMixin):
|
|
def __init__(
|
|
self,
|
|
version: str = "0.0.1",
|
|
) -> None:
|
|
# Initialize the report with empty groups
|
|
self.groups: list[list[Callable[[], None]]] = [] # Store groups of charts
|
|
self.current_group: list[
|
|
Callable[[], None]
|
|
] = [] # Store current group of charts being created
|
|
|
|
# file like buffer to create the PDF
|
|
self._buffer = BytesIO()
|
|
self.pdf_pages: PdfPages | None = None
|
|
|
|
self.version = version # overwrite version as test for constant output of pdf for test
|
|
self.current_time = to_datetime(
|
|
as_string="YYYY-MM-DD HH:mm:ss", in_timezone=self.config.general.timezone
|
|
)
|
|
|
|
def add_chart_to_group(self, chart_func: Callable[[], None], title: str | None) -> None:
|
|
"""Add a chart function to the current group and save it as a PNG and SVG."""
|
|
self.current_group.append(chart_func)
|
|
|
|
def finalize_group(self) -> None:
|
|
"""Finalize the current group and prepare for a new group."""
|
|
if self.current_group: # Check if current group has charts
|
|
self.groups.append(self.current_group) # Add current group to groups
|
|
else:
|
|
print("Finalizing an empty group!") # Warn if group is empty
|
|
self.current_group = [] # Reset current group for new charts
|
|
|
|
def _initialize_pdf(self) -> None:
|
|
"""Initialize the PDF."""
|
|
self._buffer = BytesIO()
|
|
self.pdf_pages = PdfPages(self._buffer, metadata={}) # Initialize PdfPages without metadata
|
|
|
|
def _save_group_to_pdf(self, group: list[Callable[[], None]]) -> None:
|
|
"""Save a group of charts to the PDF."""
|
|
if self.pdf_pages is None: # make mypy happy
|
|
raise RuntimeError("Report not initialized")
|
|
|
|
fig_count = len(group) # Number of charts in the group
|
|
|
|
if fig_count == 0:
|
|
print("Attempted to save an empty group to PDF!")
|
|
return
|
|
|
|
# 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:
|
|
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:
|
|
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))
|
|
|
|
# Add footer text with current time to each page
|
|
if self.version == "test":
|
|
current_time = "test"
|
|
else:
|
|
current_time = self.current_time
|
|
fig.text(
|
|
0.5,
|
|
0.02,
|
|
f"Generated on: {current_time} with version: {self.version}",
|
|
ha="center",
|
|
va="center",
|
|
fontsize=10,
|
|
)
|
|
|
|
# Render each chart in its corresponding axis
|
|
for idx, chart_func in enumerate(group):
|
|
plt.sca(axs[idx]) # Set current axis
|
|
chart_func() # Render the chart
|
|
|
|
# Save the figure to the PDF and clean up
|
|
for idx in range(fig_count, len(axs)):
|
|
axs[idx].set_visible(False)
|
|
|
|
self.pdf_pages.savefig(fig) # Save the figure to the PDF
|
|
plt.close(fig)
|
|
|
|
def create_line_chart_date(
|
|
self,
|
|
start_date: DateTime,
|
|
y_list: list[Union[np.ndarray, list[Optional[float]], list[float]]],
|
|
ylabel: str,
|
|
xlabel: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
labels: Optional[list[str]] = None,
|
|
markers: Optional[list[str]] = None,
|
|
line_styles: Optional[list[str]] = None,
|
|
x2label: Optional[Union[str, None]] = "Hours Since Start",
|
|
) -> None:
|
|
"""Create a line chart and add it to the current group."""
|
|
|
|
def chart() -> None:
|
|
timestamps = [
|
|
start_date.add(hours=i) for i in range(len(y_list[0]))
|
|
] # 840 timestamps at 1-hour intervals
|
|
|
|
for idx, y_data in enumerate(y_list):
|
|
label = labels[idx] if labels else None # Chart label
|
|
marker = markers[idx] if markers and idx < len(markers) else "o" # Marker style
|
|
line_style = line_styles[idx] if line_styles and idx < len(line_styles) else "-"
|
|
plt.plot(
|
|
mdates.date2num(timestamps),
|
|
np.asarray(y_data, dtype=float),
|
|
label=label,
|
|
marker=marker,
|
|
linestyle=line_style,
|
|
) # Plot line
|
|
|
|
# Format the time axis
|
|
plt.gca().xaxis.set_major_formatter(
|
|
mdates.DateFormatter("%Y-%m-%d", tz=self.config.general.timezone)
|
|
) # Show date and time
|
|
plt.gca().xaxis.set_major_locator(
|
|
mdates.DayLocator(interval=1, tz=self.config.general.timezone)
|
|
) # Major ticks every day
|
|
plt.gca().xaxis.set_minor_locator(
|
|
mdates.HourLocator(interval=2, tz=self.config.general.timezone)
|
|
)
|
|
# Minor ticks every 6 hours
|
|
plt.gca().xaxis.set_minor_formatter(
|
|
mdates.DateFormatter("%H", tz=self.config.general.timezone)
|
|
)
|
|
# plt.gcf().autofmt_xdate(rotation=45, which="major")
|
|
# Auto-format the x-axis for readability
|
|
|
|
# Move major tick labels further down to avoid collision with minor tick labels
|
|
for plt_label in plt.gca().get_xticklabels(which="major"):
|
|
plt_label.set_y(-0.04)
|
|
|
|
# Add labels, title, and legend
|
|
if xlabel:
|
|
plt.xlabel(xlabel)
|
|
plt.ylabel(ylabel)
|
|
if title:
|
|
plt.title(title)
|
|
if labels:
|
|
plt.legend()
|
|
plt.grid(True)
|
|
|
|
# Add vertical line for the current date if within the axis range
|
|
current_time = pendulum.now(self.config.general.timezone)
|
|
if timestamps[0].subtract(hours=2) <= current_time <= timestamps[-1]:
|
|
plt.axvline(mdates.date2num(current_time), color="r", linestyle="--", label="Now")
|
|
plt.text(
|
|
mdates.date2num(current_time),
|
|
plt.ylim()[1],
|
|
"Now",
|
|
color="r",
|
|
ha="center",
|
|
va="bottom",
|
|
)
|
|
|
|
# Add a second x-axis on top
|
|
ax1 = plt.gca()
|
|
ax2 = ax1.twiny()
|
|
ax2.set_xlim(ax1.get_xlim()) # Align the second axis with the first
|
|
|
|
# Generate integer hour labels
|
|
hours_since_start = [(t - timestamps[0]).total_seconds() / 3600 for t in timestamps]
|
|
# ax2.set_xticks(timestamps[::48]) # Set ticks every 12 hours
|
|
# ax2.set_xticklabels([f"{int(h)}" for h in hours_since_start[::48]])
|
|
# ax2.set_xticks(timestamps[:: len(timestamps) // 24]) # Select 10 evenly spaced ticks
|
|
ax2.set_xticks(
|
|
mdates.date2num(timestamps[:: len(timestamps) // 12])
|
|
) # Select 10 evenly spaced ticks
|
|
# ax2.set_xticklabels([f"{int(h)}" for h in hours_since_start[:: len(timestamps) // 24]])
|
|
ax2.set_xticklabels([f"{int(h)}" for h in hours_since_start[:: len(timestamps) // 12]])
|
|
if x2label:
|
|
ax2.set_xlabel(x2label)
|
|
|
|
# Ensure ax1 and ax2 are aligned
|
|
# assert ax1.get_xlim() == ax2.get_xlim(), "ax1 and ax2 are not aligned"
|
|
|
|
self.add_chart_to_group(chart, title) # Add chart function to current group
|
|
|
|
def create_line_chart(
|
|
self,
|
|
start_hour: Optional[int],
|
|
y_list: list[Union[np.ndarray, list[Optional[float]], list[float]]],
|
|
title: str,
|
|
xlabel: str,
|
|
ylabel: str,
|
|
labels: Optional[list[str]] = None,
|
|
markers: Optional[list[str]] = None,
|
|
line_styles: Optional[list[str]] = None,
|
|
) -> None:
|
|
"""Create a line chart and add it to the current group."""
|
|
|
|
def chart() -> None:
|
|
nonlocal start_hour # Allow modifying `x` within the nested function
|
|
if start_hour is None:
|
|
start_hour = 0
|
|
first_element = y_list[0]
|
|
x: np.ndarray
|
|
# Case 1: y_list contains np.ndarray elements
|
|
if isinstance(first_element, np.ndarray):
|
|
x = np.arange(
|
|
start_hour, start_hour + len(first_element)
|
|
) # Start at x and extend by ndarray length
|
|
# Case 2: y_list contains float elements (1D list)
|
|
elif isinstance(first_element, float):
|
|
x = np.arange(
|
|
start_hour, start_hour + len(y_list)
|
|
) # Start at x and extend by list length
|
|
# Case 3: y_list is a nested list of floats
|
|
elif isinstance(first_element, list) and all(
|
|
isinstance(i, float) for i in first_element
|
|
):
|
|
max_len = max(len(sublist) for sublist in y_list)
|
|
x = np.arange(
|
|
start_hour, start_hour + max_len
|
|
) # Start at x and extend by max sublist length
|
|
else:
|
|
print(f"Unsupported y_list structure: {type(y_list)}, {y_list}")
|
|
raise TypeError(
|
|
"y_list elements must be np.ndarray, float, or a nested list of floats"
|
|
)
|
|
|
|
for idx, y_data in enumerate(y_list):
|
|
label = labels[idx] if labels else None # Chart label
|
|
marker = markers[idx] if markers and idx < len(markers) else "o" # Marker style
|
|
line_style = (
|
|
line_styles[idx] if line_styles and idx < len(line_styles) else "-"
|
|
) # Line style
|
|
plt.plot(
|
|
x,
|
|
np.asarray(y_data, dtype=float),
|
|
label=label,
|
|
marker=marker,
|
|
linestyle=line_style,
|
|
) # Plot line
|
|
|
|
plt.title(title) # Set title
|
|
plt.xlabel(xlabel) # Set x-axis label
|
|
plt.ylabel(ylabel) # Set y-axis label
|
|
if labels:
|
|
plt.legend() # Show legend if labels are provided
|
|
plt.grid(True) # Show grid
|
|
plt.xlim(x[0] - 0.5, x[-1] + 0.5) # Adjust x-limits
|
|
|
|
self.add_chart_to_group(chart, title) # Add chart function to current group
|
|
|
|
def create_scatter_plot(
|
|
self,
|
|
x: np.ndarray,
|
|
y: np.ndarray,
|
|
title: str,
|
|
xlabel: str,
|
|
ylabel: str,
|
|
c: Optional[np.ndarray] = None,
|
|
) -> None:
|
|
"""Create a scatter plot and add it to the current group."""
|
|
|
|
def chart() -> None:
|
|
scatter = plt.scatter(x, y, c=c, cmap="viridis") # Create scatter plot
|
|
plt.title(title) # Set title
|
|
plt.xlabel(xlabel) # Set x-axis label
|
|
plt.ylabel(ylabel) # Set y-axis label
|
|
if c is not None:
|
|
plt.colorbar(scatter, label="Constraint") # Add colorbar if color data is provided
|
|
plt.grid(True) # Show grid
|
|
|
|
self.add_chart_to_group(chart, title) # Add chart function to current group
|
|
|
|
def create_bar_chart(
|
|
self,
|
|
labels: list[str],
|
|
values_list: Sequence[Sequence[int | float]],
|
|
title: str,
|
|
ylabel: str,
|
|
xlabels: Optional[list[str]] = None,
|
|
label_names: Optional[list[str]] = None,
|
|
colors: Optional[list[str]] = None,
|
|
bar_width: float = 0.35,
|
|
bottom: Optional[int] = None,
|
|
) -> None:
|
|
"""Create a bar chart and add it to the current group."""
|
|
|
|
def chart() -> None:
|
|
num_groups = len(values_list) # Number of data groups
|
|
num_bars = len(labels) # Number of bars (categories)
|
|
# Calculate the positions for each bar group on the x-axis
|
|
x = np.arange(num_bars) # x positions for bars
|
|
offset = np.linspace(
|
|
-bar_width * (num_groups - 1) / 2, bar_width * (num_groups - 1) / 2, num_groups
|
|
) # Bar offsets
|
|
for i, values in enumerate(values_list):
|
|
bottom_use = None
|
|
if bottom == i + 1: # Set bottom if specified
|
|
bottom_use = 1
|
|
color = colors[i] if colors and i < len(colors) else None # Bar color
|
|
label_name = label_names[i] if label_names else None # Bar label
|
|
plt.bar(
|
|
x + offset[i],
|
|
values,
|
|
bar_width,
|
|
label=label_name,
|
|
color=color,
|
|
zorder=2,
|
|
alpha=0.6,
|
|
bottom=bottom_use,
|
|
) # Create bar
|
|
if xlabels:
|
|
plt.xticks(x, labels) # Add custom labels to the x-axis
|
|
plt.title(title) # Set title
|
|
plt.ylabel(ylabel) # Set y-axis label
|
|
|
|
if colors and label_names:
|
|
plt.legend() # Show legend if colors are provided
|
|
plt.grid(True, zorder=0) # Show grid in the background
|
|
plt.xlim(-0.5, len(labels) - 0.5) # Set x-axis limits
|
|
|
|
self.add_chart_to_group(chart, title) # Add chart function to current group
|
|
|
|
def create_violin_plot(
|
|
self, data_list: list[np.ndarray], labels: list[str], title: str, xlabel: str, ylabel: str
|
|
) -> None:
|
|
"""Create a violin plot and add it to the current group."""
|
|
|
|
def chart() -> None:
|
|
plt.violinplot(data_list, showmeans=True, showmedians=True) # Create violin plot
|
|
plt.xticks(np.arange(1, len(labels) + 1), labels) # Set x-ticks and labels
|
|
plt.title(title) # Set title
|
|
plt.xlabel(xlabel) # Set x-axis label
|
|
plt.ylabel(ylabel) # Set y-axis label
|
|
plt.grid(True) # Show grid
|
|
|
|
self.add_chart_to_group(chart, title) # 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:
|
|
if self.pdf_pages is None: # make mypy happy
|
|
raise RuntimeError("Report not initialized")
|
|
|
|
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, title) # 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:
|
|
if self.pdf_pages is None: # make mypy happy
|
|
raise RuntimeError("Report not initialized")
|
|
|
|
# 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, title) # Treat the JSON page as a "chart" in the group
|
|
|
|
def generate_pdf(self) -> bytes:
|
|
"""Generate the PDF report with all the added chart groups.
|
|
|
|
Returns:
|
|
PDF report as bytes object
|
|
"""
|
|
self._initialize_pdf() # Initialize the PDF
|
|
|
|
buffer = BytesIO()
|
|
self.pdf_pages = PdfPages(buffer, metadata={})
|
|
|
|
for group in self.groups:
|
|
self._save_group_to_pdf(group)
|
|
|
|
self.pdf_pages.close()
|
|
return buffer.getvalue()
|
|
|
|
|
|
def genetic_prepare_visualize(
|
|
solution: GeneticSolution,
|
|
) -> bytes:
|
|
global debug_visualize
|
|
|
|
report = GeneticVisualizationReport()
|
|
next_full_hour_date = get_ems().start_datetime
|
|
start_hour = solution.start_hour
|
|
# Group 1:
|
|
report.create_line_chart_date(
|
|
next_full_hour_date,
|
|
[
|
|
solution.parameters.ems.total_load[start_hour:],
|
|
],
|
|
title="Load Profile",
|
|
# xlabel="Hours", # not enough space
|
|
ylabel="Load (Wh)",
|
|
labels=["Total Load (Wh)"],
|
|
)
|
|
report.create_line_chart_date(
|
|
next_full_hour_date,
|
|
[
|
|
solution.parameters.ems.pv_forecast_wh[start_hour:],
|
|
],
|
|
title="PV Forecast",
|
|
# xlabel="Hours", # not enough space
|
|
ylabel="PV Generation (Wh)",
|
|
)
|
|
|
|
report.create_line_chart_date(
|
|
next_full_hour_date,
|
|
[
|
|
np.full(
|
|
len(solution.parameters.ems.total_load) - start_hour,
|
|
solution.parameters.ems.feed_in_tariff_per_wh[start_hour:]
|
|
if isinstance(solution.parameters.ems.feed_in_tariff_per_wh, list)
|
|
else solution.parameters.ems.feed_in_tariff_per_wh,
|
|
)
|
|
],
|
|
title="Remuneration",
|
|
# xlabel="Hours", # not enough space
|
|
ylabel="amount/Wh",
|
|
x2label=None, # not enough space
|
|
)
|
|
if solution.parameters.temperature_forecast:
|
|
report.create_line_chart_date(
|
|
next_full_hour_date,
|
|
[
|
|
solution.parameters.temperature_forecast[start_hour:],
|
|
],
|
|
title="Temperature Forecast",
|
|
# xlabel="Hours", # not enough space
|
|
ylabel="°C",
|
|
x2label=None, # not enough space
|
|
)
|
|
report.finalize_group()
|
|
|
|
# Group 2:
|
|
report.create_line_chart_date(
|
|
start_date=next_full_hour_date, # start_date
|
|
y_list=[
|
|
solution.result.load_wh_per_hour,
|
|
solution.result.home_appliance_wh_per_hour,
|
|
solution.result.grid_feed_in_wh_per_hour,
|
|
solution.result.grid_consumption_wh_per_hour,
|
|
solution.result.losses_per_hour,
|
|
],
|
|
title="Energy Flow per Hour",
|
|
# xlabel="Date", # not enough space
|
|
ylabel="Energy (Wh)",
|
|
labels=[
|
|
"Load (Wh)",
|
|
"Household Device (Wh)",
|
|
"Grid Feed-in (Wh)",
|
|
"Grid Consumption (Wh)",
|
|
"Losses (Wh)",
|
|
],
|
|
markers=["o", "o", "x", "^", "^"],
|
|
line_styles=["-", "--", ":", "-.", "-"],
|
|
)
|
|
report.finalize_group()
|
|
|
|
# Group 3:
|
|
report.create_line_chart_date(
|
|
start_date=next_full_hour_date,
|
|
y_list=[solution.result.battery_soc_per_hour, solution.result.ev_soc_per_hour],
|
|
title="Battery SOC",
|
|
# xlabel="Date", # not enough space
|
|
ylabel="%",
|
|
labels=[
|
|
"Battery SOC (%)",
|
|
"Electric Vehicle SOC (%)",
|
|
],
|
|
markers=["o", "x"],
|
|
)
|
|
report.create_line_chart_date(
|
|
start_date=next_full_hour_date, # start_date
|
|
y_list=[solution.parameters.ems.electricity_price_per_wh[start_hour:]],
|
|
# title="Electricity Price", # not enough space
|
|
# xlabel="Date", # not enough space
|
|
ylabel="Electricity Price (amount/Wh)",
|
|
x2label=None, # not enough space
|
|
)
|
|
|
|
labels = list(
|
|
item
|
|
for sublist in zip(
|
|
list(str(i) for i in range(0, 23, 2)), list(str(" ") for i in range(0, 23, 2))
|
|
)
|
|
for item in sublist
|
|
)
|
|
labels = labels[start_hour:] + labels
|
|
|
|
report.create_bar_chart(
|
|
labels=labels,
|
|
values_list=[
|
|
solution.ac_charge[start_hour:],
|
|
solution.dc_charge[start_hour:],
|
|
solution.discharge_allowed[start_hour:],
|
|
],
|
|
title="AC/DC Charging and Discharge Overview",
|
|
ylabel="Relative Power (0-1) / Discharge (0 or 1)",
|
|
label_names=["AC Charging (relative)", "DC Charging (relative)", "Discharge Allowed"],
|
|
colors=["blue", "green", "red"],
|
|
bottom=3,
|
|
xlabels=labels,
|
|
)
|
|
report.finalize_group()
|
|
|
|
# Group 4:
|
|
|
|
report.create_line_chart_date(
|
|
next_full_hour_date, # start_date
|
|
[
|
|
solution.result.costs_per_hour,
|
|
solution.result.revenue_per_hour,
|
|
],
|
|
title="Financial Balance per Hour",
|
|
# xlabel="Date", # not enough space
|
|
ylabel="Amount",
|
|
labels=["Costs", "Revenue"],
|
|
)
|
|
|
|
extra_data = solution.extra_data
|
|
if extra_data:
|
|
report.create_scatter_plot(
|
|
x=np.array(extra_data["losses"]),
|
|
y=np.array(extra_data["balance"]),
|
|
title="Scatter Plot",
|
|
xlabel="losses",
|
|
ylabel="balance",
|
|
c=np.array(extra_data["constraints"]),
|
|
)
|
|
|
|
values_list = [
|
|
[
|
|
solution.result.total_costs,
|
|
solution.result.total_revenue,
|
|
solution.result.total_balance,
|
|
]
|
|
]
|
|
labels = ["Total Costs [amount]", "Total Revenue [amount]", "Total Balance [amount]"]
|
|
|
|
report.create_bar_chart(
|
|
labels=labels,
|
|
values_list=values_list,
|
|
title="Financial Overview",
|
|
ylabel="Amount",
|
|
xlabels=["Total Costs [amount]", "Total Revenue [amount]", "Total Balance [amount]"],
|
|
)
|
|
|
|
report.finalize_group()
|
|
|
|
# Group 1: Scatter plot of losses vs balance with color-coded constraints
|
|
if extra_data:
|
|
f1 = np.array(extra_data["losses"]) # Losses
|
|
f2 = np.array(extra_data["balance"]) # Balance
|
|
n1 = np.array(extra_data["constraints"]) # Constraints
|
|
|
|
# Filter data where 'constraints' < 0.01
|
|
filtered_indices = n1 < 0.01
|
|
filtered_losses = f1[filtered_indices]
|
|
filtered_balance = f2[filtered_indices]
|
|
|
|
# Group 2: Violin plot for filtered losses
|
|
if filtered_losses.size > 0:
|
|
report.create_violin_plot(
|
|
data_list=[filtered_losses], # Data for filtered losses
|
|
labels=["Filtered Losses"], # Label for the violin plot
|
|
title="Violin Plot for Filtered Losses (Constraint < 0.01)",
|
|
xlabel="Losses",
|
|
ylabel="Values",
|
|
)
|
|
else:
|
|
print("No data available for filtered losses violin plot (Constraint < 0.01)")
|
|
|
|
# Group 3: Violin plot for filtered balance
|
|
if extra_data:
|
|
if filtered_balance.size > 0:
|
|
report.create_violin_plot(
|
|
data_list=[filtered_balance], # Data for filtered balance
|
|
labels=["Filtered Balance"], # Label for the violin plot
|
|
title="Violin Plot for Filtered Balance (Constraint < 0.01)",
|
|
xlabel="Balance",
|
|
ylabel="Values",
|
|
)
|
|
else:
|
|
print("No data available for filtered balance violin plot (Constraint < 0.01)")
|
|
|
|
if filtered_balance.size > 0 or filtered_losses.size > 0:
|
|
report.finalize_group()
|
|
|
|
if solution.fitness_history:
|
|
if debug_visualize or solution.fixed_seed:
|
|
report.create_line_chart(
|
|
start_hour=0,
|
|
y_list=[
|
|
np.array(solution.fitness_history["avg"]),
|
|
np.array(solution.fitness_history["max"]),
|
|
np.array(solution.fitness_history["min"]),
|
|
],
|
|
title=f"DEBUG: Generation Fitness for seed {solution.fixed_seed}",
|
|
xlabel="Generation",
|
|
ylabel="Fitness",
|
|
labels=[
|
|
"avg",
|
|
"max",
|
|
"min",
|
|
],
|
|
markers=[".", ".", "."],
|
|
)
|
|
report.finalize_group()
|
|
|
|
# Generate the PDF report
|
|
pdf = report.generate_pdf()
|
|
|
|
# Return pdf (a bytes object)
|
|
return pdf
|
|
|
|
|
|
def genetic_generate_example_report(filename: str = "example_report.pdf") -> None:
|
|
"""Generate example visualization report."""
|
|
global debug_visualize
|
|
|
|
report = GeneticVisualizationReport("test")
|
|
x_hours = 0 # Define x-axis start values (e.g., hours)
|
|
|
|
# Group 1: Adding charts to be displayed on the same page
|
|
report.create_line_chart(
|
|
x_hours,
|
|
[np.array([10, 20, 30, 40])],
|
|
title="Load Profile",
|
|
xlabel="Hours",
|
|
ylabel="Load (Wh)",
|
|
)
|
|
report.create_line_chart(
|
|
x_hours,
|
|
[np.array([5, 15, 25, 35])],
|
|
title="PV Forecast",
|
|
xlabel="Hours",
|
|
ylabel="PV Generation (Wh)",
|
|
)
|
|
report.create_line_chart(
|
|
x_hours,
|
|
[np.array([5, 15, 25, 35])],
|
|
title="PV Forecast",
|
|
xlabel="Hours",
|
|
ylabel="PV Generation (Wh)",
|
|
)
|
|
# Note: If there are only 3 charts per page, the first is as wide as the page
|
|
|
|
report.finalize_group() # Finalize the first group of charts
|
|
|
|
# Group 2: Adding more charts to be displayed on another page
|
|
report.create_line_chart(
|
|
x_hours,
|
|
[np.array([0.2, 0.25, 0.3, 0.35])],
|
|
title="Electricity Price",
|
|
xlabel="Hours",
|
|
ylabel="Price (€/Wh)",
|
|
)
|
|
report.create_bar_chart(
|
|
["Costs", "Revenue", "Balance"],
|
|
[[500.0], [600.0], [100.0]],
|
|
title="Financial Overview",
|
|
ylabel="Euro",
|
|
label_names=["AC Charging (relative)", "DC Charging (relative)", "Discharge Allowed"],
|
|
colors=["red", "green", "blue"],
|
|
)
|
|
report.create_scatter_plot(
|
|
np.array([5, 6, 7, 8]),
|
|
np.array([100, 200, 150, 250]),
|
|
title="Scatter Plot",
|
|
xlabel="Losses",
|
|
ylabel="Balance",
|
|
c=np.array([0.1, 0.2, 0.3, 0.4]),
|
|
)
|
|
report.finalize_group() # Finalize the second group of charts
|
|
|
|
# Group 3: Adding a violin plot
|
|
data = [np.random.normal(0, std, 100) for std in range(1, 5)] # Example data for violin plot
|
|
report.create_violin_plot(
|
|
data,
|
|
labels=["Group 1", "Group 2", "Group 3", "Group 4"],
|
|
title="Violin Plot",
|
|
xlabel="Groups",
|
|
ylabel="Values",
|
|
)
|
|
data = [np.random.normal(0, 1, 100)] # Example data for violin plot
|
|
report.create_violin_plot(
|
|
data, labels=["Group 1"], title="Violin Plot", xlabel="Group", ylabel="Values"
|
|
)
|
|
|
|
report.finalize_group() # Finalize the third group of charts
|
|
|
|
debug_visualize = True # set level for example report
|
|
|
|
if debug_visualize:
|
|
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()
|
|
|
|
report.create_line_chart_date(
|
|
pendulum.now().subtract(hours=0),
|
|
[list(np.random.random(840))],
|
|
title="test",
|
|
xlabel="test",
|
|
ylabel="test",
|
|
)
|
|
report.finalize_group()
|
|
# Generate the PDF report (a bytes object)
|
|
pdf = report.generate_pdf()
|
|
|
|
# write to file
|
|
with open(filename, "wb") as f:
|
|
f.write(pdf)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
genetic_generate_example_report()
|