mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-30 13:23:41 +00:00
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.
127 lines
3.6 KiB
Python
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)
|