189 lines
8.3 KiB
Python
189 lines
8.3 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:
|
|
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, {})))
|