hass.tibber_prices/custom_components/tibber_prices/const.py

"""Constants for the Tibber Price Analytics integration."""

import json
import logging
from collections.abc import Sequence
from pathlib import Path
from typing import Any

import aiofiles

from homeassistant.const import (
    CURRENCY_DOLLAR,
    CURRENCY_EURO,
    UnitOfPower,
    UnitOfTime,
)
from homeassistant.core import HomeAssistant

DOMAIN = "tibber_prices"
CONF_EXTENDED_DESCRIPTIONS = "extended_descriptions"
CONF_BEST_PRICE_FLEX = "best_price_flex"
CONF_PEAK_PRICE_FLEX = "peak_price_flex"
CONF_BEST_PRICE_MIN_DISTANCE_FROM_AVG = "best_price_min_distance_from_avg"
CONF_PEAK_PRICE_MIN_DISTANCE_FROM_AVG = "peak_price_min_distance_from_avg"
CONF_BEST_PRICE_MIN_PERIOD_LENGTH = "best_price_min_period_length"
CONF_PEAK_PRICE_MIN_PERIOD_LENGTH = "peak_price_min_period_length"
CONF_PRICE_RATING_THRESHOLD_LOW = "price_rating_threshold_low"
CONF_PRICE_RATING_THRESHOLD_HIGH = "price_rating_threshold_high"
CONF_PRICE_TREND_THRESHOLD_RISING = "price_trend_threshold_rising"
CONF_PRICE_TREND_THRESHOLD_FALLING = "price_trend_threshold_falling"
CONF_VOLATILITY_THRESHOLD_MODERATE = "volatility_threshold_moderate"
CONF_VOLATILITY_THRESHOLD_HIGH = "volatility_threshold_high"
CONF_VOLATILITY_THRESHOLD_VERY_HIGH = "volatility_threshold_very_high"
CONF_BEST_PRICE_MAX_LEVEL = "best_price_max_level"
CONF_PEAK_PRICE_MIN_LEVEL = "peak_price_min_level"
CONF_BEST_PRICE_MAX_LEVEL_GAP_COUNT = "best_price_max_level_gap_count"
CONF_PEAK_PRICE_MAX_LEVEL_GAP_COUNT = "peak_price_max_level_gap_count"
CONF_ENABLE_MIN_PERIODS_BEST = "enable_min_periods_best"
CONF_MIN_PERIODS_BEST = "min_periods_best"
CONF_RELAXATION_STEP_BEST = "relaxation_step_best"
CONF_ENABLE_MIN_PERIODS_PEAK = "enable_min_periods_peak"
CONF_MIN_PERIODS_PEAK = "min_periods_peak"
CONF_RELAXATION_STEP_PEAK = "relaxation_step_peak"

ATTRIBUTION = "Data provided by Tibber"

