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