Coverage for custom_components/supernotify/scenario.py: 93%
266 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import logging
4import re
5from collections.abc import Generator
6from typing import TYPE_CHECKING, Any
8from homeassistant.const import CONF_ENABLED, STATE_OFF, STATE_ON, STATE_UNKNOWN
9from homeassistant.core import (
10 Context,
11 HomeAssistant,
12 callback,
13)
14from homeassistant.helpers import condition
15from homeassistant.helpers import issue_registry as ir
17from custom_components.supernotify.people import PeopleRegistry
19from .const import (
20 ATTR_MEDIA,
21 CONF_EXPOSE_STATE,
22 CONF_NOTIFY_ENTITY,
23 CONF_REFRESH,
24 CONF_REFRESH_INTERVAL,
25 PRIORITY_MEDIUM,
26 SCENARIO_STATE_REFRESH_DEFAULT,
27)
28from .model import DeliveryCustomization
30if TYPE_CHECKING:
31 from homeassistant.helpers.typing import ConfigType
33 from .binary_sensor import SupernotifyScenarioBinarySensor
34 from .delivery import Delivery, DeliveryRegistry
35 from .hass_api import HomeAssistantAPI
36 from .schema import ConditionsFunc
38from contextlib import contextmanager
40# type: ignore[attr-defined,unused-ignore]
41from homeassistant.components.trace import async_store_trace
42from homeassistant.components.trace.models import ActionTrace
43from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_NAME, CONF_ALIAS, CONF_CONDITIONS
45from .compat import vol
46from .const import ATTR_ENABLED, CONF_ACTION_GROUP_NAMES, CONF_DELIVERY, CONF_MEDIA
47from .model import ConditionVariables
49_LOGGER = logging.getLogger(__name__)
52class ScenarioRegistry:
53 def __init__(
54 self, scenario_configs: ConfigType, scenario_control: ConfigType | None, people_registry: PeopleRegistry
55 ) -> None:
56 self._config: ConfigType = scenario_configs or {}
57 self.scenarios: dict[str, Scenario] = {}
58 self.scenario_control = scenario_control or {}
59 self._people_registry: PeopleRegistry = people_registry
60 # Populated by binary_sensor.py's async_setup_entry once the platform is loaded (after
61 # initialize() below) - see register_entity/unregister_entity. Empty (and harmless to
62 # look up against) before then, e.g. in tests that build ScenarioRegistry directly
63 # without a config entry.
64 self._entities: dict[str, SupernotifyScenarioBinarySensor] = {}
65 # Shared occupancy/ConditionVariables snapshot for the scenario currently being batch
66 # refreshed - set for the duration of async_refresh_scenario_states()'s loop, read by
67 # scenario_is_on() so determine_occupancy() runs once per refresh instead of once per
68 # scenario. None outside of a batch refresh (each scenario_is_on() call then computes
69 # its own, e.g. a single entity being read on demand).
70 self._batch_cvars: ConditionVariables | None = None
72 async def initialize(
73 self,
74 delivery_registry: DeliveryRegistry,
75 mobile_actions: ConfigType,
76 hass_api: HomeAssistantAPI,
77 ) -> None:
79 for scenario_name, scenario_definition in self._config.items():
80 scenario = Scenario(scenario_name, scenario_definition, delivery_registry, hass_api)
81 if await scenario.validate(valid_action_group_names=list(mobile_actions)):
82 self.scenarios[scenario_name] = scenario
83 else:
84 _LOGGER.warning("SUPERNOTIFY Scenario %s failed to validate, ignoring", scenario.name)
85 self._hass_api: HomeAssistantAPI = hass_api
86 self._scenario_cond_entities = self._collect_scenario_condition_entities()
87 self._scenario_by_entity = self._index_scenarios_by_entity()
89 # Keep the scenario binary_sensors' state current: react to their condition entities
90 # (immediate, and only for the scenarios that depend on the entity that changed), plus
91 # a periodic sweep for conditions no entity change announces - time windows, sun, and
92 # templates whose dependencies could not be extracted.
93 # Evaluating conditions costs whatever the conditions cost, so the whole mechanism is
94 # switchable: `scenario_control: {enabled: false}` subscribes to nothing and starts no
95 # timer, and `refresh_interval: 0` keeps the reactive path without the sweep.
96 if self.scenario_state_enabled:
97 scenario_watch: set[str] = set(self._scenario_by_entity)
98 if scenario_watch:
99 hass_api.subscribe_state(sorted(scenario_watch), self.async_refresh_scenario_states)
100 if self.scenario_state_interval:
101 hass_api.subscribe_interval(self.scenario_state_interval, self.async_refresh_scenario_states)
102 else:
103 _LOGGER.info("SUPERNOTIFY Scenario state refresh is off, scenario binary_sensors will be unknown")
105 def register_entity(self, name: str, entity: SupernotifyScenarioBinarySensor) -> None:
106 """Called by SupernotifyScenarioBinarySensor.async_added_to_hass()."""
107 self._entities[name] = entity
109 def unregister_entity(self, name: str) -> None:
110 """Called by SupernotifyScenarioBinarySensor.async_will_remove_from_hass()."""
111 self._entities.pop(name, None)
113 @callback
114 def async_refresh_entity(self, name: str, context: Context | None = None) -> None:
115 """Re-publish one scenario's binary_sensor now, whether or not periodic refresh is on,
116 attributing any change of state to `context` if given"""
117 entity = self._entities.get(name)
118 if entity is not None:
119 if context is not None:
120 entity.async_set_context(context)
121 entity.async_write_ha_state()
123 def scenario_has_state(self, scenario: Scenario) -> bool:
124 """Whether a scenario has any state to report - anything that hasn't opted out with
125 expose_state. That is the evaluated state of its conditions if it has any, otherwise
126 a manual state that something outside Supernotify sets, see Scenario.manual_active."""
127 return scenario.expose_state
129 def scenario_is_on(self, scenario: Scenario) -> bool | None:
130 """`is_on` for SupernotifyScenarioBinarySensor - None maps to STATE_UNKNOWN."""
131 if scenario.is_manual:
132 return scenario.manual_active
133 state = self._scenario_state(scenario, self._batch_cvars)
134 if state == STATE_UNKNOWN:
135 return None
136 return state == STATE_ON
138 def _collect_scenario_condition_entities(self) -> dict[str, set[str]]:
139 """Entities referenced by each scenario's conditions.
141 A scenario whose conditions reference no Home Assistant entity depends
142 only on the per-notification variables (notification_priority /
143 applied_scenarios). Such a scenario is 'transient': it has no meaningful
144 state between notifications, so it is left as STATE_UNKNOWN. Extraction is
145 best-effort (templates are opaque); the periodic refresh is the safety net.
146 """
147 mapping: dict[str, set[str]] = {}
148 for name, scenario in self.scenarios.items():
149 ents: set[str] = set()
150 for cond in scenario.conditions_config or []:
151 try:
152 ents |= condition.async_extract_entities(cond)
153 except Exception:
154 _LOGGER.debug("SUPERNOTIFY could not extract entities for scenario %s", name)
155 mapping[name] = ents
156 return mapping
158 @property
159 def scenario_state_enabled(self) -> bool:
160 return bool(self.scenario_control.get(CONF_REFRESH, False))
162 @property
163 def scenario_state_interval(self) -> int:
164 return int(self.scenario_control.get(CONF_REFRESH_INTERVAL, SCENARIO_STATE_REFRESH_DEFAULT))
166 def _index_scenarios_by_entity(self) -> dict[str, set[str]]:
167 """Reverse of _collect_scenario_condition_entities: entity -> scenarios depending on it.
169 Used to re-evaluate only the scenarios a state change can actually affect, instead of
170 the whole registry on every event.
171 """
172 index: dict[str, set[str]] = {}
173 for name, entities in self._scenario_cond_entities.items():
174 scenario = self.scenarios.get(name)
175 if scenario is not None and not scenario.expose_state:
176 continue
177 for entity_id in entities:
178 index.setdefault(entity_id, set()).add(name)
179 return index
181 def _scenario_state(self, scenario: Scenario, cvars: ConditionVariables | None = None) -> str:
182 """State to expose for a scenario binary_sensor.
184 - refresh not switched on in `scenario_control` -> STATE_UNKNOWN, since nothing
185 would keep an evaluated state current and it would stay as it was at startup;
186 - no conditions at all (manual/emergency-only scenario) -> STATE_UNKNOWN
187 (state is undefined outside of a notification);
188 - otherwise ON/OFF from a neutral evaluation (current occupancy, medium
189 priority), the same basis as enquire_active_scenarios() - including a
190 scenario whose conditions reference no HA entity (e.g. a pure now()/date
191 template): those are exactly what the periodic sweep in initialize() exists
192 to keep current, so they get evaluated for real rather than stuck at UNKNOWN.
193 A condition keyed off notification_priority/applied_scenarios/message/title
194 instead will always evaluate the same way here, since those are neutral
195 placeholders rather than real values - a known, accepted limitation rather
196 than something this method tries to detect and suppress.
197 """
198 if not self.scenario_state_enabled or not scenario.expose_state:
199 return STATE_UNKNOWN
200 if not scenario.conditions_config:
201 return STATE_UNKNOWN
202 if cvars is None:
203 occupiers = self._people_registry.determine_occupancy()
204 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
205 return STATE_ON if scenario.evaluate(cvars) else STATE_OFF
207 @callback
208 def async_refresh_scenario_states(self, *args: Any) -> None:
209 """Ask each affected scenario's binary_sensor entity to re-read and re-publish its state.
211 Triggered by the 1-minute timer (time/date scenarios and any dependency
212 not captured by entity extraction) and by state changes of the scenarios'
213 condition entities (immediate reactivity). The entity's own `is_on`
214 property (via scenario_is_on() above) does the actual (pure, in-memory)
215 evaluation on read; this only decides which entities need to refresh, and
216 is a no-op for a scenario with no entity registered yet (e.g. before the
217 binary_sensor platform has finished loading).
218 """
219 if not self.scenario_state_enabled:
220 return
221 names: set[str] | None = None
222 context: Context | None = None
223 if args:
224 event = args[0]
225 entity_id = getattr(event, "data", {}).get("entity_id") if hasattr(event, "data") else None
226 if entity_id is not None:
227 names = self._scenario_by_entity.get(entity_id, set())
228 if not names:
229 return
230 # so a scenario turning on or off is attributed, in the logbook and history, to
231 # whatever changed the condition entity - the periodic sweep has no such cause
232 context = getattr(event, "context", None)
234 # Computed once for the whole batch (see _batch_cvars/scenario_is_on) rather than once
235 # per scenario below - determine_occupancy() is only dict lookups, not real I/O, but
236 # doing it once per refresh instead of once per scenario is the correct scale for a
237 # mechanism meant to run on every relevant state change and every periodic sweep.
238 occupiers = self._people_registry.determine_occupancy()
239 self._batch_cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
240 refreshed: int = 0
241 try:
242 for name in self.scenarios if names is None else names:
243 entity = self._entities.get(name)
244 if entity is not None:
245 if context is not None:
246 entity.async_set_context(context)
247 entity.async_write_ha_state()
248 refreshed += 1
249 finally:
250 self._batch_cvars = None
251 _LOGGER.debug("SUPERNOTIFY Refreshed state of %s scenario binary_sensors", refreshed)
254class Scenario:
255 def __init__(
256 self, name: str, scenario_definition: dict[str, Any], delivery_registry: DeliveryRegistry, hass_api: HomeAssistantAPI
257 ) -> None:
258 self.hass_api: HomeAssistantAPI = hass_api
259 self.delivery_registry = delivery_registry
260 self.enabled: bool = scenario_definition.get(CONF_ENABLED, True)
261 # as configured, which enabled can be overridden from at runtime by the scenario switch
262 self.config_enabled: bool = self.enabled
263 self.expose_state: bool = scenario_definition.get(CONF_EXPOSE_STATE, True)
264 self.name: str = name
265 self.alias: str | None = scenario_definition.get(CONF_ALIAS)
266 self.notify_entity: str | None = scenario_definition.get(CONF_NOTIFY_ENTITY)
267 # set once its notify entity is made, see exposed_notify_entities() in notify.py
268 self.notify_entity_id: str | None = None
269 self.conditions: ConditionsFunc | None = None
270 self.conditions_config: list[ConfigType] | None = scenario_definition.get(CONF_CONDITIONS)
271 # With no conditions to evaluate, whether the scenario applies is set from outside, by
272 # the state of its binary_sensor - see SupernotifyScenarioManualBinarySensor
273 self.manual_active: bool = False
274 self.media: dict[str, Any] | None = scenario_definition.get(CONF_MEDIA)
275 self.action_groups: list[str] = scenario_definition.get(CONF_ACTION_GROUP_NAMES, [])
276 self._config_delivery: dict[str, DeliveryCustomization]
277 self.delivery_overrides: dict[str, DeliveryCustomization] = {}
278 self._delivery_selector: dict[str, str] = {}
279 self.last_trace: ActionTrace | None = None
280 self.startup_issue_count: int = 0
282 delivery_data = scenario_definition.get(CONF_DELIVERY)
283 if isinstance(delivery_data, list):
284 # a bare list of deliveries implies enabling
285 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for list %s", self.name, delivery_data)
286 self._config_delivery = {k: DeliveryCustomization(config=None, default_enabled=True) for k in delivery_data}
287 elif isinstance(delivery_data, str) and delivery_data:
288 # a bare list of deliveries implies enabled delivery
289 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery default enabled for single %s", self.name, delivery_data)
290 self._config_delivery = {delivery_data: DeliveryCustomization(config=None, default_enabled=True)}
291 elif isinstance(delivery_data, dict):
292 # whereas a dict may be used to tune or restrict
293 _LOGGER.debug("SUPERNOTIFY Scenario %s delivery selection %s", self.name, delivery_data)
294 self._config_delivery = {}
295 for k, v in delivery_data.items():
296 # a wildcard/regex pattern with no explicit enabled: only apply as an
297 # override to deliveries already selected elsewhere, don't force-enable
298 # every delivery it happens to match (e.g. selection: scenario deliveries)
299 self._config_delivery[k] = DeliveryCustomization(
300 config=v, default_enabled=True if k in delivery_registry.deliveries else None
301 )
302 elif delivery_data:
303 _LOGGER.warning("SUPERNOTIFY Unable to interpret scenario %s delivery data %s", self.name, delivery_data)
304 self._config_delivery = {}
305 else:
306 _LOGGER.warning("SUPERNOTIFY No delivery definitions for scenario %s", self.name)
307 self._config_delivery = {}
309 @property
310 def is_manual(self) -> bool:
311 """A scenario with no conditions, which only applies when its manual state is on"""
312 return not self.conditions_config
314 async def validate(self, valid_action_group_names: list[str] | None = None) -> bool:
315 """Validate Home Assistant conditiion definition at initiation"""
316 if self.conditions_config:
317 error: str | None = None
318 try:
319 # note: basic template syntax within conditions already validated by voluptuous checks
320 self.conditions = await self.hass_api.build_conditions(
321 self.conditions_config, strict=True, validate=True, name=self.name
322 )
323 except vol.Invalid as vi:
324 _LOGGER.error(
325 f"SUPERNOTIFY Condition definition for scenario {self.name} fails Home Assistant schema check {vi}"
326 )
327 error = str(vi)
328 except Exception as e:
329 _LOGGER.error(
330 "SUPERNOTIFY Disabling scenario %s with error validating %s: %s", self.name, self.conditions_config, e
331 )
332 error = str(e)
333 if error is not None:
334 self.startup_issue_count += 1
335 self.hass_api.raise_issue(
336 f"scenario_{self.name}_condition",
337 is_fixable=False,
338 issue_key="scenario_condition",
339 issue_map={"scenario": self.name, "error": error},
340 severity=ir.IssueSeverity.ERROR,
341 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
342 )
344 for name_or_pattern, config in self._config_delivery.items():
345 matched: bool = False
346 delivery: Delivery | None = self.delivery_registry.deliveries.get(name_or_pattern)
347 if delivery:
348 self.delivery_overrides[delivery.name] = config
349 self._delivery_selector[delivery.name] = name_or_pattern
350 matched = True
351 else:
352 # look for a wildcard match instead
353 for delivery_name in self.delivery_registry.deliveries:
354 if re.fullmatch(name_or_pattern, delivery_name):
355 if self._delivery_selector.get(delivery_name) == delivery_name:
356 _LOGGER.debug(
357 f"SUPERNOTIFY Scenario {self.name} ignoring '{name_or_pattern}' shadowing explicit delivery {delivery_name}"
358 )
359 else:
360 _LOGGER.debug(
361 f"SUPERNOTIFY Scenario {self.name} delivery '{name_or_pattern}' matched {delivery_name}"
362 )
363 self.delivery_overrides[delivery_name] = config
364 self._delivery_selector[delivery_name] = name_or_pattern
365 matched = True
366 if not matched:
367 _LOGGER.error(f"SUPERNOTIFY Scenario {self.name} has delivery {name_or_pattern} not found")
368 self.startup_issue_count += 1
369 self.hass_api.raise_issue(
370 f"scenario_{self.name}_delivery_{name_or_pattern.replace('.', 'DOT').replace('*', 'STAR')}",
371 is_fixable=False,
372 issue_key="scenario_delivery",
373 issue_map={
374 "scenario": self.name,
375 "delivery": name_or_pattern,
376 "deliveries": ", ".join(self.delivery_registry.deliveries),
377 },
378 severity=ir.IssueSeverity.WARNING,
379 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
380 )
382 if valid_action_group_names is not None:
383 invalid_action_groups: list[str] = []
384 for action_group_name in self.action_groups:
385 if action_group_name not in valid_action_group_names:
386 _LOGGER.error(f"SUPERNOTIFY Unknown action group {action_group_name} removed from scenario {self.name}")
387 invalid_action_groups.append(action_group_name)
388 self.startup_issue_count += 1
389 self.hass_api.raise_issue(
390 f"scenario_{self.name}_action_group_{action_group_name}",
391 is_fixable=False,
392 issue_key="scenario_action_group",
393 issue_map={
394 "scenario": self.name,
395 "action_group": action_group_name,
396 "action_groups": ", ".join(valid_action_group_names),
397 },
398 severity=ir.IssueSeverity.WARNING,
399 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/scenarios/",
400 )
401 for action_group_name in invalid_action_groups:
402 self.action_groups.remove(action_group_name)
404 return self.startup_issue_count == 0
406 def enabling_deliveries(self) -> list[str]:
407 # default_enabled (see __init__/DeliveryCustomization) already resolves whether an
408 # omitted `enabled` key should count as enabling: True for a directly-named delivery
409 # (matching the list/string delivery config forms), None for a wildcard/regex match
410 # so it doesn't force-select every delivery it happens to match. An explicit
411 # `enabled: None` (e.g. just to carry a priority override) is left as None too, not
412 # upgraded to the default - only a real `enabled: true` should land here.
413 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is True]
415 def relevant_deliveries(self) -> list[str]:
416 return [
417 del_name
418 for del_name, del_config in self.delivery_overrides.items()
419 if del_config.enabled or del_config.enabled is None
420 ]
422 def disabling_deliveries(self) -> list[str]:
423 return [del_name for del_name, del_config in self.delivery_overrides.items() if del_config.enabled is False]
425 def delivery_customization(self, delivery_name: str) -> DeliveryCustomization | None:
426 return self.delivery_overrides.get(delivery_name)
428 def attributes(self, include_condition: bool = True, include_trace: bool = False) -> dict[str, Any]:
429 """Return scenario attributes"""
430 attrs = {
431 ATTR_NAME: self.name,
432 ATTR_ENABLED: self.enabled,
433 ATTR_MEDIA: self.media,
434 "action_groups": self.action_groups,
435 "delivery": self.delivery_overrides,
436 }
437 if self.alias:
438 attrs[ATTR_FRIENDLY_NAME] = self.alias
439 if include_condition:
440 attrs["conditions"] = self.conditions_config
441 if include_trace and self.last_trace:
442 attrs["trace"] = self.last_trace.as_extended_dict()
443 return attrs
445 def delivery_config(self, delivery_name: str) -> DeliveryCustomization | None:
446 return self.delivery_overrides.get(delivery_name)
448 def contents(self, minimal: bool = False, **_kwargs: Any) -> dict[str, Any]:
449 """Archive friendly view of scenario"""
450 return self.attributes(include_condition=False, include_trace=not minimal)
452 def evaluate(self, condition_variables: ConditionVariables) -> bool:
453 """Evaluate scenario conditions"""
454 result: bool | None = False
455 if self.enabled and self.is_manual:
456 return self.manual_active
457 if self.enabled and self.conditions:
458 try:
459 result = self.hass_api.evaluate_conditions(self.conditions, condition_variables)
460 if result is None:
461 _LOGGER.warning(f"SUPERNOTIFY Scenario {self.name} condition empty result")
462 except Exception as e:
463 _LOGGER.error(
464 "SUPERNOTIFY Scenario %s condition eval failed: %s, vars: %s",
465 self.name,
466 e,
467 condition_variables.as_dict() if condition_variables else {},
468 )
469 return result if result is not None else False
471 async def trace(self, condition_variables: ConditionVariables) -> bool:
472 """Trace scenario condition execution"""
473 result: bool | None = False
474 trace: ActionTrace | None = None
475 if self.enabled and self.is_manual:
476 return self.manual_active
477 if self.enabled and self.conditions:
478 result, trace = await self.hass_api.trace_conditions(
479 self.conditions, condition_variables, trace_name=f"scenario_{self.name}"
480 )
481 if trace:
482 self.last_trace = trace
483 return result if result is not None else False
486@contextmanager
487def trace_action(
488 hass: HomeAssistant,
489 item_id: str,
490 config: dict[str, Any],
491 context: Context | None = None,
492 stored_traces: int = 5,
493) -> Generator[ActionTrace]:
494 """Trace execution of a scenario."""
495 trace = ActionTrace(item_id, config, None, context or Context())
496 async_store_trace(hass, trace, stored_traces)
498 try:
499 yield trace
500 except Exception as ex:
501 if item_id:
502 trace.set_error(ex)
503 raise
504 finally:
505 if item_id:
506 trace.finished()