Coverage for custom_components/autoarm/notifier.py: 99%
134 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-26 20:57 +0000
1import logging
2from typing import Any
4from homeassistant.components.alarm_control_panel.const import AlarmControlPanelState
5from homeassistant.const import CONF_ACTION, CONF_ACTIONS, CONF_SERVICE, CONF_SOURCE, CONF_STATE, CONF_TARGET
6from homeassistant.core import Context, HomeAssistant
7from homeassistant.helpers import config_validation as cv
8from homeassistant.helpers import translation
10from custom_components.autoarm.const import (
11 ALARM_STATES,
12 CONF_SCENARIO,
13 CONF_SUPERNOTIFY,
14 DOMAIN,
15 MOBILE_ACTION_STATES,
16 NOTIFY_COMMON,
17 NOTIFY_PRIORITY_SOURCES,
18 SUPERNOTIFY_ACTION,
19 SUPERNOTIFY_MOBILE_ACTIONS,
20 ChangeSource,
21)
22from custom_components.autoarm.helpers import AppHealthTracker, friendly_state
24_LOGGER = logging.getLogger(__name__)
27class Notifier:
28 def __init__(
29 self,
30 notify_profiles: dict[str, dict[str, Any]] | None,
31 hass: HomeAssistant,
32 app_health_tracker: AppHealthTracker,
33 notify_action: str | None,
34 notify_targets: list[str] | None = None,
35 notify_data: dict[str, Any] | None = None,
36 notify_target: dict[str, Any] | None = None,
37 notify_custom_target: list[str] | None = None,
38 notify_deliveries: list[str] | None = None,
39 notify_priorities: dict[str, str] | None = None,
40 ) -> None:
41 self.notify_profiles: dict[str, dict[str, Any]] = notify_profiles or {}
42 self.hass: HomeAssistant = hass
43 self.app_health_tracker: AppHealthTracker = app_health_tracker
44 self.notify_action: str | None = notify_action
45 # targets typed in, as addresses or entity ids, wherever in the options they were put
46 self.notify_targets: list[str] = list(dict.fromkeys([*(notify_targets or []), *(notify_custom_target or [])]))
47 # targets picked from Home Assistant's own entities, devices, areas and labels
48 self.notify_target: dict[str, Any] = notify_target or {}
49 self.notify_deliveries: list[str] = notify_deliveries or []
50 # Supernotify priority for each kind of change, keyed as NOTIFY_PRIORITY_SOURCES
51 self.notify_priorities: dict[str, str] = notify_priorities or {}
52 # extra service data configured via the options UI's ObjectSelector, lowest priority so
53 # YAML profile data (common or per-profile) can still override individual keys
54 self.notify_data: dict[str, Any] = notify_data or {}
55 if (
56 (self.notify_action or self.notify_profiles.get(NOTIFY_COMMON, {}).get(CONF_SERVICE)) == "notify.send_message"
57 and not self.notify_targets
58 and not self.notify_target
59 and not any(profile.get(CONF_TARGET) for profile in self.notify_profiles.values())
60 ):
61 _LOGGER.warning(
62 "AUTOARM No notifications will be sent, as notify.send_message needs Notification Targets, set in the options"
63 )
65 async def _mobile_actions(
66 self, actions: list[dict[str, Any]], state: AlarmControlPanelState | None
67 ) -> list[dict[str, Any]]:
68 """Leave out actions for the state the alarm is already in, and give Auto Arm's own a translated title"""
69 titles: dict[str, str] = await translation.async_get_translations(
70 self.hass, self.hass.config.language, "selector", [DOMAIN]
71 )
72 selected: list[dict[str, Any]] = []
73 for action in actions:
74 action_key: str = str(action.get(CONF_ACTION))
75 if state is not None and MOBILE_ACTION_STATES.get(action_key) == state:
76 continue
77 title: str | None = titles.get(f"component.{DOMAIN}.selector.mobile_action.options.{action_key.lower()}")
78 selected.append({"title": title, **action} if title else action)
79 return selected
81 async def notify(
82 self,
83 source: ChangeSource,
84 from_state: AlarmControlPanelState | None = None,
85 to_state: AlarmControlPanelState | None = None,
86 message: str | None = None,
87 title: str | None = None,
88 context: Context | None = None,
89 source_name: str | None = None,
90 ) -> None:
92 try:
93 selected_profile: dict[str, Any] | None = None
94 selected_profile_name: str | None = None
95 config_by_state_pickiness = sorted(
96 self.notify_profiles, key=lambda v: len(self.notify_profiles[v].get(CONF_STATE, ALARM_STATES))
97 )
98 for profile_name in config_by_state_pickiness:
99 if profile_name == NOTIFY_COMMON:
100 continue
101 profile: dict[str, Any] = self.notify_profiles[profile_name]
102 if profile.get(CONF_SOURCE) and source not in profile.get(CONF_SOURCE, []):
103 _LOGGER.debug(
104 "AUTOARM Notification not selected for %s profile for source match on %s", profile_name, source
105 )
106 continue
107 only_for_states: list[AlarmControlPanelState] | None = profile.get(CONF_STATE)
108 if only_for_states and from_state not in only_for_states and to_state not in only_for_states:
109 _LOGGER.debug(
110 "AUTOARM Notification not selected for %s profile for state match on %s->%s",
111 profile_name,
112 from_state,
113 to_state,
114 )
115 continue
116 selected_profile = profile
117 selected_profile_name = profile_name
118 break
119 if selected_profile is None:
120 _LOGGER.debug("AUTOARM No profile selected for %s notification: %s", source, message)
121 return
123 # separately merge base dict and data sub-dict as cheap and nasty semi-deep-merge
124 base_profile = self.notify_profiles.get(NOTIFY_COMMON, {})
125 base_profile_data = base_profile.get("data", {})
126 merged_profile = dict(base_profile)
127 # options-configured data is the lowest priority layer, YAML profiles can override any key
128 merged_profile_data = dict(self.notify_data)
129 merged_profile_data.update(base_profile_data)
130 if selected_profile is not None:
131 selected_profile_data: dict[str, Any] = selected_profile.get("data", {})
132 merged_profile.update(selected_profile)
133 merged_profile_data.update(selected_profile_data)
134 merged_profile["data"] = merged_profile_data
136 data = merged_profile.get("data", {})
137 if "source" in data and data["source"] is None:
138 data["source"] = str(source)
139 if "profile" in data and data["profile"] is None:
140 data["profile"] = selected_profile_name
141 # a profile's own service wins, then the action chosen in the options, then the common profile's
142 notify_action: str | None = (
143 selected_profile.get(CONF_SERVICE) or self.notify_action or merged_profile.get(CONF_SERVICE)
144 )
145 supernotify_action: bool = notify_action == SUPERNOTIFY_ACTION
146 if supernotify_action:
147 data.setdefault(CONF_ACTIONS, SUPERNOTIFY_MOBILE_ACTIONS)
148 # a priority in a profile or the extra data is kept
149 for kind, priority in self.notify_priorities.items():
150 if source in NOTIFY_PRIORITY_SOURCES.get(kind, []):
151 data.setdefault("priority", priority)
152 if isinstance(data.get(CONF_ACTIONS), list):
153 data[CONF_ACTIONS] = await self._mobile_actions(data[CONF_ACTIONS], to_state)
154 if (merged_profile.get(CONF_SUPERNOTIFY) or supernotify_action) and merged_profile.get(CONF_SCENARIO):
155 data["apply_scenarios"] = merged_profile.get(CONF_SCENARIO)
156 if supernotify_action and to_state:
157 # for Supernotify scenarios to pick up, it ignores any that aren't configured there
158 state_scenarios: list[str] = [f"{DOMAIN}-{to_state}"]
159 if to_state.startswith("armed_"):
160 state_scenarios.append(f"{DOMAIN}-armed")
161 data["apply_scenarios"] = [*cv.ensure_list(data.get("apply_scenarios")), *state_scenarios]
163 # a profile's own targets take over from all of those chosen in the options
164 notify_targets: list[str] | None = merged_profile.get(CONF_TARGET, self.notify_targets)
165 picked_target: dict[str, Any] = {} if CONF_TARGET in merged_profile else self.notify_target
166 if notify_action is None:
167 _LOGGER.debug("AUTOARM Notifications disabled, no notification action")
168 return
169 if notify_action == "notify.send_message" and not notify_targets and not picked_target:
170 _LOGGER.debug("AUTOARM Notifications disabled, no targets for notify.send_message")
171 return
173 if title is None:
174 title = f"Alarm now {friendly_state(to_state)}" if to_state else "Alarm Panel Change"
175 if message is None:
176 if from_state and to_state:
177 if source == ChangeSource.TOD:
178 changed_by = f"{source_name} Times of the Day sensor" if source_name else "Times of the Day sensor"
179 else:
180 changed_by = source.capitalize()
181 message = (
182 f"Alarm state changed from {friendly_state(from_state)} to {friendly_state(to_state)} by {changed_by}"
183 )
184 else:
185 message = "Alarm control panel operation complete"
187 if notify_action and merged_profile:
188 service_data: dict[str, Any] = {"message": message, "title": title}
189 if supernotify_action:
190 # supernotify.notify takes the data fields at the top level
191 service_data.update(data)
192 if self.notify_deliveries and "delivery" not in service_data:
193 service_data["delivery"] = self.notify_deliveries
194 service_data.setdefault("delivery_selection", "explicit")
195 elif data and notify_action != "notify.send_message":
196 # new notify entities can't handle nested data section
197 service_data["data"] = data
198 domain, action = notify_action.split(".", 1)
199 target: dict[str, Any] | None = None
200 if domain == "notify" and action == "send_message":
201 # notify.send_message is entity-based and targets via entity_id,
202 # not a "target" key in service_data like legacy notify services
203 target = dict(picked_target)
204 if entity_ids := [*cv.ensure_list(picked_target.get("entity_id")), *(notify_targets or [])]:
205 target["entity_id"] = entity_ids
206 elif supernotify_action and picked_target:
207 # the same pair of target fields as supernotify.notify itself
208 service_data["target"] = picked_target
209 if notify_targets:
210 service_data["custom_target"] = notify_targets
211 elif notify_targets:
212 service_data["target"] = notify_targets
214 _LOGGER.debug("AUTOARM Notifying %s.%s with %s target=%s", domain, action, service_data, target)
215 try:
216 await self.hass.services.async_call(
217 domain,
218 action,
219 service_data=service_data,
220 target=target,
221 context=context,
222 )
223 except Exception as e:
224 _LOGGER.error("AUTOARM notify failure, %s.%s: %s [%s]", domain, action, service_data, e)
225 raise
226 else:
227 _LOGGER.debug("AUTOARM Skipped notification, service: %s, data: %s", self.notify_action, merged_profile)
229 except Exception:
230 self.app_health_tracker.record_runtime_error()
231 _LOGGER.exception("AUTOARM %s failed", self.notify_action)