"""UI-only installation and reconfiguration: rooms, source priorities, weights.""" from __future__ import annotations from typing import Any import voluptuous as vol from homeassistant import config_entries from homeassistant.core import callback from homeassistant.helpers import selector from .const import ( CONF_ROOM_MAX_AGE, CONF_WEATHER_MAX_AGE, DOMAIN, CONF_CLIMATE, CONF_HEATER, CONF_OUTSIDE, CONF_ROOMS, CONF_WEIGHTS, CONF_HORIZON, CONF_MIN_ON, CONF_MIN_OFF, CONF_SWITCH_COST, CONF_HEAT_COST, CONF_BAND, DEFAULTS, WEATHER_SOURCES, WEATHER_MAIN, WEATHER_EXTRA, ) from .weather_sources import sources_for def _entity_picker(multiple: bool = False): return selector.EntitySelector( selector.EntitySelectorConfig(domain="sensor", multiple=multiple, **({"reorder": True} if multiple else {})) ) def _required(key: str, data: dict): return vol.Required(key, default=data[key]) if key in data else vol.Required(key) def _main_schema(config: dict[str, Any] | None = None, detailed: bool = False): data = config or {} schema: dict[Any, Any] = { _required(CONF_CLIMATE, data): selector.EntitySelector( selector.EntitySelectorConfig(domain="climate")), _required(CONF_HEATER, data): selector.EntitySelector( selector.EntitySelectorConfig(domain="switch")), _required(CONF_ROOMS, data): _entity_picker(multiple=True), } for key in (CONF_ROOM_MAX_AGE, CONF_WEATHER_MAX_AGE): schema[vol.Required(key, default=data.get(key, DEFAULTS[key]))] = vol.All( vol.Coerce(int), vol.Range(min=5, max=1440)) if detailed: for key, allowed in { CONF_HORIZON: vol.In([60, 90, 120, 150, 180]), CONF_MIN_ON: vol.All(vol.Coerce(int), vol.Range(min=5, max=60)), CONF_MIN_OFF: vol.All(vol.Coerce(int), vol.Range(min=5, max=60)), CONF_SWITCH_COST: vol.All(vol.Coerce(float), vol.Range(min=0, max=5)), CONF_HEAT_COST: vol.All(vol.Coerce(float), vol.Range(min=0, max=5)), CONF_BAND: vol.All(vol.Coerce(float), vol.Range(min=0.05, max=0.7)), }.items(): schema[vol.Required(key, default=data.get(key, DEFAULTS[key]))] = allowed return vol.Schema(schema) def _weather_schema(config: dict[str, Any], metrics: tuple[str, ...]): schema = {} for metric in metrics: key = WEATHER_SOURCES[metric] selected = sources_for(config, metric) if metric == "temperature": # A temperature source is mandatory: other weather inputs are optional. marker = vol.Required(key, default=selected) if selected else vol.Required(key) else: marker = vol.Optional(key, default=selected) schema[marker] = _entity_picker(multiple=True) return vol.Schema(schema) def _weight_schema(rooms: list[str], existing: dict[str, float] | None = None): weights = existing or {} return vol.Schema({ vol.Required(f"weight__{room}", default=weights.get(room, 1.0)): vol.All(vol.Coerce(float), vol.Range(min=0.1, max=10)) for room in rooms }) def _valid_rooms(rooms: list[str]) -> bool: return bool(rooms) and len(rooms) <= 12 and len(rooms) == len(set(rooms)) def _valid_weather(rooms: list[str], data: dict) -> bool: sources = sources_for(data, "temperature") return bool(sources) and len(sources) <= 10 and not set(sources).intersection(rooms) class HeatingMPCConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Create a native observation-only MPC installation.""" VERSION = 2 def __init__(self): self._draft: dict[str, Any] = {} async def async_step_user(self, user_input: dict[str, Any] | None = None): if user_input is not None: if not _valid_rooms(user_input[CONF_ROOMS]): return self.async_show_form(step_id="user", data_schema=_main_schema(), errors={"base": "invalid_rooms"}) await self.async_set_unique_id("home_heating") self._abort_if_unique_id_configured() self._draft = user_input return await self.async_step_weather() return self.async_show_form(step_id="user", data_schema=_main_schema()) async def async_step_weather(self, user_input: dict[str, Any] | None = None): if user_input is not None: if not _valid_weather(self._draft[CONF_ROOMS], user_input): return self.async_show_form(step_id="weather", data_schema=_weather_schema( user_input, WEATHER_MAIN), errors={"base": "invalid_weather"}) self._draft.update(user_input) return await self.async_step_weather_extra() return self.async_show_form(step_id="weather", data_schema=_weather_schema( self._draft, WEATHER_MAIN)) async def async_step_weather_extra(self, user_input: dict[str, Any] | None = None): if user_input is not None: self._draft.update(user_input) return await self.async_step_weights() return self.async_show_form(step_id="weather_extra", data_schema=_weather_schema( self._draft, WEATHER_EXTRA)) async def async_step_weights(self, user_input: dict[str, Any] | None = None): if user_input is not None: self._draft[CONF_WEIGHTS] = { room: float(user_input[f"weight__{room}"]) for room in self._draft[CONF_ROOMS] } return self.async_create_entry(title="Heating MPC", data=self._draft) return self.async_show_form(step_id="weights", data_schema=_weight_schema( self._draft[CONF_ROOMS])) @staticmethod @callback def async_get_options_flow(config_entry: config_entries.ConfigEntry): return HeatingMPCOptionsFlow() class HeatingMPCOptionsFlow(config_entries.OptionsFlow): """Edit sensors, ranked alternative weather providers and MPC parameters.""" def __init__(self): self._draft: dict[str, Any] = {} async def async_step_init(self, user_input: dict[str, Any] | None = None): config = {**self.config_entry.data, **self.config_entry.options} if user_input is not None: if not _valid_rooms(user_input[CONF_ROOMS]): return self.async_show_form(step_id="init", data_schema=_main_schema( config, detailed=True), errors={"base": "invalid_rooms"}) self._draft = user_input return await self.async_step_weather() return self.async_show_form(step_id="init", data_schema=_main_schema( config, detailed=True)) async def async_step_weather(self, user_input: dict[str, Any] | None = None): config = {**self.config_entry.data, **self.config_entry.options, **self._draft} if user_input is not None: if not _valid_weather(self._draft[CONF_ROOMS], user_input): return self.async_show_form(step_id="weather", data_schema=_weather_schema( user_input, WEATHER_MAIN), errors={"base": "invalid_weather"}) self._draft.update(user_input) return await self.async_step_weather_extra() return self.async_show_form(step_id="weather", data_schema=_weather_schema( config, WEATHER_MAIN)) async def async_step_weather_extra(self, user_input: dict[str, Any] | None = None): config = {**self.config_entry.data, **self.config_entry.options, **self._draft} if user_input is not None: self._draft.update(user_input) return await self.async_step_weights() return self.async_show_form(step_id="weather_extra", data_schema=_weather_schema( config, WEATHER_EXTRA)) async def async_step_weights(self, user_input: dict[str, Any] | None = None): if user_input is not None: self._draft[CONF_WEIGHTS] = { room: float(user_input[f"weight__{room}"]) for room in self._draft[CONF_ROOMS] } # Prevent an old single-temperature sensor from overriding new sources. self._draft.pop(CONF_OUTSIDE, None) return self.async_create_entry(title="", data=self._draft) config = {**self.config_entry.data, **self.config_entry.options} return self.async_show_form(step_id="weights", data_schema=_weight_schema( self._draft[CONF_ROOMS], config.get(CONF_WEIGHTS, {})))