Files
EOS/src/akkudoktoreos/prediction/elecpriceabc.py
Bobby Noelte bd38b3c5ef
Some checks failed
docker-build / platform-excludes (push) Has been cancelled
pre-commit / pre-commit (push) Has been cancelled
Run Pytest on Pull Request / test (push) Has been cancelled
docker-build / build (push) Has been cancelled
docker-build / merge (push) Has been cancelled
Close stale pull requests/issues / Find Stale issues and PRs (push) Has been cancelled
fix: logging, prediction update, multiple bugs (#584)
* Fix logging configuration issues that made logging stop operation. Switch to Loguru
  logging (from Python logging). Enable console and file logging with different log levels.
  Add logging documentation.

* Fix logging configuration and EOS configuration out of sync. Added tracking support
  for nested value updates of Pydantic models. This used to update the logging configuration
  when the EOS configurationm for logging is changed. Should keep logging config and EOS
  config in sync as long as all changes to the EOS logging configuration are done by
  set_nested_value(), which is the case for the REST API.

* Fix energy management task looping endlessly after the second update when trying to update
  the last_update datetime.

* Fix get_nested_value() to correctly take values from the dicts in a Pydantic model instance.

* Fix usage of model classes instead of model instances in nested value access when evaluation
  the value type that is associated to each key.

* Fix illegal json format in prediction documentation for PVForecastAkkudoktor provider.

* Fix documentation qirks and add EOS Connect to integrations.

* Support deprecated fields in configuration in documentation generation and EOSdash.

* Enhance EOSdash demo to show BrightSky humidity data (that is often missing)

* Update documentation reference to German EOS installation videos.

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
2025-06-10 22:00:28 +02:00

72 lines
2.7 KiB
Python

"""Abstract and base classes for electricity price predictions.
Notes:
- Ensure appropriate API keys or configurations are set up if required by external data sources.
"""
from abc import abstractmethod
from typing import List, Optional
from pydantic import Field, computed_field
from akkudoktoreos.prediction.predictionabc import PredictionProvider, PredictionRecord
class ElecPriceDataRecord(PredictionRecord):
"""Represents a electricity price data record containing various price attributes at a specific datetime.
Attributes:
date_time (Optional[AwareDatetime]): The datetime of the record.
"""
elecprice_marketprice_wh: Optional[float] = Field(
None, description="Electricity market price per Wh (€/Wh)"
)
# Computed fields
@computed_field # type: ignore[prop-decorator]
@property
def elecprice_marketprice_kwh(self) -> Optional[float]:
"""Electricity market price per kWh (€/kWh).
Convenience attribute calculated from `elecprice_marketprice_wh`.
"""
if self.elecprice_marketprice_wh is None:
return None
return self.elecprice_marketprice_wh * 1000.0
class ElecPriceProvider(PredictionProvider):
"""Abstract base class for electricity price providers.
WeatherProvider is a thread-safe singleton, ensuring only one instance of this class is created.
Configuration variables:
electricity price_provider (str): Prediction provider for electricity price.
Attributes:
hours (int, optional): The number of hours into the future for which predictions are generated.
historic_hours (int, optional): The number of past hours for which historical data is retained.
latitude (float, optional): The latitude in degrees, must be within -90 to 90.
longitude (float, optional): The longitude in degrees, must be within -180 to 180.
start_datetime (datetime, optional): The starting datetime for predictions, defaults to the current datetime if unspecified.
end_datetime (datetime, computed): The datetime representing the end of the prediction range,
calculated based on `start_datetime` and `hours`.
keep_datetime (datetime, computed): The earliest datetime for retaining historical data, calculated
based on `start_datetime` and `historic_hours`.
"""
# overload
records: List[ElecPriceDataRecord] = Field(
default_factory=list, description="List of ElecPriceDataRecord records"
)
@classmethod
@abstractmethod
def provider_id(cls) -> str:
return "ElecPriceProvider"
def enabled(self) -> bool:
return self.provider_id() == self.config.elecprice.provider