Coverage for custom_components/appliance_autonotify/config_flow.py: 100%
153 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-14 15:14 +0000
1"""Config flow for Appliance Auto Notify, one config entry per appliance"""
3from __future__ import annotations
5from typing import Any
7from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlowWithReload
8from homeassistant.const import CONF_DEVICE_ID, CONF_NAME, CONF_TYPE, UnitOfPower, UnitOfTime
9from homeassistant.core import HomeAssistant, callback
10from homeassistant.data_entry_flow import section
11from homeassistant.helpers.selector import (
12 BooleanSelector,
13 EntitySelector,
14 EntitySelectorConfig,
15 NumberSelector,
16 NumberSelectorConfig,
17 NumberSelectorMode,
18 SelectOptionDict,
19 SelectSelector,
20 SelectSelectorConfig,
21 SelectSelectorMode,
22 TargetSelector,
23 TextSelector,
24 TextSelectorConfig,
25)
27from .compat import vol
28from .const import (
29 CONF_AUTO_DISCOVER,
30 CONF_CUSTOM_TARGET,
31 CONF_DASHBOARD,
32 CONF_DELIVERIES,
33 CONF_END_MESSAGE,
34 CONF_END_TITLE,
35 CONF_GRACE_PERIOD,
36 CONF_MIN_DURATION,
37 CONF_NOTIFY_END,
38 CONF_NOTIFY_PHASE,
39 CONF_NOTIFY_PROGRESS,
40 CONF_NOTIFY_START,
41 CONF_POWER_ENTITY,
42 CONF_PROGRAM_IN_TITLE,
43 CONF_REMINDER,
44 CONF_REMINDER_REPEATS,
45 CONF_START_MESSAGE,
46 CONF_START_TITLE,
47 CONF_SWITCH_ENTITY,
48 CONF_SWITCH_OFF_BUTTON,
49 CONF_TARGET,
50 CONF_TARGETS,
51 CONF_THRESHOLD,
52 DEFAULT_END_MESSAGE,
53 DEFAULT_GRACE_PERIOD,
54 DEFAULT_MIN_DURATION,
55 DEFAULT_REMINDER_REPEATS,
56 DEFAULT_START_MESSAGE,
57 DEFAULT_THRESHOLD,
58 DEFAULT_TITLE,
59 DISCOVERY_TITLE,
60 DOMAIN,
61 NAME_PLACEHOLDER,
62 PROGRAM_PLACEHOLDER,
63 SUPERNOTIFY_MISSING,
64 SUPERNOTIFY_PRESENT,
65 TYPE_DISCOVERY,
66 TYPE_POWER,
67)
68from .dashboards import dashboards
69from .discovery import find_appliances
70from .supernotify_api import deliveries_by_transport, supernotify_available
72STEP_POWER = "power"
73STEP_DISCOVERY = "discovery"
74STEP_SETTINGS = "settings"
75SECTION_START = "start"
76SECTION_END = "end"
77SECTION_NOTIFY = "notify"
79# a switch left as it comes isn't a setting either
80DEFAULT_SWITCHES: dict[str, bool] = {
81 CONF_NOTIFY_START: False,
82 CONF_NOTIFY_END: True,
83 CONF_NOTIFY_PROGRESS: False,
84 CONF_PROGRAM_IN_TITLE: True,
85}
87DEFAULT_TEXTS: dict[str, str] = {
88 CONF_START_TITLE: DEFAULT_TITLE,
89 CONF_START_MESSAGE: DEFAULT_START_MESSAGE,
90 CONF_END_TITLE: DEFAULT_TITLE,
91 CONF_END_MESSAGE: DEFAULT_END_MESSAGE,
92}
95def supernotify_status(hass: HomeAssistant) -> str:
96 return SUPERNOTIFY_PRESENT if supernotify_available(hass) else SUPERNOTIFY_MISSING
99def discovery_schema(auto_discover: bool = True) -> vol.Schema:
100 return vol.Schema({vol.Required(CONF_AUTO_DISCOVER, default=auto_discover): BooleanSelector()})
103async def settings_schema(hass: HomeAssistant, progress: bool = True, phase: bool | None = None) -> vol.Schema:
104 fields: dict[vol.Marker, Any] = {
105 vol.Required(CONF_NOTIFY_START, default=False): BooleanSelector(),
106 vol.Required(CONF_NOTIFY_END, default=True): BooleanSelector(),
107 }
108 if phase is not None:
109 # only for an appliance with phases, and on or off to begin with by how much they matter for its type
110 fields[vol.Required(CONF_NOTIFY_PHASE, default=phase)] = BooleanSelector()
111 if progress:
112 fields[vol.Required(CONF_NOTIFY_PROGRESS, default=False)] = BooleanSelector()
113 if supernotify_available(hass):
114 # only a Live Activity has a title that changes as the cycle goes on
115 fields[vol.Required(CONF_PROGRAM_IN_TITLE, default=True)] = BooleanSelector()
116 notify: dict[vol.Marker, Any] = {}
117 end: dict[vol.Marker, Any] = {
118 vol.Optional(CONF_END_TITLE): TextSelector(),
119 vol.Optional(CONF_END_MESSAGE): TextSelector(TextSelectorConfig(multiline=True)),
120 }
121 if supernotify_available(hass):
122 # a choice of what's there, and not a box to type in, so a notification can't be made to open anything else
123 choices = [SelectOptionDict(value=path, label=title) for path, title in dashboards(hass).items()]
124 if choices:
125 fields[vol.Optional(CONF_DASHBOARD)] = SelectSelector(
126 SelectSelectorConfig(options=choices, mode=SelectSelectorMode.DROPDOWN)
127 )
128 # the same pair of target fields as supernotify.notify itself
129 notify[vol.Optional(CONF_TARGET)] = TargetSelector()
130 notify[vol.Optional(CONF_CUSTOM_TARGET)] = TextSelector(TextSelectorConfig(multiple=True))
131 deliveries = sorted({name for names in (await deliveries_by_transport(hass)).values() for name in names})
132 notify[vol.Optional(CONF_DELIVERIES)] = SelectSelector(SelectSelectorConfig(options=deliveries, multiple=True))
133 # buttons and reminders go with the end notification, which only Supernotify can give buttons
134 end[vol.Optional(CONF_REMINDER)] = NumberSelector(
135 NumberSelectorConfig(min=0, max=720, step=1, mode=NumberSelectorMode.BOX, unit_of_measurement=UnitOfTime.MINUTES)
136 )
137 end[vol.Required(CONF_REMINDER_REPEATS, default=DEFAULT_REMINDER_REPEATS)] = NumberSelector(
138 NumberSelectorConfig(min=1, max=20, step=1, mode=NumberSelectorMode.BOX)
139 )
140 else:
141 notify[vol.Optional(CONF_TARGETS)] = EntitySelector(EntitySelectorConfig(domain="notify", multiple=True))
142 sections: dict[str, dict[vol.Marker, Any]] = {
143 SECTION_START: {
144 vol.Optional(CONF_START_TITLE): TextSelector(),
145 vol.Optional(CONF_START_MESSAGE): TextSelector(TextSelectorConfig(multiline=True)),
146 },
147 SECTION_END: end,
148 SECTION_NOTIFY: notify,
149 }
150 # all tuning, so out of the way until it's wanted
151 fields.update({
152 vol.Required(name): section(vol.Schema(contents), {"collapsed": True}) for name, contents in sections.items()
153 })
154 return vol.Schema(fields)
157def sectioned(schema: vol.Schema, values: dict[str, Any]) -> dict[str, Any]:
158 """Settings are stored flat, the sections are only for display"""
159 shown: dict[str, Any] = {}
160 for name, field in schema.schema.items():
161 if isinstance(field, section):
162 shown[str(name)] = {str(key): values[key] for key in field.schema.schema if key in values}
163 elif name in values:
164 shown[str(name)] = values[name]
165 return shown
168def flattened(user_input: dict[str, Any]) -> dict[str, Any]:
169 """The settings of a form, with those in sections alongside the rest"""
170 settings: dict[str, Any] = {}
171 for name, value in user_input.items():
172 settings.update(value if isinstance(value, dict) else {name: value})
173 return settings
176def power_schema(with_name: bool) -> vol.Schema:
177 fields: dict[vol.Marker, Any] = {}
178 if with_name:
179 fields[vol.Required(CONF_NAME)] = TextSelector()
180 fields[vol.Required(CONF_POWER_ENTITY)] = EntitySelector(EntitySelectorConfig(domain="sensor", device_class="power"))
181 fields[vol.Required(CONF_THRESHOLD, default=DEFAULT_THRESHOLD)] = NumberSelector(
182 NumberSelectorConfig(min=0, step="any", mode=NumberSelectorMode.BOX, unit_of_measurement=UnitOfPower.WATT)
183 )
184 fields[vol.Required(CONF_GRACE_PERIOD, default=DEFAULT_GRACE_PERIOD)] = NumberSelector(
185 NumberSelectorConfig(min=0, max=3600, step=1, mode=NumberSelectorMode.BOX, unit_of_measurement=UnitOfTime.SECONDS)
186 )
187 fields[vol.Required(CONF_MIN_DURATION, default=DEFAULT_MIN_DURATION)] = NumberSelector(
188 NumberSelectorConfig(min=0, max=3600, step=1, mode=NumberSelectorMode.BOX, unit_of_measurement=UnitOfTime.SECONDS)
189 )
190 fields[vol.Optional(CONF_SWITCH_ENTITY)] = EntitySelector(EntitySelectorConfig(domain="switch"))
191 fields[vol.Required(CONF_SWITCH_OFF_BUTTON, default=True)] = BooleanSelector()
192 return vol.Schema(fields)
195class AppliancesConfigFlow(ConfigFlow, domain=DOMAIN):
196 VERSION = 1
198 def __init__(self) -> None:
199 self._device_id: str | None = None
200 self._name: str = ""
202 @staticmethod
203 @callback
204 def async_get_options_flow(config_entry: ConfigEntry) -> AppliancesOptionsFlow:
205 return AppliancesOptionsFlow()
207 async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
208 """Automatic discovery the first time, and from then on an appliance to describe by hand"""
209 if any(entry.data.get(CONF_TYPE) == TYPE_DISCOVERY for entry in self._async_current_entries()):
210 return await self.async_step_power()
211 if user_input is not None:
212 await self.async_set_unique_id(TYPE_DISCOVERY)
213 self._abort_if_unique_id_configured()
214 # the appliances themselves are set up as this entry is, each as its own entry to change or disable
215 return self.async_create_entry(title=DISCOVERY_TITLE, data={CONF_TYPE: TYPE_DISCOVERY}, options=user_input)
216 configured = self._async_current_ids()
217 found = [a.name for a in find_appliances(self.hass) if a.device_id not in configured]
218 if found:
219 found_text = "Appliances found:\n\n" + "\n".join(f"- {name}" for name in found)
220 else:
221 found_text = "No appliances could be automatically installed."
222 return self.async_show_form(
223 step_id="user",
224 data_schema=discovery_schema(),
225 description_placeholders={"found": found_text, "supernotify": supernotify_status(self.hass)},
226 )
228 async def async_step_power(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
229 """An appliance not found by itself, with a power monitor to tell when it's running"""
230 if user_input is not None:
231 await self.async_set_unique_id(user_input[CONF_POWER_ENTITY])
232 self._abort_if_unique_id_configured()
233 return self.async_create_entry(
234 title=user_input[CONF_NAME],
235 data={CONF_TYPE: TYPE_POWER},
236 options={k: v for k, v in user_input.items() if k != CONF_NAME},
237 )
238 return self.async_show_form(step_id=STEP_POWER, data_schema=power_schema(with_name=True))
240 async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
241 """An appliance found with automatic discovery on, so set up without asking"""
242 # a discovery of the same appliance may be waiting, and goes once this entry exists
243 await self.async_set_unique_id(import_data[CONF_DEVICE_ID], raise_on_progress=False)
244 self._abort_if_unique_id_configured()
245 return self.async_create_entry(title=import_data[CONF_NAME], data={CONF_DEVICE_ID: import_data[CONF_DEVICE_ID]})
247 async def async_step_integration_discovery(self, discovery_info: dict[str, Any]) -> ConfigFlowResult:
248 """An appliance found with automatic discovery off, so offered to be added"""
249 self._device_id = discovery_info[CONF_DEVICE_ID]
250 self._name = discovery_info[CONF_NAME]
251 await self.async_set_unique_id(self._device_id)
252 self._abort_if_unique_id_configured()
253 self.context["title_placeholders"] = {"name": self._name}
254 return await self.async_step_confirm()
256 async def async_step_confirm(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
257 if user_input is not None:
258 return self.async_create_entry(title=self._name, data={CONF_DEVICE_ID: self._device_id})
259 self._set_confirm_only()
260 return self.async_show_form(
261 step_id="confirm",
262 description_placeholders={"name": self._name, "supernotify": supernotify_status(self.hass)},
263 )
266class AppliancesOptionsFlow(OptionsFlowWithReload):
267 def __init__(self) -> None:
268 self._settings: dict[str, Any] = {}
270 def _name(self) -> str:
271 # the name notifications are sent with, which follows the device rather than the entry's title
272 watcher = getattr(self.config_entry, "runtime_data", None)
273 return watcher.name if watcher is not None else self.config_entry.title
275 def _phase_default(self) -> bool | None:
276 return getattr(getattr(self.config_entry, "runtime_data", None), "phase_default", None)
278 async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
279 if self.config_entry.data.get(CONF_TYPE) == TYPE_DISCOVERY:
280 return await self.async_step_discovery()
281 return await self.async_step_settings()
283 async def async_step_discovery(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
284 if user_input is not None:
285 return self.async_create_entry(data=user_input)
286 return self.async_show_form(
287 step_id=STEP_DISCOVERY,
288 data_schema=discovery_schema(self.config_entry.options.get(CONF_AUTO_DISCOVER, True)),
289 )
291 async def async_step_settings(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
292 name = self._name()
293 # as they'll be sent, so there's the real text to edit rather than a template or an empty box
294 defaults = {key: text.replace(NAME_PLACEHOLDER, name) for key, text in DEFAULT_TEXTS.items()}
295 if user_input is not None:
296 # text left as the default isn't a setting, so it goes on following the name of the appliance
297 unset = (
298 defaults
299 | DEFAULT_SWITCHES
300 | {CONF_NOTIFY_PHASE: self._phase_default(), CONF_REMINDER_REPEATS: DEFAULT_REMINDER_REPEATS}
301 )
302 settings = {
303 key: value
304 for key, value in flattened(user_input).items()
305 if (value or value is False) and value != unset.get(key)
306 }
307 if self.config_entry.data.get(CONF_TYPE) == TYPE_POWER:
308 self._settings = settings
309 return await self.async_step_power()
310 return self.async_create_entry(data=settings)
311 # a power monitored appliance has no progress to report
312 schema = await settings_schema(
313 self.hass, progress=self.config_entry.data.get(CONF_TYPE) != TYPE_POWER, phase=self._phase_default()
314 )
315 shown = dict(self.config_entry.options)
316 if shown.get(CONF_DASHBOARD) not in dashboards(self.hass):
317 # removed since it was chosen
318 shown.pop(CONF_DASHBOARD, None)
319 for key, text in defaults.items():
320 shown[key] = str(shown.get(key) or text).replace(NAME_PLACEHOLDER, name)
321 return self.async_show_form(
322 step_id=STEP_SETTINGS,
323 data_schema=self.add_suggested_values_to_schema(schema, sectioned(schema, shown)),
324 description_placeholders={
325 "name": name,
326 "supernotify": supernotify_status(self.hass),
327 # as it's to be typed, which a translation can't say for itself
328 "program_placeholder": PROGRAM_PLACEHOLDER,
329 },
330 )
332 async def async_step_power(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
333 if user_input is not None:
334 return self.async_create_entry(data=self._settings | user_input)
335 return self.async_show_form(
336 step_id=STEP_POWER,
337 data_schema=self.add_suggested_values_to_schema(power_schema(with_name=False), self.config_entry.options),
338 description_placeholders={"name": self.config_entry.title},
339 )