# Integration name should match manifest.json
DEFAULT_NAME = "Tibber Price Information & Ratings"
DEFAULT_EXTENDED_DESCRIPTIONS = False
DEFAULT_BEST_PRICE_FLEX = 15  # 15% flexibility for best price (user-facing, percent)
# Peak price flexibility is set to -20 (20%) to allow for more adaptive detection of expensive periods.
# This is intentionally more flexible than best price (15%) because peak price periods can be more variable,
# and users may benefit from earlier warnings about expensive periods, even if they are less sharply defined.
# The negative sign indicates that the threshold is set below the MAX price (e.g., -20% means MAX * 0.8), not above the average price.
# A higher percentage allows for more conservative detection, reducing false negatives for peak price warnings.
DEFAULT_PEAK_PRICE_FLEX = -20  # 20% flexibility for peak price (user-facing, percent)
DEFAULT_BEST_PRICE_MIN_DISTANCE_FROM_AVG = 5  # 5% minimum distance from daily average (ensures significance)
DEFAULT_PEAK_PRICE_MIN_DISTANCE_FROM_AVG = 5  # 5% minimum distance from daily average (ensures significance)
DEFAULT_BEST_PRICE_MIN_PERIOD_LENGTH = 60  # 60 minutes minimum period length for best price (user-facing, minutes)
# Note: Peak price warnings are allowed for shorter periods (30 min) than best price periods (60 min).
# This asymmetry is intentional: shorter peak periods are acceptable for alerting users to brief expensive spikes,
# while best price periods require longer duration to ensure meaningful savings and avoid recommending short,
# impractical windows.
DEFAULT_PEAK_PRICE_MIN_PERIOD_LENGTH = 30  # 30 minutes minimum period length for peak price (user-facing, minutes)
DEFAULT_PRICE_RATING_THRESHOLD_LOW = -10  # Default rating threshold low percentage
DEFAULT_PRICE_RATING_THRESHOLD_HIGH = 10  # Default rating threshold high percentage
DEFAULT_PRICE_TREND_THRESHOLD_RISING = 5  # Default trend threshold for rising prices (%)
DEFAULT_PRICE_TREND_THRESHOLD_FALLING = -5  # Default trend threshold for falling prices (%, negative value)
DEFAULT_VOLATILITY_THRESHOLD_MODERATE = 5.0  # Default threshold for MODERATE volatility (ct/øre)
DEFAULT_VOLATILITY_THRESHOLD_HIGH = 15.0  # Default threshold for HIGH volatility (ct/øre)
DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH = 30.0  # Default threshold for VERY_HIGH volatility (ct/øre)
DEFAULT_BEST_PRICE_MAX_LEVEL = "cheap"  # Default: prefer genuinely cheap periods, relax to "any" if needed
DEFAULT_PEAK_PRICE_MIN_LEVEL = "expensive"  # Default: prefer genuinely expensive periods, relax to "any" if needed
DEFAULT_BEST_PRICE_MAX_LEVEL_GAP_COUNT = 1  # Default: allow 1 level gap (e.g., CHEAP→NORMAL→CHEAP stays together)
DEFAULT_PEAK_PRICE_MAX_LEVEL_GAP_COUNT = 1  # Default: allow 1 level gap for peak price periods
MIN_INTERVALS_FOR_GAP_TOLERANCE = 6  # Minimum period length (in 15-min intervals = 1.5h) required for gap tolerance
DEFAULT_ENABLE_MIN_PERIODS_BEST = True  # Default: minimum periods feature enabled for best price
DEFAULT_MIN_PERIODS_BEST = 2  # Default: require at least 2 best price periods (when enabled)
DEFAULT_RELAXATION_STEP_BEST = 25  # Default: 25% of original threshold per relaxation step for best price
DEFAULT_ENABLE_MIN_PERIODS_PEAK = True  # Default: minimum periods feature enabled for peak price
DEFAULT_MIN_PERIODS_PEAK = 2  # Default: require at least 2 peak price periods (when enabled)
DEFAULT_RELAXATION_STEP_PEAK = 25  # Default: 25% of original threshold per relaxation step for peak price

# Home types
HOME_TYPE_APARTMENT = "APARTMENT"
HOME_TYPE_ROWHOUSE = "ROWHOUSE"
HOME_TYPE_HOUSE = "HOUSE"
HOME_TYPE_COTTAGE = "COTTAGE"

# Mapping for home types to their localized names
HOME_TYPES = {
    HOME_TYPE_APARTMENT: "Apartment",
    HOME_TYPE_ROWHOUSE: "Rowhouse",
    HOME_TYPE_HOUSE: "House",
    HOME_TYPE_COTTAGE: "Cottage",
}

# Currency mapping: ISO code -> (major_symbol, minor_symbol, minor_name)
# For currencies with Home Assistant constants, use those; otherwise define custom ones
CURRENCY_INFO = {
    "EUR": (CURRENCY_EURO, "ct", "cents"),
    "NOK": ("kr", "øre", "øre"),
    "SEK": ("kr", "öre", "öre"),
    "DKK": ("kr", "øre", "øre"),
    "USD": (CURRENCY_DOLLAR, "¢", "cents"),
    "GBP": ("£", "p", "pence"),
}


def get_currency_info(currency_code: str | None) -> tuple[str, str, str]:
    """
    Get currency information for a given ISO currency code.

    Args:
        currency_code: ISO 4217 currency code (e.g., 'EUR', 'NOK', 'SEK')

    Returns:
        Tuple of (major_symbol, minor_symbol, minor_name)
        Defaults to EUR if currency is not recognized

    """
    if not currency_code:
        currency_code = "EUR"

    return CURRENCY_INFO.get(currency_code.upper(), CURRENCY_INFO["EUR"])


