Coverage for custom_components/camera_autonotify/config_flow.py: 100%
117 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:32 +0000
1"""Config flow for Camera AutoNotify, one config entry per camera, or zone"""
3from __future__ import annotations
5from dataclasses import fields
6from typing import Any
8from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult, OptionsFlowWithReload
9from homeassistant.const import CONF_DEVICE_ID, CONF_NAME, CONF_TYPE, 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 Selector,
20 SelectSelector,
21 SelectSelectorConfig,
22 SelectSelectorMode,
23 TargetSelector,
24 TextSelector,
25 TextSelectorConfig,
26)
28from .compat import vol
29from .const import (
30 CONF_ALERT_ONCE,
31 CONF_AUTO_DISCOVER,
32 CONF_COOLDOWN,
33 CONF_CUSTOM_TARGET,
34 CONF_DELIVERIES,
35 CONF_DISCOVER_ZONES,
36 CONF_FINAL_DELAY,
37 CONF_FINAL_UPDATE,
38 CONF_INITIAL_DELAY,
39 CONF_LABELS,
40 CONF_SEVERITIES,
41 CONF_SOURCE,
42 CONF_SOURCE_ENTRY_ID,
43 CONF_SOURCE_NAME,
44 CONF_TARGET,
45 CONF_TARGETS,
46 CONF_TIMEOUT,
47 CONF_UPDATE_SUB_LABEL,
48 CONF_ZONE_MULTI,
49 CONF_ZONE_ORDER,
50 CONF_ZONES,
51 DISCOVERY_TITLE,
52 DOMAIN,
53 SEVERITIES,
54 SUPERNOTIFY_MISSING,
55 SUPERNOTIFY_PRESENT,
56 TYPE_DISCOVERY,
57 TYPE_ZONE,
58)
59from .discovery import find_cameras
60from .frigate import camera_choices, frigate_config, zone_labels
61from .supernotify_api import deliveries_by_transport, supernotify_available
62from .tracker import Settings
64STEP_DISCOVERY = "discovery"
65STEP_SETTINGS = "settings"
66SECTION_TIMING = "timing"
67SECTION_NOTIFY = "notify"
68# a setting left as it comes isn't stored, so it follows any better default of a later version
69DEFAULTS: dict[str, Any] = {
70 f.name: sorted(value) if isinstance(value := getattr(Settings(), f.name), frozenset) else value for f in fields(Settings)
71}
72# what a camera or zone's entry holds, as it's found
73ENTRY_DATA = (CONF_DEVICE_ID, CONF_TYPE, CONF_SOURCE, CONF_SOURCE_ENTRY_ID, CONF_SOURCE_NAME)
76def supernotify_status(hass: HomeAssistant) -> str:
77 return SUPERNOTIFY_PRESENT if supernotify_available(hass) else SUPERNOTIFY_MISSING
80def discovery_schema(auto_discover: bool = True, discover_zones: bool = False) -> vol.Schema:
81 return vol.Schema({
82 vol.Required(CONF_AUTO_DISCOVER, default=auto_discover): BooleanSelector(),
83 vol.Required(CONF_DISCOVER_ZONES, default=discover_zones): BooleanSelector(),
84 })
87# as the base class, since type checkers can't all take one registered by a decorator as a type
88def _seconds(maximum: int) -> Selector:
89 return NumberSelector(
90 NumberSelectorConfig(min=0, max=maximum, step=1, mode=NumberSelectorMode.BOX, unit_of_measurement=UnitOfTime.SECONDS)
91 )
94def _choose(options: list[str], translation_key: str | None = None) -> Selector:
95 config = SelectSelectorConfig(options=options, multiple=True, mode=SelectSelectorMode.DROPDOWN)
96 if translation_key:
97 config["translation_key"] = translation_key
98 else:
99 # what's typed is kept too, for a label or zone Frigate's config doesn't list, such as one added since
100 config["custom_value"] = True
101 return SelectSelector(config)
104async def settings_schema(hass: HomeAssistant, labels: list[str], zones: list[str] | None) -> vol.Schema:
105 """The blueprint's filters, then its timing and who is told, out of the way until wanted. Zones only for a
106 camera, since a zone's entry is the zone"""
107 fields_: dict[vol.Marker, Any] = {
108 vol.Required(CONF_SEVERITIES, default=DEFAULTS[CONF_SEVERITIES]): _choose(list(SEVERITIES), CONF_SEVERITIES),
109 vol.Optional(CONF_LABELS): _choose(labels),
110 }
111 if zones is not None:
112 fields_[vol.Optional(CONF_ZONES)] = _choose(zones)
113 fields_[vol.Required(CONF_ZONE_MULTI, default=False)] = BooleanSelector()
114 fields_[vol.Required(CONF_ZONE_ORDER, default=False)] = BooleanSelector()
115 timing: dict[vol.Marker, Any] = {
116 vol.Required(CONF_COOLDOWN, default=DEFAULTS[CONF_COOLDOWN]): _seconds(3600),
117 vol.Required(CONF_INITIAL_DELAY, default=DEFAULTS[CONF_INITIAL_DELAY]): _seconds(60),
118 vol.Required(CONF_FINAL_UPDATE, default=DEFAULTS[CONF_FINAL_UPDATE]): BooleanSelector(),
119 vol.Required(CONF_FINAL_DELAY, default=DEFAULTS[CONF_FINAL_DELAY]): _seconds(300),
120 vol.Required(CONF_TIMEOUT, default=DEFAULTS[CONF_TIMEOUT]): _seconds(3600),
121 vol.Required(CONF_ALERT_ONCE, default=DEFAULTS[CONF_ALERT_ONCE]): BooleanSelector(),
122 vol.Required(CONF_UPDATE_SUB_LABEL, default=DEFAULTS[CONF_UPDATE_SUB_LABEL]): BooleanSelector(),
123 }
124 notify: dict[vol.Marker, Any] = {}
125 if supernotify_available(hass):
126 # the same pair of target fields as supernotify.notify itself
127 notify[vol.Optional(CONF_TARGET)] = TargetSelector()
128 notify[vol.Optional(CONF_CUSTOM_TARGET)] = TextSelector(TextSelectorConfig(multiple=True))
129 deliveries = sorted({name for names in (await deliveries_by_transport(hass)).values() for name in names})
130 notify[vol.Optional(CONF_DELIVERIES)] = SelectSelector(SelectSelectorConfig(options=deliveries, multiple=True))
131 else:
132 notify[vol.Optional(CONF_TARGETS)] = EntitySelector(EntitySelectorConfig(domain="notify", multiple=True))
133 fields_[vol.Required(SECTION_TIMING)] = section(vol.Schema(timing), {"collapsed": True})
134 fields_[vol.Required(SECTION_NOTIFY)] = section(vol.Schema(notify), {"collapsed": True})
135 return vol.Schema(fields_)
138def sectioned(schema: vol.Schema, values: dict[str, Any]) -> dict[str, Any]:
139 """Settings are stored flat, the sections are only for display"""
140 shown: dict[str, Any] = {}
141 for name, field in schema.schema.items():
142 if isinstance(field, section):
143 shown[str(name)] = {str(key): values[key] for key in field.schema.schema if key in values}
144 elif name in values:
145 shown[str(name)] = values[name]
146 return shown
149def flattened(user_input: dict[str, Any]) -> dict[str, Any]:
150 """The settings of a form, with those in sections alongside the rest"""
151 settings: dict[str, Any] = {}
152 for name, value in user_input.items():
153 settings.update(value if isinstance(value, dict) and name in (SECTION_TIMING, SECTION_NOTIFY) else {name: value})
154 return settings
157class CameraAutoNotifyConfigFlow(ConfigFlow, domain=DOMAIN):
158 VERSION = 1
159 MINOR_VERSION = 1
161 def __init__(self) -> None:
162 self._found: dict[str, Any] = {}
164 @staticmethod
165 @callback
166 def async_get_options_flow(config_entry: ConfigEntry) -> CameraAutoNotifyOptionsFlow:
167 return CameraAutoNotifyOptionsFlow()
169 async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
170 """Automatic discovery, the cameras themselves set up as they're found"""
171 await self.async_set_unique_id(TYPE_DISCOVERY)
172 self._abort_if_unique_id_configured()
173 if user_input is not None:
174 # the cameras are set up as this entry is, each as its own entry to change or delete
175 return self.async_create_entry(title=DISCOVERY_TITLE, data={CONF_TYPE: TYPE_DISCOVERY}, options=user_input)
176 found = [camera.name for camera in find_cameras(self.hass)]
177 if found:
178 found_text = "Cameras found:\n\n" + "\n".join(f"- {name}" for name in found)
179 else:
180 found_text = "No Frigate cameras found yet."
181 return self.async_show_form(
182 step_id="user",
183 data_schema=discovery_schema(),
184 description_placeholders={"found": found_text, "supernotify": supernotify_status(self.hass)},
185 )
187 async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
188 """A camera found with automatic discovery on, so set up without asking"""
189 # a discovery of the same camera may be waiting, and goes once this entry exists
190 await self.async_set_unique_id(import_data[CONF_DEVICE_ID], raise_on_progress=False)
191 self._abort_if_unique_id_configured()
192 return self.async_create_entry(
193 title=import_data[CONF_NAME], data={k: v for k, v in import_data.items() if k in ENTRY_DATA}
194 )
196 async def async_step_integration_discovery(self, discovery_info: dict[str, Any]) -> ConfigFlowResult:
197 """A camera found with automatic discovery off, or a zone, so offered to be added"""
198 self._found = dict(discovery_info)
199 await self.async_set_unique_id(discovery_info[CONF_DEVICE_ID])
200 self._abort_if_unique_id_configured()
201 self.context["title_placeholders"] = {"name": discovery_info[CONF_NAME]}
202 return await self.async_step_confirm()
204 async def async_step_confirm(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
205 if user_input is not None:
206 return self.async_create_entry(
207 title=self._found[CONF_NAME], data={k: v for k, v in self._found.items() if k in ENTRY_DATA}
208 )
209 self._set_confirm_only()
210 return self.async_show_form(
211 step_id="confirm",
212 description_placeholders={"name": self._found[CONF_NAME], "supernotify": supernotify_status(self.hass)},
213 )
216class CameraAutoNotifyOptionsFlow(OptionsFlowWithReload):
217 async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
218 if self.config_entry.data.get(CONF_TYPE) == TYPE_DISCOVERY:
219 return await self.async_step_discovery()
220 return await self.async_step_settings()
222 async def async_step_discovery(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
223 if user_input is not None:
224 return self.async_create_entry(data=user_input)
225 options = self.config_entry.options
226 return self.async_show_form(
227 step_id=STEP_DISCOVERY,
228 data_schema=discovery_schema(options.get(CONF_AUTO_DISCOVER, True), options.get(CONF_DISCOVER_ZONES, False)),
229 )
231 async def async_step_settings(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
232 if user_input is not None:
233 settings = {
234 key: value
235 for key, value in flattened(user_input).items()
236 if (value or value is False or value == 0) and value != DEFAULTS.get(key)
237 }
238 return self.async_create_entry(data=settings)
239 data = self.config_entry.data
240 config = frigate_config(self.hass, data[CONF_SOURCE_ENTRY_ID])
241 if data.get(CONF_TYPE) == TYPE_ZONE:
242 labels, zones = zone_labels(config, data[CONF_SOURCE_NAME]), None
243 else:
244 labels, zones = camera_choices(config, data[CONF_SOURCE_NAME])
245 schema = await settings_schema(self.hass, labels, zones)
246 return self.async_show_form(
247 step_id=STEP_SETTINGS,
248 data_schema=self.add_suggested_values_to_schema(schema, sectioned(schema, dict(self.config_entry.options))),
249 description_placeholders={"name": self.config_entry.title, "supernotify": supernotify_status(self.hass)},
250 )