fix: cached_method deprecated and test

cachebox deprecated the method decorator. Used cached instead.

Fix cache integration tests that were accessing real world addresses.

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
This commit is contained in:
Bobby Noelte
2025-11-13 15:13:37 +01:00
parent cab3a3dd21
commit 8da137f8f1
5 changed files with 41 additions and 53 deletions

View File

@@ -179,64 +179,35 @@ class CacheEnergyManagementStore(SingletonMixin):
raise AttributeError(f"'{self.cache.__class__.__name__}' object has no method 'clear'")
def cachemethod_energy_management(method: TCallable) -> TCallable:
"""Decorator for in memory caching the result of an instance method.
def cache_energy_management(callable: TCallable) -> TCallable:
"""Decorator for in memory caching the result of a callable.
This decorator caches the method's result in `CacheEnergyManagementStore`, ensuring
that subsequent calls with the same arguments return the cached result until the
This decorator caches the method or function's result in `CacheEnergyManagementStore`,
ensuring that subsequent calls with the same arguments return the cached result until the
next energy management start.
Args:
method (Callable): The instance method to be decorated.
Returns:
Callable: The wrapped method with caching functionality.
Example:
>>> class MyClass:
>>> @cachemethod_energy_management
>>> def expensive_method(self, param: str) -> str:
>>> # Perform expensive computation
>>> return f"Computed {param}"
"""
@cachebox.cachedmethod(
cache=CacheEnergyManagementStore().cache, callback=cache_energy_management_store_callback
)
@functools.wraps(method)
def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
result = method(self, *args, **kwargs)
return result
return wrapper
def cache_energy_management(func: TCallable) -> TCallable:
"""Decorator for in memory caching the result of a standalone function.
This decorator caches the function's result in `CacheEnergyManagementStore`, ensuring
that subsequent calls with the same arguments return the cached result until the
next energy management start.
Args:
func (Callable): The function to be decorated.
callable (Callable): The function or method to be decorated.
Returns:
Callable: The wrapped function with caching functionality.
Example:
>>> @cache_until_next_update
>>> def expensive_function(param: str) -> str:
>>> # Perform expensive computation
>>> return f"Computed {param}"
.. code-block:: python
@cache_energy_management
def expensive_function(param: str) -> str:
# Perform expensive computation
return f"Computed {param}"
"""
@cachebox.cached(
cache=CacheEnergyManagementStore().cache, callback=cache_energy_management_store_callback
)
@functools.wraps(func)
@functools.wraps(callable)
def wrapper(*args: Any, **kwargs: Any) -> Any:
result = func(*args, **kwargs)
result = callable(*args, **kwargs)
return result
return wrapper

View File

@@ -5,7 +5,7 @@ from pathlib import Path
import numpy as np
from scipy.interpolate import RegularGridInterpolator
from akkudoktoreos.core.cache import cachemethod_energy_management
from akkudoktoreos.core.cache import cache_energy_management
from akkudoktoreos.core.coreabc import SingletonMixin
@@ -24,7 +24,7 @@ class SelfConsumptionProbabilityInterpolator:
points = np.array([np.full_like(partial_loads, load_1h_power), partial_loads]).T
return points, partial_loads
@cachemethod_energy_management
@cache_energy_management
def calculate_self_consumption(self, load_1h_power: float, pv_power: float) -> float:
"""Calculate the PV self-consumption rate using RegularGridInterpolator.

View File

@@ -16,7 +16,6 @@ from akkudoktoreos.core.cache import (
CacheFileStore,
cache_energy_management,
cache_in_file,
cachemethod_energy_management,
)
from akkudoktoreos.utils.datetimeutil import compare_datetimes, to_datetime, to_duration
@@ -64,10 +63,10 @@ class TestCacheEnergyManagementStore:
class TestCacheUntilUpdateDecorators:
def test_cachemethod_energy_management(self, cache_energy_management_store):
"""Test that cachemethod_energy_management caches method results."""
"""Test that cache_energy_management caches method results."""
class MyClass:
@cachemethod_energy_management
@cache_energy_management
def compute(self, value: int) -> int:
return value * 2
@@ -102,7 +101,7 @@ class TestCacheUntilUpdateDecorators:
"""Test that caching works for different arguments."""
class MyClass:
@cachemethod_energy_management
@cache_energy_management
def compute(self, value: int) -> int:
return value * 2
@@ -123,7 +122,7 @@ class TestCacheUntilUpdateDecorators:
"""Test that cache is cleared between EMS update cycles."""
class MyClass:
@cachemethod_energy_management
@cache_energy_management
def compute(self, value: int) -> int:
return value * 2

View File

@@ -173,11 +173,20 @@ def test_request_forecast_status_codes(
provider._request_forecast()
@patch("requests.get")
@patch("akkudoktoreos.core.cache.CacheFileStore")
def test_cache_integration(mock_cache, provider):
def test_cache_integration(mock_cache, mock_get, provider, sample_akkudoktor_1_json):
"""Test caching of 8-day electricity price data."""
# Mock response object
mock_response = Mock()
mock_response.status_code = 200
mock_response.content = json.dumps(sample_akkudoktor_1_json)
mock_get.return_value = mock_response
# Mock cache object
mock_cache_instance = mock_cache.return_value
mock_cache_instance.get.return_value = None # Simulate no cache
provider._update_data(force_update=True)
mock_cache_instance.create.assert_called_once()
mock_cache_instance.get.assert_called_once()

View File

@@ -167,11 +167,20 @@ def test_request_forecast_status_codes(
provider._request_forecast()
@patch("requests.get")
@patch("akkudoktoreos.core.cache.CacheFileStore")
def test_cache_integration(mock_cache, provider):
def test_cache_integration(mock_cache, mock_get, provider, sample_energycharts_json):
"""Test caching of 8-day electricity price data."""
# Mock response object
mock_response = Mock()
mock_response.status_code = 200
mock_response.content = json.dumps(sample_energycharts_json)
mock_get.return_value = mock_response
# Mock cache object
mock_cache_instance = mock_cache.return_value
mock_cache_instance.get.return_value = None # Simulate no cache
provider._update_data(force_update=True)
mock_cache_instance.create.assert_called_once()
mock_cache_instance.get.assert_called_once()
@@ -195,7 +204,7 @@ def test_key_to_array_resampling(provider):
@pytest.mark.skip(reason="For development only")
def test_akkudoktor_development_forecast_data(provider):
def test_energycharts_development_forecast_data(provider):
"""Fetch data from real Energy-Charts server."""
# Preset, as this is usually done by update_data()
provider.ems_start_datetime = to_datetime("2024-10-26 00:00:00")