def format_price_unit_major(currency_code: str | None) -> str:
    """
    Format the price unit string with major currency unit (e.g., '€/kWh').

    Args:
        currency_code: ISO 4217 currency code (e.g., 'EUR', 'NOK', 'SEK')

    Returns:
        Formatted unit string like '€/kWh' or 'kr/kWh'

    """
    major_symbol, _, _ = get_currency_info(currency_code)
    return f"{major_symbol}/{UnitOfPower.KILO_WATT}{UnitOfTime.HOURS}"


def format_price_unit_minor(currency_code: str | None) -> str:
    """
    Format the price unit string with minor currency unit (e.g., 'ct/kWh').

    Args:
        currency_code: ISO 4217 currency code (e.g., 'EUR', 'NOK', 'SEK')

    Returns:
        Formatted unit string like 'ct/kWh' or 'øre/kWh'

    """
    _, minor_symbol, _ = get_currency_info(currency_code)
    return f"{minor_symbol}/{UnitOfPower.KILO_WATT}{UnitOfTime.HOURS}"


# Price level constants from Tibber API
PRICE_LEVEL_VERY_CHEAP = "VERY_CHEAP"
PRICE_LEVEL_CHEAP = "CHEAP"
PRICE_LEVEL_NORMAL = "NORMAL"
PRICE_LEVEL_EXPENSIVE = "EXPENSIVE"
PRICE_LEVEL_VERY_EXPENSIVE = "VERY_EXPENSIVE"

# Price rating constants (calculated values)
PRICE_RATING_LOW = "LOW"
PRICE_RATING_NORMAL = "NORMAL"
PRICE_RATING_HIGH = "HIGH"

# Price volatility levels (based on spread between min and max)
VOLATILITY_LOW = "LOW"
VOLATILITY_MODERATE = "MODERATE"
VOLATILITY_HIGH = "HIGH"
VOLATILITY_VERY_HIGH = "VERY_HIGH"

# Sensor options (lowercase versions for ENUM device class)
# NOTE: These constants define the valid enum options, but they are not used directly
# in sensor.py due to import timing issues. Instead, the options are defined inline
# in the SensorEntityDescription objects. Keep these in sync with sensor.py!
PRICE_LEVEL_OPTIONS = [
    PRICE_LEVEL_VERY_CHEAP.lower(),
    PRICE_LEVEL_CHEAP.lower(),
    PRICE_LEVEL_NORMAL.lower(),
    PRICE_LEVEL_EXPENSIVE.lower(),
    PRICE_LEVEL_VERY_EXPENSIVE.lower(),
]

PRICE_RATING_OPTIONS = [
    PRICE_RATING_LOW.lower(),
    PRICE_RATING_NORMAL.lower(),
    PRICE_RATING_HIGH.lower(),
]

VOLATILITY_OPTIONS = [
    VOLATILITY_LOW.lower(),
    VOLATILITY_MODERATE.lower(),
    VOLATILITY_HIGH.lower(),
    VOLATILITY_VERY_HIGH.lower(),
]

# Valid options for best price maximum level filter
# Sorted from cheap to expensive: user selects "up to how expensive"
BEST_PRICE_MAX_LEVEL_OPTIONS = [
    "any",  # No filter, allow all price levels
    PRICE_LEVEL_VERY_CHEAP.lower(),  # Only show if level ≤ VERY_CHEAP
    PRICE_LEVEL_CHEAP.lower(),  # Only show if level ≤ CHEAP
    PRICE_LEVEL_NORMAL.lower(),  # Only show if level ≤ NORMAL
    PRICE_LEVEL_EXPENSIVE.lower(),  # Only show if level ≤ EXPENSIVE
]

# Valid options for peak price minimum level filter (AND-linked with volatility filter)
# Sorted from expensive to cheap: user selects "starting from how expensive"
PEAK_PRICE_MIN_LEVEL_OPTIONS = [
    "any",  # No filter, allow all price levels
    PRICE_LEVEL_EXPENSIVE.lower(),  # Only show if level ≥ EXPENSIVE
    PRICE_LEVEL_NORMAL.lower(),  # Only show if level ≥ NORMAL
    PRICE_LEVEL_CHEAP.lower(),  # Only show if level ≥ CHEAP
    PRICE_LEVEL_VERY_CHEAP.lower(),  # Only show if level ≥ VERY_CHEAP
]

