buzeraandClaude Opus 5.5
30f78fb6a9
Optional weather source fields used the current selection as the form default. A cleared field that the frontend omits was refilled with the old sources by voluptuous, and an options key missing from the submission fell back to the original setup data. Optional providers could therefore not be removed in the UI. Pre-fill with suggested values instead, and save omitted optional lists as empty so they override older entry data. No UI strings changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
199 lines
9.0 KiB
Python
199 lines
9.0 KiB
Python
"""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:
|
|
# A default would re-insert old sources whenever a cleared field is
|
|
# omitted from the submission; a suggested value only pre-fills the form.
|
|
marker = vol.Optional(key, description={"suggested_value": selected})
|
|
schema[marker] = _entity_picker(multiple=True)
|
|
return vol.Schema(schema)
|
|
|
|
|
|
def _with_cleared_weather(user_input: dict[str, Any], metrics: tuple[str, ...]) -> dict[str, Any]:
|
|
"""Store an omitted optional source list as empty, so it overrides older data."""
|
|
result = dict(user_input)
|
|
for metric in metrics:
|
|
result.setdefault(WEATHER_SOURCES[metric], [])
|
|
return result
|
|
|
|
|
|
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(_with_cleared_weather(user_input, WEATHER_MAIN))
|
|
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(_with_cleared_weather(user_input, WEATHER_EXTRA))
|
|
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(_with_cleared_weather(user_input, WEATHER_MAIN))
|
|
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(_with_cleared_weather(user_input, WEATHER_EXTRA))
|
|
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, {})))
|