-
Notifications
You must be signed in to change notification settings - Fork 41
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
- Loading branch information
Showing
4 changed files
with
2,265 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,218 @@ | ||
"""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__) | ||
|
||
|
||
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 | ||
from akkudoktoreos.config.config import get_config | ||
|
||
return get_config() | ||
|
||
|
||
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 | ||
from akkudoktoreos.prediction.prediction import get_prediction | ||
|
||
return get_prediction() | ||
|
||
|
||
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 | ||
from akkudoktoreos.core.ems import get_ems | ||
|
||
return get_ems() | ||
|
||
|
||
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 |
Oops, something went wrong.