# Relaxation level constants (for period filter relaxation)
# These describe which filter relaxation was applied to find a period
RELAXATION_NONE = "none"  # No relaxation, normal filters
RELAXATION_VOLATILITY_ANY = "volatility_any"  # Volatility filter disabled
RELAXATION_ALL_FILTERS_OFF = "all_filters_off"  # All filters disabled (last resort)

# Mapping for comparing price levels (used for sorting)
PRICE_LEVEL_MAPPING = {
    PRICE_LEVEL_VERY_CHEAP: -2,
    PRICE_LEVEL_CHEAP: -1,
    PRICE_LEVEL_NORMAL: 0,
    PRICE_LEVEL_EXPENSIVE: 1,
    PRICE_LEVEL_VERY_EXPENSIVE: 2,
}

# Mapping for comparing price ratings (used for sorting)
PRICE_RATING_MAPPING = {
    PRICE_RATING_LOW: -1,
    PRICE_RATING_NORMAL: 0,
    PRICE_RATING_HIGH: 1,
}

# Mapping for comparing volatility levels (used for sorting)
VOLATILITY_MAPPING = {
    VOLATILITY_LOW: 0,
    VOLATILITY_MODERATE: 1,
    VOLATILITY_HIGH: 2,
    VOLATILITY_VERY_HIGH: 3,
}

LOGGER = logging.getLogger(__package__)

# Path to custom translations directory
CUSTOM_TRANSLATIONS_DIR = Path(__file__).parent / "custom_translations"

# Path to standard translations directory
TRANSLATIONS_DIR = Path(__file__).parent / "translations"

# Cache for translations to avoid repeated file reads
_TRANSLATIONS_CACHE: dict[str, dict] = {}

# Cache for standard translations (config flow, home_types, etc.)
_STANDARD_TRANSLATIONS_CACHE: dict[str, dict] = {}


async def async_load_translations(hass: HomeAssistant, language: str) -> dict:
    """
    Load translations from file asynchronously.

    Args:
        hass: HomeAssistant instance
        language: The language code to load

    Returns:
        The loaded translations as a dictionary

    """
    # Use a key that includes the language parameter
    cache_key = f"{DOMAIN}_translations_{language}"

    # Check if we have an instance in hass.data
    if cache_key in hass.data:
        return hass.data[cache_key]

    # Check the module-level cache
    if language in _TRANSLATIONS_CACHE:
        return _TRANSLATIONS_CACHE[language]

    # Determine the file path
    file_path = CUSTOM_TRANSLATIONS_DIR / f"{language}.json"
    if not file_path.exists():
        # Fall back to English if requested language not found
        file_path = CUSTOM_TRANSLATIONS_DIR / "en.json"
        if not file_path.exists():
            LOGGER.debug("No custom translations found at %s", file_path)
            empty_cache = {}
            _TRANSLATIONS_CACHE[language] = empty_cache
            hass.data[cache_key] = empty_cache
            return empty_cache

    try:
        # Read the file asynchronously
        async with aiofiles.open(file_path, encoding="utf-8") as f:
            content = await f.read()
            translations = json.loads(content)

            # Store in both caches for future calls
            _TRANSLATIONS_CACHE[language] = translations
            hass.data[cache_key] = translations

            return translations

    except (OSError, json.JSONDecodeError) as err:
        LOGGER.warning("Error loading custom translations file: %s", err)
        empty_cache = {}
        _TRANSLATIONS_CACHE[language] = empty_cache
        hass.data[cache_key] = empty_cache
        return empty_cache

    except Exception:  # pylint: disable=broad-except
        LOGGER.exception("Unexpected error loading custom translations")
        empty_cache = {}
        _TRANSLATIONS_CACHE[language] = empty_cache
        hass.data[cache_key] = empty_cache
        return empty_cache


