"""Optimizer area mapping helpers.""" from __future__ import annotations from collections.abc import Collection, Mapping from typing import Any from homeassistant.util import slugify from .const import CONF_AREA_MAPPING, LOGGER type AreaMapping = dict[str, tuple[str, ...]] DEFAULT_AREA_MAPPINGS: dict[str, dict[str, list[str]]] = { "4886699": { "west1": [ "1.1.1", "1.1.2", "1.1.4", "1.1.5", "1.1.6", "1.1.9", "1.1.10", "1.1.11", "1.1.12", ], "west2": ["1.1.3", "1.1.13", "1.1.14"], "gaube": ["1.1.7", "1.1.8"], "süd": ["1.1.15", "1.1.16"], "ost1": [ "1.2.1", "1.2.2", "1.2.3", "1.2.4", "1.2.7", "1.2.8", "1.2.9", "1.2.10", "1.2.11", "1.2.12", "1.2.13", "1.2.14", "1.2.17", "1.2.18", ], "ost2": ["1.2.5", "1.2.6", "1.2.15", "1.2.16"], } } class AreaMappingError(ValueError): """An area mapping could not be validated.""" def __init__(self, translation_key: str) -> None: """Initialize the mapping error.""" super().__init__(translation_key) self.translation_key = translation_key def default_area_mapping(site_id: str) -> dict[str, list[str]]: """Return a mutable copy of the site's default area mapping.""" return { area: list(optimizer_ids) for area, optimizer_ids in DEFAULT_AREA_MAPPINGS.get(site_id, {}).items() } def configured_area_mapping(site_id: str, options: Mapping[str, Any]) -> AreaMapping: """Return the configured mapping or the site-specific default.""" value = options.get(CONF_AREA_MAPPING, default_area_mapping(site_id)) try: return parse_area_mapping(value) except AreaMappingError: LOGGER.exception("Stored optimizer area mapping is invalid") return {} def parse_area_mapping( value: Any, available_optimizer_ids: Collection[str] | None = None, ) -> AreaMapping: """Validate and normalize an optimizer area mapping.""" if not isinstance(value, dict): raise AreaMappingError("invalid_area_mapping") available = ( set(available_optimizer_ids) if available_optimizer_ids is not None else None ) mapping: AreaMapping = {} assigned_optimizer_ids: set[str] = set() area_slugs: set[str] = set() for raw_area, raw_optimizer_ids in value.items(): if not isinstance(raw_area, str) or not (area := raw_area.strip()): raise AreaMappingError("invalid_area_mapping") area_slug = slugify(area) if not area_slug or area_slug in area_slugs: raise AreaMappingError("duplicate_area") area_slugs.add(area_slug) if not isinstance(raw_optimizer_ids, list) or not raw_optimizer_ids: raise AreaMappingError("empty_area") optimizer_ids: list[str] = [] for raw_optimizer_id in raw_optimizer_ids: if not isinstance(raw_optimizer_id, str) or not ( optimizer_id := raw_optimizer_id.strip() ): raise AreaMappingError("invalid_area_mapping") if available is not None and optimizer_id not in available: raise AreaMappingError("unknown_optimizer") if optimizer_id in assigned_optimizer_ids: raise AreaMappingError("duplicate_optimizer") assigned_optimizer_ids.add(optimizer_id) optimizer_ids.append(optimizer_id) mapping[area] = tuple(optimizer_ids) return mapping def serialize_area_mapping(mapping: AreaMapping) -> dict[str, list[str]]: """Convert the normalized mapping into config-entry-safe data.""" return {area: list(optimizer_ids) for area, optimizer_ids in mapping.items()}