buzeraandClaude Opus 5.5
ed8f9fe2e2
Opening a window adds heat exchange that the envelope model mistook for
poor insulation or a constant cooling bias.
windows.py adds windows as config subentries ("Add window"): a contact
sensor (optionally inverted; unknown/unavailable stays unknown, never
"closed"), its zone, the direction its outside faces (default 345°,
editable) and an opening factor. Each poll computes the share of the
interval each window was open (one change located by its timestamp, a
round trip is unknown), the wind exposure with proper circular angles
(meteorological "from" direction vs the window's facing) and a recovery
weight after closing that decays towards 0, pulling the air back
towards its pre-ventilation temperature. A bounded event log records
opening/closing, duration, temperatures before/lowest/at close/30 min
after, weather and other open windows.
While a zone's window is open, unknown or recovering, the base and
extended models skip it; a separate ventilation learner fits the
closed-window model's residual with natural exchange, wind speed, a
directional term bounded by the wind term (a leeward window still
ventilates), recovery, and optional gust excess and cross-ventilation
terms. Forecasts assume an open window closes after a configurable time
(default 15 min), then recovers. New poll outcome "ventilation".
Weather gains optional wind direction (degrees, 360 -> 0) and wind gust
sources with the same priorities and fallback; knots and ft/s accepted.
On a synthetic week with ventilation events every four hours and a
daily outdoor swing, the configured zone keeps loss/heat/bias close to
the truth and recovers the combined natural+wind effect within 15 %,
recovery within 15 % and direction within 30 %. The same data without
the window configured produces a -0.013 °C/5 min cooling bias and an
11 % inflated boiler effect.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
447 lines
24 KiB
Python
447 lines
24 KiB
Python
"""UI-only installation and reconfiguration: zones, source priorities, weights."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
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_ROOM_STALE_KEEP, CONF_WEATHER_MAX_AGE,
|
|
DOMAIN, CONF_CLIMATE, CONF_HEATER, CONF_OUTSIDE, CONF_ROOMS,
|
|
CONF_HORIZON, CONF_MIN_ON, CONF_MIN_OFF,
|
|
CONF_SWITCH_COST, CONF_HEAT_COST, CONF_BAND, CONF_INTERNAL_HALF_LIFE, DEFAULTS,
|
|
CONF_RECOVERY_MINUTES, CONF_VENT_CROSS, CONF_VENT_GUST, CONF_VENT_WIND, CONF_WINDOW_OPEN_MINUTES,
|
|
WEATHER_SOURCES, WEATHER_MAIN, WEATHER_EXTRA,
|
|
)
|
|
from .heat_sources import (
|
|
BASELINE_AUTO, BASELINE_MANUAL, DEFAULT_RESPONSE_MINUTES, MAX_POWER_W, RESPONSE_MINUTES_RANGE,
|
|
SOURCE_BASELINE_MODE, SOURCE_BASELINE_W, SOURCE_ENABLED, SOURCE_FRACTION, SOURCE_INCLUDED_IN,
|
|
SOURCE_KEY, SOURCE_NAME, SOURCE_RESPONSE, SOURCE_SENSOR, SOURCE_ZONE, SUBENTRY_HEAT_SOURCE,
|
|
HeatSource, build_sources, validate_source,
|
|
)
|
|
from .weather_sources import sources_for
|
|
from .windows import (
|
|
DEFAULT_AZIMUTH, OPENING_RANGE, SUBENTRY_WINDOW, WINDOW_AZIMUTH, WINDOW_ENABLED, WINDOW_INVERTED,
|
|
WINDOW_KEY, WINDOW_NAME, WINDOW_OPENING, WINDOW_SENSOR, WINDOW_ZONE, Window, build_windows,
|
|
validate_window,
|
|
)
|
|
from .zones import (
|
|
COMFORT_WEIGHT_RANGE, SENSOR_WEIGHT_RANGE, SUBENTRY_ZONE, TARGET_OFFSET_RANGE,
|
|
ZONE_COMFORT_WEIGHT, ZONE_KEY, ZONE_NAME, ZONE_NEIGHBORS, ZONE_SENSOR_WEIGHTS,
|
|
ZONE_SENSORS, ZONE_TARGET_OFFSET, Zone, build_zones, legacy_zone_data, validate_zone,
|
|
)
|
|
|
|
|
|
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):
|
|
"""Setup asks for room sensors (one zone each); Configure manages zones separately."""
|
|
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")),
|
|
}
|
|
if not detailed:
|
|
schema[_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))
|
|
schema[vol.Required(CONF_ROOM_STALE_KEEP, default=bool(
|
|
data.get(CONF_ROOM_STALE_KEEP, DEFAULTS[CONF_ROOM_STALE_KEEP])))] = selector.BooleanSelector()
|
|
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)),
|
|
CONF_INTERNAL_HALF_LIFE: vol.All(vol.Coerce(int), vol.Range(min=0, max=720)),
|
|
CONF_WINDOW_OPEN_MINUTES: vol.All(vol.Coerce(int), vol.Range(min=5, max=120)),
|
|
CONF_RECOVERY_MINUTES: vol.All(vol.Coerce(int), vol.Range(min=5, max=120)),
|
|
}.items():
|
|
schema[vol.Required(key, default=data.get(key, DEFAULTS[key]))] = allowed
|
|
for key in (CONF_VENT_WIND, CONF_VENT_GUST, CONF_VENT_CROSS):
|
|
schema[vol.Required(key, default=bool(data.get(key, DEFAULTS[key])))] = selector.BooleanSelector()
|
|
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)
|
|
|
|
|
|
def entry_zones(entry: config_entries.ConfigEntry) -> list[Zone]:
|
|
"""Zones of a config entry, from its zone subentries."""
|
|
return build_zones((sub.data, sub.subentry_id) for sub in entry.subentries.values()
|
|
if sub.subentry_type == SUBENTRY_ZONE)
|
|
|
|
|
|
def entry_sources(entry: config_entries.ConfigEntry) -> list[HeatSource]:
|
|
zones = {zone.key for zone in entry_zones(entry)}
|
|
return build_sources(((sub.data, sub.subentry_id) for sub in entry.subentries.values()
|
|
if sub.subentry_type == SUBENTRY_HEAT_SOURCE), zones)
|
|
|
|
|
|
def _number(lo: float, hi: float, step: float) -> selector.NumberSelector:
|
|
return selector.NumberSelector(selector.NumberSelectorConfig(
|
|
min=lo, max=hi, step=step, mode=selector.NumberSelectorMode.BOX))
|
|
|
|
|
|
class HeatingMPCConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|
"""Create a native observation-only MPC installation."""
|
|
|
|
# 3: rooms and comfort weights moved into zone subentries.
|
|
VERSION = 3
|
|
|
|
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:
|
|
rooms = self._draft.pop(CONF_ROOMS)
|
|
weights = {room: float(user_input[f"weight__{room}"]) for room in rooms}
|
|
names = {}
|
|
for room in rooms:
|
|
state = self.hass.states.get(room)
|
|
names[room] = state.name if state else room
|
|
# Each chosen sensor becomes its own zone; zones can be edited later.
|
|
zones = [{"data": data, "subentry_type": SUBENTRY_ZONE, "title": data[ZONE_NAME],
|
|
"unique_id": data[ZONE_KEY]}
|
|
for data in legacy_zone_data(rooms, weights, names)]
|
|
return self.async_create_entry(title="Heating MPC", data=self._draft, subentries=zones)
|
|
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()
|
|
|
|
@classmethod
|
|
@callback
|
|
def async_get_supported_subentry_types(cls, config_entry: config_entries.ConfigEntry):
|
|
return {SUBENTRY_ZONE: ZoneSubentryFlow, SUBENTRY_HEAT_SOURCE: HeatSourceSubentryFlow,
|
|
SUBENTRY_WINDOW: WindowSubentryFlow}
|
|
|
|
|
|
class HeatingMPCOptionsFlow(config_entries.OptionsFlow):
|
|
"""Edit sensors, ranked alternative weather providers and MPC parameters."""
|
|
|
|
def __init__(self):
|
|
self._draft: dict[str, Any] = {}
|
|
|
|
def _zone_sensors(self) -> list[str]:
|
|
return [sensor for zone in entry_zones(self.config_entry) for sensor in zone.sensors]
|
|
|
|
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:
|
|
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._zone_sensors(), 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))
|
|
# 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)
|
|
return self.async_show_form(step_id="weather_extra", data_schema=_weather_schema(
|
|
config, WEATHER_EXTRA))
|
|
|
|
|
|
class ZoneSubentryFlow(config_entries.ConfigSubentryFlow):
|
|
"""Add or edit a thermal zone: its sensors, comfort weight and neighbours."""
|
|
|
|
def __init__(self) -> None:
|
|
self._draft: dict[str, Any] = {}
|
|
|
|
def _other_zones(self, key: str | None) -> list[Zone]:
|
|
return [zone for zone in entry_zones(self._get_entry()) if zone.key != key]
|
|
|
|
def _schema(self, data: dict[str, Any], key: str | None) -> vol.Schema:
|
|
schema: dict[Any, Any] = {
|
|
vol.Required(ZONE_NAME, default=data.get(ZONE_NAME, "")): selector.TextSelector(),
|
|
vol.Required(ZONE_SENSORS, default=list(data.get(ZONE_SENSORS, []))):
|
|
_entity_picker(multiple=True),
|
|
vol.Required(ZONE_COMFORT_WEIGHT, default=data.get(ZONE_COMFORT_WEIGHT, 1.0)):
|
|
_number(*COMFORT_WEIGHT_RANGE, 0.1),
|
|
vol.Required(ZONE_TARGET_OFFSET, default=data.get(ZONE_TARGET_OFFSET, 0.0)):
|
|
_number(*TARGET_OFFSET_RANGE, 0.1),
|
|
}
|
|
others = self._other_zones(key)
|
|
if others:
|
|
schema[vol.Optional(ZONE_NEIGHBORS, description={
|
|
"suggested_value": [n for n in data.get(ZONE_NEIGHBORS, []) if n in {z.key for z in others}]})] = \
|
|
selector.SelectSelector(selector.SelectSelectorConfig(
|
|
options=[selector.SelectOptionDict(value=z.key, label=z.name) for z in others],
|
|
multiple=True, mode=selector.SelectSelectorMode.LIST))
|
|
return vol.Schema(schema)
|
|
|
|
async def _zone_form(self, step_id: str, user_input: dict[str, Any] | None,
|
|
existing: dict[str, Any], key: str | None):
|
|
if user_input is not None:
|
|
entry = self._get_entry()
|
|
config = {**entry.data, **entry.options}
|
|
weather = {s for metric in ("temperature",) for s in sources_for(config, metric)}
|
|
error = validate_zone(user_input, self._other_zones(key), weather)
|
|
if error:
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(user_input, key),
|
|
errors={"base": error})
|
|
self._draft = {**existing, **user_input, ZONE_NAME: user_input[ZONE_NAME].strip()}
|
|
self._draft.setdefault(ZONE_NEIGHBORS, [])
|
|
if len(user_input[ZONE_SENSORS]) > 1:
|
|
return await self.async_step_sensor_weights()
|
|
self._draft[ZONE_SENSOR_WEIGHTS] = {user_input[ZONE_SENSORS][0]: 1.0}
|
|
return self._finish()
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(existing, key))
|
|
|
|
async def async_step_user(self, user_input: dict[str, Any] | None = None):
|
|
if not self._draft.get(ZONE_KEY):
|
|
self._draft = {ZONE_KEY: f"zone_{uuid4().hex[:10]}"}
|
|
return await self._zone_form("user", user_input, self._draft, None)
|
|
|
|
async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None):
|
|
subentry = self._get_reconfigure_subentry()
|
|
existing = dict(subentry.data)
|
|
existing.setdefault(ZONE_KEY, subentry.subentry_id)
|
|
return await self._zone_form("reconfigure", user_input, existing, existing[ZONE_KEY])
|
|
|
|
async def async_step_sensor_weights(self, user_input: dict[str, Any] | None = None):
|
|
sensors = self._draft[ZONE_SENSORS]
|
|
if user_input is not None:
|
|
self._draft[ZONE_SENSOR_WEIGHTS] = {s: float(user_input[f"weight__{s}"]) for s in sensors}
|
|
return self._finish()
|
|
weights = self._draft.get(ZONE_SENSOR_WEIGHTS) or {}
|
|
return self.async_show_form(step_id="sensor_weights", data_schema=vol.Schema({
|
|
vol.Required(f"weight__{s}", default=weights.get(s, 1.0)): _number(*SENSOR_WEIGHT_RANGE, 0.1)
|
|
for s in sensors}))
|
|
|
|
def _finish(self):
|
|
data = {key: self._draft[key] for key in (
|
|
ZONE_KEY, ZONE_NAME, ZONE_SENSORS, ZONE_SENSOR_WEIGHTS, ZONE_COMFORT_WEIGHT,
|
|
ZONE_TARGET_OFFSET, ZONE_NEIGHBORS)}
|
|
data[ZONE_COMFORT_WEIGHT] = float(data[ZONE_COMFORT_WEIGHT])
|
|
data[ZONE_TARGET_OFFSET] = float(data[ZONE_TARGET_OFFSET])
|
|
if self.source == config_entries.SOURCE_RECONFIGURE:
|
|
return self.async_update_and_abort(self._get_entry(), self._get_reconfigure_subentry(),
|
|
title=data[ZONE_NAME], data=data)
|
|
return self.async_create_entry(title=data[ZONE_NAME], data=data, unique_id=data[ZONE_KEY])
|
|
|
|
|
|
class HeatSourceSubentryFlow(config_entries.ConfigSubentryFlow):
|
|
"""Add or edit an electrical heat source (UPS, PC, server) in a zone."""
|
|
|
|
def _others(self, key: str | None) -> list[HeatSource]:
|
|
return [s for s in entry_sources(self._get_entry()) if s.key != key]
|
|
|
|
def _schema(self, data: dict[str, Any], key: str | None) -> vol.Schema:
|
|
zones = entry_zones(self._get_entry())
|
|
parents = [s for s in self._others(key) if s.included_in is None]
|
|
schema: dict[Any, Any] = {
|
|
vol.Required(SOURCE_NAME, default=data.get(SOURCE_NAME, "")): selector.TextSelector(),
|
|
vol.Required(SOURCE_SENSOR, **({"default": data[SOURCE_SENSOR]} if data.get(SOURCE_SENSOR) else {})):
|
|
selector.EntitySelector(selector.EntitySelectorConfig(domain="sensor", device_class="power")),
|
|
vol.Required(SOURCE_ZONE, **({"default": data[SOURCE_ZONE]} if data.get(SOURCE_ZONE) else {})):
|
|
selector.SelectSelector(selector.SelectSelectorConfig(
|
|
options=[selector.SelectOptionDict(value=z.key, label=z.name) for z in zones])),
|
|
vol.Required(SOURCE_FRACTION, default=data.get(SOURCE_FRACTION, 1.0)): _number(0.0, 1.0, 0.05),
|
|
vol.Required(SOURCE_BASELINE_MODE, default=data.get(SOURCE_BASELINE_MODE, BASELINE_AUTO)):
|
|
selector.SelectSelector(selector.SelectSelectorConfig(
|
|
options=[BASELINE_AUTO, BASELINE_MANUAL], translation_key=SOURCE_BASELINE_MODE)),
|
|
vol.Optional(SOURCE_BASELINE_W, description={"suggested_value": data.get(SOURCE_BASELINE_W)}):
|
|
_number(0.0, MAX_POWER_W, 1.0),
|
|
vol.Required(SOURCE_RESPONSE, default=data.get(SOURCE_RESPONSE, DEFAULT_RESPONSE_MINUTES)):
|
|
_number(*RESPONSE_MINUTES_RANGE, 1.0),
|
|
vol.Required(SOURCE_ENABLED, default=data.get(SOURCE_ENABLED, True)): selector.BooleanSelector(),
|
|
}
|
|
if parents:
|
|
schema[vol.Optional(SOURCE_INCLUDED_IN, description={
|
|
"suggested_value": data.get(SOURCE_INCLUDED_IN)})] = selector.SelectSelector(
|
|
selector.SelectSelectorConfig(options=[
|
|
selector.SelectOptionDict(value=s.key, label=s.name) for s in parents]))
|
|
return vol.Schema(schema)
|
|
|
|
async def _form(self, step_id: str, user_input: dict[str, Any] | None, existing: dict[str, Any]):
|
|
key = existing[SOURCE_KEY]
|
|
if not entry_zones(self._get_entry()):
|
|
return self.async_abort(reason="no_zones")
|
|
if user_input is not None:
|
|
data = {**existing, **user_input, SOURCE_KEY: key}
|
|
data.setdefault(SOURCE_INCLUDED_IN, None)
|
|
error = validate_source(data, self._others(key))
|
|
if error:
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(data, key),
|
|
errors={"base": error})
|
|
data[SOURCE_NAME] = data[SOURCE_NAME].strip()
|
|
for number in (SOURCE_FRACTION, SOURCE_RESPONSE):
|
|
data[number] = float(data[number])
|
|
if data.get(SOURCE_BASELINE_W) is not None:
|
|
data[SOURCE_BASELINE_W] = float(data[SOURCE_BASELINE_W])
|
|
if self.source == config_entries.SOURCE_RECONFIGURE:
|
|
return self.async_update_and_abort(self._get_entry(), self._get_reconfigure_subentry(),
|
|
title=data[SOURCE_NAME], data=data)
|
|
return self.async_create_entry(title=data[SOURCE_NAME], data=data, unique_id=key)
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(existing, key))
|
|
|
|
async def async_step_user(self, user_input: dict[str, Any] | None = None):
|
|
if not hasattr(self, "_key"):
|
|
self._key = f"source_{uuid4().hex[:10]}"
|
|
return await self._form("user", user_input, {SOURCE_KEY: self._key})
|
|
|
|
async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None):
|
|
subentry = self._get_reconfigure_subentry()
|
|
existing = dict(subentry.data)
|
|
existing.setdefault(SOURCE_KEY, subentry.subentry_id)
|
|
return await self._form("reconfigure", user_input, existing)
|
|
|
|
|
|
def entry_windows(entry: config_entries.ConfigEntry) -> list[Window]:
|
|
zones = {zone.key for zone in entry_zones(entry)}
|
|
return build_windows(((sub.data, sub.subentry_id) for sub in entry.subentries.values()
|
|
if sub.subentry_type == SUBENTRY_WINDOW), zones)
|
|
|
|
|
|
class WindowSubentryFlow(config_entries.ConfigSubentryFlow):
|
|
"""Add or edit a window contact sensor with its zone and orientation."""
|
|
|
|
def _schema(self, data: dict[str, Any]) -> vol.Schema:
|
|
zones = entry_zones(self._get_entry())
|
|
return vol.Schema({
|
|
vol.Required(WINDOW_NAME, default=data.get(WINDOW_NAME, "")): selector.TextSelector(),
|
|
vol.Required(WINDOW_SENSOR, **({"default": data[WINDOW_SENSOR]} if data.get(WINDOW_SENSOR) else {})):
|
|
selector.EntitySelector(selector.EntitySelectorConfig(domain="binary_sensor")),
|
|
vol.Required(WINDOW_ZONE, **({"default": data[WINDOW_ZONE]} if data.get(WINDOW_ZONE) else {})):
|
|
selector.SelectSelector(selector.SelectSelectorConfig(
|
|
options=[selector.SelectOptionDict(value=z.key, label=z.name) for z in zones])),
|
|
vol.Required(WINDOW_INVERTED, default=data.get(WINDOW_INVERTED, False)): selector.BooleanSelector(),
|
|
vol.Required(WINDOW_AZIMUTH, default=data.get(WINDOW_AZIMUTH, DEFAULT_AZIMUTH)):
|
|
_number(0.0, 359.0, 1.0),
|
|
vol.Required(WINDOW_OPENING, default=data.get(WINDOW_OPENING, 1.0)): _number(*OPENING_RANGE, 0.1),
|
|
vol.Required(WINDOW_ENABLED, default=data.get(WINDOW_ENABLED, True)): selector.BooleanSelector(),
|
|
})
|
|
|
|
async def _form(self, step_id: str, user_input: dict[str, Any] | None, existing: dict[str, Any]):
|
|
key = existing[WINDOW_KEY]
|
|
if not entry_zones(self._get_entry()):
|
|
return self.async_abort(reason="no_zones")
|
|
if user_input is not None:
|
|
data = {**existing, **user_input, WINDOW_KEY: key}
|
|
error = validate_window(data, [w for w in entry_windows(self._get_entry()) if w.key != key])
|
|
if error:
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(data),
|
|
errors={"base": error})
|
|
data[WINDOW_NAME] = data[WINDOW_NAME].strip()
|
|
data[WINDOW_AZIMUTH] = float(data[WINDOW_AZIMUTH])
|
|
data[WINDOW_OPENING] = float(data[WINDOW_OPENING])
|
|
if self.source == config_entries.SOURCE_RECONFIGURE:
|
|
return self.async_update_and_abort(self._get_entry(), self._get_reconfigure_subentry(),
|
|
title=data[WINDOW_NAME], data=data)
|
|
return self.async_create_entry(title=data[WINDOW_NAME], data=data, unique_id=key)
|
|
return self.async_show_form(step_id=step_id, data_schema=self._schema(existing))
|
|
|
|
async def async_step_user(self, user_input: dict[str, Any] | None = None):
|
|
if not hasattr(self, "_key"):
|
|
self._key = f"window_{uuid4().hex[:10]}"
|
|
return await self._form("user", user_input, {WINDOW_KEY: self._key})
|
|
|
|
async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None):
|
|
subentry = self._get_reconfigure_subentry()
|
|
existing = dict(subentry.data)
|
|
existing.setdefault(WINDOW_KEY, subentry.subentry_id)
|
|
return await self._form("reconfigure", user_input, existing)
|