async def async_load_standard_translations(hass: HomeAssistant, language: str) -> dict:
    """
    Load standard translations from the translations directory asynchronously.

    These are the config flow and home_types translations used in the UI.

    Args:
        hass: HomeAssistant instance
        language: The language code to load

    Returns:
        The loaded translations as a dictionary

    """
    cache_key = f"{DOMAIN}_standard_translations_{language}"

    # Check if we have an instance in hass.data
    if cache_key in hass.data:
        return hass.data[cache_key]

    # Check the module-level cache
    if language in _STANDARD_TRANSLATIONS_CACHE:
        return _STANDARD_TRANSLATIONS_CACHE[language]

    # Determine the file path
    file_path = TRANSLATIONS_DIR / f"{language}.json"
    if not file_path.exists():
        # Fall back to English if requested language not found
        file_path = TRANSLATIONS_DIR / "en.json"
        if not file_path.exists():
            LOGGER.debug("No standard translations found at %s", file_path)
            empty_cache = {}
            _STANDARD_TRANSLATIONS_CACHE[language] = empty_cache
            hass.data[cache_key] = empty_cache
            return empty_cache

    try:
        # Read the file asynchronously
        async with aiofiles.open(file_path, encoding="utf-8") as f:
            content = await f.read()
            translations = json.loads(content)
            # Store in both caches for future calls
            _STANDARD_TRANSLATIONS_CACHE[language] = translations
            hass.data[cache_key] = translations
            return translations

    except (OSError, json.JSONDecodeError) as err:
        LOGGER.warning("Error loading standard translations file: %s", err)
        empty_cache = {}
        _STANDARD_TRANSLATIONS_CACHE[language] = empty_cache
        hass.data[cache_key] = empty_cache
        return empty_cache

    except Exception:  # pylint: disable=broad-except
        LOGGER.exception("Unexpected error loading standard translations")
        empty_cache = {}
        _STANDARD_TRANSLATIONS_CACHE[language] = empty_cache
        hass.data[cache_key] = empty_cache
        return empty_cache


async def async_get_translation(
    hass: HomeAssistant,
    path: Sequence[str],
    language: str = "en",
) -> Any:
    """
    Get a translation value by path asynchronously.

    Checks standard translations first, then custom translations.

    Args:
        hass: HomeAssistant instance
        path: A sequence of keys defining the path to the translation value
        language: The language code (defaults to English)

    Returns:
        The translation value if found, None otherwise

    """
    # Try standard translations first (config flow, home_types, etc.)
    translations = await async_load_standard_translations(hass, language)

    # Navigate to the requested path
    current = translations
    for key in path:
        if not isinstance(current, dict) or key not in current:
            break
        current = current.get(key)
    else:
        # If we successfully navigated to the end, return the value
        return current

    # Fall back to custom translations if not found in standard translations
    translations = await async_load_translations(hass, language)

    # Navigate to the requested path
    current = translations
    for key in path:
        if not isinstance(current, dict) or key not in current:
            return None
        current = current[key]

    return current


def get_translation(
    path: Sequence[str],
    language: str = "en",
) -> Any:
    """
    Get a translation value by path synchronously from the cache.

    This function only accesses the cached translations to avoid blocking I/O.
    Checks standard translations first, then custom translations.

    Args:
        path: A sequence of keys defining the path to the translation value
        language: The language code (defaults to English)

    Returns:
        The translation value if found in cache, None otherwise

    """

    def _navigate_dict(d: dict, keys: Sequence[str]) -> Any:
        """Navigate through nested dict following the keys path."""
        current = d
        for key in keys:
            if not isinstance(current, dict) or key not in current:
                return None
            current = current[key]
        return current

    def _get_from_cache(cache: dict[str, dict], lang: str) -> Any:
        """Get translation from cache with fallback to English."""
        if lang in cache:
            result = _navigate_dict(cache[lang], path)
            if result is not None:
                return result
        # Fallback to English if not found in requested language
        if lang != "en" and "en" in cache:
            result = _navigate_dict(cache["en"], path)
            if result is not None:
                return result
        return None

    # Try standard translations first
    result = _get_from_cache(_STANDARD_TRANSLATIONS_CACHE, language)
    if result is not None:
        return result

    # Fall back to custom translations
    result = _get_from_cache(_TRANSLATIONS_CACHE, language)
    if result is not None:
        return result

    # Log the missing key for debugging
    LOGGER.debug("Translation key '%s' not found for language %s", path, language)
    return None


