Coverage for custom_components / supernotify / transports / generic.py: 96%
195 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +0000
1from __future__ import annotations
3import logging
4from dataclasses import dataclass, field
5from typing import TYPE_CHECKING, Any
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
13from custom_components.supernotify.common import ensure_list
14from custom_components.supernotify.const import (
15 ATTR_ACTION_URL,
16 ATTR_ACTIONS,
17 ATTR_MEDIA,
18 ATTR_MEDIA_SNAPSHOT_URL,
19 ATTR_PRIORITY,
20 OPTION_DATA_KEYS_SELECT,
21 OPTION_GENERIC_DOMAIN_STYLE,
22 OPTION_MESSAGE_USAGE,
23 OPTION_RAW,
24 OPTION_SIMPLIFY_TEXT,
25 OPTION_STRIP_URLS,
26 OPTION_TARGET_CATEGORIES,
27 PRIORITY_CRITICAL,
28 PRIORITY_HIGH,
29 PRIORITY_LOW,
30 PRIORITY_MEDIUM,
31 PRIORITY_MINIMUM,
32 PRIORITY_VALUES,
33 TRANSPORT_GENERIC,
34)
35from custom_components.supernotify.model import (
36 DataFilter,
37 DebugTrace,
38 MessageOnlyPolicy,
39 Target,
40 TargetRequired,
41 TransportConfig,
42 TransportFeature,
43)
44from custom_components.supernotify.transport import (
45 Transport,
46)
48if TYPE_CHECKING:
49 from custom_components.supernotify.delivery import Delivery
50 from custom_components.supernotify.envelope import Envelope
51 from custom_components.supernotify.hass_api import HomeAssistantAPI
53_LOGGER = logging.getLogger(__name__)
54"""
55Replaced by reuse of original service schema to prune out fields
57DATA_FIELDS_ALLOWED_BY_DOMAIN = {
58 "light": [
59 "transition",
60 "rgb_color",
61 "color_temp_kelvin",
62 "brightness_pct",
63 "brightness_step_pct",
64 "effect",
65 "rgbw_color",
66 "rgbww_color",
67 "color_name",
68 "hs_color",
69 "xy_color",
70 "color_temp",
71 "brightness",
72 "brightness_step",
73 "white",
74 "profile",
75 "flash",
76 ],
77 "siren": ["tone", "duration", "volume_level"],
78 "mqtt": ["topic", "payload", "evaluate_payload", "qos", "retain"],
79 "script": ["variables", "wait", "wait_template"],
80 "ntfy": [
81 "title",
82 "message",
83 "markdown",
84 "tags",
85 "priority",
86 "click",
87 "delay",
88 "attach",
89 "attach_file",
90 "filename",
91 "email",
92 "call",
93 "icon",
94 "action",
95 "sequence_id",
96 ],
97 "tts": ["cache", "options", "message", "language", "media_player_entity_id", "entity_id", "target"],
98} """
101class GenericTransport(Transport):
102 """Call any service, including non-notify ones, like switch.turn_on or mqtt.publish"""
104 name = TRANSPORT_GENERIC
106 def __init__(self, *args: Any, **kwargs: Any) -> None:
107 super().__init__(*args, **kwargs)
109 @property
110 def supported_features(self) -> TransportFeature:
111 return TransportFeature.MESSAGE | TransportFeature.TITLE
113 @property
114 def default_config(self) -> TransportConfig:
115 config = TransportConfig()
116 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
117 config.delivery_defaults.options = {
118 OPTION_SIMPLIFY_TEXT: False,
119 OPTION_STRIP_URLS: False,
120 OPTION_RAW: False,
121 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
122 OPTION_DATA_KEYS_SELECT: None,
123 OPTION_GENERIC_DOMAIN_STYLE: None,
124 }
125 return config
127 def validate_action(self, action: str | None) -> bool:
128 if action is not None and "." in action:
129 return True
130 _LOGGER.warning("SUPERNOTIFY generic transport must have a qualified action name, e.g. notify.foo")
131 return False
133 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002
134 # inputs
135 data: dict[str, Any] = envelope.data or {}
136 core_action_data: dict[str, Any] = envelope.core_action_data(force_message=False)
137 raw_mode: bool = envelope.delivery.options.get(OPTION_RAW, False)
138 qualified_action: str | None = envelope.delivery.action
139 split_action = (
140 qualified_action.split(".", 1) if qualified_action and "." in qualified_action else [None, qualified_action]
141 )
142 domain: str | None = split_action[0]
143 service: str | None = split_action[1]
145 equiv_domain: str | None = domain
146 if envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE):
147 equiv_domain = envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE)
148 _LOGGER.debug("SUPERNOTIFY Handling %s generic message as if it was %s", domain, equiv_domain)
150 # outputs
151 action_data: dict[str, Any] = {}
152 target_data: dict[str, Any] | None = {}
153 build_targets: bool = False
154 prune_data: bool = True
155 mini_envelopes: list[MiniEnvelope] = [] # only used for script and ntfy
157 if raw_mode:
158 action_data = core_action_data
159 action_data.update(data)
160 build_targets = True
161 elif equiv_domain == "notify":
162 action_data = core_action_data
163 if qualified_action == "notify.send_message":
164 # amongst the wild west of notifty handling, at least care for the modern core one
165 action_data = core_action_data
166 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
167 prune_data = False
168 else:
169 action_data = core_action_data
170 action_data[ATTR_DATA] = data
171 build_targets = True
172 elif equiv_domain == "input_text":
173 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
174 if "value" in data:
175 action_data = {"value": data["value"]}
176 else:
177 action_data = {"value": core_action_data[ATTR_MESSAGE]}
178 elif equiv_domain == "switch":
179 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
180 elif equiv_domain == "mqtt":
181 action_data = data
182 if "payload" not in action_data:
183 action_data["payload"] = envelope.message
184 # add `payload:` with empty value for empty topic
185 elif equiv_domain == "tts":
186 action_data = core_action_data
187 action_data.update(data)
188 build_targets = True
189 elif equiv_domain == "notify_events":
190 mini_envelopes.extend(notify_events(envelope.message, envelope.title, core_action_data, data, envelope.delivery))
191 elif qualified_action == "ntfy.publish":
192 mini_envelopes.extend(ntfy(core_action_data, data, envelope.target, envelope.delivery, self.hass_api))
193 elif equiv_domain in ("siren", "light"):
194 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}
195 action_data = data
196 elif equiv_domain == "rest_command":
197 action_data = data
198 elif equiv_domain == "script":
199 mini_envelopes.extend(
200 script(qualified_action, core_action_data, data, envelope.target, envelope.delivery, self.hass_api)
201 )
202 else:
203 action_data = core_action_data
204 action_data.update(data)
205 build_targets = True
207 if mini_envelopes:
208 results: list[bool] = [
209 await self.call_action(
210 envelope, qualified_action, action_data=mini_envelope.action_data, target_data=mini_envelope.target_data
211 )
212 for mini_envelope in mini_envelopes
213 ]
214 return all(results)
216 if build_targets:
217 all_targets: list[str] = []
218 if OPTION_TARGET_CATEGORIES in envelope.delivery.options:
219 for category in ensure_list(envelope.delivery.options.get(OPTION_TARGET_CATEGORIES, [])):
220 all_targets.extend(envelope.target.for_category(category))
221 else:
222 all_targets = envelope.target.resolved_targets()
223 if len(all_targets) == 1:
224 action_data[ATTR_TARGET] = all_targets[0]
225 elif len(all_targets) >= 1:
226 action_data[ATTR_TARGET] = all_targets
228 if prune_data and action_data:
229 action_data = envelope.customize_data(action_data)
230 if not raw_mode and domain and service:
231 # use the service schema to remove unsupported fields or force type
232 action_data = self.context.hass_api.coerce_schema(domain, service, action_data)
234 return await self.call_action(envelope, qualified_action, action_data=action_data, target_data=target_data or None)
237@dataclass
238class MiniEnvelope:
239 action_data: dict[str, Any] = field(default_factory=dict)
240 target_data: dict[str, Any] | None = None
243def script(
244 qualified_action: str | None,
245 core_action_data: dict[str, Any],
246 data: dict[str, Any],
247 target: Target,
248 delivery: Delivery, # noqa: ARG001
249 hass_api: HomeAssistantAPI,
250) -> list[MiniEnvelope]:
251 """Customize `data` for script integration"""
252 results: list[MiniEnvelope] = []
253 if qualified_action in ("script.turn_on", "script.turn_off"):
254 action_data = {}
255 action_data["variables"] = core_action_data
256 if "variables" in data:
257 action_data["variables"].update(data.pop("variables"))
258 action_data["variables"].update(data)
259 action_data = hass_api.coerce_schema("script", qualified_action.replace("script.", ""), action_data)
260 results.append(MiniEnvelope(action_data=action_data, target_data={ATTR_ENTITY_ID: target.domain_entity_ids("script")}))
261 else:
262 action_data = core_action_data
263 action_data.update(data)
264 results.append(MiniEnvelope(action_data=action_data))
266 return results
269def ntfy(
270 core_action_data: dict[str, Any],
271 data: dict[str, Any],
272 target: Target,
273 delivery: Delivery,
274 hass_api: HomeAssistantAPI,
275) -> list[MiniEnvelope]:
276 """Customize `data` for ntfy integration"""
277 results: list[MiniEnvelope] = []
278 action_data: dict[str, Any] = dict(core_action_data)
279 action_data.update(data)
280 action_data = hass_api.coerce_schema("ntfy", "publish", action_data)
282 if ATTR_PRIORITY in action_data and action_data[ATTR_PRIORITY] in PRIORITY_VALUES:
283 action_data[ATTR_PRIORITY] = PRIORITY_VALUES.get(action_data[ATTR_PRIORITY], 3)
285 media = action_data.pop(ATTR_MEDIA, {})
286 if media and media.get(ATTR_MEDIA_SNAPSHOT_URL) and "attach" not in action_data:
287 action_data["attach"] = media.get(ATTR_MEDIA_SNAPSHOT_URL)
288 actions = action_data.pop(ATTR_ACTIONS, [])
289 if len(actions) > 0:
290 first_action = actions[0]
291 if first_action.get(ATTR_ACTION_URL) and "click" not in action_data:
292 action_data["click"] = first_action.get(ATTR_ACTION_URL)
294 if target.email and "email" not in action_data:
295 for email in target.email:
296 call_data: dict[str, Any] = dict(action_data)
297 if len(results) == 1 and len(target.email) == 1:
298 results[0].action_data["email"] = email
299 else:
300 call_data["email"] = email
301 results.append(MiniEnvelope(action_data=call_data))
302 if target.phone and "call" not in action_data:
303 for phone in target.phone:
304 call_data = dict(action_data)
305 if len(results) == 1 and len(target.phone) == 1:
306 results[0].action_data["call"] = phone
307 else:
308 call_data["call"] = phone
309 results.append(MiniEnvelope(action_data=call_data))
310 notify_entities = target.domain_entity_ids(NOTIFY_DOMAIN)
312 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)
313 action_data = DataFilter(rules).apply(action_data)
315 if not results or notify_entities:
316 if len(results) == 1:
317 results[0].target_data = {"entity_id": notify_entities}
318 else:
319 results.append(MiniEnvelope(action_data=dict(action_data), target_data={"entity_id": notify_entities}))
321 return results
324def notify_events(
325 message: str | None,
326 title: str | None,
327 core_action_data: dict[str, Any],
328 data: dict[str, Any],
329 delivery: Delivery,
330) -> list[MiniEnvelope]:
331 """Customize `data` for notify_events integration"""
332 results: list[MiniEnvelope] = []
333 input_data: dict[str, Any] = dict(core_action_data)
334 input_data.update(data)
336 action_data: dict[str, Any] = {}
337 action_data[ATTR_MESSAGE] = message
338 priority_mapping: dict[str, str] = {
339 PRIORITY_MINIMUM: "lowest",
340 PRIORITY_LOW: "low",
341 PRIORITY_MEDIUM: "normal",
342 PRIORITY_HIGH: "high",
343 PRIORITY_CRITICAL: "highest",
344 }
345 if title:
346 action_data.setdefault(ATTR_DATA, {})
347 action_data[ATTR_DATA][ATTR_TITLE] = title
349 if ATTR_DATA in input_data:
350 # notify_events is schema-less for action
351 action_data[ATTR_DATA] = input_data[ATTR_DATA]
353 if ATTR_PRIORITY in input_data and input_data[ATTR_PRIORITY] in PRIORITY_VALUES:
354 action_data.setdefault(ATTR_DATA, {})
355 action_data[ATTR_DATA][ATTR_PRIORITY] = priority_mapping.get(input_data[ATTR_PRIORITY])
356 elif ATTR_PRIORITY in input_data and input_data[ATTR_PRIORITY] in priority_mapping.values():
357 action_data.setdefault(ATTR_DATA, {})
358 action_data[ATTR_DATA][ATTR_PRIORITY] = input_data[ATTR_PRIORITY]
360 if "token" in input_data:
361 action_data.setdefault(ATTR_DATA, {})
362 action_data[ATTR_DATA]["token"] = input_data["token"]
363 if "level" in input_data:
364 action_data.setdefault(ATTR_DATA, {})
365 action_data[ATTR_DATA]["level"] = input_data["level"]
367 if input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL) and "images" not in input_data:
368 action_data.setdefault(ATTR_DATA, {})
369 action_data[ATTR_DATA].setdefault("images", [])
370 action_data[ATTR_DATA]["images"].append({"url": input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)})
372 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)
373 action_data = DataFilter(rules).apply(action_data)
374 results.append(MiniEnvelope(action_data=dict(action_data)))
376 return results