jpawlowski/hass.tibber_prices
1
0
Fork 0
mirror of https://github.com/jpawlowski/hass.tibber_prices.git synced 2026-03-29 21:03:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

refactor(naming): complete class naming convention alignment

Browse source 
Renamed 25 public classes + 1 Enum to include TibberPrices prefix
following Home Assistant integration naming standards.

All classes now follow pattern: TibberPrices{SemanticPurpose}
No package hierarchy in names (import path is namespace).

Key changes:
- Coordinator module: DataFetcher, DataTransformer, ListenerManager,
  PeriodCalculator, TimeService (203 usages), CacheData
- Config flow: CannotConnectError, InvalidAuthError
- Entity utils: IconContext
- Sensor calculators: BaseCalculator + 8 subclasses
- Period handlers: 5 NamedTuples (PeriodConfig, PeriodData,
  PeriodStatistics, ThresholdConfig, IntervalCriteria)
- Period handlers: SpikeCandidateContext (dataclass → NamedTuple)
- API: QueryType Enum

Documentation updates:
- AGENTS.md: Added Pyright code generation guidelines
- planning/class-naming-refactoring-plan.md: Complete execution log

Quality metrics:
- 0 Pyright errors (strict type checking)
- 0 Ruff errors (linting + formatting)
- All hassfest checks passed
- 79 files validated

Impact: Aligns with HA Core standards (TibberDataCoordinator pattern).
No user-facing changes - internal refactor only.
This commit is contained in:
Julian Pawlowski 2025-11-20 11:22:53 +00:00
parent 07f5990e06
commit c2b9908e69
55 changed files with 407 additions and 366 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
AGENTS.md
View file

@ -1516,6 +1516,25 @@ venv = ".venv"
typeCheckingMode = "basic"
```
**CRITICAL: When generating code, always aim for Pyright `basic` mode compliance:**
✅ **DO:**
- Add type hints to all function signatures (parameters + return types)
- Use proper type annotations: `dict[str, Any]`, `list[dict]`, `str | None`
- Handle Optional types explicitly (None-checks before use)
- Use TYPE_CHECKING imports for type-only dependencies
- Prefer explicit returns over implicit `None`
❌ **DON'T:**
- Leave functions without return type hints
- Ignore potential `None` values in Optional types
- Use `Any` as escape hatch (only when truly needed)
- Create functions that could return different types based on runtime logic
**Goal:** Generated code should pass `./scripts/type-check` on first try, minimizing post-generation fixes.
**See also:** "Ruff Code Style Guidelines" section below for complementary code style rules that ensure `./scripts/lint` compliance.
### When Type Errors Are Acceptable
**Use `type: ignore` comments sparingly and ONLY when:**
@ -2066,6 +2085,8 @@ Understanding **how** good documentation emerges is as important as knowing what
These rules ensure generated code passes `./scripts/lint` on first try. Ruff enforces these automatically.
**See also:** "Linting Best Practices" section above for Pyright type checking guidelines that ensure `./scripts/type-check` compliance.
**String Formatting:**
```python

22
custom_components/tibber_prices/api/client.py
View file

@ -24,10 +24,10 @@ from .helpers import (
verify_graphql_response,
verify_response_or_raise,
)
from .queries import QueryType
from .queries import TibberPricesQueryType
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
@ -46,7 +46,7 @@ class TibberPricesApiClient:
self._session = session
self._version = version
self._request_semaphore = asyncio.Semaphore(2) # Max 2 concurrent requests
self.time: TimeService | None = None # Set externally by coordinator
self.time: TibberPricesTimeService # Set externally by coordinator (always initialized before use)
self._last_request_time = None # Set on first request
self._min_request_interval = timedelta(seconds=1) # Min 1 second between requests
self._max_retries = 5
@ -130,7 +130,7 @@ class TibberPricesApiClient:
}
"""
},
query_type=QueryType.USER,
query_type=TibberPricesQueryType.USER,
)
async def async_get_price_info(self, home_ids: set[str]) -> dict:
@ -183,7 +183,7 @@ class TibberPricesApiClient:
data = await self._api_wrapper(
data={"query": query},
query_type=QueryType.PRICE_INFO,
query_type=TibberPricesQueryType.PRICE_INFO,
)
# Parse aliased response
@ -234,7 +234,7 @@ class TibberPricesApiClient:
}
}}}}}"""
},
query_type=QueryType.DAILY_RATING,
query_type=TibberPricesQueryType.DAILY_RATING,
)
homes = data.get("viewer", {}).get("homes", [])
@ -266,7 +266,7 @@ class TibberPricesApiClient:
}
}}}}}"""
},
query_type=QueryType.HOURLY_RATING,
query_type=TibberPricesQueryType.HOURLY_RATING,
)
homes = data.get("viewer", {}).get("homes", [])
@ -298,7 +298,7 @@ class TibberPricesApiClient:
}
}}}}}"""
},
query_type=QueryType.MONTHLY_RATING,
query_type=TibberPricesQueryType.MONTHLY_RATING,
)
homes = data.get("viewer", {}).get("homes", [])
@ -322,7 +322,7 @@ class TibberPricesApiClient:
self,
headers: dict[str, str],
data: dict,
query_type: QueryType,
query_type: TibberPricesQueryType,
) -> dict[str, Any]:
"""Make an API request with comprehensive error handling for network issues."""
_LOGGER.debug("Making API request with data: %s", data)
@ -443,7 +443,7 @@ class TibberPricesApiClient:
self,
headers: dict[str, str],
data: dict,
query_type: QueryType,
query_type: TibberPricesQueryType,
) -> Any:
"""Handle a single API request with rate limiting."""
async with self._request_semaphore:
@ -547,7 +547,7 @@ class TibberPricesApiClient:
self,
data: dict | None = None,
headers: dict | None = None,
query_type: QueryType = QueryType.USER,
query_type: TibberPricesQueryType = TibberPricesQueryType.USER,
) -> Any:
"""Get information from the API with rate limiting and retry logic."""
headers = headers or prepare_headers(self._access_token, self._version)

8
custom_components/tibber_prices/api/helpers.py
View file

