Coverage for custom_components/supernotify/transports/generic.py: 97%
208 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import logging
4from dataclasses import dataclass, field
5from typing import TYPE_CHECKING, Any, ClassVar
7from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
8from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
10# ATTR_VARIABLES from script.const has import issues
11from homeassistant.const import ATTR_ENTITY_ID
12from homeassistant.helpers import config_validation as cv
13from homeassistant.helpers.typing import ConfigType
15from custom_components.supernotify.common import ensure_list
16from custom_components.supernotify.const import (
17 ATTR_ACTION_URL,
18 ATTR_ACTIONS,
19 ATTR_MEDIA,
20 ATTR_MEDIA_SNAPSHOT_URL,
21 ATTR_PRIORITY,
22 PRIORITY_CRITICAL,
23 PRIORITY_HIGH,
24 PRIORITY_LOW,
25 PRIORITY_MEDIUM,
26 PRIORITY_MINIMUM,
27 PRIORITY_VALUES,
28 TRANSPORT_GENERIC,
29)
30from custom_components.supernotify.model import (
31 DataFilter,
32 DebugTrace,
33 MessageOnlyPolicy,
34 SelectionRule,
35 Target,
36 TargetRequired,
37 TransportConfig,
38 TransportFeature,
39)
40from custom_components.supernotify.options import (
41 OPTION_DATA_KEYS_SELECT,
42 OPTION_MESSAGE_USAGE,
43 OPTION_SIMPLIFY_TEXT,
44 OPTION_STRIP_URLS,
45 OPTION_TARGET_CATEGORIES,
46 DeliveryOption,
47)
48from custom_components.supernotify.transport import (
49 Transport,
50)
52if TYPE_CHECKING:
53 from custom_components.supernotify.delivery import Delivery
54 from custom_components.supernotify.envelope import Envelope
55 from custom_components.supernotify.hass_api import HomeAssistantAPI
57_LOGGER = logging.getLogger(__name__)
59OPTION_RAW = "raw"
60OPTION_GENERIC_DOMAIN_STYLE = "handle_as_domain"
63class GenericTransport(Transport):
64 """Call any service, including non-notify ones, like switch.turn_on or mqtt.publish"""
66 name = TRANSPORT_GENERIC
67 declared_options: ClassVar[list[DeliveryOption]] = [
68 DeliveryOption(OPTION_RAW, "Don't apply domain specific data handling and pruning rules", value_type=cv.boolean),
69 DeliveryOption(OPTION_GENERIC_DOMAIN_STYLE, "Treat the action call in the same way as a known domain"),
70 DeliveryOption(
71 OPTION_DATA_KEYS_SELECT,
72 "Prune the data block by including/excluding values or by regex pattern",
73 value_type=SelectionRule,
74 ),
75 ]
77 def __init__(self, *args: Any, **kwargs: Any) -> None:
78 super().__init__(*args, **kwargs)
80 @property
81 def supported_features(self) -> TransportFeature:
82 return TransportFeature.MESSAGE | TransportFeature.TITLE
84 @property
85 def default_config(self) -> TransportConfig:
86 config = TransportConfig()
87 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
88 config.delivery_defaults.inclusion = self.inclusion_mode
89 config.delivery_defaults.options = {
90 OPTION_SIMPLIFY_TEXT: False,
91 OPTION_STRIP_URLS: False,
92 OPTION_RAW: False,
93 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
94 OPTION_DATA_KEYS_SELECT: None,
95 OPTION_GENERIC_DOMAIN_STYLE: None,
96 }
97 return config
99 def validate_action(self, action: str | None) -> bool:
100 if action is not None and "." in action:
101 return True
102 _LOGGER.warning("SUPERNOTIFY Generic transport must have a qualified action name, e.g. notify.foo")
103 return False
105 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
106 # entirely delivery-driven (bring-your-own-action) - there's no transport-level
107 # prerequisite to check. A transport-level default action still gates whether
108 # build_standard_deliveries() below produces something usable; if not,
109 # DeliveryRegistry prunes this transport entirely once it's confirmed no delivery
110 # (explicit or auto) uses it
111 return True
113 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
114 # with no default action configured, there's nothing to auto-generate a delivery
115 # from - validate_action()/Delivery.initialize() reject it before it's ever used
116 action = self.delivery_defaults.action
117 if action is None or "." not in action:
118 return {}
119 return {self.name: {}}
121 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
122 # inputs
123 data: dict[str, Any] = envelope.data or {}
124 core_action_data: dict[str, Any] = envelope.core_action_data(force_message=False)
125 raw_mode: bool = envelope.delivery.options.get(OPTION_RAW, False)
126 qualified_action: str | None = envelope.delivery.action
127 split_action = (
128 qualified_action.split(".", 1) if qualified_action and "." in qualified_action else [None, qualified_action]
129 )
130 domain: str | None = split_action[0]
131 service: str | None = split_action[1]
133 equiv_domain: str | None = domain
134 if envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE):
135 equiv_domain = envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE)
136 _LOGGER.debug("SUPERNOTIFY Handling %s generic message as if it was %s", domain, equiv_domain)
138 # outputs
139 action_data: dict[str, Any] = {}
140 target_data: dict[str, Any] | None = {}
141 build_targets: bool = False
142 prune_data: bool = True
143 mini_envelopes: list[MiniEnvelope] = [] # only used for script and ntfy
145 if raw_mode:
146 action_data = core_action_data
147 action_data.update(data)
148 build_targets = True
149 elif equiv_domain == "notify":
150 action_data = core_action_data
151 if qualified_action == "notify.send_message":
152 # amongst the wild west of notifty handling, at least care for the modern core one
153 action_data = core_action_data
154 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
155 prune_data = False
156 else:
157 action_data = core_action_data
158 action_data[ATTR_DATA] = data
159 build_targets = True
160 elif equiv_domain == "input_text":
161 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
162 if "value" in data:
163 action_data = {"value": data["value"]}
164 else:
165 action_data = {"value": core_action_data[ATTR_MESSAGE]}
166 elif equiv_domain == "switch":
167 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
168 elif equiv_domain == "mqtt":
169 action_data = data
170 if "payload" not in action_data:
171 action_data["payload"] = envelope.message
172 # add `payload:` with empty value for empty topic
173 elif equiv_domain == "tts":
174 action_data = core_action_data
175 action_data.update(data)
176 build_targets = True
177 elif equiv_domain == "notify_events":
178 mini_envelopes.extend(
179 notify_events(envelope.message, envelope.title, core_action_data, data, envelope.delivery, envelope.priority)
180 )
181 elif qualified_action == "ntfy.publish":
182 mini_envelopes.extend(
183 ntfy(core_action_data, data, envelope.target, envelope.delivery, envelope.priority, self.hass_api)
184 )
185 elif equiv_domain in ("siren", "light"):
186 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
187 action_data = data
188 elif equiv_domain == "rest_command":
189 action_data = data
190 elif equiv_domain == "script":
191 mini_envelopes.extend(
192 script(qualified_action, core_action_data, data, envelope.target, envelope.delivery, self.hass_api)
193 )
194 else:
195 action_data = core_action_data
196 action_data.update(data)
197 build_targets = True
199 if mini_envelopes:
200 results: list[bool] = [
201 await self.call_action(
202 envelope, qualified_action, action_data=mini_envelope.action_data, target_data=mini_envelope.target_data
203 )
204 for mini_envelope in mini_envelopes
205 ]
206 return all(results)
208 if build_targets:
209 all_targets: list[str] = []
210 if OPTION_TARGET_CATEGORIES in envelope.delivery.options:
211 for category in ensure_list(envelope.delivery.options.get(OPTION_TARGET_CATEGORIES, [])):
212 all_targets.extend(envelope.target.for_category(category))
213 else:
214 all_targets = envelope.target.resolved_targets()
215 if len(all_targets) == 1:
216 action_data[ATTR_TARGET] = all_targets[0]
217 elif len(all_targets) >= 1:
218 action_data[ATTR_TARGET] = all_targets
220 if prune_data and action_data:
221 action_data = envelope.customize_data(action_data)
222 if not raw_mode and domain and service:
223 # use the service schema to remove unsupported fields or force type
224 action_data = self.context.hass_api.coerce_schema(domain, service, action_data)
226 return await self.call_action(envelope, qualified_action, action_data=action_data, target_data=target_data or None)
229@dataclass
230class MiniEnvelope:
231 action_data: dict[str, Any] = field(default_factory=dict)
232 target_data: dict[str, Any] | None = None
235def script(
236 qualified_action: str | None,
237 core_action_data: dict[str, Any],
238 data: dict[str, Any],
239 target: Target,
240 delivery: Delivery,
241 hass_api: HomeAssistantAPI,
242) -> list[MiniEnvelope]:
243 """Customize `data` for script integration"""
244 results: list[MiniEnvelope] = []
245 if qualified_action in ("script.turn_on", "script.turn_off"):
246 action_data = {}
247 action_data["variables"] = core_action_data
248 if "variables" in data:
249 action_data["variables"].update(data.pop("variables"))
250 action_data["variables"].update(data)
251 action_data = hass_api.coerce_schema("script", qualified_action.replace("script.", ""), action_data)
252 results.append(MiniEnvelope(action_data=action_data, target_data={ATTR_ENTITY_ID: target.domain_entity_ids("script")}))
253 else:
254 action_data = core_action_data
255 action_data.update(data)
256 results.append(MiniEnvelope(action_data=action_data))
258 return results
261def ntfy(
262 core_action_data: dict[str, Any],
263 data: dict[str, Any],
264 target: Target,
265 delivery: Delivery,
266 priority: str | None,
267 hass_api: HomeAssistantAPI,
268) -> list[MiniEnvelope]:
269 """Customize `data` for ntfy integration"""
270 results: list[MiniEnvelope] = []
271 action_data: dict[str, Any] = dict(core_action_data)
272 action_data.update(data)
273 action_data = hass_api.coerce_schema("ntfy", "publish", action_data)
275 if priority and priority in PRIORITY_VALUES:
276 action_data[ATTR_PRIORITY] = PRIORITY_VALUES.get(priority, 3)
278 media = action_data.pop(ATTR_MEDIA, {})
279 if media and media.get(ATTR_MEDIA_SNAPSHOT_URL) and "attach" not in action_data:
280 action_data["attach"] = media.get(ATTR_MEDIA_SNAPSHOT_URL)
281 actions = action_data.pop(ATTR_ACTIONS, [])
282 if len(actions) > 0:
283 first_action = actions[0]
284 if first_action.get(ATTR_ACTION_URL) and "click" not in action_data:
285 action_data["click"] = first_action.get(ATTR_ACTION_URL)
287 if target.email and "email" not in action_data:
288 for email in target.email:
289 call_data: dict[str, Any] = dict(action_data)
290 if len(results) == 1 and len(target.email) == 1:
291 results[0].action_data["email"] = email
292 else:
293 call_data["email"] = email
294 results.append(MiniEnvelope(action_data=call_data))
295 if target.phone and "call" not in action_data:
296 for phone in target.phone:
297 call_data = dict(action_data)
298 if len(results) == 1 and len(target.phone) == 1:
299 results[0].action_data["call"] = phone
300 else:
301 call_data["call"] = phone
302 results.append(MiniEnvelope(action_data=call_data))
303 notify_entities = target.domain_entity_ids(NOTIFY_DOMAIN)
305 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)
306 action_data = DataFilter(rules).apply(action_data)
308 if not results or notify_entities:
309 if len(results) == 1:
310 results[0].target_data = {"entity_id": notify_entities}
311 else:
312 results.append(MiniEnvelope(action_data=dict(action_data), target_data={"entity_id": notify_entities}))
314 return results
317def notify_events(
318 message: str | None,
319 title: str | None,
320 core_action_data: dict[str, Any],
321 data: dict[str, Any],
322 delivery: Delivery,
323 priority: str | None,
324) -> list[MiniEnvelope]:
325 """Customize `data` for notify_events integration"""
326 results: list[MiniEnvelope] = []
327 input_data: dict[str, Any] = dict(core_action_data)
328 input_data.update(data)
330 action_data: dict[str, Any] = {}
331 action_data[ATTR_MESSAGE] = message
332 priority_mapping: dict[str, str] = {
333 PRIORITY_MINIMUM: "lowest",
334 PRIORITY_LOW: "low",
335 PRIORITY_MEDIUM: "normal",
336 PRIORITY_HIGH: "high",
337 PRIORITY_CRITICAL: "highest",
338 }
339 if title:
340 action_data.setdefault(ATTR_DATA, {})
341 action_data[ATTR_DATA][ATTR_TITLE] = title
343 if ATTR_DATA in input_data:
344 # notify_events is schema-less for action
345 action_data[ATTR_DATA] = input_data[ATTR_DATA]
347 if priority and priority in PRIORITY_VALUES:
348 action_data.setdefault(ATTR_DATA, {})
349 action_data[ATTR_DATA][ATTR_PRIORITY] = priority_mapping.get(priority)
350 elif priority and priority in priority_mapping.values():
351 action_data.setdefault(ATTR_DATA, {})
352 action_data[ATTR_DATA][ATTR_PRIORITY] = priority
354 if "token" in input_data:
355 action_data.setdefault(ATTR_DATA, {})
356 action_data[ATTR_DATA]["token"] = input_data["token"]
357 if "level" in input_data:
358 action_data.setdefault(ATTR_DATA, {})
359 action_data[ATTR_DATA]["level"] = input_data["level"]
361 if input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL) and "images" not in input_data:
362 action_data.setdefault(ATTR_DATA, {})
363 action_data[ATTR_DATA].setdefault("images", [])
364 action_data[ATTR_DATA]["images"].append({"url": input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)})
366 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)
367 action_data = DataFilter(rules).apply(action_data)
368 results.append(MiniEnvelope(action_data=dict(action_data)))
370 return results