# Convenience functions for backward compatibility and common usage patterns
async def async_get_entity_description(
    hass: HomeAssistant,
    entity_type: str,
    entity_key: str,
    language: str = "en",
    field: str = "description",
) -> str | None:
    """
    Get a specific field from the entity's custom translations asynchronously.

    Args:
        hass: HomeAssistant instance
        entity_type: The type of entity (sensor, binary_sensor, etc.)
        entity_key: The key of the entity
        language: The language code (defaults to English)
        field: The field to retrieve (description, long_description, usage_tips)

    Returns:
        The requested field's value if found, None otherwise

    """
    entity_data = await async_get_translation(hass, [entity_type, entity_key], language)

    # Handle the case where entity_data is a string (for description field)
    if isinstance(entity_data, str) and field == "description":
        return entity_data

    # Handle the case where entity_data is a dict
    if isinstance(entity_data, dict) and field in entity_data:
        return entity_data[field]

    return None


def get_entity_description(
    entity_type: str,
    entity_key: str,
    language: str = "en",
    field: str = "description",
) -> str | None:
    """
    Get entity description synchronously from the cache.

    This function only accesses the cached translations to avoid blocking I/O.

    Args:
        entity_type: The type of entity
        entity_key: The key of the entity
        language: The language code
        field: The field to retrieve

    Returns:
        The requested field's value if found in cache, None otherwise

    """
    entity_data = get_translation([entity_type, entity_key], language)

    # Handle the case where entity_data is a string (for description field)
    if isinstance(entity_data, str) and field == "description":
        return entity_data

    # Handle the case where entity_data is a dict
    if isinstance(entity_data, dict) and field in entity_data:
        return entity_data[field]

    return None


async def async_get_price_level_translation(
    hass: HomeAssistant,
    level: str,
    language: str = "en",
) -> str | None:
    """
    Get a localized translation for a price level asynchronously.

    Args:
        hass: HomeAssistant instance
        level: The price level (e.g., VERY_CHEAP, NORMAL, etc.)
        language: The language code (defaults to English)

    Returns:
        The localized price level if found, None otherwise

    """
    return await async_get_translation(hass, ["sensor", "price_level", "price_levels", level], language)


def get_price_level_translation(
    level: str,
    language: str = "en",
) -> str | None:
    """
    Get a localized translation for a price level synchronously from the cache.

    This function only accesses the cached translations to avoid blocking I/O.

    Args:
        level: The price level (e.g., VERY_CHEAP, NORMAL, etc.)
        language: The language code (defaults to English)

    Returns:
        The localized price level if found in cache, None otherwise

    """
    return get_translation(["sensor", "price_level", "price_levels", level], language)


async def async_get_home_type_translation(
    hass: HomeAssistant,
    home_type: str,
    language: str = "en",
) -> str | None:
    """
    Get a localized translation for a home type asynchronously.

    Args:
        hass: HomeAssistant instance
        home_type: The home type (e.g., APARTMENT, HOUSE, etc.)
        language: The language code (defaults to English)

    Returns:
        The localized home type if found, None otherwise

    """
    return await async_get_translation(hass, ["home_types", home_type], language)


def get_home_type_translation(
    home_type: str,
    language: str = "en",
) -> str | None:
    """
    Get a localized translation for a home type synchronously from the cache.

    This function only accesses the cached translations to avoid blocking I/O.

    Args:
        home_type: The home type (e.g., APARTMENT, HOUSE, etc.)
        language: The language code (defaults to English)

    Returns:
        The localized home type if found in cache, fallback to HOME_TYPES dict, or None

    """
    translated = get_translation(["home_types", home_type], language)
    if translated:
        return translated
    fallback = HOME_TYPES.get(home_type)
    LOGGER.debug(
        "No translation found for home type '%s' in language '%s', using fallback: %s. "
        "Available caches: standard=%s, custom=%s",
        home_type,
        language,
        fallback,
        list(_STANDARD_TRANSLATIONS_CACHE.keys()),
        list(_TRANSLATIONS_CACHE.keys()),
    )
    return fallback