Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-08-25 06:52:23 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Reasonable defaults, isolate tests, EOS_LOGGING_LEVEL, EOS_CONFIG_DIR

Browse Source 
* Add EOS_CONFIG_DIR to set config dir (relative path to EOS_DIR or
   absolute path).
    - config_folder_path read-only
    - config_file_path read-only
 * Default values to support app start with empty config:
    - latitude/longitude (Berlin)
    - optimization_ev_available_charge_rates_percent (null, so model
      default value is used)
    - Enable Akkudoktor electricity price forecast (docker-compose).
 * Fix some endpoints (empty data, remove unused params, fix types).
 * cacheutil: Use cache dir. Closes #240
 * Support EOS_LOGGING_LEVEL environment variable to set log level.
 * tests: All tests use separate temporary config
    - Add pytest switch --check-config-side-effect to check user
      config file existence after each test. Will also fail if user config
      existed before test execution (but will only check after the test has
      run).
      Enable flag in github workflow.
    - Globally mock platformdirs in config module. Now no longer required
      to patch individually.
      Function calls to config instance (e.g. merge_settings_from_dict)
      were unaffected previously.
 * Set Berlin as default location (default config/docker-compose).
This commit is contained in:
Dominique Lasserre
2024-12-30 13:41:39 +01:00
committed by GitHub
parent 267a9bf427
commit 75987db9e1
29 changed files with 373 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
src/akkudoktoreos/config/config.py
View File

