Add core abstract and base classes.

b0661 · b0661 · commit 4675ec1c302a · 2024-11-25T22:35:15.000+01:00
Signed-off-by: Bobby Noelte &lt;b0661n0e17e@gmail.com&gt;
diff --git a/src/akkudoktoreos/core/coreabc.py b/src/akkudoktoreos/core/coreabc.py
@@ -0,0 +1,159 @@
+"""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, Type
+
+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):
+        """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):
+        """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 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, BaseModel):
+            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):
+        """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
diff --git a/src/akkudoktoreos/core/pydantic.py b/src/akkudoktoreos/core/pydantic.py
@@ -0,0 +1,225 @@
+"""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,
+    )
+
+    # Override Pydantic’s serialization for all DateTime fields
+    def model_dump(self, *args, **kwargs) -> 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(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()
+            >>> 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()
+
+        # 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)
