Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-09-20 10:41:14 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Improve Configuration and Prediction Usability (#220)

Browse Source 
* Update utilities in utils submodule.
* Add base configuration modules.
* Add server base configuration modules.
* Add devices base configuration modules.
* Add optimization base configuration modules.
* Add utils base configuration modules.
* Add prediction abstract and base classes plus tests.
* Add PV forecast to prediction submodule.
   The PV forecast modules are adapted from the class_pvforecast module and
   replace it.
* Add weather forecast to prediction submodule.
   The modules provide classes and methods to retrieve, manage, and process weather forecast data
   from various sources. Includes are structured representations of weather data and utilities
   for fetching forecasts for specific locations and time ranges.
   BrightSky and ClearOutside are currently supported.
* Add electricity price forecast to prediction submodule.
* Adapt fastapi server to base config and add fasthtml server.
* Add ems to core submodule.
* Adapt genetic to config.
* Adapt visualize to config.
* Adapt common test fixtures to config.
* Add load forecast to prediction submodule.
* Add core abstract and base classes.
* Adapt single test optimization to config.
* Adapt devices to config.

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
This commit is contained in:
Bobby Noelte
2024-12-15 14:40:03 +01:00
committed by GitHub
parent a5e637ab4c
commit aa334d0b61
80 changed files with 29048 additions and 2451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
src/akkudoktoreos/core/coreabc.py Normal file
View File

@@ -0,0 +1,275 @@
"""Abstract and base classes for EOS core.
This module provides foundational classes for handling configuration and prediction functionality
in EOS. It includes base classes that provide convenient access to global
configuration and prediction instances through properties.
Classes:
- ConfigMixin: Mixin class for managing and accessing global configuration.
- PredictionMixin: Mixin class for managing and accessing global prediction data.
- SingletonMixin: Mixin class to create singletons.
"""
import threading
from typing import Any, ClassVar, Dict, Optional, Type
from pendulum import DateTime
from pydantic import computed_field
from akkudoktoreos.utils.logutil import get_logger
logger = get_logger(__name__)
config_eos: Any = None
prediction_eos: Any = None
devices_eos: Any = None
ems_eos: Any = None
class ConfigMixin:
"""Mixin class for managing EOS configuration data.
This class serves as a foundational component for EOS-related classes requiring access
to the global EOS configuration. It provides a `config` property that dynamically retrieves
the configuration instance, ensuring up-to-date access to configuration settings.
Usage:
Subclass this base class to gain access to the `config` attribute, which retrieves the
global configuration instance lazily to avoid import-time circular dependencies.
Attributes:
config (ConfigEOS): Property to access the global EOS configuration.
Example:
```python
class MyEOSClass(ConfigMixin):
def my_method(self):
if self.config.myconfigval:
```
"""
@property
def config(self) -> Any:
"""Convenience method/ attribute to retrieve the EOS onfiguration data.
Returns:
ConfigEOS: The configuration.
"""
# avoid circular dependency at import time
global config_eos
if config_eos is None:
from akkudoktoreos.config.config import get_config
config_eos = get_config()
return config_eos
class PredictionMixin:
"""Mixin class for managing EOS prediction data.
This class serves as a foundational component for EOS-related classes requiring access
to global prediction data. It provides a `prediction` property that dynamically retrieves
the prediction instance, ensuring up-to-date access to prediction results.
Usage:
Subclass this base class to gain access to the `prediction` attribute, which retrieves the
global prediction instance lazily to avoid import-time circular dependencies.
Attributes:
prediction (Prediction): Property to access the global EOS prediction data.
Example:
```python
class MyOptimizationClass(PredictionMixin):
def analyze_myprediction(self):
prediction_data = self.prediction.mypredictionresult
# Perform analysis
```
"""
@property
def prediction(self) -> Any:
"""Convenience method/ attribute to retrieve the EOS prediction data.
Returns:
Prediction: The prediction.
"""
# avoid circular dependency at import time
global prediction_eos
if prediction_eos is None:
from akkudoktoreos.prediction.prediction import get_prediction
prediction_eos = get_prediction()
return prediction_eos
class DevicesMixin:
"""Mixin class for managing EOS devices simulation data.
This class serves as a foundational component for EOS-related classes requiring access
to global devices simulation data. It provides a `devices` property that dynamically retrieves
the devices instance, ensuring up-to-date access to devices simulation results.
Usage:
Subclass this base class to gain access to the `devices` attribute, which retrieves the
global devices instance lazily to avoid import-time circular dependencies.
Attributes:
devices (Devices): Property to access the global EOS devices simulation data.
Example:
```python
class MyOptimizationClass(DevicesMixin):
def analyze_mydevicesimulation(self):
device_simulation_data = self.devices.mydevicesresult
# Perform analysis
```
"""
@property
def devices(self) -> Any:
"""Convenience method/ attribute to retrieve the EOS devices simulation data.
Returns:
Devices: The devices simulation.
"""
# avoid circular dependency at import time
global devices_eos
if devices_eos is None:
from akkudoktoreos.devices.devices import get_devices
devices_eos = get_devices()
return devices_eos
class EnergyManagementSystemMixin:
"""Mixin class for managing EOS energy management system.
This class serves as a foundational component for EOS-related classes requiring access
to global energy management system. It provides a `ems` property that dynamically retrieves
the energy management system instance, ensuring up-to-date access to energy management system
control.
Usage:
Subclass this base class to gain access to the `ems` attribute, which retrieves the
global EnergyManagementSystem instance lazily to avoid import-time circular dependencies.
Attributes:
ems (EnergyManagementSystem): Property to access the global EOS energy management system.
Example:
```python
class MyOptimizationClass(EnergyManagementSystemMixin):
def analyze_myprediction(self):
ems_data = self.ems.the_ems_method()
# Perform analysis
```
"""
@property
def ems(self) -> Any:
"""Convenience method/ attribute to retrieve the EOS energy management system.
Returns:
EnergyManagementSystem: The energy management system.
"""
# avoid circular dependency at import time
global ems_eos
if ems_eos is None:
from akkudoktoreos.core.ems import get_ems
ems_eos = get_ems()
return ems_eos
class StartMixin(EnergyManagementSystemMixin):
"""A mixin to manage the start datetime for energy management.
Provides property:
- `start_datetime`: The starting datetime of the current or latest energy management.
"""
# Computed field for start_datetime
@computed_field # type: ignore[prop-decorator]
@property
def start_datetime(self) -> Optional[DateTime]:
"""Returns the start datetime of the current or latest energy management.
Returns:
DateTime: The starting datetime of the current or latest energy management, or None.
"""
return self.ems.start_datetime
class SingletonMixin:
"""A thread-safe singleton mixin class.
Ensures that only one instance of the derived class is created, even when accessed from multiple
threads. This mixin is intended to be combined with other classes, such as Pydantic models,
to make them singletons.
Attributes:
_instances (Dict[Type, Any]): A dictionary holding instances of each singleton class.
_lock (threading.Lock): A lock to synchronize access to singleton instance creation.
Usage:
- Inherit from `SingletonMixin` alongside other classes to make them singletons.
- Avoid using `__init__` to reinitialize the singleton instance after it has been created.
Example:
class MySingletonModel(SingletonMixin, PydanticBaseModel):
name: str
instance1 = MySingletonModel(name="Instance 1")
instance2 = MySingletonModel(name="Instance 2")
assert instance1 is instance2 # True
print(instance1.name) # Output: "Instance 1"
"""
_lock: ClassVar[threading.Lock] = threading.Lock()
_instances: ClassVar[Dict[Type, Any]] = {}
def __new__(cls: Type["SingletonMixin"], *args: Any, **kwargs: Any) -> "SingletonMixin":
"""Creates or returns the singleton instance of the class.
Ensures thread-safe instance creation by locking during the first instantiation.
Args:
*args: Positional arguments for instance creation (ignored if instance exists).
**kwargs: Keyword arguments for instance creation (ignored if instance exists).
Returns:
SingletonMixin: The singleton instance of the derived class.
"""
if cls not in cls._instances:
with cls._lock:
if cls not in cls._instances:
instance = super().__new__(cls)
cls._instances[cls] = instance
return cls._instances[cls]
@classmethod
def reset_instance(cls) -> None:
"""Resets the singleton instance, forcing it to be recreated on next access."""
with cls._lock:
if cls in cls._instances:
del cls._instances[cls]
logger.debug(f"{cls.__name__} singleton instance has been reset.")
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initializes the singleton instance if it has not been initialized previously.
Further calls to `__init__` are ignored for the singleton instance.
Args:
*args: Positional arguments for initialization.
**kwargs: Keyword arguments for initialization.
"""
if not hasattr(self, "_initialized"):
super().__init__(*args, **kwargs)
self._initialized = True

1307
src/akkudoktoreos/core/dataabc.py Normal file
View File

File diff suppressed because it is too large Load Diff

424
src/akkudoktoreos/core/ems.py Normal file
View File

@@ -0,0 +1,424 @@
from typing import Any, ClassVar, Dict, Optional, Union
import numpy as np
from numpydantic import NDArray, Shape
from pendulum import DateTime
from pydantic import ConfigDict, Field, computed_field, field_validator, model_validator
from typing_extensions import Self
from akkudoktoreos.core.coreabc import ConfigMixin, PredictionMixin, SingletonMixin
from akkudoktoreos.core.pydantic import PydanticBaseModel
from akkudoktoreos.devices.battery import PVAkku
from akkudoktoreos.devices.generic import HomeAppliance
from akkudoktoreos.devices.inverter import Wechselrichter
from akkudoktoreos.utils.datetimeutil import to_datetime
from akkudoktoreos.utils.logutil import get_logger
from akkudoktoreos.utils.utils import NumpyEncoder
logger = get_logger(__name__)
class EnergieManagementSystemParameters(PydanticBaseModel):
pv_prognose_wh: list[float] = Field(
description="An array of floats representing the forecasted photovoltaic output in watts for different time intervals."
)
strompreis_euro_pro_wh: list[float] = Field(
description="An array of floats representing the electricity price in euros per watt-hour for different time intervals."
)
einspeiseverguetung_euro_pro_wh: list[float] | float = Field(
description="A float or array of floats representing the feed-in compensation in euros per watt-hour."
)
preis_euro_pro_wh_akku: float = Field(
description="A float representing the cost of battery energy per watt-hour."
)
gesamtlast: list[float] = Field(
description="An array of floats representing the total load (consumption) in watts for different time intervals."
)
@model_validator(mode="after")
def validate_list_length(self) -> Self:
pv_prognose_length = len(self.pv_prognose_wh)
if (
pv_prognose_length != len(self.strompreis_euro_pro_wh)
or pv_prognose_length != len(self.gesamtlast)
or (
isinstance(self.einspeiseverguetung_euro_pro_wh, list)
and pv_prognose_length != len(self.einspeiseverguetung_euro_pro_wh)
)
):
raise ValueError("Input lists have different lengths")
return self
class SimulationResult(PydanticBaseModel):
"""This object contains the results of the simulation and provides insights into various parameters over the entire forecast period."""
Last_Wh_pro_Stunde: list[Optional[float]] = Field(description="TBD")
EAuto_SoC_pro_Stunde: list[Optional[float]] = Field(
description="The state of charge of the EV for each hour."
)
Einnahmen_Euro_pro_Stunde: list[Optional[float]] = Field(
description="The revenue from grid feed-in or other sources in euros per hour."
)
Gesamt_Verluste: float = Field(
description="The total losses in watt-hours over the entire period."
)
Gesamtbilanz_Euro: float = Field(
description="The total balance of revenues minus costs in euros."
)
Gesamteinnahmen_Euro: float = Field(description="The total revenues in euros.")
Gesamtkosten_Euro: float = Field(description="The total costs in euros.")
Home_appliance_wh_per_hour: list[Optional[float]] = Field(
description="The energy consumption of a household appliance in watt-hours per hour."
)
Kosten_Euro_pro_Stunde: list[Optional[float]] = Field(
description="The costs in euros per hour."
)
Netzbezug_Wh_pro_Stunde: list[Optional[float]] = Field(
description="The grid energy drawn in watt-hours per hour."
)
Netzeinspeisung_Wh_pro_Stunde: list[Optional[float]] = Field(
description="The energy fed into the grid in watt-hours per hour."
)
Verluste_Pro_Stunde: list[Optional[float]] = Field(
description="The losses in watt-hours per hour."
)
akku_soc_pro_stunde: list[Optional[float]] = Field(
description="The state of charge of the battery (not the EV) in percentage per hour."
)
@field_validator(
"Last_Wh_pro_Stunde",
"Netzeinspeisung_Wh_pro_Stunde",
"akku_soc_pro_stunde",
"Netzbezug_Wh_pro_Stunde",
"Kosten_Euro_pro_Stunde",
"Einnahmen_Euro_pro_Stunde",
"EAuto_SoC_pro_Stunde",
"Verluste_Pro_Stunde",
"Home_appliance_wh_per_hour",
mode="before",
)
def convert_numpy(cls, field: Any) -> Any:
return NumpyEncoder.convert_numpy(field)[0]
class EnergieManagementSystem(SingletonMixin, ConfigMixin, PredictionMixin, PydanticBaseModel):
# Disable validation on assignment to speed up simulation runs.
model_config = ConfigDict(
validate_assignment=False,
)
# Start datetime.
_start_datetime: ClassVar[Optional[DateTime]] = None
@computed_field # type: ignore[prop-decorator]
@property
def start_datetime(self) -> DateTime:
"""The starting datetime of the current or latest energy management."""
if EnergieManagementSystem._start_datetime is None:
EnergieManagementSystem.set_start_datetime()
return EnergieManagementSystem._start_datetime
@classmethod
def set_start_datetime(cls, start_datetime: Optional[DateTime] = None) -> DateTime:
if start_datetime is None:
start_datetime = to_datetime()
cls._start_datetime = start_datetime.set(minute=0, second=0, microsecond=0)
return cls._start_datetime
# -------------------------
# TODO: Take from prediction
# -------------------------
gesamtlast: Optional[NDArray[Shape["*"], float]] = Field(
default=None,
description="An array of floats representing the total load (consumption) in watts for different time intervals.",
)
pv_prognose_wh: Optional[NDArray[Shape["*"], float]] = Field(
default=None,
description="An array of floats representing the forecasted photovoltaic output in watts for different time intervals.",
)
strompreis_euro_pro_wh: Optional[NDArray[Shape["*"], float]] = Field(
default=None,
description="An array of floats representing the electricity price in euros per watt-hour for different time intervals.",
)
einspeiseverguetung_euro_pro_wh_arr: Optional[NDArray[Shape["*"], float]] = Field(
default=None,
description="An array of floats representing the feed-in compensation in euros per watt-hour.",
)
# -------------------------
# TODO: Move to devices
# -------------------------
akku: Optional[PVAkku] = Field(default=None, description="TBD.")
eauto: Optional[PVAkku] = Field(default=None, description="TBD.")
home_appliance: Optional[HomeAppliance] = Field(default=None, description="TBD.")
wechselrichter: Optional[Wechselrichter] = Field(default=None, description="TBD.")
# -------------------------
# TODO: Move to devices
# -------------------------
ac_charge_hours: Optional[NDArray[Shape["*"], float]] = Field(default=None, description="TBD")
dc_charge_hours: Optional[NDArray[Shape["*"], float]] = Field(default=None, description="TBD")
ev_charge_hours: Optional[NDArray[Shape["*"], float]] = Field(default=None, description="TBD")
def set_parameters(
self,
parameters: EnergieManagementSystemParameters,
eauto: Optional[PVAkku] = None,
home_appliance: Optional[HomeAppliance] = None,
wechselrichter: Optional[Wechselrichter] = None,
) -> None:
self.gesamtlast = np.array(parameters.gesamtlast, float)
self.pv_prognose_wh = np.array(parameters.pv_prognose_wh, float)
self.strompreis_euro_pro_wh = np.array(parameters.strompreis_euro_pro_wh, float)
self.einspeiseverguetung_euro_pro_wh_arr = (
parameters.einspeiseverguetung_euro_pro_wh
if isinstance(parameters.einspeiseverguetung_euro_pro_wh, list)
else np.full(len(self.gesamtlast), parameters.einspeiseverguetung_euro_pro_wh, float)
)
if wechselrichter is not None:
self.akku = wechselrichter.akku
else:
self.akku = None
self.eauto = eauto
self.home_appliance = home_appliance
self.wechselrichter = wechselrichter
self.ac_charge_hours = np.full(self.config.prediction_hours, 0.0)
self.dc_charge_hours = np.full(self.config.prediction_hours, 1.0)
self.ev_charge_hours = np.full(self.config.prediction_hours, 0.0)
def set_akku_discharge_hours(self, ds: np.ndarray) -> None:
if self.akku is not 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
def set_ev_charge_hours(self, ds: np.ndarray) -> None:
self.ev_charge_hours = ds
def set_home_appliance_start(self, ds: int, global_start_hour: int = 0) -> None:
if self.home_appliance is not None:
self.home_appliance.set_starting_time(ds, global_start_hour=global_start_hour)
def reset(self) -> None:
if self.eauto:
self.eauto.reset()
if self.akku:
self.akku.reset()
def run(
self,
start_hour: Optional[int] = None,
force_enable: Optional[bool] = False,
force_update: Optional[bool] = False,
) -> None:
"""Run energy management.
Sets `start_datetime` to current hour, updates the configuration and the prediction, and
starts simulation at current hour.
Args:
start_hour (int, optional): Hour to take as start time for the energy management. Defaults
to now.
force_enable (bool, optional): If True, forces to update even if disabled. This
is mostly relevant to prediction providers.
force_update (bool, optional): If True, forces to update the data even if still cached.
"""
self.set_start_hour(start_hour=start_hour)
self.config.update()
# Check for run definitions
if self.start_datetime is None:
error_msg = "Start datetime unknown."
logger.error(error_msg)
raise ValueError(error_msg)
if self.config.prediction_hours is None:
error_msg = "Prediction hours unknown."
logger.error(error_msg)
raise ValueError(error_msg)
if self.config.optimisation_hours is None:
error_msg = "Optimisation hours unknown."
logger.error(error_msg)
raise ValueError(error_msg)
self.prediction.update_data(force_enable=force_enable, force_update=force_update)
# TODO: Create optimisation problem that calls into devices.update_data() for simulations.
def set_start_hour(self, start_hour: Optional[int] = None) -> None:
"""Sets start datetime to given hour.
Args:
start_hour (int, optional): Hour to take as start time for the energy management. Defaults
to now.
"""
if start_hour is None:
self.set_start_datetime()
else:
start_datetime = to_datetime().set(hour=start_hour, minute=0, second=0, microsecond=0)
self.set_start_datetime(start_datetime)
def simuliere_ab_jetzt(self) -> dict[str, Any]:
jetzt = to_datetime().now()
start_stunde = jetzt.hour
return self.simuliere(start_stunde)
def simuliere(self, start_stunde: int) -> dict[str, Any]:
"""hour.
akku_soc_pro_stunde begin of the hour, initial hour state!
last_wh_pro_stunde integral of last hour (end state)
"""
# Check for simulation integrity
if (
self.gesamtlast is None
or self.pv_prognose_wh is None
or self.strompreis_euro_pro_wh is None
or self.ev_charge_hours is None
or self.ac_charge_hours is None
or self.dc_charge_hours is None
or self.einspeiseverguetung_euro_pro_wh_arr is None
):
error_msg = (
f"Mandatory data missing - "
f"Load Curve: {self.gesamtlast}, "
f"PV Forecast: {self.pv_prognose_wh}, "
f"Electricity Price: {self.strompreis_euro_pro_wh}, "
f"EV Charge Hours: {self.ev_charge_hours}, "
f"AC Charge Hours: {self.ac_charge_hours}, "
f"DC Charge Hours: {self.dc_charge_hours}, "
f"Feed-in tariff: {self.einspeiseverguetung_euro_pro_wh_arr}"
)
logger.error(error_msg)
raise ValueError(error_msg)
lastkurve_wh = self.gesamtlast
if not (len(lastkurve_wh) == len(self.pv_prognose_wh) == len(self.strompreis_euro_pro_wh)):
error_msg = f"Array sizes do not match: Load Curve = {len(lastkurve_wh)}, PV Forecast = {len(self.pv_prognose_wh)}, Electricity Price = {len(self.strompreis_euro_pro_wh)}"
logger.error(error_msg)
raise ValueError(error_msg)
# Optimized total hours calculation
ende = len(lastkurve_wh)
total_hours = ende - start_stunde
# Pre-allocate arrays for the results, optimized for speed
last_wh_pro_stunde = np.full((total_hours), np.nan)
netzeinspeisung_wh_pro_stunde = np.full((total_hours), np.nan)
netzbezug_wh_pro_stunde = np.full((total_hours), np.nan)
kosten_euro_pro_stunde = np.full((total_hours), np.nan)
einnahmen_euro_pro_stunde = np.full((total_hours), np.nan)
akku_soc_pro_stunde = np.full((total_hours), np.nan)
eauto_soc_pro_stunde = np.full((total_hours), np.nan)
verluste_wh_pro_stunde = np.full((total_hours), np.nan)
home_appliance_wh_per_hour = np.full((total_hours), np.nan)
# Set initial state
if self.akku:
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):
stunde_since_now = stunde - start_stunde
# Accumulate loads and PV generation
verbrauch = self.gesamtlast[stunde]
verluste_wh_pro_stunde[stunde_since_now] = 0.0
# Home appliances
if self.home_appliance:
ha_load = self.home_appliance.get_load_for_hour(stunde)
verbrauch += ha_load
home_appliance_wh_per_hour[stunde_since_now] = ha_load
# E-Auto handling
if self.eauto:
if 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
eauto_soc_pro_stunde[stunde_since_now] = self.eauto.ladezustand_in_prozent()
# Process inverter logic
netzeinspeisung, netzbezug, verluste, eigenverbrauch = (0.0, 0.0, 0.0, 0.0)
if self.akku:
self.akku.set_charge_allowed_for_hour(self.dc_charge_hours[stunde], stunde)
if self.wechselrichter:
erzeugung = self.pv_prognose_wh[stunde]
netzeinspeisung, netzbezug, verluste, eigenverbrauch = (
self.wechselrichter.energie_verarbeiten(erzeugung, verbrauch, stunde)
)
# AC PV Battery Charge
if self.akku and 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())
verbrauch += geladene_menge
verbrauch += verluste_wh
netzbezug += geladene_menge
netzbezug += verluste_wh
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
verluste_wh_pro_stunde[stunde_since_now] += verluste
last_wh_pro_stunde[stunde_since_now] = verbrauch
# Financial calculations
kosten_euro_pro_stunde[stunde_since_now] = (
netzbezug * self.strompreis_euro_pro_wh[stunde]
)
einnahmen_euro_pro_stunde[stunde_since_now] = (
netzeinspeisung * self.einspeiseverguetung_euro_pro_wh_arr[stunde]
)
# Akku SOC tracking
if self.akku:
akku_soc_pro_stunde[stunde_since_now] = self.akku.ladezustand_in_prozent()
else:
akku_soc_pro_stunde[stunde_since_now] = 0.0
# Total cost and return
gesamtkosten_euro = np.nansum(kosten_euro_pro_stunde) - np.nansum(einnahmen_euro_pro_stunde)
# Prepare output dictionary
out: Dict[str, Union[np.ndarray, float]] = {
"Last_Wh_pro_Stunde": last_wh_pro_stunde,
"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,
"EAuto_SoC_pro_Stunde": eauto_soc_pro_stunde,
"Gesamteinnahmen_Euro": np.nansum(einnahmen_euro_pro_stunde),
"Gesamtkosten_Euro": np.nansum(kosten_euro_pro_stunde),
"Verluste_Pro_Stunde": verluste_wh_pro_stunde,
"Gesamt_Verluste": np.nansum(verluste_wh_pro_stunde),
"Home_appliance_wh_per_hour": home_appliance_wh_per_hour,
}
return out
# Initialize the Energy Management System, it is a singleton.
ems = EnergieManagementSystem()
def get_ems() -> EnergieManagementSystem:
"""Gets the EOS Energy Management System."""
return ems

226
src/akkudoktoreos/core/pydantic.py Normal file
View File

@@ -0,0 +1,226 @@
"""Module for managing and serializing Pydantic-based models with custom support.
This module introduces the `PydanticBaseModel` class, which extends Pydantic’s `BaseModel` to facilitate
custom serialization and deserialization for `pendulum.DateTime` objects. The main features include
automatic handling of `pendulum.DateTime` fields, custom serialization to ISO 8601 format, and utility
methods for converting model instances to and from dictionary and JSON formats.
Key Classes:
- PendulumDateTime: A custom type adapter that provides serialization and deserialization
functionality for `pendulum.DateTime` objects, converting them to ISO 8601 strings and back.
- PydanticBaseModel: A base model class for handling prediction records or configuration data
with automatic Pendulum DateTime handling and additional methods for JSON and dictionary
conversion.
Classes:
PendulumDateTime(TypeAdapter[pendulum.DateTime]): Type adapter for `pendulum.DateTime` fields
with ISO 8601 serialization. Includes:
- serialize: Converts `pendulum.DateTime` instances to ISO 8601 string.
- deserialize: Converts ISO 8601 strings to `pendulum.DateTime` instances.
- is_iso8601: Validates if a string matches the ISO 8601 date format.
PydanticBaseModel(BaseModel): Extends `pydantic.BaseModel` to handle `pendulum.DateTime` fields
and adds convenience methods for dictionary and JSON serialization. Key methods:
- model_dump: Dumps the model, converting `pendulum.DateTime` fields to ISO 8601.
- model_construct: Constructs a model instance with automatic deserialization of
`pendulum.DateTime` fields from ISO 8601.
- to_dict: Serializes the model instance to a dictionary.
- from_dict: Constructs a model instance from a dictionary.
- to_json: Converts the model instance to a JSON string.
- from_json: Creates a model instance from a JSON string.
Usage Example:
# Define custom settings in a model using PydanticBaseModel
class PredictionCommonSettings(PydanticBaseModel):
prediction_start: pendulum.DateTime = Field(...)
# Serialize a model instance to a dictionary or JSON
config = PredictionCommonSettings(prediction_start=pendulum.now())
config_dict = config.to_dict()
config_json = config.to_json()
# Deserialize from dictionary or JSON
new_config = PredictionCommonSettings.from_dict(config_dict)
restored_config = PredictionCommonSettings.from_json(config_json)
Dependencies:
- `pendulum`: Required for handling timezone-aware datetime fields.
- `pydantic`: Required for model and validation functionality.
Notes:
- This module enables custom handling of Pendulum DateTime fields within Pydantic models,
which is particularly useful for applications requiring consistent ISO 8601 datetime formatting
and robust timezone-aware datetime support.
"""
import json
import re
from typing import Any, Type
import pendulum
from pydantic import BaseModel, ConfigDict, TypeAdapter
# Custom type adapter for Pendulum DateTime fields
class PendulumDateTime(TypeAdapter[pendulum.DateTime]):
@classmethod
def serialize(cls, value: Any) -> str:
"""Convert pendulum.DateTime to ISO 8601 string."""
if isinstance(value, pendulum.DateTime):
return value.to_iso8601_string()
raise ValueError(f"Expected pendulum.DateTime, got {type(value)}")
@classmethod
def deserialize(cls, value: Any) -> pendulum.DateTime:
"""Convert ISO 8601 string to pendulum.DateTime."""
if isinstance(value, str) and cls.is_iso8601(value):
try:
return pendulum.parse(value)
except pendulum.parsing.exceptions.ParserError as e:
raise ValueError(f"Invalid date format: {value}") from e
elif isinstance(value, pendulum.DateTime):
return value
raise ValueError(f"Expected ISO 8601 string or pendulum.DateTime, got {type(value)}")
@staticmethod
def is_iso8601(value: str) -> bool:
"""Check if the string is a valid ISO 8601 date string."""
iso8601_pattern = (
r"^(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d{1,3})?(?:Z|[+-]\d{2}:\d{2})?)$"
)
return bool(re.match(iso8601_pattern, value))
class PydanticBaseModel(BaseModel):
"""Base model class with automatic serialization and deserialization of `pendulum.DateTime` fields.
This model serializes pendulum.DateTime objects to ISO 8601 strings and
deserializes ISO 8601 strings to pendulum.DateTime objects.
"""
# Enable custom serialization globally in config
model_config = ConfigDict(
arbitrary_types_allowed=True,
use_enum_values=True,
validate_assignment=True,
)
# Override Pydantic’s serialization for all DateTime fields
def model_dump(self, *args: Any, **kwargs: Any) -> dict:
"""Custom dump method to handle serialization for DateTime fields."""
result = super().model_dump(*args, **kwargs)
for key, value in result.items():
if isinstance(value, pendulum.DateTime):
result[key] = PendulumDateTime.serialize(value)
return result
@classmethod
def model_construct(cls, data: dict) -> "PydanticBaseModel":
"""Custom constructor to handle deserialization for DateTime fields."""
for key, value in data.items():
if isinstance(value, str) and PendulumDateTime.is_iso8601(value):
data[key] = PendulumDateTime.deserialize(value)
return super().model_construct(data)
def reset_optional(self) -> "PydanticBaseModel":
"""Resets all optional fields in the model to None.
Iterates through all model fields and sets any optional (non-required)
fields to None. The modification is done in-place on the current instance.
Returns:
PydanticBaseModel: The current instance with all optional fields
reset to None.
Example:
>>> settings = PydanticBaseModel(name="test", optional_field="value")
>>> settings.reset_optional()
>>> assert settings.optional_field is None
"""
for field_name, field in self.model_fields.items():
if field.is_required is False: # Check if field is optional
setattr(self, field_name, None)
return self
def to_dict(self) -> dict:
"""Convert this PredictionRecord instance to a dictionary representation.
Returns:
dict: A dictionary where the keys are the field names of the PydanticBaseModel,
and the values are the corresponding field values.
"""
return self.model_dump()
@classmethod
def from_dict(cls: Type["PydanticBaseModel"], data: dict) -> "PydanticBaseModel":
"""Create a PydanticBaseModel instance from a dictionary.
Args:
data (dict): A dictionary containing data to initialize the PydanticBaseModel.
Keys should match the field names defined in the model.
Returns:
PydanticBaseModel: An instance of the PydanticBaseModel populated with the data.
Notes:
Works with derived classes by ensuring the `cls` argument is used to instantiate the object.
"""
return cls.model_validate(data)
@classmethod
def from_dict_with_reset(cls, data: dict | None = None) -> "PydanticBaseModel":
"""Creates a new instance with reset optional fields, then updates from dict.
First creates an instance with default values, resets all optional fields
to None, then updates the instance with the provided dictionary data if any.
Args:
data (dict | None): Dictionary containing field values to initialize
the instance with. Defaults to None.
Returns:
PydanticBaseModel: A new instance with all optional fields initially
reset to None and then updated with provided data.
Example:
>>> data = {"name": "test", "optional_field": "value"}
>>> settings = PydanticBaseModel.from_dict_with_reset(data)
>>> # All non-specified optional fields will be None
"""
# Create instance with model defaults
instance = cls()
# Reset all optional fields to None
instance.reset_optional()
# Update with provided data if any
if data:
# Use model_validate to ensure proper type conversion and validation
updated_instance = instance.model_validate({**instance.model_dump(), **data})
return updated_instance
return instance
def to_json(self) -> str:
"""Convert the PydanticBaseModel instance to a JSON string.
Returns:
str: The JSON representation of the instance.
"""
return self.model_dump_json()
@classmethod
def from_json(cls: Type["PydanticBaseModel"], json_str: str) -> "PydanticBaseModel":
"""Create an instance of the PydanticBaseModel class or its subclass from a JSON string.
Args:
json_str (str): JSON string to parse and convert into a PydanticBaseModel instance.
Returns:
PydanticBaseModel: A new instance of the class, populated with data from the JSON string.
Notes:
Works with derived classes by ensuring the `cls` argument is used to instantiate the object.
"""
data = json.loads(json_str)
return cls.model_validate(data)