@ -11,9 +11,9 @@ from homeassistant.const import __version__ as ha_version
if TYPE_CHECKING:
import aiohttp
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .queries import QueryType
from .queries import TibberPricesQueryType
from .exceptions import (
TibberPricesApiClientAuthenticationError,
@ -50,7 +50,7 @@ def verify_response_or_raise(response: aiohttp.ClientResponse) -> None:
response.raise_for_status()
async def verify_graphql_response(response_json: dict, query_type: QueryType) -> None:
async def verify_graphql_response(response_json: dict, query_type: TibberPricesQueryType) -> None:
"""Verify the GraphQL response for errors and data completeness, including empty data."""
if "errors" in response_json:
errors = response_json["errors"]
@ -252,7 +252,7 @@ def prepare_headers(access_token: str, version: str) -> dict[str, str]:
}
def flatten_price_info(subscription: dict, currency: str | None = None, *, time: TimeService) -> dict:
def flatten_price_info(subscription: dict, currency: str | None = None, *, time: TibberPricesTimeService) -> dict:
"""
Transform and flatten priceInfo from full API data structure.

2
custom_components/tibber_prices/api/queries.py
View file

@ -5,7 +5,7 @@ from __future__ import annotations
from enum import Enum
class QueryType(Enum):
class TibberPricesQueryType(Enum):
"""Types of queries that can be made to the API."""
PRICE_INFO = "price_info"

24
custom_components/tibber_prices/binary_sensor/attributes.py
View file

@ -7,7 +7,7 @@ from typing import TYPE_CHECKING
from custom_components.tibber_prices.entity_utils import add_icon_color_attribute
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
if TYPE_CHECKING:
from datetime import datetime
@ -19,14 +19,14 @@ if TYPE_CHECKING:
def get_tomorrow_data_available_attributes(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> dict | None:
"""
Build attributes for tomorrow_data_available sensor.
Args:
coordinator_data: Coordinator data dict
time: TimeService instance
time: TibberPricesTimeService instance
Returns:
Attributes dict with intervals_available and data_status
@ -59,7 +59,7 @@ def get_tomorrow_data_available_attributes(
def get_price_intervals_attributes(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
reverse_sort: bool,
) -> dict | None:
"""
@ -72,7 +72,7 @@ def get_price_intervals_attributes(
Args:
coordinator_data: Coordinator data dict
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
reverse_sort: True for peak_price (highest first), False for best_price (lowest first)
Returns:
@ -117,7 +117,7 @@ def get_price_intervals_attributes(
return build_final_attributes_simple(current_period, period_summaries, time=time)
def build_no_periods_result(*, time: TimeService) -> dict:
def build_no_periods_result(*, time: TibberPricesTimeService) -> dict:
"""
Build result when no periods exist (not filtered, just none available).
@ -214,7 +214,7 @@ def build_final_attributes_simple(
current_period: dict | None,
period_summaries: list[dict],
*,
time: TimeService,
time: TibberPricesTimeService,
) -> dict:
"""
Build the final attributes dictionary from coordinator's period summaries.
@ -237,7 +237,7 @@ def build_final_attributes_simple(
Args:
current_period: The current or next period (already complete from coordinator)
period_summaries: All period summaries from coordinator
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Complete attributes dict with all fields
@ -286,7 +286,7 @@ async def build_async_extra_state_attributes( # noqa: PLR0913
translation_key: str | None,
hass: HomeAssistant,
*,
time: TimeService,
time: TibberPricesTimeService,
config_entry: TibberPricesConfigEntry,
sensor_attrs: dict | None = None,
is_on: bool | None = None,
@ -300,7 +300,7 @@ async def build_async_extra_state_attributes( # noqa: PLR0913
entity_key: Entity key (e.g., "best_price_period")
translation_key: Translation key for entity
hass: Home Assistant instance
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
config_entry: Config entry with options (keyword-only)
sensor_attrs: Sensor-specific attributes (keyword-only)
is_on: Binary sensor state (keyword-only)
@ -349,7 +349,7 @@ def build_sync_extra_state_attributes( # noqa: PLR0913
translation_key: str | None,
hass: HomeAssistant,
*,
time: TimeService,
time: TibberPricesTimeService,
config_entry: TibberPricesConfigEntry,
sensor_attrs: dict | None = None,
is_on: bool | None = None,
@ -363,7 +363,7 @@ def build_sync_extra_state_attributes( # noqa: PLR0913
entity_key: Entity key (e.g., "best_price_period")
translation_key: Translation key for entity
hass: Home Assistant instance
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
config_entry: Config entry with options (keyword-only)
sensor_attrs: Sensor-specific attributes (keyword-only)
is_on: Binary sensor state (keyword-only)

6
custom_components/tibber_prices/binary_sensor/core.py
View file

@ -27,7 +27,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
@ -65,12 +65,12 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
self._time_sensitive_remove_listener = None
@callback
def _handle_time_sensitive_update(self, time_service: TimeService) -> None:
def _handle_time_sensitive_update(self, time_service: TibberPricesTimeService) -> None:
"""
Handle time-sensitive update from coordinator.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
# Store TimeService from Timer #2 for calculations during this update cycle

8
custom_components/tibber_prices/config_flow.py
View file

@ -27,17 +27,17 @@ from .config_flow_handlers.subentry_flow import (
)
from .config_flow_handlers.user_flow import TibberPricesFlowHandler as ConfigFlow
from .config_flow_handlers.validators import (
CannotConnectError,
InvalidAuthError,
TibberPricesCannotConnectError,
TibberPricesInvalidAuthError,
validate_api_token,
)
__all__ = [
"CannotConnectError",
"ConfigFlow",
"InvalidAuthError",
"OptionsFlowHandler",
"SubentryFlowHandler",
"TibberPricesCannotConnectError",
"TibberPricesInvalidAuthError",
"get_best_price_schema",
"get_options_init_schema",
"get_peak_price_schema",

8
custom_components/tibber_prices/config_flow_handlers/__init__.py
View file

@ -42,15 +42,15 @@ from custom_components.tibber_prices.config_flow_handlers.user_flow import (
TibberPricesFlowHandler,
)
from custom_components.tibber_prices.config_flow_handlers.validators import (
CannotConnectError,
InvalidAuthError,
TibberPricesCannotConnectError,
TibberPricesInvalidAuthError,
validate_api_token,
)
__all__ = [
"CannotConnectError",
"InvalidAuthError",
"TibberPricesCannotConnectError",
"TibberPricesFlowHandler",
"TibberPricesInvalidAuthError",
"TibberPricesOptionsFlowHandler",
"TibberPricesSubentryFlowHandler",
"get_best_price_schema",

2
custom_components/tibber_prices/config_flow_handlers/options_flow.py
View file

@ -146,7 +146,7 @@ class TibberPricesOptionsFlowHandler(OptionsFlow):
errors["base"] = "invalid_yaml_syntax"
# Test service call with parsed parameters
if not errors and parsed:
if not errors and parsed and isinstance(parsed, dict):
try:
# Add entry_id to service call data
service_data = {**parsed, "entry_id": self.config_entry.entry_id}

12
custom_components/tibber_prices/config_flow_handlers/user_flow.py
View file

@ -16,8 +16,8 @@ from custom_components.tibber_prices.config_flow_handlers.subentry_flow import (
TibberPricesSubentryFlowHandler,
)
from custom_components.tibber_prices.config_flow_handlers.validators import (
CannotConnectError,
InvalidAuthError,
TibberPricesCannotConnectError,
TibberPricesInvalidAuthError,
validate_api_token,
)
from custom_components.tibber_prices.const import DOMAIN, LOGGER
@ -84,10 +84,10 @@ class TibberPricesFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is not None:
try:
viewer = await validate_api_token(self.hass, user_input[CONF_ACCESS_TOKEN])
except InvalidAuthError as exception:
except TibberPricesInvalidAuthError as exception:
LOGGER.warning(exception)
_errors["base"] = "auth"
except CannotConnectError as exception:
except TibberPricesCannotConnectError as exception:
LOGGER.error(exception)
_errors["base"] = "connection"
else:
@ -137,10 +137,10 @@ class TibberPricesFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is not None:
try:
viewer = await validate_api_token(self.hass, user_input[CONF_ACCESS_TOKEN])
except InvalidAuthError as exception:
except TibberPricesInvalidAuthError as exception:
LOGGER.warning(exception)
_errors["base"] = "auth"
except CannotConnectError as exception:
except TibberPricesCannotConnectError as exception:
LOGGER.error(exception)
_errors["base"] = "connection"
else:

14
custom_components/tibber_prices/config_flow_handlers/validators.py
View file

@ -23,11 +23,11 @@ MAX_FLEX_PERCENTAGE = 100.0
MAX_MIN_PERIODS = 10 # Arbitrary upper limit for sanity
class InvalidAuthError(HomeAssistantError):
class TibberPricesInvalidAuthError(HomeAssistantError):
"""Error to indicate invalid authentication."""
class CannotConnectError(HomeAssistantError):
class TibberPricesCannotConnectError(HomeAssistantError):
"""Error to indicate we cannot connect."""
@ -43,8 +43,8 @@ async def validate_api_token(hass: HomeAssistant, token: str) -> dict:
dict with viewer data on success
Raises:
InvalidAuthError: Invalid token
CannotConnectError: API connection failed
TibberPricesInvalidAuthError: Invalid token
TibberPricesCannotConnectError: API connection failed
"""
try:
@ -57,11 +57,11 @@ async def validate_api_token(hass: HomeAssistant, token: str) -> dict:
result = await client.async_get_viewer_details()
return result["viewer"]
except TibberPricesApiClientAuthenticationError as exception:
raise InvalidAuthError from exception
raise TibberPricesInvalidAuthError from exception
except TibberPricesApiClientCommunicationError as exception:
raise CannotConnectError from exception
raise TibberPricesCannotConnectError from exception
except TibberPricesApiClientError as exception:
raise CannotConnectError from exception
raise TibberPricesCannotConnectError from exception
def validate_threshold_range(value: float, min_val: float, max_val: float) -> bool:

4
custom_components/tibber_prices/coordinator/__init__.py
View file

@ -22,12 +22,12 @@ from .constants import (
TIME_SENSITIVE_ENTITY_KEYS,
)
from .core import TibberPricesDataUpdateCoordinator
from .time_service import TimeService
from .time_service import TibberPricesTimeService
__all__ = [
"MINUTE_UPDATE_ENTITY_KEYS",
"STORAGE_VERSION",
"TIME_SENSITIVE_ENTITY_KEYS",
"TibberPricesDataUpdateCoordinator",
"TimeService",
"TibberPricesTimeService",
]

20
custom_components/tibber_prices/coordinator/cache.py
View file

@ -10,12 +10,12 @@ if TYPE_CHECKING:
from homeassistant.helpers.storage import Store
from .time_service import TimeService
from .time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
class CacheData(NamedTuple):
class TibberPricesCacheData(NamedTuple):
"""Cache data structure."""
price_data: dict[str, Any] | None
@ -29,8 +29,8 @@ async def load_cache(
store: Store,
log_prefix: str,
*,
time: TimeService,
) -> CacheData:
time: TibberPricesTimeService,
) -> TibberPricesCacheData:
"""Load cached data from storage."""
try:
stored = await store.async_load()
@ -51,7 +51,7 @@ async def load_cache(
last_midnight_check = time.parse_datetime(last_midnight_check_str)
_LOGGER.debug("%s Cache loaded successfully", log_prefix)
return CacheData(
return TibberPricesCacheData(
price_data=cached_price_data,
user_data=cached_user_data,
last_price_update=last_price_update,
@ -63,7 +63,7 @@ async def load_cache(
except OSError as ex:
_LOGGER.warning("%s Failed to load cache: %s", log_prefix, ex)
return CacheData(
return TibberPricesCacheData(
price_data=None,
user_data=None,
last_price_update=None,
@ -72,9 +72,9 @@ async def load_cache(
)
async def store_cache(
async def save_cache(
store: Store,
cache_data: CacheData,
cache_data: TibberPricesCacheData,
log_prefix: str,
) -> None:
"""Store cache data."""
@ -94,10 +94,10 @@ async def store_cache(
def is_cache_valid(
cache_data: CacheData,
cache_data: TibberPricesCacheData,
log_prefix: str,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> bool:
"""
Validate if cached price data is still current.

42
custom_components/tibber_prices/coordinator/core.py
View file

@ -17,6 +17,8 @@ if TYPE_CHECKING:
from homeassistant.config_entries import ConfigEntry
from .listeners import TimeServiceCallback
from custom_components.tibber_prices import const as _const
from custom_components.tibber_prices.api import (
TibberPricesApiClient,
@ -34,11 +36,11 @@ from .constants import (
STORAGE_VERSION,
UPDATE_INTERVAL,
)
from .data_fetching import DataFetcher
from .data_transformation import DataTransformer
from .listeners import ListenerManager
from .periods import PeriodCalculator
from .time_service import TimeService
from .data_fetching import TibberPricesDataFetcher
from .data_transformation import TibberPricesDataTransformer
from .listeners import TibberPricesListenerManager
from .periods import TibberPricesPeriodCalculator
from .time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
@ -134,28 +136,28 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Track if this is the main entry (first one created)
self._is_main_entry = not self._has_existing_main_coordinator()
# Initialize time service (single source of truth for datetime operations)
self.time = TimeService()
# Initialize time service (single source of truth for all time operations)
self.time = TibberPricesTimeService()
# Set time on API client (needed for rate limiting)
self.api.time = self.time
# Initialize helper modules
self._listener_manager = ListenerManager(hass, self._log_prefix)
self._data_fetcher = DataFetcher(
self._listener_manager = TibberPricesListenerManager(hass, self._log_prefix)
self._data_fetcher = TibberPricesDataFetcher(
api=self.api,
store=self._store,
log_prefix=self._log_prefix,
user_update_interval=timedelta(days=1),
time=self.time,
)
self._data_transformer = DataTransformer(
self._data_transformer = TibberPricesDataTransformer(
config_entry=config_entry,
log_prefix=self._log_prefix,
perform_turnover_fn=self._perform_midnight_turnover,
time=self.time,
)
self._period_calculator = PeriodCalculator(
self._period_calculator = TibberPricesPeriodCalculator(
config_entry=config_entry,
log_prefix=self._log_prefix,
)
@ -191,7 +193,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
await self.async_request_refresh()
@callback
def async_add_time_sensitive_listener(self, update_callback: CALLBACK_TYPE) -> CALLBACK_TYPE:
def async_add_time_sensitive_listener(self, update_callback: TimeServiceCallback) -> CALLBACK_TYPE:
"""
Listen for time-sensitive updates that occur every quarter-hour.
@ -205,18 +207,18 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
return self._listener_manager.async_add_time_sensitive_listener(update_callback)
@callback
def _async_update_time_sensitive_listeners(self, time_service: TimeService) -> None:
def _async_update_time_sensitive_listeners(self, time_service: TibberPricesTimeService) -> None:
"""
Update all time-sensitive entities without triggering a full coordinator update.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
self._listener_manager.async_update_time_sensitive_listeners(time_service)
@callback
def async_add_minute_update_listener(self, update_callback: CALLBACK_TYPE) -> CALLBACK_TYPE:
def async_add_minute_update_listener(self, update_callback: TimeServiceCallback) -> CALLBACK_TYPE:
"""
Listen for minute-by-minute updates for timing sensors.
@ -230,12 +232,12 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
return self._listener_manager.async_add_minute_update_listener(update_callback)
@callback
def _async_update_minute_listeners(self, time_service: TimeService) -> None:
def _async_update_minute_listeners(self, time_service: TibberPricesTimeService) -> None:
"""
Update all minute-update entities without triggering a full coordinator update.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
self._listener_manager.async_update_minute_listeners(time_service)
@ -259,7 +261,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Each timer has its own TimeService instance - no shared state between timers
# This timer updates 30+ time-sensitive entities at quarter-hour boundaries
# (Timer #3 handles timing entities separately - no overlap)
time_service = TimeService()
time_service = TibberPricesTimeService()
now = time_service.now()
# Update shared coordinator time (used by Timer #1 and other operations)
@ -311,7 +313,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Timer #2 updates 30+ time-sensitive entities (prices, levels, timestamps)
# Timer #3 updates 6 timing entities (remaining_minutes, progress, next_in_minutes)
# NO overlap - entities are registered with either Timer #2 OR Timer #3, never both
time_service = TimeService()
time_service = TibberPricesTimeService()
# Only log at debug level to avoid log spam (this runs every 30 seconds)
self._log("debug", "[Timer #3] 30-second refresh for timing sensors")
@ -447,7 +449,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._log("debug", "[Timer #1] DataUpdateCoordinator check triggered")
# Create TimeService with fresh reference time for this update cycle
self.time = TimeService()
self.time = TibberPricesTimeService()
current_time = self.time.now()
# Update helper modules with fresh TimeService instance

12
custom_components/tibber_prices/coordinator/data_fetching.py
View file

@ -28,12 +28,12 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.api import TibberPricesApiClient
from .time_service import TimeService
from .time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
class DataFetcher:
class TibberPricesDataFetcher:
"""Handles data fetching, caching, and main/subentry coordination."""
def __init__(
@ -42,14 +42,14 @@ class DataFetcher:
store: Any,
log_prefix: str,
user_update_interval: timedelta,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""Initialize the data fetcher."""
self.api = api
self._store = store
self._log_prefix = log_prefix
self._user_update_interval = user_update_interval
self.time = time
self.time: TibberPricesTimeService = time
# Cached data
self._cached_price_data: dict[str, Any] | None = None
@ -84,14 +84,14 @@ class DataFetcher:
async def store_cache(self, last_midnight_check: datetime | None = None) -> None:
"""Store cache data."""
cache_data = cache.CacheData(
cache_data = cache.TibberPricesCacheData(
price_data=self._cached_price_data,
user_data=self._cached_user_data,
last_price_update=self._last_price_update,
last_user_update=self._last_user_update,
last_midnight_check=last_midnight_check,
)
await cache.store_cache(self._store, cache_data, self._log_prefix)
await cache.save_cache(self._store, cache_data, self._log_prefix)
async def update_user_data_if_needed(self, current_time: datetime) -> None:
"""Update user data if needed (daily check)."""

8
custom_components/tibber_prices/coordinator/data_transformation.py
View file

@ -14,12 +14,12 @@ if TYPE_CHECKING:
from homeassistant.config_entries import ConfigEntry
from .time_service import TimeService
from .time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
class DataTransformer:
class TibberPricesDataTransformer:
"""Handles data transformation, enrichment, and period calculations."""
def __init__(
@ -27,13 +27,13 @@ class DataTransformer:
config_entry: ConfigEntry,
log_prefix: str,
perform_turnover_fn: Callable[[dict[str, Any]], dict[str, Any]],
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""Initialize the data transformer."""
self.config_entry = config_entry
self._log_prefix = log_prefix
self._perform_turnover_fn = perform_turnover_fn
self.time = time
self.time: TibberPricesTimeService = time
# Transformation cache
self._cached_transformed_data: dict[str, Any] | None = None

10
custom_components/tibber_prices/coordinator/helpers.py
View file

@ -10,7 +10,7 @@ if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
from .time_service import TimeService
from .time_service import TibberPricesTimeService
from custom_components.tibber_prices.const import DOMAIN
@ -57,7 +57,7 @@ def needs_tomorrow_data(
return False
def perform_midnight_turnover(price_info: dict[str, Any], *, time: TimeService) -> dict[str, Any]:
def perform_midnight_turnover(price_info: dict[str, Any], *, time: TibberPricesTimeService) -> dict[str, Any]:
"""
Perform midnight turnover on price data.
@ -69,7 +69,7 @@ def perform_midnight_turnover(price_info: dict[str, Any], *, time: TimeService)
Args:
price_info: The price info dict with 'today', 'tomorrow', 'yesterday' keys
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Updated price_info with rotated day data
@ -103,7 +103,7 @@ def perform_midnight_turnover(price_info: dict[str, Any], *, time: TimeService)
return price_info
def parse_all_timestamps(price_data: dict[str, Any], *, time: TimeService) -> dict[str, Any]:
def parse_all_timestamps(price_data: dict[str, Any], *, time: TibberPricesTimeService) -> dict[str, Any]:
"""
Parse all API timestamp strings to datetime objects.
@ -114,7 +114,7 @@ def parse_all_timestamps(price_data: dict[str, Any], *, time: TimeService) -> di
Args:
price_data: Raw API data with string timestamps
time: TimeService for parsing
time: TibberPricesTimeService for parsing
Returns:
Same structure but with datetime objects instead of strings

28
custom_components/tibber_prices/coordinator/listeners.py
View file

@ -11,16 +11,20 @@ from homeassistant.helpers.event import async_track_utc_time_change
from .constants import QUARTER_HOUR_BOUNDARIES
if TYPE_CHECKING:
from collections.abc import Callable
from datetime import datetime
from homeassistant.core import HomeAssistant
from .time_service import TimeService
from .time_service import TibberPricesTimeService
# Callback type that accepts TibberPricesTimeService parameter
TimeServiceCallback = Callable[[TibberPricesTimeService], None]
_LOGGER = logging.getLogger(__name__)
class ListenerManager:
class TibberPricesListenerManager:
"""Manages listeners and scheduling for coordinator updates."""
def __init__(self, hass: HomeAssistant, log_prefix: str) -> None:
@ -29,8 +33,8 @@ class ListenerManager:
self._log_prefix = log_prefix
# Listener lists
self._time_sensitive_listeners: list[CALLBACK_TYPE] = []
self._minute_update_listeners: list[CALLBACK_TYPE] = []
self._time_sensitive_listeners: list[TimeServiceCallback] = []
self._minute_update_listeners: list[TimeServiceCallback] = []
# Timer cancellation callbacks
self._quarter_hour_timer_cancel: CALLBACK_TYPE | None = None
@ -45,7 +49,7 @@ class ListenerManager:
getattr(_LOGGER, level)(prefixed_message, *args, **kwargs)
@callback
def async_add_time_sensitive_listener(self, update_callback: CALLBACK_TYPE) -> CALLBACK_TYPE:
def async_add_time_sensitive_listener(self, update_callback: TimeServiceCallback) -> CALLBACK_TYPE:
"""
Listen for time-sensitive updates that occur every quarter-hour.
@ -66,12 +70,12 @@ class ListenerManager:
return remove_listener
@callback
def async_update_time_sensitive_listeners(self, time_service: TimeService) -> None:
def async_update_time_sensitive_listeners(self, time_service: TibberPricesTimeService) -> None:
"""
Update all time-sensitive entities without triggering a full coordinator update.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
for update_callback in self._time_sensitive_listeners:
@ -84,7 +88,7 @@ class ListenerManager:
)
@callback
def async_add_minute_update_listener(self, update_callback: CALLBACK_TYPE) -> CALLBACK_TYPE:
def async_add_minute_update_listener(self, update_callback: TimeServiceCallback) -> CALLBACK_TYPE:
"""
Listen for minute-by-minute updates for timing sensors.
@ -105,12 +109,12 @@ class ListenerManager:
return remove_listener
@callback
def async_update_minute_listeners(self, time_service: TimeService) -> None:
def async_update_minute_listeners(self, time_service: TibberPricesTimeService) -> None:
"""
Update all minute-update entities without triggering a full coordinator update.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
for update_callback in self._minute_update_listeners:
@ -124,7 +128,7 @@ class ListenerManager:
def schedule_quarter_hour_refresh(
self,
handler_callback: CALLBACK_TYPE,
handler_callback: Callable[[datetime], None],
) -> None:
"""Schedule the next quarter-hour entity refresh using Home Assistant's time tracking."""
# Cancel any existing timer
@ -151,7 +155,7 @@ class ListenerManager:
def schedule_minute_refresh(
self,
handler_callback: CALLBACK_TYPE,
handler_callback: Callable[[datetime], None],
) -> None:
"""Schedule 30-second entity refresh for timing sensors."""
# Cancel any existing timer

20
custom_components/tibber_prices/coordinator/period_handlers/__init__.py
View file

@ -33,11 +33,11 @@ from .types import (
INDENT_L3,
INDENT_L4,
INDENT_L5,
IntervalCriteria,
PeriodConfig,
PeriodData,
PeriodStatistics,
ThresholdConfig,
TibberPricesIntervalCriteria,
TibberPricesPeriodConfig,
TibberPricesPeriodData,
TibberPricesPeriodStatistics,
TibberPricesThresholdConfig,
)
__all__ = [
@ -47,11 +47,11 @@ __all__ = [
"INDENT_L3",
"INDENT_L4",
"INDENT_L5",
"IntervalCriteria",
"PeriodConfig",
"PeriodData",
"PeriodStatistics",
"ThresholdConfig",
"TibberPricesIntervalCriteria",
"TibberPricesPeriodConfig",
"TibberPricesPeriodData",
"TibberPricesPeriodStatistics",
"TibberPricesThresholdConfig",
"calculate_periods",
"calculate_periods_with_relaxation",
"filter_price_outliers",

14
custom_components/tibber_prices/coordinator/period_handlers/core.py
View file

@ -5,9 +5,9 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .types import PeriodConfig
from .types import TibberPricesPeriodConfig
from .outlier_filtering import (
filter_price_outliers,
@ -23,7 +23,7 @@ from .period_building import (
from .period_statistics import (
extract_period_summaries,
)
from .types import ThresholdConfig
from .types import TibberPricesThresholdConfig
# Flex limits to prevent degenerate behavior (see docs/development/period-calculation-theory.md)
MAX_SAFE_FLEX = 0.50 # 50% - hard cap: above this, period detection becomes unreliable
@ -33,8 +33,8 @@ MAX_OUTLIER_FLEX = 0.25 # 25% - cap for outlier filtering: above this, spike de
def calculate_periods(
all_prices: list[dict],
*,
config: PeriodConfig,
time: TimeService,
config: TibberPricesPeriodConfig,
time: TibberPricesTimeService,
) -> dict[str, Any]:
"""
Calculate price periods (best or peak) from price data.
@ -55,7 +55,7 @@ def calculate_periods(
all_prices: All price data points from yesterday/today/tomorrow
config: Period configuration containing reverse_sort, flex, min_distance_from_avg,
min_period_length, threshold_low, and threshold_high
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Dict with:
@ -183,7 +183,7 @@ def calculate_periods(
# Step 8: Extract lightweight period summaries (no full price data)
# Note: Filtering for current/future is done here based on end date,
# not start date. This preserves periods that started yesterday but end today.
thresholds = ThresholdConfig(
thresholds = TibberPricesThresholdConfig(
threshold_low=threshold_low,
threshold_high=threshold_high,
threshold_volatility_moderate=config.threshold_volatility_moderate,

4
custom_components/tibber_prices/coordinator/period_handlers/level_filtering.py
View file

@ -14,7 +14,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .types import IntervalCriteria
from .types import TibberPricesIntervalCriteria
from custom_components.tibber_prices.const import PRICE_LEVEL_MAPPING
@ -104,7 +104,7 @@ def apply_level_filter(
def check_interval_criteria(
price: float,
criteria: IntervalCriteria,
criteria: TibberPricesIntervalCriteria,
) -> tuple[bool, bool]:
"""
Check if interval meets flex and minimum distance criteria.

9
custom_components/tibber_prices/coordinator/period_handlers/outlier_filtering.py
View file

@ -15,7 +15,7 @@ Uses statistical methods:
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import NamedTuple
_LOGGER = logging.getLogger(__name__)
@ -32,8 +32,7 @@ ZIGZAG_TAIL_WINDOW = 6 # Skip zigzag/cluster detection for last ~1.5h (6 interv
INDENT_L0 = "" # All logs in this module (no indentation needed)
@dataclass(slots=True)
class SpikeCandidateContext:
class TibberPricesSpikeCandidateContext(NamedTuple):
"""Container for spike validation parameters."""
current: dict
@ -183,7 +182,7 @@ def _detect_zigzag_pattern(window: list[dict], context_std_dev: float) -> bool:
def _validate_spike_candidate(
candidate: SpikeCandidateContext,
candidate: TibberPricesSpikeCandidateContext,
) -> bool:
"""Run stability, symmetry, and zigzag checks before smoothing."""
avg_before = sum(x["total"] for x in candidate.context_before) / len(candidate.context_before)
@ -308,7 +307,7 @@ def filter_price_outliers(
# SPIKE CANDIDATE DETECTED - Now validate
remaining_intervals = len(intervals) - (i + 1)
analysis_window = [*context_before[-2:], current, *context_after[:2]]
candidate_context = SpikeCandidateContext(
candidate_context = TibberPricesSpikeCandidateContext(
current=current,
context_before=context_before,
context_after=context_after,

18
custom_components/tibber_prices/coordinator/period_handlers/period_building.py
View file

@ -10,13 +10,13 @@ from custom_components.tibber_prices.const import PRICE_LEVEL_MAPPING
if TYPE_CHECKING:
from datetime import date
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .level_filtering import (
apply_level_filter,
check_interval_criteria,
)
from .types import IntervalCriteria
from .types import TibberPricesIntervalCriteria
_LOGGER = logging.getLogger(__name__)
@ -25,7 +25,7 @@ INDENT_L0 = "" # Entry point / main function
def split_intervals_by_day(
all_prices: list[dict], *, time: TimeService
all_prices: list[dict], *, time: TibberPricesTimeService
) -> tuple[dict[date, list[dict]], dict[date, float]]:
"""Split intervals by day and calculate average price per day."""
intervals_by_day: dict[date, list[dict]] = {}
@ -60,7 +60,7 @@ def build_periods( # noqa: PLR0913, PLR0915, PLR0912 - Complex period building
reverse_sort: bool,
level_filter: str | None = None,
gap_count: int = 0,
time: TimeService,
time: TibberPricesTimeService,
) -> list[list[dict]]:
"""
Build periods, allowing periods to cross midnight (day boundary).
@ -75,7 +75,7 @@ def build_periods( # noqa: PLR0913, PLR0915, PLR0912 - Complex period building
reverse_sort: True for peak price (high prices), False for best price (low prices)
level_filter: Level filter string ("cheap", "expensive", "any", None)
gap_count: Number of allowed consecutive intervals deviating by exactly 1 level step
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
ref_prices = price_context["ref_prices"]
@ -130,7 +130,7 @@ def build_periods( # noqa: PLR0913, PLR0915, PLR0912 - Complex period building
ref_date = period_start_date if period_start_date is not None else date_key
# Check flex and minimum distance criteria (using smoothed price and period start date reference)
criteria = IntervalCriteria(
criteria = TibberPricesIntervalCriteria(
ref_price=ref_prices[ref_date],
avg_price=avg_prices[ref_date],
flex=flex,
@ -230,14 +230,14 @@ def build_periods( # noqa: PLR0913, PLR0915, PLR0912 - Complex period building
def filter_periods_by_min_length(
periods: list[list[dict]], min_period_length: int, *, time: TimeService
periods: list[list[dict]], min_period_length: int, *, time: TibberPricesTimeService
) -> list[list[dict]]:
"""Filter periods to only include those meeting the minimum length requirement."""
min_intervals = time.minutes_to_intervals(min_period_length)
return [period for period in periods if len(period) >= min_intervals]
def add_interval_ends(periods: list[list[dict]], *, time: TimeService) -> None:
def add_interval_ends(periods: list[list[dict]], *, time: TibberPricesTimeService) -> None:
"""Add interval_end to each interval in-place."""
interval_duration = time.get_interval_duration()
for period in periods:
@ -247,7 +247,7 @@ def add_interval_ends(periods: list[list[dict]], *, time: TimeService) -> None:
interval["interval_end"] = start + interval_duration
def filter_periods_by_end_date(periods: list[list[dict]], *, time: TimeService) -> list[list[dict]]:
def filter_periods_by_end_date(periods: list[list[dict]], *, time: TibberPricesTimeService) -> list[list[dict]]:
"""
Filter periods to keep only relevant ones for today and tomorrow.

6
custom_components/tibber_prices/coordinator/period_handlers/period_overlap.py
View file

@ -6,7 +6,7 @@ import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
_LOGGER = logging.getLogger(__name__)
@ -16,7 +16,7 @@ INDENT_L1 = " " # Nested logic / loop iterations
INDENT_L2 = " " # Deeper nesting
def recalculate_period_metadata(periods: list[dict], *, time: TimeService) -> None:
def recalculate_period_metadata(periods: list[dict], *, time: TibberPricesTimeService) -> None:
"""
Recalculate period metadata after merging periods.
@ -28,7 +28,7 @@ def recalculate_period_metadata(periods: list[dict], *, time: TimeService) -> No
Args:
periods: List of period summary dicts (mutated in-place)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
if not periods:

26
custom_components/tibber_prices/coordinator/period_handlers/period_statistics.py
View file

@ -7,12 +7,12 @@ from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from datetime import datetime
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .types import (
PeriodData,
PeriodStatistics,
ThresholdConfig,
TibberPricesPeriodData,
TibberPricesPeriodStatistics,
TibberPricesThresholdConfig,
)
from custom_components.tibber_prices.utils.price import (
aggregate_period_levels,
@ -115,8 +115,8 @@ def calculate_period_price_statistics(period_price_data: list[dict]) -> dict[str
def build_period_summary_dict(
period_data: PeriodData,
stats: PeriodStatistics,
period_data: TibberPricesPeriodData,
stats: TibberPricesPeriodStatistics,
*,
reverse_sort: bool,
) -> dict:
@ -176,9 +176,9 @@ def extract_period_summaries(
periods: list[list[dict]],
all_prices: list[dict],
price_context: dict[str, Any],
thresholds: ThresholdConfig,
thresholds: TibberPricesThresholdConfig,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> list[dict]:
"""
Extract complete period summaries with all aggregated attributes.
@ -199,12 +199,12 @@ def extract_period_summaries(
all_prices: All price data from the API (enriched with level, difference, rating_level)
price_context: Dictionary with ref_prices and avg_prices per day
thresholds: Threshold configuration for calculations
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
from .types import ( # noqa: PLC0415 - Avoid circular import
PeriodData,
PeriodStatistics,
TibberPricesPeriodData,
TibberPricesPeriodStatistics,
)
# Build lookup dictionary for full price data by timestamp
@ -284,7 +284,7 @@ def extract_period_summaries(
level_gap_count = sum(1 for interval in period if interval.get("is_level_gap", False))
# Build period data and statistics objects
period_data = PeriodData(
period_data = TibberPricesPeriodData(
start_time=start_time,
end_time=end_time,
period_length=len(period),
@ -292,7 +292,7 @@ def extract_period_summaries(
total_periods=total_periods,
)
stats = PeriodStatistics(
stats = TibberPricesPeriodStatistics(
aggregated_level=aggregated_level,
aggregated_rating=aggregated_rating,
rating_difference_pct=rating_difference_pct,

30
custom_components/tibber_prices/coordinator/period_handlers/relaxation.py
View file

@ -9,9 +9,9 @@ if TYPE_CHECKING:
from collections.abc import Callable
from datetime import date
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .types import PeriodConfig
from .types import TibberPricesPeriodConfig
from .period_overlap import (
recalculate_period_metadata,
@ -60,13 +60,13 @@ def group_periods_by_day(periods: list[dict]) -> dict[date, list[dict]]:
return periods_by_day
def group_prices_by_day(all_prices: list[dict], *, time: TimeService) -> dict[date, list[dict]]:
def group_prices_by_day(all_prices: list[dict], *, time: TibberPricesTimeService) -> dict[date, list[dict]]:
"""
Group price intervals by the day they belong to (today and future only).
Args:
all_prices: List of price dicts with "startsAt" timestamp
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Dict mapping date to list of price intervals for that day (only today and future)
@ -87,7 +87,7 @@ def group_prices_by_day(all_prices: list[dict], *, time: TimeService) -> dict[da
def check_min_periods_per_day(
periods: list[dict], min_periods: int, all_prices: list[dict], *, time: TimeService
periods: list[dict], min_periods: int, all_prices: list[dict], *, time: TibberPricesTimeService
) -> bool:
"""
Check if minimum periods requirement is met for each day individually.
@ -99,7 +99,7 @@ def check_min_periods_per_day(
periods: List of period summary dicts
min_periods: Minimum number of periods required per day
all_prices: All available price intervals (used to determine which days have data)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
True if every day with price data has at least min_periods, False otherwise
@ -171,12 +171,12 @@ def mark_periods_with_relaxation(
def calculate_periods_with_relaxation( # noqa: PLR0913, PLR0915 - Per-day relaxation requires many parameters and statements
all_prices: list[dict],
*,
config: PeriodConfig,
config: TibberPricesPeriodConfig,
enable_relaxation: bool,
min_periods: int,
max_relaxation_attempts: int,
should_show_callback: Callable[[str | None], bool],
time: TimeService,
time: TibberPricesTimeService,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""
Calculate periods with optional per-day filter relaxation.
@ -201,7 +201,7 @@ def calculate_periods_with_relaxation( # noqa: PLR0913, PLR0915 - Per-day relax
should_show_callback: Callback function(level_override) -> bool
Returns True if periods should be shown with given filter overrides. Pass None
to use original configured filter values.
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Tuple of (periods_result, relaxation_metadata):
@ -273,7 +273,11 @@ def calculate_periods_with_relaxation( # noqa: PLR0913, PLR0915 - Per-day relax
# Validate we have price data for today/future
today = time.now().date()
future_prices = [p for p in all_prices if time.get_interval_time(p).date() >= today]
future_prices = [
p
for p in all_prices
if (interval_time := time.get_interval_time(p)) is not None and interval_time.date() >= today
]
if not future_prices:
# No price data for today/future
@ -394,13 +398,13 @@ def calculate_periods_with_relaxation( # noqa: PLR0913, PLR0915 - Per-day relax
def relax_all_prices( # noqa: PLR0913 - Comprehensive filter relaxation requires many parameters and statements
all_prices: list[dict],
config: PeriodConfig,
config: TibberPricesPeriodConfig,
min_periods: int,
max_relaxation_attempts: int,
should_show_callback: Callable[[str | None], bool],
baseline_periods: list[dict],
*,
time: TimeService,
time: TibberPricesTimeService,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""
Relax filters for all prices until min_periods per day is reached.
@ -416,7 +420,7 @@ def relax_all_prices( # noqa: PLR0913 - Comprehensive filter relaxation require
max_relaxation_attempts: Maximum flex levels to try
should_show_callback: Callback to check if a flex level should be shown
baseline_periods: Baseline periods (before relaxation)
time: TimeService instance
time: TibberPricesTimeService instance
Returns:
Tuple of (result_dict, metadata_dict)

10
custom_components/tibber_prices/coordinator/period_handlers/types.py
View file

@ -24,7 +24,7 @@ INDENT_L4 = " " # Period-by-period analysis
INDENT_L5 = " " # Segment details
class PeriodConfig(NamedTuple):
class TibberPricesPeriodConfig(NamedTuple):
"""Configuration for period calculation."""
reverse_sort: bool
@ -40,7 +40,7 @@ class PeriodConfig(NamedTuple):
gap_count: int = 0 # Number of allowed consecutive deviating intervals
class PeriodData(NamedTuple):
class TibberPricesPeriodData(NamedTuple):
"""Data for building a period summary."""
start_time: datetime
@ -50,7 +50,7 @@ class PeriodData(NamedTuple):
total_periods: int
class PeriodStatistics(NamedTuple):
class TibberPricesPeriodStatistics(NamedTuple):
"""Calculated statistics for a period."""
aggregated_level: str | None
@ -65,7 +65,7 @@ class PeriodStatistics(NamedTuple):
period_price_diff_pct: float | None
class ThresholdConfig(NamedTuple):
class TibberPricesThresholdConfig(NamedTuple):
"""Threshold configuration for period calculations."""
threshold_low: float | None
@ -76,7 +76,7 @@ class ThresholdConfig(NamedTuple):
reverse_sort: bool
class IntervalCriteria(NamedTuple):
class TibberPricesIntervalCriteria(NamedTuple):
"""Criteria for checking if an interval qualifies for a period."""
ref_price: float

12
custom_components/tibber_prices/coordinator/periods.py
View file

@ -13,10 +13,10 @@ from typing import TYPE_CHECKING, Any
from custom_components.tibber_prices import const as _const
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .period_handlers import (
PeriodConfig,
TibberPricesPeriodConfig,
calculate_periods_with_relaxation,
)
@ -26,7 +26,7 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__)
class PeriodCalculator:
class TibberPricesPeriodCalculator:
"""Handles period calculations with level filtering and gap tolerance."""
def __init__(
@ -37,7 +37,7 @@ class PeriodCalculator:
"""Initialize the period calculator."""
self.config_entry = config_entry
self._log_prefix = log_prefix
self.time: TimeService # Set by coordinator before first use
self.time: TibberPricesTimeService # Set by coordinator before first use
self._config_cache: dict[str, dict[str, Any]] | None = None
self._config_cache_valid = False
@ -602,7 +602,7 @@ class PeriodCalculator:
_const.CONF_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
_const.DEFAULT_BEST_PRICE_MAX_LEVEL_GAP_COUNT,
)
best_period_config = PeriodConfig(
best_period_config = TibberPricesPeriodConfig(
reverse_sort=False,
flex=best_config["flex"],
min_distance_from_avg=best_config["min_distance_from_avg"],
@ -670,7 +670,7 @@ class PeriodCalculator:
_const.CONF_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
_const.DEFAULT_PEAK_PRICE_MAX_LEVEL_GAP_COUNT,
)
peak_period_config = PeriodConfig(
peak_period_config = TibberPricesPeriodConfig(
reverse_sort=True,
flex=peak_config["flex"],
min_distance_from_avg=peak_config["min_distance_from_avg"],

19
custom_components/tibber_prices/coordinator/time_service.py
View file

@ -45,7 +45,7 @@ _INTERVALS_PER_DAY = 24 * _INTERVALS_PER_HOUR # 96
_BOUNDARY_TOLERANCE_SECONDS = 2
class TimeService:
class TibberPricesTimeService:
"""
Centralized time service for Tibber Prices integration.
@ -669,6 +669,10 @@ class TimeService:
# - Normal day: 24 hours (96 intervals)
#
tz = self._reference_time.tzinfo # Get timezone from reference time
if tz is None:
# Should never happen - dt_util.now() always returns timezone-aware datetime
msg = "Reference time has no timezone information"
raise ValueError(msg)
# Create naive datetimes for midnight of target and next day
start_naive = datetime.combine(target_date, datetime.min.time())
@ -678,8 +682,9 @@ class TimeService:
# Localize to get correct DST offset for each date
if hasattr(tz, "localize"):
# pytz timezone - use localize() to handle DST correctly
start_midnight_local = tz.localize(start_naive)
end_midnight_local = tz.localize(end_naive)
# Type checker doesn't understand hasattr runtime check, but this is safe
start_midnight_local = tz.localize(start_naive) # type: ignore[attr-defined]
end_midnight_local = tz.localize(end_naive) # type: ignore[attr-defined]
else:
# zoneinfo or other timezone - can use replace directly
start_midnight_local = start_naive.replace(tzinfo=tz)
@ -767,9 +772,9 @@ class TimeService:
# Time-Travel Support
# -------------------------------------------------------------------------
def with_reference_time(self, new_time: datetime) -> TimeService:
def with_reference_time(self, new_time: datetime) -> TibberPricesTimeService:
"""
Create new TimeService with different reference time.
Create new TibberPricesTimeService with different reference time.
Used for time-travel testing: inject simulated "now".
@ -777,7 +782,7 @@ class TimeService:
new_time: New reference time.
Returns:
New TimeService instance with updated reference time.
New TibberPricesTimeService instance with updated reference time.
Example:
# Simulate being at 14:30 on 2025-11-19
@ -785,4 +790,4 @@ class TimeService:
future_service = time_service.with_reference_time(simulated_time)
"""
return TimeService(reference_time=new_time)
return TibberPricesTimeService(reference_time=new_time)

6
custom_components/tibber_prices/entity_utils/helpers.py
View file

@ -19,7 +19,7 @@ from custom_components.tibber_prices.const import get_price_level_translation
if TYPE_CHECKING:
from datetime import datetime
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from homeassistant.core import HomeAssistant
@ -91,7 +91,7 @@ def find_rolling_hour_center_index(
current_time: datetime,
hour_offset: int,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> int | None:
"""
Find the center index for the rolling hour window.
@ -100,7 +100,7 @@ def find_rolling_hour_center_index(
all_prices: List of all price interval dictionaries with 'startsAt' key
current_time: Current datetime to find the current interval
hour_offset: Number of hours to offset from current interval (can be negative)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Index of the center interval for the rolling hour window, or None if not found

22
custom_components/tibber_prices/entity_utils/icons.py
View file

@ -8,7 +8,7 @@ from datetime import timedelta
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from custom_components.tibber_prices.const import (
BINARY_SENSOR_ICON_MAPPING,
@ -26,14 +26,14 @@ _INTERVAL_MINUTES = 15 # Tibber's 15-minute intervals
@dataclass
class IconContext:
class TibberPricesIconContext:
"""Context data for dynamic icon selection."""
is_on: bool | None = None
coordinator_data: dict | None = None
has_future_periods_callback: Callable[[], bool] | None = None
period_is_active_callback: Callable[[], bool] | None = None
time: TimeService | None = None
time: TibberPricesTimeService | None = None
if TYPE_CHECKING:
@ -53,7 +53,7 @@ def get_dynamic_icon(
key: str,
value: Any,
*,
context: IconContext | None = None,
context: TibberPricesIconContext | None = None,
) -> str | None:
"""
Get dynamic icon based on sensor state.
@ -69,7 +69,7 @@ def get_dynamic_icon(
Icon string or None if no dynamic icon applies
"""
ctx = context or IconContext()
ctx = context or TibberPricesIconContext()
# Try various icon sources in order
return (
@ -173,7 +173,7 @@ def get_price_sensor_icon(
key: str,
coordinator_data: dict | None,
*,
time: TimeService | None,
time: TibberPricesTimeService | None,
) -> str | None:
"""
Get icon for current price sensors (dynamic based on price level).
@ -185,7 +185,7 @@ def get_price_sensor_icon(
Args:
key: Entity description key
coordinator_data: Coordinator data for price level lookups
time: TimeService instance (required for determining current interval)
time: TibberPricesTimeService instance (required for determining current interval)
Returns:
Icon string or None if not a current price sensor
@ -300,7 +300,7 @@ def get_price_level_for_icon(
coordinator_data: dict,
*,
interval_offset: int | None = None,
time: TimeService,
time: TibberPricesTimeService,
) -> str | None:
"""
Get the price level for icon determination.
@ -310,7 +310,7 @@ def get_price_level_for_icon(
Args:
coordinator_data: Coordinator data
interval_offset: Interval offset (0=current, 1=next, -1=previous)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Price level string or None if not found
@ -336,7 +336,7 @@ def get_rolling_hour_price_level_for_icon(
coordinator_data: dict,
*,
hour_offset: int = 0,
time: TimeService,
time: TibberPricesTimeService,
) -> str | None:
"""
Get the aggregated price level for rolling hour icon determination.
@ -349,7 +349,7 @@ def get_rolling_hour_price_level_for_icon(
Args:
coordinator_data: Coordinator data
hour_offset: Hour offset (0=current hour, 1=next hour)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Aggregated price level string or None if not found

6
custom_components/tibber_prices/sensor/attributes/__init__.py
View file

@ -19,7 +19,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.core import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from custom_components.tibber_prices.data import TibberPricesConfigEntry
from homeassistant.core import HomeAssistant
@ -171,7 +171,7 @@ def build_extra_state_attributes( # noqa: PLR0913
config_entry: TibberPricesConfigEntry,
coordinator_data: dict,
sensor_attrs: dict | None = None,
time: TimeService | None = None,
time: TibberPricesTimeService,
) -> dict[str, Any] | None:
"""
Build extra state attributes for sensors.
@ -189,7 +189,7 @@ def build_extra_state_attributes( # noqa: PLR0913
config_entry: Config entry with options (keyword-only)
coordinator_data: Coordinator data dict (keyword-only)
sensor_attrs: Sensor-specific attributes (keyword-only)
time: TimeService instance (optional, creates new if not provided)
time: TibberPricesTimeService instance (required)
Returns:
Complete attributes dict or None if no data available

12
custom_components/tibber_prices/sensor/attributes/daily_stat.py
View file

@ -8,11 +8,13 @@ from custom_components.tibber_prices.const import PRICE_RATING_MAPPING
from homeassistant.const import PERCENTAGE
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from datetime import datetime
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def _get_day_midnight_timestamp(key: str, *, time: TimeService) -> str:
"""Get midnight timestamp for a given day sensor key."""
def _get_day_midnight_timestamp(key: str, *, time: TibberPricesTimeService) -> datetime:
"""Get midnight timestamp for a given day sensor key (returns datetime object)."""
# Determine which day based on sensor key
if key.startswith("yesterday") or key == "average_price_yesterday":
day = "yesterday"
@ -65,7 +67,7 @@ def add_statistics_attributes(
key: str,
cached_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add attributes for statistics and rating sensors.
@ -74,7 +76,7 @@ def add_statistics_attributes(
attributes: Dictionary to add attributes to
key: The sensor entity key
cached_data: Dictionary containing cached sensor data
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# Data timestamp sensor - shows API fetch time

14
custom_components/tibber_prices/sensor/attributes/future.py
View file

@ -9,7 +9,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.core import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
# Constants
MAX_FORECAST_INTERVALS = 8 # Show up to 8 future intervals (2 hours with 15-min intervals)
@ -20,7 +20,7 @@ def add_next_avg_attributes(
key: str,
coordinator: TibberPricesDataUpdateCoordinator,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add attributes for next N hours average price sensors.
@ -29,7 +29,7 @@ def add_next_avg_attributes(
attributes: Dictionary to add attributes to
key: The sensor entity key
coordinator: The data update coordinator
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# Extract hours from sensor key (e.g., "next_avg_3h" -> 3)
@ -70,7 +70,7 @@ def add_price_forecast_attributes(
attributes: dict,
coordinator: TibberPricesDataUpdateCoordinator,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add forecast attributes for the price forecast sensor.
@ -78,7 +78,7 @@ def add_price_forecast_attributes(
Args:
attributes: Dictionary to add attributes to
coordinator: The data update coordinator
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
future_prices = get_future_prices(coordinator, max_intervals=MAX_FORECAST_INTERVALS, time=time)
@ -164,7 +164,7 @@ def get_future_prices(
coordinator: TibberPricesDataUpdateCoordinator,
max_intervals: int | None = None,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> list[dict] | None:
"""
Get future price data for multiple upcoming intervals.
@ -172,7 +172,7 @@ def get_future_prices(
Args:
coordinator: The data update coordinator
max_intervals: Maximum number of future intervals to return
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
List of upcoming price intervals with timestamps and prices

14
custom_components/tibber_prices/sensor/attributes/interval.py
View file

@ -16,7 +16,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.core import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .metadata import get_current_interval_data
@ -28,7 +28,7 @@ def add_current_interval_price_attributes( # noqa: PLR0913
native_value: Any,
cached_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add attributes for current interval price sensors.
@ -39,7 +39,7 @@ def add_current_interval_price_attributes( # noqa: PLR0913
coordinator: The data update coordinator
native_value: The current native value of the sensor
cached_data: Dictionary containing cached sensor data
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
price_info = coordinator.data.get("priceInfo", {}) if coordinator.data else {}
@ -137,7 +137,7 @@ def add_level_attributes_for_sensor( # noqa: PLR0913
coordinator: TibberPricesDataUpdateCoordinator,
native_value: Any,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add price level attributes based on sensor type.
@ -148,7 +148,7 @@ def add_level_attributes_for_sensor( # noqa: PLR0913
interval_data: Interval data for next/previous sensors
coordinator: The data update coordinator
native_value: The current native value of the sensor
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# For interval-based level sensors (next/previous), use interval data
@ -191,7 +191,7 @@ def add_rating_attributes_for_sensor( # noqa: PLR0913
coordinator: TibberPricesDataUpdateCoordinator,
native_value: Any,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add price rating attributes based on sensor type.
@ -202,7 +202,7 @@ def add_rating_attributes_for_sensor( # noqa: PLR0913
interval_data: Interval data for next/previous sensors
coordinator: The data update coordinator
native_value: The current native value of the sensor
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# For interval-based rating sensors (next/previous), use interval data

6
custom_components/tibber_prices/sensor/attributes/metadata.py
View file

@ -10,20 +10,20 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.core import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def get_current_interval_data(
coordinator: TibberPricesDataUpdateCoordinator,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> dict | None:
"""
Get current interval's price data.
Args:
coordinator: The data update coordinator
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current interval data or None if not found

6
custom_components/tibber_prices/sensor/attributes/timing.py
View file

@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Any
from custom_components.tibber_prices.entity_utils import add_icon_color_attribute
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
# Timer #3 triggers every 30 seconds
TIMER_30_SEC_BOUNDARY = 30
@ -35,7 +35,7 @@ def add_period_timing_attributes(
key: str,
state_value: Any = None,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add timestamp and icon_color attributes for best_price/peak_price timing sensors.
@ -48,7 +48,7 @@ def add_period_timing_attributes(
attributes: Dictionary to add attributes to
key: The sensor entity key (e.g., "best_price_end_time")
state_value: Current sensor value for icon_color calculation
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# Determine if this is a quarter-hour or 30-second update sensor

4
custom_components/tibber_prices/sensor/attributes/trend.py
View file

@ -5,7 +5,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from .timing import add_period_timing_attributes
from .volatility import add_volatility_attributes
@ -17,7 +17,7 @@ def _add_timing_or_volatility_attributes(
cached_data: dict,
native_value: Any = None,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""Add attributes for timing or volatility sensors."""
if key.endswith("_volatility"):

14
custom_components/tibber_prices/sensor/attributes/volatility.py
View file

@ -8,14 +8,14 @@ from typing import TYPE_CHECKING
from custom_components.tibber_prices.utils.price import calculate_volatility_level
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def add_volatility_attributes(
attributes: dict,
cached_data: dict,
*,
time: TimeService, # noqa: ARG001
time: TibberPricesTimeService, # noqa: ARG001
) -> None:
"""
Add attributes for volatility sensors.
@ -23,7 +23,7 @@ def add_volatility_attributes(
Args:
attributes: Dictionary to add attributes to
cached_data: Dictionary containing cached sensor data
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
if cached_data.get("volatility_attributes"):
@ -34,7 +34,7 @@ def get_prices_for_volatility(
volatility_type: str,
price_info: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> list[float]:
"""
Get price list for volatility calculation based on type.
@ -42,7 +42,7 @@ def get_prices_for_volatility(
Args:
volatility_type: One of "today", "tomorrow", "next_24h", "today_tomorrow"
price_info: Price information dictionary from coordinator data
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
List of prices to analyze
@ -88,7 +88,7 @@ def add_volatility_type_attributes(
price_info: dict,
thresholds: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add type-specific attributes for volatility sensors.
@ -98,7 +98,7 @@ def add_volatility_type_attributes(
volatility_type: Type of volatility calculation
price_info: Price information dictionary from coordinator data
thresholds: Volatility thresholds configuration
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# Add timestamp for calendar day volatility sensors (midnight of the day)

6
custom_components/tibber_prices/sensor/attributes/window_24h.py
View file

@ -8,7 +8,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.core import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def _update_extreme_interval(extreme_interval: dict | None, price_data: dict, key: str) -> dict:
@ -43,7 +43,7 @@ def add_average_price_attributes(
key: str,
coordinator: TibberPricesDataUpdateCoordinator,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> None:
"""
Add attributes for trailing and leading average/min/max price sensors.
@ -52,7 +52,7 @@ def add_average_price_attributes(
attributes: Dictionary to add attributes to
key: The sensor entity key
coordinator: The data update coordinator
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
"""
# Determine if this is trailing or leading

36
custom_components/tibber_prices/sensor/calculators/__init__.py
View file

@ -10,24 +10,24 @@ All calculators inherit from BaseCalculator and have access to coordinator data.
from __future__ import annotations
from .base import BaseCalculator
from .daily_stat import DailyStatCalculator
from .interval import IntervalCalculator
from .metadata import MetadataCalculator
from .rolling_hour import RollingHourCalculator
from .timing import TimingCalculator
from .trend import TrendCalculator
from .volatility import VolatilityCalculator
from .window_24h import Window24hCalculator
from .base import TibberPricesBaseCalculator
from .daily_stat import TibberPricesDailyStatCalculator
from .interval import TibberPricesIntervalCalculator
from .metadata import TibberPricesMetadataCalculator
from .rolling_hour import TibberPricesRollingHourCalculator
from .timing import TibberPricesTimingCalculator
from .trend import TibberPricesTrendCalculator
from .volatility import TibberPricesVolatilityCalculator
from .window_24h import TibberPricesWindow24hCalculator
__all__ = [
"BaseCalculator",
"DailyStatCalculator",
"IntervalCalculator",
"MetadataCalculator",
"RollingHourCalculator",
"TimingCalculator",
"TrendCalculator",
"VolatilityCalculator",
"Window24hCalculator",
"TibberPricesBaseCalculator",
"TibberPricesDailyStatCalculator",
"TibberPricesIntervalCalculator",
"TibberPricesMetadataCalculator",
"TibberPricesRollingHourCalculator",
"TibberPricesTimingCalculator",
"TibberPricesTrendCalculator",
"TibberPricesVolatilityCalculator",
"TibberPricesWindow24hCalculator",
]

2
custom_components/tibber_prices/sensor/calculators/base.py
View file

@ -12,7 +12,7 @@ if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
class BaseCalculator:
class TibberPricesBaseCalculator:
"""
Base class for all sensor value calculators.

4
custom_components/tibber_prices/sensor/calculators/daily_stat.py
View file

@ -16,7 +16,7 @@ from custom_components.tibber_prices.sensor.helpers import (
aggregate_rating_data,
)
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
if TYPE_CHECKING:
from collections.abc import Callable
@ -26,7 +26,7 @@ if TYPE_CHECKING:
)
class DailyStatCalculator(BaseCalculator):
class TibberPricesDailyStatCalculator(TibberPricesBaseCalculator):
"""
Calculator for daily statistics.

4
custom_components/tibber_prices/sensor/calculators/interval.py
View file

@ -6,7 +6,7 @@ from typing import TYPE_CHECKING
from custom_components.tibber_prices.utils.price import find_price_data_for_interval
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator import (
@ -14,7 +14,7 @@ if TYPE_CHECKING:
)
class IntervalCalculator(BaseCalculator):
class TibberPricesIntervalCalculator(TibberPricesBaseCalculator):
"""
Calculator for interval-based sensors.

4
custom_components/tibber_prices/sensor/calculators/metadata.py
View file

@ -2,10 +2,10 @@
from __future__ import annotations
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
class MetadataCalculator(BaseCalculator):
class TibberPricesMetadataCalculator(TibberPricesBaseCalculator):
"""
Calculator for home metadata, metering point, and subscription data.

8
custom_components/tibber_prices/sensor/calculators/rolling_hour.py
View file

@ -15,10 +15,10 @@ from custom_components.tibber_prices.sensor.helpers import (
aggregate_rating_data,
)
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
class RollingHourCalculator(BaseCalculator):
class TibberPricesRollingHourCalculator(TibberPricesBaseCalculator):
"""
Calculator for rolling hour values (5-interval windows).
@ -75,9 +75,9 @@ class RollingHourCalculator(BaseCalculator):
if not window_data:
return None
return self._aggregate_window_data(window_data, value_type)
return self.aggregate_window_data(window_data, value_type)
def _aggregate_window_data(
def aggregate_window_data(
self,
window_data: list[dict],
value_type: str,

4
custom_components/tibber_prices/sensor/calculators/timing.py
View file

@ -16,12 +16,12 @@ The calculator provides smart defaults:
from datetime import datetime
from .base import BaseCalculator # Constants
from .base import TibberPricesBaseCalculator # Constants
PROGRESS_GRACE_PERIOD_SECONDS = 60 # Show 100% for 1 minute after period ends
class TimingCalculator(BaseCalculator):
class TibberPricesTimingCalculator(TibberPricesBaseCalculator):
"""
Calculator for period timing sensors.

4
custom_components/tibber_prices/sensor/calculators/trend.py
View file

@ -21,7 +21,7 @@ from custom_components.tibber_prices.utils.price import (
find_price_data_for_interval,
)
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator import (
@ -32,7 +32,7 @@ if TYPE_CHECKING:
MIN_HOURS_FOR_LATER_HALF = 3 # Minimum hours needed to calculate later half average
class TrendCalculator(BaseCalculator):
class TibberPricesTrendCalculator(TibberPricesBaseCalculator):
"""
Calculator for price trend sensors.

4
custom_components/tibber_prices/sensor/calculators/volatility.py
View file

@ -11,13 +11,13 @@ from custom_components.tibber_prices.sensor.attributes import (
)
from custom_components.tibber_prices.utils.price import calculate_volatility_level
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
if TYPE_CHECKING:
from typing import Any
class VolatilityCalculator(BaseCalculator):
class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator):
"""
Calculator for price volatility analysis.

4
custom_components/tibber_prices/sensor/calculators/window_24h.py
View file

@ -6,13 +6,13 @@ from typing import TYPE_CHECKING
from custom_components.tibber_prices.entity_utils import get_price_value
from .base import BaseCalculator
from .base import TibberPricesBaseCalculator
if TYPE_CHECKING:
from collections.abc import Callable
class Window24hCalculator(BaseCalculator):
class TibberPricesWindow24hCalculator(TibberPricesBaseCalculator):
"""
Calculator for 24-hour sliding window statistics.

53
custom_components/tibber_prices/sensor/core.py
View file

@ -27,7 +27,10 @@ from custom_components.tibber_prices.entity_utils import (
find_rolling_hour_center_index,
get_price_value,
)
from custom_components.tibber_prices.entity_utils.icons import IconContext, get_dynamic_icon
from custom_components.tibber_prices.entity_utils.icons import (
TibberPricesIconContext,
get_dynamic_icon,
)
from custom_components.tibber_prices.utils.average import (
calculate_next_n_hours_avg,
)
@ -50,14 +53,14 @@ from .attributes import (
get_prices_for_volatility,
)
from .calculators import (
DailyStatCalculator,
IntervalCalculator,
MetadataCalculator,
RollingHourCalculator,
TimingCalculator,
TrendCalculator,
VolatilityCalculator,
Window24hCalculator,
TibberPricesDailyStatCalculator,
TibberPricesIntervalCalculator,
TibberPricesMetadataCalculator,
TibberPricesRollingHourCalculator,
TibberPricesTimingCalculator,
TibberPricesTrendCalculator,
TibberPricesVolatilityCalculator,
TibberPricesWindow24hCalculator,
)
from .chart_data import (
build_chart_data_attributes,
@ -73,7 +76,7 @@ if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator import (
TibberPricesDataUpdateCoordinator,
)
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
HOURS_IN_DAY = 24
LAST_HOUR_OF_DAY = 23
@ -95,14 +98,14 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{entity_description.key}"
self._attr_has_entity_name = True
# Instantiate calculators
self._metadata_calculator = MetadataCalculator(coordinator)
self._volatility_calculator = VolatilityCalculator(coordinator)
self._window_24h_calculator = Window24hCalculator(coordinator)
self._rolling_hour_calculator = RollingHourCalculator(coordinator)
self._daily_stat_calculator = DailyStatCalculator(coordinator)
self._interval_calculator = IntervalCalculator(coordinator)
self._timing_calculator = TimingCalculator(coordinator)
self._trend_calculator = TrendCalculator(coordinator)
self._metadata_calculator = TibberPricesMetadataCalculator(coordinator)
self._volatility_calculator = TibberPricesVolatilityCalculator(coordinator)
self._window_24h_calculator = TibberPricesWindow24hCalculator(coordinator)
self._rolling_hour_calculator = TibberPricesRollingHourCalculator(coordinator)
self._daily_stat_calculator = TibberPricesDailyStatCalculator(coordinator)
self._interval_calculator = TibberPricesIntervalCalculator(coordinator)
self._timing_calculator = TibberPricesTimingCalculator(coordinator)
self._trend_calculator = TibberPricesTrendCalculator(coordinator)
self._value_getter: Callable | None = self._get_value_getter()
self._time_sensitive_remove_listener: Callable | None = None
self._minute_update_remove_listener: Callable | None = None
@ -146,12 +149,12 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
self._minute_update_remove_listener = None
@callback
def _handle_time_sensitive_update(self, time_service: TimeService) -> None:
def _handle_time_sensitive_update(self, time_service: TibberPricesTimeService) -> None:
"""
Handle time-sensitive update from coordinator.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
# Store TimeService from Timer #2 for calculations during this update cycle
@ -166,12 +169,12 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
self.async_write_ha_state()
@callback
def _handle_minute_update(self, time_service: TimeService) -> None:
def _handle_minute_update(self, time_service: TibberPricesTimeService) -> None:
"""
Handle minute-by-minute update from coordinator.
Args:
time_service: TimeService instance with reference time for this update cycle
time_service: TibberPricesTimeService instance with reference time for this update cycle
"""
# Store TimeService from Timer #3 for calculations during this update cycle
@ -267,7 +270,7 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
if not window_data:
return None
return self._aggregate_window_data(window_data, value_type)
return self._rolling_hour_calculator.aggregate_window_data(window_data, value_type)
# ========================================================================
# INTERVAL-BASED VALUE METHODS
@ -470,7 +473,7 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
return en_translations["sensor"]["current_interval_price_rating"]["price_levels"][level]
return level
def _get_next_avg_n_hours_value(self, *, hours: int) -> float | None:
def _get_next_avg_n_hours_value(self, hours: int) -> float | None:
"""
Get average price for next N hours starting from next interval.
@ -803,7 +806,7 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
icon = get_dynamic_icon(
key=key,
value=value,
context=IconContext(
context=TibberPricesIconContext(
coordinator_data=self.coordinator.data,
period_is_active_callback=period_is_active_callback,
time=self.coordinator.time,

6
custom_components/tibber_prices/sensor/helpers.py
View file

@ -22,7 +22,7 @@ from datetime import timedelta
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from custom_components.tibber_prices.entity_utils.helpers import get_price_value
from custom_components.tibber_prices.utils.price import (
@ -134,7 +134,7 @@ def get_hourly_price_value(
*,
hour_offset: int,
in_euro: bool,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Get price for current hour or with offset.
@ -146,7 +146,7 @@ def get_hourly_price_value(
price_info: Price information dict with 'today' and 'tomorrow' keys
hour_offset: Hour offset from current time (positive=future, negative=past)
in_euro: If True, return price in major currency (EUR), else minor (cents/øre)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Price value, or None if not found

53
custom_components/tibber_prices/sensor/value_getters.py
View file

@ -15,29 +15,30 @@ from custom_components.tibber_prices.utils.average import (
if TYPE_CHECKING:
from collections.abc import Callable
from datetime import datetime
from custom_components.tibber_prices.sensor.calculators.daily_stat import DailyStatCalculator
from custom_components.tibber_prices.sensor.calculators.interval import IntervalCalculator
from custom_components.tibber_prices.sensor.calculators.metadata import MetadataCalculator
from custom_components.tibber_prices.sensor.calculators.rolling_hour import RollingHourCalculator
from custom_components.tibber_prices.sensor.calculators.timing import TimingCalculator
from custom_components.tibber_prices.sensor.calculators.trend import TrendCalculator
from custom_components.tibber_prices.sensor.calculators.volatility import VolatilityCalculator
from custom_components.tibber_prices.sensor.calculators.window_24h import Window24hCalculator
from custom_components.tibber_prices.sensor.calculators.daily_stat import TibberPricesDailyStatCalculator
from custom_components.tibber_prices.sensor.calculators.interval import TibberPricesIntervalCalculator
from custom_components.tibber_prices.sensor.calculators.metadata import TibberPricesMetadataCalculator
from custom_components.tibber_prices.sensor.calculators.rolling_hour import TibberPricesRollingHourCalculator
from custom_components.tibber_prices.sensor.calculators.timing import TibberPricesTimingCalculator
from custom_components.tibber_prices.sensor.calculators.trend import TibberPricesTrendCalculator
from custom_components.tibber_prices.sensor.calculators.volatility import TibberPricesVolatilityCalculator
from custom_components.tibber_prices.sensor.calculators.window_24h import TibberPricesWindow24hCalculator
def get_value_getter_mapping( # noqa: PLR0913 - needs all calculators as parameters
interval_calculator: IntervalCalculator,
rolling_hour_calculator: RollingHourCalculator,
daily_stat_calculator: DailyStatCalculator,
window_24h_calculator: Window24hCalculator,
trend_calculator: TrendCalculator,
timing_calculator: TimingCalculator,
volatility_calculator: VolatilityCalculator,
metadata_calculator: MetadataCalculator,
interval_calculator: TibberPricesIntervalCalculator,
rolling_hour_calculator: TibberPricesRollingHourCalculator,
daily_stat_calculator: TibberPricesDailyStatCalculator,
window_24h_calculator: TibberPricesWindow24hCalculator,
trend_calculator: TibberPricesTrendCalculator,
timing_calculator: TibberPricesTimingCalculator,
volatility_calculator: TibberPricesVolatilityCalculator,
metadata_calculator: TibberPricesMetadataCalculator,
get_next_avg_n_hours_value: Callable[[int], float | None],
get_price_forecast_value: Callable[[], str | None],
get_data_timestamp: Callable[[], str | None],
get_data_timestamp: Callable[[], datetime | None],
get_chart_data_export_value: Callable[[], str | None],
) -> dict[str, Callable]:
"""
@ -154,7 +155,7 @@ def get_value_getter_mapping( # noqa: PLR0913 - needs all calculators as parame
day="tomorrow", value_type="rating"
),
# ================================================================
# 24H WINDOW SENSORS (trailing/leading from current) - via Window24hCalculator
# 24H WINDOW SENSORS (trailing/leading from current) - via TibberPricesWindow24hCalculator
# ================================================================
# Trailing and leading average sensors
"trailing_price_average": lambda: window_24h_calculator.get_24h_window_value(
@ -180,14 +181,14 @@ def get_value_getter_mapping( # noqa: PLR0913 - needs all calculators as parame
# FUTURE FORECAST SENSORS
# ================================================================
# Future average sensors (next N hours from next interval)
"next_avg_1h": lambda: get_next_avg_n_hours_value(hours=1),
"next_avg_2h": lambda: get_next_avg_n_hours_value(hours=2),
"next_avg_3h": lambda: get_next_avg_n_hours_value(hours=3),
"next_avg_4h": lambda: get_next_avg_n_hours_value(hours=4),
"next_avg_5h": lambda: get_next_avg_n_hours_value(hours=5),
"next_avg_6h": lambda: get_next_avg_n_hours_value(hours=6),
"next_avg_8h": lambda: get_next_avg_n_hours_value(hours=8),
"next_avg_12h": lambda: get_next_avg_n_hours_value(hours=12),
"next_avg_1h": lambda: get_next_avg_n_hours_value(1),
"next_avg_2h": lambda: get_next_avg_n_hours_value(2),
"next_avg_3h": lambda: get_next_avg_n_hours_value(3),
"next_avg_4h": lambda: get_next_avg_n_hours_value(4),
"next_avg_5h": lambda: get_next_avg_n_hours_value(5),
"next_avg_6h": lambda: get_next_avg_n_hours_value(6),
"next_avg_8h": lambda: get_next_avg_n_hours_value(8),
"next_avg_12h": lambda: get_next_avg_n_hours_value(12),
# Current and next trend change sensors
"current_price_trend": trend_calculator.get_current_trend_value,
"next_price_trend_change": trend_calculator.get_next_trend_change_value,

50
custom_components/tibber_prices/utils/average.py
View file

@ -6,7 +6,7 @@ from datetime import datetime, timedelta
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def calculate_trailing_24h_avg(all_prices: list[dict], interval_start: datetime) -> float:
@ -16,7 +16,7 @@ def calculate_trailing_24h_avg(all_prices: list[dict], interval_start: datetime)
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate average for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Average price for the 24 hours preceding the interval (not including the interval itself)
@ -49,7 +49,7 @@ def calculate_leading_24h_avg(all_prices: list[dict], interval_start: datetime)
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate average for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Average price for up to 24 hours following the interval (including the interval itself)
@ -78,14 +78,14 @@ def calculate_leading_24h_avg(all_prices: list[dict], interval_start: datetime)
def calculate_current_trailing_avg(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour average for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour average price, or None if unavailable
@ -110,14 +110,14 @@ def calculate_current_trailing_avg(
def calculate_current_leading_avg(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour average for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour average price, or None if unavailable
@ -143,7 +143,7 @@ def calculate_trailing_24h_min(
all_prices: list[dict],
interval_start: datetime,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float:
"""
Calculate trailing 24-hour minimum price for a given interval.
@ -151,7 +151,7 @@ def calculate_trailing_24h_min(
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate minimum for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Minimum price for the 24 hours preceding the interval (not including the interval itself)
@ -181,7 +181,7 @@ def calculate_trailing_24h_max(
all_prices: list[dict],
interval_start: datetime,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float:
"""
Calculate trailing 24-hour maximum price for a given interval.
@ -189,7 +189,7 @@ def calculate_trailing_24h_max(
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate maximum for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Maximum price for the 24 hours preceding the interval (not including the interval itself)
@ -219,7 +219,7 @@ def calculate_leading_24h_min(
all_prices: list[dict],
interval_start: datetime,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float:
"""
Calculate leading 24-hour minimum price for a given interval.
@ -227,7 +227,7 @@ def calculate_leading_24h_min(
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate minimum for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Minimum price for up to 24 hours following the interval (including the interval itself)
@ -257,7 +257,7 @@ def calculate_leading_24h_max(
all_prices: list[dict],
interval_start: datetime,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float:
"""
Calculate leading 24-hour maximum price for a given interval.
@ -265,7 +265,7 @@ def calculate_leading_24h_max(
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate maximum for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Maximum price for up to 24 hours following the interval (including the interval itself)
@ -294,14 +294,14 @@ def calculate_leading_24h_max(
def calculate_current_trailing_min(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour minimum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour minimum price, or None if unavailable
@ -326,14 +326,14 @@ def calculate_current_trailing_min(
def calculate_current_trailing_max(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour maximum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour maximum price, or None if unavailable
@ -358,14 +358,14 @@ def calculate_current_trailing_max(
def calculate_current_leading_min(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour minimum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour minimum price, or None if unavailable
@ -390,14 +390,14 @@ def calculate_current_leading_min(
def calculate_current_leading_max(
coordinator_data: dict,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour maximum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour maximum price, or None if unavailable
@ -423,7 +423,7 @@ def calculate_next_n_hours_avg(
coordinator_data: dict,
hours: int,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate average price for the next N hours starting from the next interval.
@ -434,7 +434,7 @@ def calculate_next_n_hours_avg(
Args:
coordinator_data: The coordinator data containing priceInfo
hours: Number of hours to look ahead (1, 2, 3, 4, 5, 6, 8, 12, etc.)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Average price for the next N hours, or None if insufficient data

8
custom_components/tibber_prices/utils/price.py
View file

@ -8,7 +8,7 @@ from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TimeService
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
from custom_components.tibber_prices.const import (
DEFAULT_VOLATILITY_THRESHOLD_HIGH,
@ -281,7 +281,7 @@ def enrich_price_info_with_differences(
price_info: Dictionary with 'yesterday', 'today', 'tomorrow' keys
threshold_low: Low threshold percentage for rating_level (defaults to -10)
threshold_high: High threshold percentage for rating_level (defaults to 10)
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Updated price_info dict with 'difference' and 'rating_level' added
@ -324,7 +324,7 @@ def find_price_data_for_interval(
price_info: Any,
target_time: datetime,
*,
time: TimeService,
time: TibberPricesTimeService,
) -> dict | None:
"""
Find the price data for a specific 15-minute interval timestamp.
@ -332,7 +332,7 @@ def find_price_data_for_interval(
Args:
price_info: The price info dictionary from Tibber API
target_time: The target timestamp to find price data for
time: TimeService instance (required)
time: TibberPricesTimeService instance (required)
Returns:
Price data dict if found, None otherwise