@@ -14,7 +14,7 @@ import shutil
from pathlib import Path
from typing import Any, ClassVar, List, Optional
import platformdirs
from platformdirs import user_config_dir, user_data_dir
from pydantic import Field, ValidationError, computed_field
# settings
@@ -40,6 +40,24 @@ from akkudoktoreos.utils.utils import UtilsCommonSettings
logger = get_logger(__name__)
def get_absolute_path(
basepath: Optional[Path | str], subpath: Optional[Path | str]
) -> Optional[Path]:
"""Get path based on base path."""
if isinstance(basepath, str):
basepath = Path(basepath)
if subpath is None:
return basepath
if isinstance(subpath, str):
subpath = Path(subpath)
if subpath.is_absolute():
return subpath
if basepath is not None:
return basepath.joinpath(subpath)
return None
class ConfigCommonSettings(SettingsBaseModel):
"""Settings for common configuration."""
@@ -60,22 +78,14 @@ class ConfigCommonSettings(SettingsBaseModel):
@property
def data_output_path(self) -> Optional[Path]:
"""Compute data_output_path based on data_folder_path."""
if self.data_output_subpath is None:
return self.data_folder_path
if self.data_folder_path and self.data_output_subpath:
return self.data_folder_path.joinpath(self.data_output_subpath)
return None
return get_absolute_path(self.data_folder_path, self.data_output_subpath)
# Computed fields
@computed_field # type: ignore[prop-decorator]
@property
def data_cache_path(self) -> Optional[Path]:
"""Compute data_cache_path based on data_folder_path."""
if self.data_cache_subpath is None:
return self.data_folder_path
if self.data_folder_path and self.data_cache_subpath:
return self.data_folder_path.joinpath(self.data_cache_subpath)
return None
return get_absolute_path(self.data_folder_path, self.data_cache_subpath)
class SettingsEOS(
@@ -114,9 +124,10 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
Initialization Process:
- Upon instantiation, the singleton instance attempts to load a configuration file in this order:
1. The directory specified by the `EOS_DIR` environment variable.
2. A platform specific default directory for EOS.
3. The current working directory.
1. The directory specified by the `EOS_CONFIG_DIR` environment variable
2. The directory specified by the `EOS_DIR` environment variable.
3. A platform specific default directory for EOS.
4. The current working directory.
- The first available configuration file found in these directories is loaded.
- If no configuration file is found, a default configuration file is created in the platform
specific default directory, and default settings are loaded into it.
@@ -150,21 +161,29 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
APP_NAME: ClassVar[str] = "net.akkudoktor.eos" # reverse order
APP_AUTHOR: ClassVar[str] = "akkudoktor"
EOS_DIR: ClassVar[str] = "EOS_DIR"
EOS_CONFIG_DIR: ClassVar[str] = "EOS_CONFIG_DIR"
ENCODING: ClassVar[str] = "UTF-8"
CONFIG_FILE_NAME: ClassVar[str] = "EOS.config.json"
_settings: ClassVar[Optional[SettingsEOS]] = None
_file_settings: ClassVar[Optional[SettingsEOS]] = None
config_folder_path: Optional[Path] = Field(
None, description="Path to EOS configuration directory."
)
config_file_path: Optional[Path] = Field(
default=None, description="Path to EOS configuration file."
)
_config_folder_path: Optional[Path] = None
_config_file_path: Optional[Path] = None
# Computed fields
@computed_field # type: ignore[prop-decorator]
@property
def config_folder_path(self) -> Optional[Path]:
"""Path to EOS configuration directory."""
return self._config_folder_path
@computed_field # type: ignore[prop-decorator]
@property
def config_file_path(self) -> Optional[Path]:
"""Path to EOS configuration file."""
return self._config_file_path
@computed_field # type: ignore[prop-decorator]
@property
def config_default_file_path(self) -> Path:
@@ -297,7 +316,7 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
pass
# From platform specific default path
try:
data_dir = platformdirs.user_data_dir(self.APP_NAME, self.APP_AUTHOR)
data_dir = Path(user_data_dir(self.APP_NAME, self.APP_AUTHOR))
if data_dir is not None:
data_dir.mkdir(parents=True, exist_ok=True)
self.data_folder_path = data_dir
@@ -308,47 +327,27 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
data_dir = Path.cwd()
self.data_folder_path = data_dir
def _config_folder_path(self) -> Optional[Path]:
"""Finds the first directory containing a valid configuration file.
def _get_config_file_path(self) -> tuple[Path, bool]:
"""Finds the a valid configuration file or returns the desired path for a new config file.
Returns:
Path: The path to the configuration directory, or None if not found.
tuple[Path, bool]: The path to the configuration directory and if there is already a config file there
"""
config_dirs = []
config_dir = None
env_dir = os.getenv(self.EOS_DIR)
logger.debug(f"Envionment '{self.EOS_DIR}': '{env_dir}'")
env_base_dir = os.getenv(self.EOS_DIR)
env_config_dir = os.getenv(self.EOS_CONFIG_DIR)
env_dir = get_absolute_path(env_base_dir, env_config_dir)
logger.debug(f"Envionment config dir: '{env_dir}'")
if env_dir is not None:
config_dirs.append(Path(env_dir).resolve())
config_dirs.append(Path(platformdirs.user_config_dir(self.APP_NAME)))
config_dirs.append(env_dir.resolve())
config_dirs.append(Path(user_config_dir(self.APP_NAME)))
config_dirs.append(Path.cwd())
for cdir in config_dirs:
cfile = cdir.joinpath(self.CONFIG_FILE_NAME)
if cfile.exists():
logger.debug(f"Found config file: '{cfile}'")
config_dir = cdir
break
return config_dir
def _config_file_path(self) -> Path:
"""Finds the path to the configuration file.
Returns:
Path: The path to the configuration file. May not exist.
"""
config_file = None
config_dir = self._config_folder_path()
if config_dir is None:
# There is currently no configuration file - create it in default path
env_dir = os.getenv(self.EOS_DIR)
if env_dir is not None:
config_dir = Path(env_dir).resolve()
else:
config_dir = Path(platformdirs.user_config_dir(self.APP_NAME))
config_file = config_dir.joinpath(self.CONFIG_FILE_NAME)
else:
config_file = config_dir.joinpath(self.CONFIG_FILE_NAME)
return config_file
return cfile, True
return config_dirs[0].joinpath(self.CONFIG_FILE_NAME), False
def from_config_file(self) -> None:
"""Loads the configuration file settings for EOS.
@@ -356,14 +355,14 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
Raises:
ValueError: If the configuration file is invalid or incomplete.
"""
config_file = self._config_file_path()
config_file, exists = self._get_config_file_path()
config_dir = config_file.parent
if not config_file.exists():
if not exists:
config_dir.mkdir(parents=True, exist_ok=True)
try:
shutil.copy2(self.config_default_file_path, config_file)
except Exception as exc:
logger.warning(f"Could not copy default config: {exc}. Using default copy...")
logger.warning(f"Could not copy default config: {exc}. Using default config...")
config_file = self.config_default_file_path
config_dir = config_file.parent
@@ -376,8 +375,8 @@ class ConfigEOS(SingletonMixin, SettingsEOS):
self.update()
# Everthing worked, remember the values
self.config_folder_path = config_dir
self.config_file_path = config_file
self._config_folder_path = config_dir
self._config_file_path = config_file
def to_config_file(self) -> None:
"""Saves the current configuration to the configuration file.

7
src/akkudoktoreos/data/default.config.json
View File

@@ -6,15 +6,16 @@
"data_folder_path": null,
"data_output_path": null,
"data_output_subpath": null,
"elecprice_charges": 0.21,
"elecprice_provider": null,
"elecpriceimport_file_path": null,
"latitude": null,
"latitude": 52.5,
"load_import_file_path": null,
"load_name": null,
"load_provider": null,
"loadakkudoktor_year_energy": null,
"longitude": null,
"optimization_ev_available_charge_rates_percent": [],
"longitude": 13.4,
"optimization_ev_available_charge_rates_percent": null,
"optimization_hours": 48,
"optimization_penalty": null,
"prediction_historic_hours": 48,

20
src/akkudoktoreos/optimization/genetic.py
View File

@@ -123,7 +123,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Decode the input array into ac_charge, dc_charge, and discharge arrays."""
discharge_hours_bin_np = np.array(discharge_hours_bin)
len_ac = len(self.config.optimization_ev_available_charge_rates_percent)
len_ac = len(self.possible_charge_values)
# Categorization:
# Idle: 0 .. len_ac-1
@@ -155,9 +155,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
discharge[discharge_mask] = 1 # Set Discharge states to 1
ac_charge = np.zeros_like(discharge_hours_bin_np, dtype=float)
ac_charge[ac_mask] = [
self.config.optimization_ev_available_charge_rates_percent[i] for i in ac_indices
]
ac_charge[ac_mask] = [self.possible_charge_values[i] for i in ac_indices]
# Idle is just 0, already default.
@@ -166,7 +164,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
def mutate(self, individual: list[int]) -> tuple[list[int]]:
"""Custom mutation function for the individual."""
# Calculate the number of states
len_ac = len(self.config.optimization_ev_available_charge_rates_percent)
len_ac = len(self.possible_charge_values)
if self.optimize_dc_charge:
total_states = 3 * len_ac + 2
else:
@@ -300,7 +298,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
creator.create("Individual", list, fitness=creator.FitnessMin)
self.toolbox = base.Toolbox()
len_ac = len(self.config.optimization_ev_available_charge_rates_percent)
len_ac = len(self.possible_charge_values)
# Total number of states without DC:
# Idle: len_ac states
@@ -378,10 +376,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
if eautocharge_hours_index is not None:
eautocharge_hours_float = np.array(
[
self.config.optimization_ev_available_charge_rates_percent[i]
for i in eautocharge_hours_index
],
[self.possible_charge_values[i] for i in eautocharge_hours_index],
float,
)
self.ems.set_ev_charge_hours(eautocharge_hours_float)
@@ -615,10 +610,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
start_solution
)
eautocharge_hours_float = (
[
self.config.optimization_ev_available_charge_rates_percent[i]
for i in eautocharge_hours_index
]
[self.possible_charge_values[i] for i in eautocharge_hours_index]
if eautocharge_hours_index is not None
else None
)

44
src/akkudoktoreos/server/fastapi_server.py
View File

@@ -7,6 +7,7 @@ from pathlib import Path
from typing import Annotated, Any, AsyncGenerator, Dict, List, Optional, Union
import httpx
import pandas as pd
import uvicorn
from fastapi import FastAPI, Query, Request
from fastapi.exceptions import HTTPException
@@ -412,20 +413,37 @@ class ForecastResponse(PydanticBaseModel):
@app.get("/pvforecast")
def fastapi_pvprognose(ac_power_measurement: Optional[float] = None) -> ForecastResponse:
def fastapi_pvforecast() -> ForecastResponse:
###############
# PV Forecast
###############
pvforecast_ac_power = prediction_eos["pvforecast_ac_power"]
# Fetch prices for the specified date range
pvforecast_ac_power = pvforecast_ac_power.loc[
prediction_eos.start_datetime : prediction_eos.end_datetime
]
pvforecastakkudoktor_temp_air = prediction_eos["pvforecastakkudoktor_temp_air"]
# Fetch prices for the specified date range
pvforecastakkudoktor_temp_air = pvforecastakkudoktor_temp_air.loc[
prediction_eos.start_datetime : prediction_eos.end_datetime
]
prediction_key = "pvforecast_ac_power"
pvforecast_ac_power = prediction_eos.get(prediction_key)
if pvforecast_ac_power is None:
raise HTTPException(status_code=404, detail=f"Prediction not available: {prediction_key}")
# On empty Series.loc TypeError: Cannot compare tz-naive and tz-aware datetime-like objects
if len(pvforecast_ac_power) == 0:
pvforecast_ac_power = pd.Series()
else:
# Fetch prices for the specified date range
pvforecast_ac_power = pvforecast_ac_power.loc[
prediction_eos.start_datetime : prediction_eos.end_datetime
]
prediction_key = "pvforecastakkudoktor_temp_air"
pvforecastakkudoktor_temp_air = prediction_eos.get(prediction_key)
if pvforecastakkudoktor_temp_air is None:
raise HTTPException(status_code=404, detail=f"Prediction not available: {prediction_key}")
# On empty Series.loc TypeError: Cannot compare tz-naive and tz-aware datetime-like objects
if len(pvforecastakkudoktor_temp_air) == 0:
pvforecastakkudoktor_temp_air = pd.Series()
else:
# Fetch prices for the specified date range
pvforecastakkudoktor_temp_air = pvforecastakkudoktor_temp_air.loc[
prediction_eos.start_datetime : prediction_eos.end_datetime
]
# Return both forecasts as a JSON response
return ForecastResponse(
@@ -457,8 +475,8 @@ def fastapi_optimize(
def get_pdf() -> PdfResponse:
# Endpoint to serve the generated PDF with visualization results
output_path = config_eos.data_output_path
if not output_path.is_dir():
raise ValueError(f"Output path does not exist: {output_path}.")
if output_path is None or not output_path.is_dir():
raise HTTPException(status_code=404, detail=f"Output path does not exist: {output_path}.")
file_path = output_path / "visualization_results.pdf"
if not file_path.is_file():
raise HTTPException(status_code=404, detail="No visualization result available.")

6
src/akkudoktoreos/utils/cacheutil.py
View File

@@ -47,6 +47,7 @@ from typing import (
from pendulum import DateTime, Duration
from pydantic import BaseModel, ConfigDict, Field
from akkudoktoreos.core.coreabc import ConfigMixin
from akkudoktoreos.utils.datetimeutil import compare_datetimes, to_datetime, to_duration
from akkudoktoreos.utils.logutil import get_logger
@@ -90,7 +91,7 @@ class CacheFileStoreMeta(type, Generic[T]):
return cls._instances[cls]
class CacheFileStore(metaclass=CacheFileStoreMeta):
class CacheFileStore(ConfigMixin, metaclass=CacheFileStoreMeta):
"""A key-value store that manages file-like tempfile objects to be used as cache files.
Cache files are associated with a date. If no date is specified, the cache files are
@@ -328,8 +329,9 @@ class CacheFileStore(metaclass=CacheFileStoreMeta):
# File already available
cache_file_obj = cache_item.cache_file
else:
self.config.data_cache_path.mkdir(parents=True, exist_ok=True)
cache_file_obj = tempfile.NamedTemporaryFile(
mode=mode, delete=delete, suffix=suffix
mode=mode, delete=delete, suffix=suffix, dir=self.config.data_cache_path
)
self._store[cache_file_key] = CacheFileRecord(
cache_file=cache_file_obj,

2
src/akkudoktoreos/utils/logutil.py
View File

@@ -50,6 +50,8 @@ def get_logger(
# Create a logger with the specified name
logger = logging.getLogger(name)
logger.propagate = True
if (env_level := os.getenv("EOS_LOGGING_LEVEL")) is not None:
logging_level = env_level
if logging_level == "DEBUG":
level = logging.DEBUG
elif logging_level == "INFO":