mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-29 21:03:40 +00:00
feat(types): add TypedDict documentation and BaseCalculator helpers
Phase 1.1 - TypedDict Documentation System: - Created sensor/types.py with 14 TypedDict classes documenting sensor attributes - Created binary_sensor/types.py with 3 TypedDict classes for binary sensors - Added Literal types (PriceLevel, PriceRating, VolatilityLevel, DataCompleteness) - Updated imports in sensor/attributes/__init__.py and binary_sensor/attributes.py - Changed function signatures to use dict[str, Any] for runtime flexibility - TypedDicts serve as IDE documentation, not runtime validation Phase 1.2 - BaseCalculator Improvements: - Added 8 smart data access methods to BaseCalculator: * get_intervals(day) - day-specific intervals with None-safety * intervals_today/tomorrow/yesterday - convenience properties * get_all_intervals() - combined yesterday+today+tomorrow * find_interval_at_offset(offset) - interval lookup with bounds checking * safe_get_from_interval(interval, key, default) - safe dict access * has_data() / has_price_info() - existence checks * get_day_intervals(day) - alias for consistency - Refactored 5 calculator files to use new helper methods: * daily_stat.py: -11 lines (coordinator_data checks, get_intervals usage) * interval.py: -18 lines (eliminated find_price_data_for_interval duplication) * rolling_hour.py: -3 lines (simplified interval collection) * volatility.py: -4 lines (eliminated price_info local variable) * window_24h.py: -2 lines (replaced coordinator_data check) * Total: -38 lines of duplicate code eliminated - Added noqa comment for lazy import (circular import avoidance) Type Duplication Resolution: - Identified duplication: Literal types in types.py vs string constants in const.py - Attempted solution: Derive constants from Literal types using typing.get_args() - Result: Circular import failure (const.py → sensor/types.py → sensor/__init__.py → const.py) - Final solution: Keep string constants as single source of truth - Added SYNC comments in all 3 files (const.py, sensor/types.py, binary_sensor/types.py) - Accept manual synchronization to avoid circular dependencies - Platform separation maintained (no cross-imports between sensor/ and binary_sensor/) Impact: Developers get IDE autocomplete and type hints for attribute dictionaries. Calculator code is more readable with fewer None-checks and clearer data access patterns. Type/constant duplication documented with sync requirements.
This commit is contained in:
Julian Pawlowski
parent
32857c0cc0
commit
3b11c6721e
11 changed files with 621 additions and 40 deletions
|
|
@ -6,6 +6,8 @@ from typing import TYPE_CHECKING
|
|||
|
||||
from custom_components.tibber_prices.entity_utils import add_icon_color_attribute
|
||||
|
||||
# Import TypedDict definitions for documentation (not used in signatures)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
|
||||
|
||||
|
|
@ -24,6 +26,8 @@ def get_tomorrow_data_available_attributes(
|
|||
"""
|
||||
Build attributes for tomorrow_data_available sensor.
|
||||
|
||||
Returns TomorrowDataAvailableAttributes structure.
|
||||
|
||||
Args:
|
||||
coordinator_data: Coordinator data dict
|
||||
time: TibberPricesTimeService instance
|
||||
|
|
@ -65,6 +69,8 @@ def get_price_intervals_attributes(
|
|||
"""
|
||||
Build attributes for period-based sensors (best/peak price).
|
||||
|
||||
Returns PeriodAttributes structure.
|
||||
|
||||
All data is already calculated in the coordinator - we just need to:
|
||||
1. Get period summaries from coordinator (already filtered and fully calculated)
|
||||
2. Add the current timestamp
|
||||
|
|
|
|||
171
custom_components/tibber_prices/binary_sensor/types.py
Normal file
171
custom_components/tibber_prices/binary_sensor/types.py
Normal file
|
|
@ -0,0 +1,171 @@
|
|||
"""
|
||||
Type definitions for Tibber Prices binary sensor attributes.
|
||||
|
||||
These TypedDict definitions serve as **documentation** of the attribute structure
|
||||
for each binary sensor type. They enable IDE autocomplete and type checking when
|
||||
working with attribute dictionaries.
|
||||
|
||||
NOTE: In function signatures, we still use dict[str, Any] for flexibility,
|
||||
but these TypedDict definitions document what keys and types are expected.
|
||||
|
||||
IMPORTANT: PriceLevel and PriceRating types are duplicated here to avoid
|
||||
cross-platform dependencies. Keep in sync with sensor/types.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal, TypedDict
|
||||
|
||||
# ============================================================================
|
||||
# Literal Type Definitions (Duplicated from sensor/types.py)
|
||||
# ============================================================================
|
||||
# SYNC: Keep these in sync with:
|
||||
# 1. sensor/types.py (Literal type definitions)
|
||||
# 2. const.py (runtime string constants - single source of truth)
|
||||
#
|
||||
# const.py defines:
|
||||
# - PRICE_LEVEL_VERY_CHEAP, PRICE_LEVEL_CHEAP, etc.
|
||||
# - PRICE_RATING_LOW, PRICE_RATING_NORMAL, etc.
|
||||
#
|
||||
# These types are intentionally duplicated here to avoid cross-platform imports.
|
||||
# Binary sensor attributes need these types for type safety without importing
|
||||
# from sensor/ package (maintains platform separation).
|
||||
|
||||
# Price level literals (shared with sensor platform - keep in sync!)
|
||||
PriceLevel = Literal[
|
||||
"VERY_CHEAP",
|
||||
"CHEAP",
|
||||
"NORMAL",
|
||||
"EXPENSIVE",
|
||||
"VERY_EXPENSIVE",
|
||||
]
|
||||
|
||||
# Price rating literals (shared with sensor platform - keep in sync!)
|
||||
PriceRating = Literal[
|
||||
"LOW",
|
||||
"NORMAL",
|
||||
"HIGH",
|
||||
]
|
||||
|
||||
|
||||
class BaseAttributes(TypedDict, total=False):
|
||||
"""
|
||||
Base attributes common to all binary sensors.
|
||||
|
||||
All binary sensor attributes include at minimum:
|
||||
- timestamp: ISO 8601 string indicating when the state/attributes are valid
|
||||
- error: Optional error message if something went wrong
|
||||
"""
|
||||
|
||||
timestamp: str
|
||||
error: str
|
||||
|
||||
|
||||
class TomorrowDataAvailableAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for tomorrow_data_available binary sensor.
|
||||
|
||||
Indicates whether tomorrow's price data is available from Tibber API.
|
||||
"""
|
||||
|
||||
intervals_available: int # Number of intervals available for tomorrow
|
||||
data_status: Literal["none", "partial", "full"] # Data completeness status
|
||||
|
||||
|
||||
class PeriodSummary(TypedDict, total=False):
|
||||
"""
|
||||
Structure for period summary nested in period attributes.
|
||||
|
||||
Each period summary contains all calculated information about one period.
|
||||
"""
|
||||
|
||||
# Time information (priority 1)
|
||||
start: str # ISO 8601 timestamp of period start
|
||||
end: str # ISO 8601 timestamp of period end
|
||||
duration_minutes: int # Duration in minutes
|
||||
|
||||
# Core decision attributes (priority 2)
|
||||
level: PriceLevel # Price level classification
|
||||
rating_level: PriceRating # Price rating classification
|
||||
rating_difference_pct: float # Difference from daily average (%)
|
||||
|
||||
# Price statistics (priority 3)
|
||||
price_avg: float # Average price in period (minor currency)
|
||||
price_min: float # Minimum price in period (minor currency)
|
||||
price_max: float # Maximum price in period (minor currency)
|
||||
price_spread: float # Price spread (max - min)
|
||||
volatility: float # Price volatility within period
|
||||
|
||||
# Price comparison (priority 4)
|
||||
period_price_diff_from_daily_min: float # Difference from daily min (minor currency)
|
||||
period_price_diff_from_daily_min_pct: float # Difference from daily min (%)
|
||||
|
||||
# Detail information (priority 5)
|
||||
period_interval_count: int # Number of intervals in period
|
||||
period_position: int # Period position (1-based)
|
||||
periods_total: int # Total number of periods
|
||||
periods_remaining: int # Remaining periods after this one
|
||||
|
||||
# Relaxation information (priority 6 - only if period was relaxed)
|
||||
relaxation_active: bool # Whether this period was found via relaxation
|
||||
relaxation_level: int # Relaxation level used (1-based)
|
||||
relaxation_threshold_original_pct: float # Original flex threshold (%)
|
||||
relaxation_threshold_applied_pct: float # Applied flex threshold after relaxation (%)
|
||||
|
||||
|
||||
class PeriodAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for period-based binary sensors (best_price_period, peak_price_period).
|
||||
|
||||
These sensors indicate whether the current/next cheap/expensive period is active.
|
||||
|
||||
Attributes follow priority ordering:
|
||||
1. Time information (timestamp, start, end, duration_minutes)
|
||||
2. Core decision attributes (level, rating_level, rating_difference_%)
|
||||
3. Price statistics (price_avg, price_min, price_max, price_spread, volatility)
|
||||
4. Price comparison (period_price_diff_from_daily_min, period_price_diff_from_daily_min_%)
|
||||
5. Detail information (period_interval_count, period_position, periods_total, periods_remaining)
|
||||
6. Relaxation information (only if period was relaxed)
|
||||
7. Meta information (periods list)
|
||||
"""
|
||||
|
||||
# Time information (priority 1) - start/end refer to current/next period
|
||||
start: str | None # ISO 8601 timestamp of current/next period start
|
||||
end: str | None # ISO 8601 timestamp of current/next period end
|
||||
duration_minutes: int # Duration of current/next period in minutes
|
||||
|
||||
# Core decision attributes (priority 2)
|
||||
level: PriceLevel # Price level of current/next period
|
||||
rating_level: PriceRating # Price rating of current/next period
|
||||
rating_difference_pct: float # Difference from daily average (%)
|
||||
|
||||
# Price statistics (priority 3)
|
||||
price_avg: float # Average price in current/next period (minor currency)
|
||||
price_min: float # Minimum price in current/next period (minor currency)
|
||||
price_max: float # Maximum price in current/next period (minor currency)
|
||||
price_spread: float # Price spread (max - min) in current/next period
|
||||
volatility: float # Price volatility within current/next period
|
||||
|
||||
# Price comparison (priority 4)
|
||||
period_price_diff_from_daily_min: float # Difference from daily min (minor currency)
|
||||
period_price_diff_from_daily_min_pct: float # Difference from daily min (%)
|
||||
|
||||
# Detail information (priority 5)
|
||||
period_interval_count: int # Number of intervals in current/next period
|
||||
period_position: int # Period position (1-based)
|
||||
periods_total: int # Total number of periods found
|
||||
periods_remaining: int # Remaining periods after current/next one
|
||||
|
||||
# Relaxation information (priority 6 - only if period was relaxed)
|
||||
relaxation_active: bool # Whether current/next period was found via relaxation
|
||||
relaxation_level: int # Relaxation level used (1-based)
|
||||
relaxation_threshold_original_pct: float # Original flex threshold (%)
|
||||
relaxation_threshold_applied_pct: float # Applied flex threshold after relaxation (%)
|
||||
|
||||
# Meta information (priority 7)
|
||||
periods: list[PeriodSummary] # All periods found (sorted by start time)
|
||||
|
||||
|
||||
# Union type for all binary sensor attributes (for documentation purposes)
|
||||
# In actual code, use dict[str, Any] for flexibility
|
||||
BinarySensorAttributes = TomorrowDataAvailableAttributes | PeriodAttributes
|
||||
|
|
@ -211,7 +211,14 @@ def format_price_unit_minor(currency_code: str | None) -> str:
|
|||
return f"{minor_symbol}/{UnitOfPower.KILO_WATT}{UnitOfTime.HOURS}"
|
||||
|
||||
|
||||
# Price level constants from Tibber API
|
||||
# ============================================================================
|
||||
# Price Level, Rating, and Volatility Constants
|
||||
# ============================================================================
|
||||
# IMPORTANT: These string constants are the single source of truth for
|
||||
# valid enum values. The Literal types in sensor/types.py and binary_sensor/types.py
|
||||
# should be kept in sync with these values manually.
|
||||
|
||||
# Price level constants (from Tibber API)
|
||||
PRICE_LEVEL_VERY_CHEAP = "VERY_CHEAP"
|
||||
PRICE_LEVEL_CHEAP = "CHEAP"
|
||||
PRICE_LEVEL_NORMAL = "NORMAL"
|
||||
|
|
@ -223,7 +230,7 @@ PRICE_RATING_LOW = "LOW"
|
|||
PRICE_RATING_NORMAL = "NORMAL"
|
||||
PRICE_RATING_HIGH = "HIGH"
|
||||
|
||||
# Price volatility levels (based on coefficient of variation: std_dev / mean * 100%)
|
||||
# Price volatility level constants
|
||||
VOLATILITY_LOW = "LOW"
|
||||
VOLATILITY_MODERATE = "MODERATE"
|
||||
VOLATILITY_HIGH = "HIGH"
|
||||
|
|
|
|||
|
|
@ -14,6 +14,22 @@ from custom_components.tibber_prices.entity_utils import (
|
|||
add_description_attributes,
|
||||
add_icon_color_attribute,
|
||||
)
|
||||
from custom_components.tibber_prices.sensor.types import (
|
||||
DailyStatPriceAttributes,
|
||||
DailyStatRatingAttributes,
|
||||
FutureAttributes,
|
||||
IntervalLevelAttributes,
|
||||
# Import all types for re-export
|
||||
IntervalPriceAttributes,
|
||||
IntervalRatingAttributes,
|
||||
LifecycleAttributes,
|
||||
MetadataAttributes,
|
||||
SensorAttributes,
|
||||
TimingAttributes,
|
||||
TrendAttributes,
|
||||
VolatilityAttributes,
|
||||
Window24hAttributes,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from custom_components.tibber_prices.coordinator.core import (
|
||||
|
|
@ -34,6 +50,20 @@ from .volatility import add_volatility_type_attributes, get_prices_for_volatilit
|
|||
from .window_24h import add_average_price_attributes
|
||||
|
||||
__all__ = [
|
||||
"DailyStatPriceAttributes",
|
||||
"DailyStatRatingAttributes",
|
||||
"FutureAttributes",
|
||||
"IntervalLevelAttributes",
|
||||
"IntervalPriceAttributes",
|
||||
"IntervalRatingAttributes",
|
||||
"LifecycleAttributes",
|
||||
"MetadataAttributes",
|
||||
# Type exports
|
||||
"SensorAttributes",
|
||||
"TimingAttributes",
|
||||
"TrendAttributes",
|
||||
"VolatilityAttributes",
|
||||
"Window24hAttributes",
|
||||
"add_volatility_type_attributes",
|
||||
"build_extra_state_attributes",
|
||||
"build_sensor_attributes",
|
||||
|
|
@ -47,7 +77,7 @@ def build_sensor_attributes(
|
|||
coordinator: TibberPricesDataUpdateCoordinator,
|
||||
native_value: Any,
|
||||
cached_data: dict,
|
||||
) -> dict | None:
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
Build attributes for a sensor based on its key.
|
||||
|
||||
|
|
@ -175,7 +205,7 @@ def build_extra_state_attributes( # noqa: PLR0913
|
|||
*,
|
||||
config_entry: TibberPricesConfigEntry,
|
||||
coordinator_data: dict,
|
||||
sensor_attrs: dict | None = None,
|
||||
sensor_attrs: dict[str, Any] | None = None,
|
||||
time: TibberPricesTimeService,
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -69,3 +69,126 @@ class TibberPricesBaseCalculator:
|
|||
def currency(self) -> str:
|
||||
"""Get currency code from price info."""
|
||||
return self.price_info.get("currency", "EUR")
|
||||
|
||||
# Smart data access methods with built-in None-safety
|
||||
|
||||
def get_intervals(self, day: str) -> list[dict]:
|
||||
"""
|
||||
Get price intervals for a specific day with None-safety.
|
||||
|
||||
Args:
|
||||
day: Day key ("yesterday", "today", "tomorrow").
|
||||
|
||||
Returns:
|
||||
List of interval dictionaries, empty list if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
return []
|
||||
return self.price_info.get(day, [])
|
||||
|
||||
@property
|
||||
def intervals_today(self) -> list[dict]:
|
||||
"""Get today's intervals with None-safety."""
|
||||
return self.get_intervals("today")
|
||||
|
||||
@property
|
||||
def intervals_tomorrow(self) -> list[dict]:
|
||||
"""Get tomorrow's intervals with None-safety."""
|
||||
return self.get_intervals("tomorrow")
|
||||
|
||||
@property
|
||||
def intervals_yesterday(self) -> list[dict]:
|
||||
"""Get yesterday's intervals with None-safety."""
|
||||
return self.get_intervals("yesterday")
|
||||
|
||||
def get_all_intervals(self) -> list[dict]:
|
||||
"""
|
||||
Get all available intervals (yesterday + today + tomorrow).
|
||||
|
||||
Returns:
|
||||
Combined list of all interval dictionaries.
|
||||
|
||||
"""
|
||||
return [
|
||||
*self.intervals_yesterday,
|
||||
*self.intervals_today,
|
||||
*self.intervals_tomorrow,
|
||||
]
|
||||
|
||||
def find_interval_at_offset(self, offset: int) -> dict | None:
|
||||
"""
|
||||
Find interval at given offset from current time with bounds checking.
|
||||
|
||||
Args:
|
||||
offset: Offset from current interval (0=current, 1=next, -1=previous).
|
||||
|
||||
Returns:
|
||||
Interval dictionary or None if out of bounds or unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
return None
|
||||
|
||||
from custom_components.tibber_prices.utils.price import ( # noqa: PLC0415 - avoid circular import
|
||||
find_price_data_for_interval,
|
||||
)
|
||||
|
||||
time = self.coordinator.time
|
||||
target_time = time.get_interval_offset_time(offset)
|
||||
return find_price_data_for_interval(self.price_info, target_time, time=time)
|
||||
|
||||
def safe_get_from_interval(
|
||||
self,
|
||||
interval: dict[str, Any],
|
||||
key: str,
|
||||
default: Any = None,
|
||||
) -> Any:
|
||||
"""
|
||||
Safely get a value from an interval dictionary.
|
||||
|
||||
Args:
|
||||
interval: Interval dictionary.
|
||||
key: Key to retrieve.
|
||||
default: Default value if key not found.
|
||||
|
||||
Returns:
|
||||
Value from interval or default.
|
||||
|
||||
"""
|
||||
return interval.get(key, default) if interval else default
|
||||
|
||||
def has_data(self) -> bool:
|
||||
"""
|
||||
Check if coordinator has any data available.
|
||||
|
||||
Returns:
|
||||
True if data is available, False otherwise.
|
||||
|
||||
"""
|
||||
return bool(self.coordinator_data)
|
||||
|
||||
def has_price_info(self) -> bool:
|
||||
"""
|
||||
Check if price info is available in coordinator data.
|
||||
|
||||
Returns:
|
||||
True if price info exists, False otherwise.
|
||||
|
||||
"""
|
||||
return bool(self.price_info)
|
||||
|
||||
def get_day_intervals(self, day: str) -> list[dict]:
|
||||
"""
|
||||
Get intervals for a specific day from coordinator data.
|
||||
|
||||
This is an alias for get_intervals() with consistent naming.
|
||||
|
||||
Args:
|
||||
day: Day key ("yesterday", "today", "tomorrow").
|
||||
|
||||
Returns:
|
||||
List of interval dictionaries, empty list if unavailable.
|
||||
|
||||
"""
|
||||
return self.get_intervals(day)
|
||||
|
|
|
|||
|
|
@ -65,11 +65,9 @@ class TibberPricesDailyStatCalculator(TibberPricesBaseCalculator):
|
|||
Price value in minor currency units (cents/øre), or None if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
price_info = self.price_info
|
||||
|
||||
# Get local midnight boundaries based on the requested day using TimeService
|
||||
time = self.coordinator.time
|
||||
local_midnight, local_midnight_next_day = time.get_day_boundaries(day)
|
||||
|
|
@ -78,7 +76,7 @@ class TibberPricesDailyStatCalculator(TibberPricesBaseCalculator):
|
|||
# that fall within the target day's local date boundaries
|
||||
price_intervals = []
|
||||
for day_key in ["today", "tomorrow"]:
|
||||
for price_data in price_info.get(day_key, []):
|
||||
for price_data in self.get_intervals(day_key):
|
||||
starts_at = price_data.get("startsAt") # Already datetime in local timezone
|
||||
if not starts_at:
|
||||
continue
|
||||
|
|
@ -131,11 +129,9 @@ class TibberPricesDailyStatCalculator(TibberPricesBaseCalculator):
|
|||
Aggregated level/rating value (lowercase), or None if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
price_info = self.price_info
|
||||
|
||||
# Get local midnight boundaries based on the requested day using TimeService
|
||||
time = self.coordinator.time
|
||||
local_midnight, local_midnight_next_day = time.get_day_boundaries(day)
|
||||
|
|
@ -144,7 +140,7 @@ class TibberPricesDailyStatCalculator(TibberPricesBaseCalculator):
|
|||
# that fall within the target day's local date boundaries
|
||||
day_intervals = []
|
||||
for day_key in ["yesterday", "today", "tomorrow"]:
|
||||
for price_data in price_info.get(day_key, []):
|
||||
for price_data in self.get_intervals(day_key):
|
||||
starts_at = price_data.get("startsAt") # Already datetime in local timezone
|
||||
if not starts_at:
|
||||
continue
|
||||
|
|
|
|||
|
|
@ -4,8 +4,6 @@ from __future__ import annotations
|
|||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from custom_components.tibber_prices.utils.price import find_price_data_for_interval
|
||||
|
||||
from .base import TibberPricesBaseCalculator
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -57,32 +55,27 @@ class TibberPricesIntervalCalculator(TibberPricesBaseCalculator):
|
|||
None if data unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
price_info = self.price_info
|
||||
time = self.coordinator.time
|
||||
# Use TimeService to get interval offset time
|
||||
target_time = time.get_interval_offset_time(interval_offset)
|
||||
|
||||
interval_data = find_price_data_for_interval(price_info, target_time, time=time)
|
||||
interval_data = self.find_interval_at_offset(interval_offset)
|
||||
if not interval_data:
|
||||
return None
|
||||
|
||||
# Extract value based on type
|
||||
if value_type == "price":
|
||||
price = interval_data.get("total")
|
||||
price = self.safe_get_from_interval(interval_data, "total")
|
||||
if price is None:
|
||||
return None
|
||||
price = float(price)
|
||||
return price if in_euro else round(price * 100, 2)
|
||||
|
||||
if value_type == "level":
|
||||
level = interval_data.get("level")
|
||||
level = self.safe_get_from_interval(interval_data, "level")
|
||||
return level.lower() if level else None
|
||||
|
||||
# For rating: extract rating_level
|
||||
rating = interval_data.get("rating_level")
|
||||
rating = self.safe_get_from_interval(interval_data, "rating_level")
|
||||
return rating.lower() if rating else None
|
||||
|
||||
def get_price_level_value(self) -> str | None:
|
||||
|
|
@ -117,19 +110,16 @@ class TibberPricesIntervalCalculator(TibberPricesBaseCalculator):
|
|||
Rating level (lowercase), or None if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data or rating_type != "current":
|
||||
if not self.has_data() or rating_type != "current":
|
||||
self._last_rating_difference = None
|
||||
self._last_rating_level = None
|
||||
return None
|
||||
|
||||
time = self.coordinator.time
|
||||
now = time.now()
|
||||
price_info = self.price_info
|
||||
current_interval = find_price_data_for_interval(price_info, now, time=time)
|
||||
current_interval = self.find_interval_at_offset(0)
|
||||
|
||||
if current_interval:
|
||||
rating_level = current_interval.get("rating_level")
|
||||
difference = current_interval.get("difference")
|
||||
rating_level = self.safe_get_from_interval(current_interval, "rating_level")
|
||||
difference = self.safe_get_from_interval(current_interval, "difference")
|
||||
if rating_level is not None:
|
||||
self._last_rating_difference = float(difference) if difference is not None else None
|
||||
self._last_rating_level = rating_level
|
||||
|
|
|
|||
|
|
@ -48,12 +48,11 @@ class TibberPricesRollingHourCalculator(TibberPricesBaseCalculator):
|
|||
- "rating": str (aggregated rating: "low", "normal", "high")
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
# Get all available price data
|
||||
price_info = self.price_info
|
||||
all_prices = price_info.get("yesterday", []) + price_info.get("today", []) + price_info.get("tomorrow", [])
|
||||
all_prices = self.get_all_intervals()
|
||||
|
||||
if not all_prices:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -51,11 +51,9 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator):
|
|||
Volatility level: "low", "moderate", "high", "very_high", or None if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
price_info = self.price_info
|
||||
|
||||
# Get volatility thresholds from config
|
||||
thresholds = {
|
||||
"threshold_moderate": self.config.get("volatility_threshold_moderate", 5.0),
|
||||
|
|
@ -64,7 +62,7 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator):
|
|||
}
|
||||
|
||||
# Get prices based on volatility type
|
||||
prices_to_analyze = get_prices_for_volatility(volatility_type, price_info, time=self.coordinator.time)
|
||||
prices_to_analyze = get_prices_for_volatility(volatility_type, self.price_info, time=self.coordinator.time)
|
||||
|
||||
if not prices_to_analyze:
|
||||
return None
|
||||
|
|
@ -96,7 +94,7 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator):
|
|||
|
||||
# Add type-specific attributes
|
||||
add_volatility_type_attributes(
|
||||
self._last_volatility_attributes, volatility_type, price_info, thresholds, time=self.coordinator.time
|
||||
self._last_volatility_attributes, volatility_type, self.price_info, thresholds, time=self.coordinator.time
|
||||
)
|
||||
|
||||
# Return lowercase for ENUM device class
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ class TibberPricesWindow24hCalculator(TibberPricesBaseCalculator):
|
|||
Price value in minor currency units (cents/øre), or None if unavailable.
|
||||
|
||||
"""
|
||||
if not self.coordinator_data:
|
||||
if not self.has_data():
|
||||
return None
|
||||
|
||||
value = stat_func(self.coordinator_data, time=self.coordinator.time)
|
||||
|
|
|
|||
261
custom_components/tibber_prices/sensor/types.py
Normal file
261
custom_components/tibber_prices/sensor/types.py
Normal file
|
|
@ -0,0 +1,261 @@
|
|||
"""
|
||||
Type definitions for Tibber Prices sensor attributes.
|
||||
|
||||
These TypedDict definitions serve as **documentation** of the attribute structure
|
||||
for each sensor type. They enable IDE autocomplete and type checking when working
|
||||
with attribute dictionaries.
|
||||
|
||||
NOTE: In function signatures, we still use dict[str, Any] for flexibility,
|
||||
but these TypedDict definitions document what keys and types are expected.
|
||||
|
||||
IMPORTANT: The Literal types defined here should be kept in sync with the
|
||||
string constants in const.py, which are the single source of truth for runtime values.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal, TypedDict
|
||||
|
||||
# ============================================================================
|
||||
# Literal Type Definitions
|
||||
# ============================================================================
|
||||
# SYNC: Keep these in sync with constants in const.py
|
||||
#
|
||||
# const.py defines the runtime constants (single source of truth):
|
||||
# - PRICE_LEVEL_VERY_CHEAP, PRICE_LEVEL_CHEAP, etc.
|
||||
# - PRICE_RATING_LOW, PRICE_RATING_NORMAL, etc.
|
||||
# - VOLATILITY_LOW, VOLATILITY_MODERATE, etc.
|
||||
#
|
||||
# These Literal types should mirror those constants for type safety.
|
||||
|
||||
# Price level literals (from Tibber API)
|
||||
PriceLevel = Literal[
|
||||
"VERY_CHEAP",
|
||||
"CHEAP",
|
||||
"NORMAL",
|
||||
"EXPENSIVE",
|
||||
"VERY_EXPENSIVE",
|
||||
]
|
||||
|
||||
# Price rating literals (calculated values)
|
||||
PriceRating = Literal[
|
||||
"LOW",
|
||||
"NORMAL",
|
||||
"HIGH",
|
||||
]
|
||||
|
||||
# Volatility level literals (based on coefficient of variation)
|
||||
VolatilityLevel = Literal[
|
||||
"LOW",
|
||||
"MODERATE",
|
||||
"HIGH",
|
||||
"VERY_HIGH",
|
||||
]
|
||||
|
||||
# Data completeness literals
|
||||
DataCompleteness = Literal[
|
||||
"complete",
|
||||
"partial_yesterday",
|
||||
"partial_today",
|
||||
"partial_tomorrow",
|
||||
"missing_yesterday",
|
||||
"missing_today",
|
||||
"missing_tomorrow",
|
||||
]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TypedDict Definitions
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class BaseAttributes(TypedDict, total=False):
|
||||
"""
|
||||
Base attributes common to all sensors.
|
||||
|
||||
All sensor attributes include at minimum:
|
||||
- timestamp: ISO 8601 string indicating when the state/attributes are valid
|
||||
- error: Optional error message if something went wrong
|
||||
"""
|
||||
|
||||
timestamp: str
|
||||
error: str
|
||||
|
||||
|
||||
class IntervalPriceAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for interval price sensors (current/next/previous).
|
||||
|
||||
These sensors show price information for a specific 15-minute interval.
|
||||
"""
|
||||
|
||||
level_value: int # Numeric value for price level (1-5)
|
||||
level_id: PriceLevel # String identifier for price level
|
||||
icon_color: str # Optional icon color based on level
|
||||
|
||||
|
||||
class IntervalLevelAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for interval level sensors.
|
||||
|
||||
These sensors show the price level classification for an interval.
|
||||
"""
|
||||
|
||||
icon_color: str # Icon color based on level
|
||||
|
||||
|
||||
class IntervalRatingAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for interval rating sensors.
|
||||
|
||||
These sensors show the price rating (LOW/NORMAL/HIGH) for an interval.
|
||||
"""
|
||||
|
||||
rating_value: int # Numeric value for price rating (1-3)
|
||||
rating_id: PriceRating # String identifier for price rating
|
||||
icon_color: str # Optional icon color based on rating
|
||||
|
||||
|
||||
class RollingHourAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for rolling hour sensors.
|
||||
|
||||
These sensors aggregate data across 5 intervals (2 before + current + 2 after).
|
||||
"""
|
||||
|
||||
icon_color: str # Optional icon color based on aggregated level
|
||||
|
||||
|
||||
class DailyStatPriceAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for daily statistics price sensors (min/max/avg).
|
||||
|
||||
These sensors show price statistics for a full calendar day.
|
||||
"""
|
||||
|
||||
# No additional attributes for daily price stats beyond base
|
||||
|
||||
|
||||
class DailyStatRatingAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for daily statistics rating sensors.
|
||||
|
||||
These sensors show rating statistics for a full calendar day.
|
||||
"""
|
||||
|
||||
diff_percent: str # Key is actually "diff_%" - percentage difference
|
||||
level_id: PriceRating # Rating level identifier
|
||||
level_value: int # Numeric rating value (1-3)
|
||||
|
||||
|
||||
class Window24hAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for 24-hour window sensors (trailing/leading).
|
||||
|
||||
These sensors analyze price data across a 24-hour window from current time.
|
||||
"""
|
||||
|
||||
interval_count: int # Number of intervals in the window
|
||||
|
||||
|
||||
class VolatilityAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for volatility sensors.
|
||||
|
||||
These sensors analyze price variation and spread across time periods.
|
||||
"""
|
||||
|
||||
today_spread: float # Price range for today (max - min)
|
||||
today_volatility: str # Volatility level for today
|
||||
interval_count_today: int # Number of intervals analyzed today
|
||||
tomorrow_spread: float # Price range for tomorrow (max - min)
|
||||
tomorrow_volatility: str # Volatility level for tomorrow
|
||||
interval_count_tomorrow: int # Number of intervals analyzed tomorrow
|
||||
|
||||
|
||||
class TrendAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for trend sensors.
|
||||
|
||||
These sensors analyze price trends and forecast future movements.
|
||||
Trend attributes are complex and may vary based on trend type.
|
||||
"""
|
||||
|
||||
# Trend attributes are dynamic and vary by sensor type
|
||||
# Keep flexible with total=False
|
||||
|
||||
|
||||
class TimingAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for period timing sensors (best_price/peak_price timing).
|
||||
|
||||
These sensors track timing information for best/peak price periods.
|
||||
"""
|
||||
|
||||
icon_color: str # Icon color based on timing status
|
||||
|
||||
|
||||
class FutureAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for future forecast sensors.
|
||||
|
||||
These sensors provide N-hour forecasts starting from next interval.
|
||||
"""
|
||||
|
||||
interval_count: int # Number of intervals in forecast
|
||||
hours: int # Number of hours in forecast window
|
||||
|
||||
|
||||
class LifecycleAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for lifecycle/diagnostic sensors.
|
||||
|
||||
These sensors provide system information and cache status.
|
||||
"""
|
||||
|
||||
cache_age: str # Human-readable cache age
|
||||
cache_age_minutes: int # Cache age in minutes
|
||||
cache_validity: str # Cache validity status
|
||||
last_api_fetch: str # ISO 8601 timestamp of last API fetch
|
||||
last_cache_update: str # ISO 8601 timestamp of last cache update
|
||||
data_completeness: DataCompleteness # Data completeness status
|
||||
yesterday_available: bool # Whether yesterday data exists
|
||||
today_available: bool # Whether today data exists
|
||||
tomorrow_available: bool # Whether tomorrow data exists
|
||||
tomorrow_expected_after: str # Time when tomorrow data expected
|
||||
next_api_poll: str # ISO 8601 timestamp of next API poll
|
||||
next_midnight_turnover: str # ISO 8601 timestamp of next midnight turnover
|
||||
updates_today: int # Number of API updates today
|
||||
last_turnover: str # ISO 8601 timestamp of last midnight turnover
|
||||
last_error: str # Last error message if any
|
||||
|
||||
|
||||
class MetadataAttributes(BaseAttributes, total=False):
|
||||
"""
|
||||
Attributes for metadata sensors (home info, metering point).
|
||||
|
||||
These sensors provide Tibber account and home metadata.
|
||||
Metadata attributes vary by sensor type.
|
||||
"""
|
||||
|
||||
# Metadata attributes are dynamic and vary by sensor type
|
||||
# Keep flexible with total=False
|
||||
|
||||
|
||||
# Union type for all sensor attributes (for documentation purposes)
|
||||
# In actual code, use dict[str, Any] for flexibility
|
||||
SensorAttributes = (
|
||||
IntervalPriceAttributes
|
||||
| IntervalLevelAttributes
|
||||
| IntervalRatingAttributes
|
||||
| RollingHourAttributes
|
||||
| DailyStatPriceAttributes
|
||||
| DailyStatRatingAttributes
|
||||
| Window24hAttributes
|
||||
| VolatilityAttributes
|
||||
| TrendAttributes
|
||||
| TimingAttributes
|
||||
| FutureAttributes
|
||||
| LifecycleAttributes
|
||||
| MetadataAttributes
|
||||
)
|
||||
Loading…
Reference in a new issue