hass.tibber_prices/custom_components/tibber_prices/entity_utils/helpers.py
Julian Pawlowski 625bc222ca refactor(coordinator): centralize time operations through TimeService
Introduce TimeService as single source of truth for all datetime operations,
replacing direct dt_util calls throughout the codebase. This establishes
consistent time context across update cycles and enables future time-travel
testing capability.

Core changes:
- NEW: coordinator/time_service.py with timezone-aware datetime API
- Coordinator now creates TimeService per update cycle, passes to calculators
- Timer callbacks (#2, #3) inject TimeService into entity update flow
- All sensor calculators receive TimeService via coordinator reference
- Attribute builders accept time parameter for timestamp calculations

Key patterns replaced:
- dt_util.now() → time.now() (single reference time per cycle)
- dt_util.parse_datetime() + as_local() → time.get_interval_time()
- Manual interval arithmetic → time.get_interval_offset_time()
- Manual day boundaries → time.get_day_boundaries()
- round_to_nearest_quarter_hour() → time.round_to_nearest_quarter()

Import cleanup:
- Removed dt_util imports from ~30 files (calculators, attributes, utils)
- Restricted dt_util to 3 modules: time_service.py (operations), api/client.py
  (rate limiting), entity_utils/icons.py (cosmetic updates)
- datetime/timedelta only for TYPE_CHECKING (type hints) or duration arithmetic

Interval resolution abstraction:
- Removed hardcoded MINUTES_PER_INTERVAL constant from 10+ files
- New methods: time.minutes_to_intervals(), time.get_interval_duration()
- Supports future 60-minute resolution (legacy data) via TimeService config

Timezone correctness:
- API timestamps (startsAt) already localized by data transformation
- TimeService operations preserve HA user timezone throughout
- DST transitions handled via get_expected_intervals_for_day() (future use)

Timestamp ordering preserved:
- Attribute builders generate default timestamp (rounded quarter)
- Sensors override when needed (next interval, daily midnight, etc.)
- Platform ensures timestamp stays FIRST in attribute dict

Timer integration:
- Timer #2 (quarter-hour): Creates TimeService, calls _handle_time_sensitive_update(time)
- Timer #3 (30-second): Creates TimeService, calls _handle_minute_update(time)
- Consistent time reference for all entities in same update batch

Time-travel readiness:
- TimeService.with_reference_time() enables time injection (not yet used)
- All calculations use time.now() → easy to simulate past/future states
- Foundation for debugging period calculations with historical data

Impact: Eliminates timestamp drift within update cycles (previously 60+ independent
dt_util.now() calls could differ by milliseconds). Establishes architecture for
time-based testing and debugging features.
2025-11-19 18:36:12 +00:00

127 lines
3.6 KiB
Python

"""
Common helper functions for entities across platforms.
This module provides utility functions used by both sensor and binary_sensor platforms:
- Price value conversion (major/minor currency units)
- Translation helpers (price levels, ratings)
- Time-based calculations (rolling hour center index)
These functions operate on entity-level concepts (states, translations) but are
platform-agnostic and can be used by both sensor and binary_sensor platforms.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
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 homeassistant.core import HomeAssistant
def get_price_value(price: float, *, in_euro: bool) -> float:
"""
Convert price based on unit.
Args:
price: Price value to convert
in_euro: If True, return price in euros; if False, return in cents/øre
Returns:
Price in requested unit (euros or minor currency units)
"""
return price if in_euro else round((price * 100), 2)
def translate_level(hass: HomeAssistant, level: str) -> str:
"""
Translate price level to the user's language.
Args:
hass: HomeAssistant instance for language configuration
level: Price level to translate (e.g., VERY_CHEAP, NORMAL, etc.)
Returns:
Translated level string, or original level if translation not found
"""
if not hass:
return level
language = hass.config.language or "en"
translated = get_price_level_translation(level, language)
if translated:
return translated
if language != "en":
fallback = get_price_level_translation(level, "en")
if fallback:
return fallback
return level
def translate_rating_level(rating: str) -> str:
"""
Translate price rating level to the user's language.
Args:
rating: Price rating to translate (e.g., LOW, NORMAL, HIGH)
Returns:
Translated rating string, or original rating if translation not found
Note:
Currently returns the rating as-is. Translation mapping for ratings
can be added here when needed, similar to translate_level().
"""
# For now, ratings are returned as-is
# Add translation mapping here when needed
return rating
def find_rolling_hour_center_index(
all_prices: list[dict],
current_time: datetime,
hour_offset: int,
*,
time: TimeService,
) -> int | None:
"""
Find the center index for the rolling hour window.
Args:
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)
Returns:
Index of the center interval for the rolling hour window, or None if not found
"""
# Round to nearest interval boundary to handle edge cases where HA schedules
# us slightly before the boundary (e.g., 14:59:59.999 → 15:00:00)
target_time = time.round_to_nearest_quarter(current_time)
current_idx = None
for idx, price_data in enumerate(all_prices):
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
# Exact match after rounding
if starts_at == target_time:
current_idx = idx
break
if current_idx is None:
return None
return current_idx + (hour_offset * 4)