125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""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()}
|