Coverage for custom_components/supernotify/transports/chime.py: 98%
238 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-01 15:06 +0000
1import logging
2from abc import abstractmethod
3from dataclasses import dataclass, field
4from typing import TYPE_CHECKING, Any
6import voluptuous as vol
7from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
8from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues
9 ATTR_DEVICE_ID,
10 ATTR_ENTITY_ID,
11 CONF_DOMAIN,
12 CONF_TARGET,
13)
14from homeassistant.exceptions import NoEntitySpecifiedError
15from voluptuous.humanize import humanize_error
17from custom_components.supernotify.const import (
18 ATTR_DATA,
19 ATTR_MEDIA,
20 ATTR_PRIORITY,
21 CONF_TUNE,
22 OPTION_CHIME_ALIASES,
23 OPTION_DEVICE_DISCOVERY,
24 OPTION_DEVICE_DOMAIN,
25 OPTION_DEVICE_MODEL_SELECT,
26 OPTION_TARGET_CATEGORIES,
27 OPTION_TARGET_SELECT,
28 OPTIONS_CHIME_DOMAINS,
29 RE_DEVICE_ID,
30 SELECT_EXCLUDE,
31 TRANSPORT_CHIME,
32)
33from custom_components.supernotify.model import DebugTrace, Target, TargetRequired, TransportConfig, TransportFeature
34from custom_components.supernotify.schema import CHIME_ALIASES_SCHEMA
35from custom_components.supernotify.transport import Transport
37if TYPE_CHECKING:
38 from homeassistant.helpers.typing import ConfigType
40 from custom_components.supernotify.envelope import Envelope
42RE_VALID_CHIME = r"(switch|script|group|rest_command|siren|media_player)\.[A-Za-z0-9_]+"
44_LOGGER = logging.getLogger(__name__)
46DEVICE_DOMAINS = ["alexa_devices"]
49@dataclass
50class ActionCall:
51 domain: str
52 service: str
53 action_data: dict[str, Any] | None = field(default_factory=dict)
54 target_data: dict[str, Any] | None = field(default_factory=dict)
57class ChimeTargetConfig:
58 def __init__(
59 self,
60 entity_id: str | None = None,
61 device_id: str | None = None,
62 tune: str | None = None,
63 duration: int | None = None,
64 volume: float | None = None,
65 data: dict[str, Any] | None = None,
66 domain: str | None = None,
67 **kwargs: Any,
68 ) -> None:
69 self.entity_id: str | None = entity_id
70 self.device_id: str | None = device_id
71 self.domain: str | None = None
72 self.entity_name: str | None = None
73 if self.entity_id and "." in self.entity_id:
74 self.domain, self.entity_name = self.entity_id.split(".", 1)
75 elif self.device_id:
76 self.domain = domain
77 else:
78 _LOGGER.warning(
79 "SUPERNOTIFY Invalid chime target, entity_id: %s, device_id %s, tune:%s", entity_id, device_id, tune
80 )
81 raise NoEntitySpecifiedError("ChimeTargetConfig target must be entity_id or device_id")
82 if kwargs:
83 _LOGGER.warning("SUPERNOTIFY ChimeTargetConfig ignoring unexpected args: %s", kwargs)
84 self.volume: float | None = volume
85 self.tune: str | None = tune
86 self.duration: int | None = duration
87 self.data: dict[str, Any] | None = data or {}
89 def as_dict(self, **kwargs) -> dict[str, Any]: # noqa: ARG002
90 return {
91 "entity_id": self.entity_id,
92 "device_id": self.device_id,
93 "domain": self.domain,
94 "tune": self.tune,
95 "duration": self.duration,
96 "volume": self.volume,
97 "data": self.data,
98 }
100 def __repr__(self) -> str:
101 """Return a developer-oriented string representation of this ChimeTargetConfig"""
102 if self.device_id is not None:
103 return f"ChimeTargetConfig(device_id={self.device_id})"
104 return f"ChimeTargetConfig(entity_id={self.entity_id})"
107class MiniChimeTransport:
108 domain: str
110 @abstractmethod
111 def build(
112 self,
113 target_config: ChimeTargetConfig,
114 action_data: dict[str, Any],
115 entity_name: str | None = None,
116 envelope: Envelope | None = None,
117 **_kwargs: Any,
118 ) -> ActionCall | None:
119 raise NotImplementedError()
122class RestCommandChimeTransport(MiniChimeTransport):
123 domain = "rest_command"
125 def build( # type: ignore[override]
126 self, target_config: ChimeTargetConfig, entity_name: str | None, **_kwargs: Any
127 ) -> ActionCall | None:
128 if entity_name is None:
129 _LOGGER.warning("SUPERNOTIFY rest_command chime target requires entity")
130 return None
131 output_data = target_config.data or {}
132 if target_config.data:
133 output_data.update(target_config.data)
134 return ActionCall(self.domain, entity_name, action_data=output_data)
137class SwitchChimeTransport(MiniChimeTransport):
138 domain = "switch"
140 def build(self, target_config: ChimeTargetConfig, **_kwargs: Any) -> ActionCall | None: # type: ignore[override]
141 return ActionCall(self.domain, "turn_on", target_data={ATTR_ENTITY_ID: target_config.entity_id})
144class SirenChimeTransport(MiniChimeTransport):
145 domain = "siren"
147 def build(self, target_config: ChimeTargetConfig, **_kwargs: Any) -> ActionCall | None: # type: ignore[override]
148 output_data: dict[str, Any] = {ATTR_DATA: {}}
149 if target_config.tune:
150 output_data[ATTR_DATA]["tone"] = target_config.tune
151 if target_config.duration is not None:
152 output_data[ATTR_DATA]["duration"] = target_config.duration
153 if target_config.volume is not None:
154 output_data[ATTR_DATA]["volume_level"] = target_config.volume
155 return ActionCall(
156 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
157 )
160class ScriptChimeTransport(MiniChimeTransport):
161 domain = "script"
163 def build( # type: ignore[override]
164 self,
165 target_config: ChimeTargetConfig,
166 entity_name: str | None,
167 envelope: Envelope,
168 **_kwargs: Any,
169 ) -> ActionCall | None:
170 if entity_name is None:
171 _LOGGER.warning("SUPERNOTIFY script chime target requires entity")
172 return None
173 variables: dict[str, Any] = target_config.data or {}
174 variables[ATTR_MESSAGE] = envelope.message
175 variables[ATTR_TITLE] = envelope.title
176 variables[ATTR_PRIORITY] = envelope.priority
177 variables["chime_tune"] = target_config.tune
178 variables["chime_volume"] = target_config.volume
179 variables["chime_duration"] = target_config.duration
180 output_data: dict[str, Any] = {"variables": variables}
181 if envelope.delivery.debug:
182 output_data["wait"] = envelope.delivery.debug
183 # use `turn_on` rather than direct call to run script in background
184 return ActionCall(
185 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
186 )
189class AlexaDevicesChimeTransport(MiniChimeTransport):
190 domain = "alexa_devices"
192 def build(self, target_config: ChimeTargetConfig, **_kwargs: Any) -> ActionCall | None: # type: ignore[override]
193 output_data: dict[str, Any] = {
194 "device_id": target_config.device_id,
195 "sound": target_config.tune,
196 }
197 return ActionCall(self.domain, "send_sound", action_data=output_data)
200class MediaPlayerChimeTransport(MiniChimeTransport):
201 domain = "media_player"
203 def build(self, target_config: ChimeTargetConfig, action_data: dict[str, Any], **_kwargs: Any) -> ActionCall | None: # type: ignore[override]
204 input_data = target_config.data or {}
205 if action_data:
206 input_data.update(action_data)
207 output_data: dict[str, Any] = {
208 "media": {
209 "media_content_type": input_data.get(ATTR_MEDIA, {"media_content_type": "sound"}).get(
210 "media_content_type", "sound"
211 ),
212 "media_content_id": target_config.tune,
213 }
214 }
215 if input_data.get("enqueue") is not None:
216 output_data["enqueue"] = input_data.get("enqueue")
217 if input_data.get("announce") is not None:
218 output_data["announce"] = input_data.get("announce")
220 return ActionCall(
221 self.domain, "play_media", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
222 )
225class ChimeTransport(Transport):
226 name = TRANSPORT_CHIME
228 def __init__(self, *args: Any, **kwargs: Any) -> None:
229 super().__init__(*args, **kwargs)
230 self.mini_transports: dict[str, MiniChimeTransport] = {
231 t.domain: t
232 for t in [
233 RestCommandChimeTransport(),
234 SwitchChimeTransport(),
235 SirenChimeTransport(),
236 ScriptChimeTransport(),
237 AlexaDevicesChimeTransport(),
238 MediaPlayerChimeTransport(),
239 ]
240 }
242 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]:
243 if OPTION_CHIME_ALIASES in options:
244 chime_aliases: ConfigType = build_aliases(options[OPTION_CHIME_ALIASES])
245 if chime_aliases:
246 _LOGGER.info("SUPERNOTIFY Set up %s chime aliases for %s", len(chime_aliases), delivery_name)
247 else:
248 _LOGGER.warning("SUPERNOTIFY Chime aliases for %s configured but not recognized", delivery_name)
249 else:
250 chime_aliases = {}
251 _LOGGER.debug("SUPERNOTIFY No chime aliases configured for %s", delivery_name)
252 return {"chime_aliases": chime_aliases}
254 @property
255 def supported_features(self) -> TransportFeature:
256 return TransportFeature(0)
258 def extra_attributes(self) -> dict[str, Any]:
259 return {"mini_transports": [t.domain for t in self.mini_transports.values()]}
261 @property
262 def default_config(self) -> TransportConfig:
263 config = TransportConfig()
264 config.delivery_defaults.options = {}
265 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
266 config.delivery_defaults.options = {
267 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID, ATTR_DEVICE_ID],
268 OPTION_TARGET_SELECT: [RE_VALID_CHIME, RE_DEVICE_ID],
269 OPTION_DEVICE_DISCOVERY: True,
270 OPTION_DEVICE_DOMAIN: DEVICE_DOMAINS,
271 OPTION_DEVICE_MODEL_SELECT: {SELECT_EXCLUDE: ["Speaker Group"]},
272 }
273 return config
275 def validate_action(self, action: str | None) -> bool:
276 return action is None
278 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
279 data: dict[str, Any] = {}
280 data.update(envelope.delivery.data)
281 data.update(envelope.data or {})
282 target: Target = envelope.target
284 # chime_repeat = data.pop("chime_repeat", 1)
285 chime_tune: str | None = data.pop("chime_tune", None)
286 chime_volume: float | None = data.pop("chime_volume", None)
287 chime_duration: int | None = data.pop("chime_duration", None)
289 _LOGGER.debug(
290 "SUPERNOTIFY notify_chime: %s -> %s (delivery: %s, env_data:%s, dlv_data:%s)",
291 chime_tune,
292 target.entity_ids,
293 envelope.delivery_name,
294 envelope.data,
295 envelope.delivery.data,
296 )
297 # expand groups
298 expanded_targets = {
299 e: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, entity_id=e)
300 for e in self.hass_api.expand_group(target.entity_ids)
301 }
302 expanded_targets.update({
303 d: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, device_id=d)
304 for d in target.device_ids
305 })
306 # resolve and include chime aliases
307 expanded_targets.update(
308 self.resolve_tune(chime_tune, envelope.delivery.transport_data.get("chime_aliases", {}), target)
309 ) # overwrite and extend
311 chimes = 0
312 if not expanded_targets:
313 _LOGGER.info("SUPERNOTIFY skipping chime, no targets")
314 return False
315 if debug_trace:
316 debug_trace.record_delivery_artefact(envelope.delivery.name, "expanded_targets", expanded_targets)
318 for chime_entity_config in expanded_targets.values():
319 _LOGGER.debug("SUPERNOTIFY chime %s: %s", chime_entity_config.entity_id, chime_entity_config.tune)
320 action_data: dict[str, Any] | None = None
321 try:
322 action_call: ActionCall | None = self.analyze_target(chime_entity_config, data, envelope)
323 if action_call is not None:
324 if await self.call_action(
325 envelope,
326 qualified_action=f"{action_call.domain}.{action_call.service}",
327 action_data=action_call.action_data,
328 target_data=action_call.target_data,
329 ):
330 chimes += 1
331 else:
332 _LOGGER.debug("SUPERNOTIFY Chime skipping incomplete service for %s", chime_entity_config.entity_id)
333 except Exception as e:
334 _LOGGER.error("SUPERNOTIFY Failed to chime %s: %s [%s]", chime_entity_config.entity_id, action_data)
335 if debug_trace:
336 debug_trace.record_delivery_exception(envelope.delivery.name, "analyze_target", e)
337 return chimes > 0
339 def analyze_target(self, target_config: ChimeTargetConfig, data: dict[str, Any], envelope: Envelope) -> ActionCall | None:
341 if not target_config.entity_id and not target_config.device_id:
342 _LOGGER.warning("SUPERNOTIFY Empty chime target")
343 return None
345 domain: str | None = None
346 name: str | None = None
348 # Alexa Devices use device_id not entity_id for sounds
349 # TODO: use method or delivery config vs fixed local constant for domains
350 if target_config.device_id is not None and DEVICE_DOMAINS:
351 if target_config.domain is not None and target_config.domain in DEVICE_DOMAINS:
352 _LOGGER.debug(f"SUPERNOTIFY Chime selected target {domain} for {target_config.domain}")
353 domain = target_config.domain
354 else:
355 domain = self.hass_api.domain_for_device(target_config.device_id, DEVICE_DOMAINS)
356 _LOGGER.debug(f"SUPERNOTIFY Chime selected device {domain} for {target_config.device_id}")
358 elif target_config.entity_id and "." in target_config.entity_id:
359 domain, name = target_config.entity_id.split(".", 1)
360 if not domain:
361 _LOGGER.warning("SUPERNOTIFY Unknown domain: %s", target_config)
362 return None
363 mini_transport: MiniChimeTransport | None = self.mini_transports.get(domain)
364 if mini_transport is None:
365 _LOGGER.warning(
366 "SUPERNOTIFY No matching chime domain/tune: %s, target: %s, tune: %s",
367 domain,
368 target_config.entity_id,
369 target_config.tune,
370 )
371 return None
373 action_call: ActionCall | None = mini_transport.build(
374 envelope=envelope, entity_name=name, action_data=data, target_config=target_config
375 )
376 _LOGGER.debug("SUPERNOTIFY analyze_chime->%s", action_call)
378 return action_call
380 def resolve_tune(
381 self, tune_or_alias: str | None, chime_config: dict[str, Any], target: Target | None = None
382 ) -> dict[str, ChimeTargetConfig]:
383 target_configs: dict[str, ChimeTargetConfig] = {}
384 if tune_or_alias is not None:
385 for alias_config in chime_config.get(tune_or_alias, {}).values():
386 alias_target: Target | None = alias_config.get(CONF_TARGET, None)
387 alias_kwargs: dict[str, Any] = {k: v for k, v in alias_config.items() if k != CONF_TARGET}
388 # pass through variables or data if present
389 if alias_target is not None:
390 target_configs.update({t: ChimeTargetConfig(entity_id=t, **alias_kwargs) for t in alias_target.entity_ids})
391 target_configs.update({t: ChimeTargetConfig(device_id=t, **alias_kwargs) for t in alias_target.device_ids})
392 elif alias_config[CONF_DOMAIN] in DEVICE_DOMAINS and target is not None:
393 # bulk apply to all known target devices of this domain
394 bulk_apply = {
395 dev: ChimeTargetConfig(device_id=dev, **alias_kwargs)
396 for dev in target.device_ids
397 if dev not in target_configs # don't overwrite existing specific targets
398 and ATTR_DEVICE_ID not in alias_config
399 }
400 # TODO: Constrain to device domain
401 target_configs.update(bulk_apply)
402 elif target is not None:
403 # bulk apply to all known target entities of this domain
404 bulk_apply = {
405 ent: ChimeTargetConfig(entity_id=ent, **alias_kwargs)
406 for ent in target.entity_ids
407 if ent.startswith(f"{alias_config[CONF_DOMAIN]}.")
408 and ent not in target_configs # don't overwrite existing specific targets
409 and ATTR_ENTITY_ID not in alias_config
410 }
411 target_configs.update(bulk_apply)
412 _LOGGER.debug("SUPERNOTIFY transport_chime: Resolved tune %s to %s", tune_or_alias, target_configs)
413 return target_configs
416def build_aliases(src_config: ConfigType) -> ConfigType:
417 dest_config: dict[str, Any] = {}
418 try:
419 validated: ConfigType = CHIME_ALIASES_SCHEMA({OPTION_CHIME_ALIASES: src_config})
420 for alias, alias_config in validated[OPTION_CHIME_ALIASES].items():
421 alias_config = alias_config or {}
422 for domain_or_label, domain_config in alias_config.items():
423 domain_config = domain_config or {}
424 if isinstance(domain_config, str):
425 domain_config = {CONF_TUNE: domain_config}
426 domain_config.setdefault(CONF_TUNE, alias)
427 if domain_or_label in OPTIONS_CHIME_DOMAINS:
428 domain_config.setdefault(CONF_DOMAIN, domain_or_label)
430 try:
431 if domain_config.get(CONF_TARGET):
432 domain_config[CONF_TARGET] = Target(domain_config[CONF_TARGET])
433 if not domain_config[CONF_TARGET].has_targets():
434 _LOGGER.warning("SUPERNOTIFY chime alias %s has empty target", alias)
435 elif domain_config[CONF_TARGET].has_unknown_targets():
436 _LOGGER.warning("SUPERNOTIFY chime alias %s has unknown targets", alias)
437 dest_config.setdefault(alias, {})
438 dest_config[alias][domain_or_label] = domain_config
439 except Exception as e:
440 _LOGGER.exception("SUPERNOTIFY chime alias %s has invalid target: %s", alias, e)
442 except vol.Invalid as ve:
443 _LOGGER.error("SUPERNOTIFY Chime alias configuration error: %s", ve)
444 _LOGGER.error("SUPERNOTIFY %s", humanize_error(src_config, ve))
445 except Exception as e:
446 _LOGGER.exception("SUPERNOTIFY Chime alias unexpected error: %s", e)
447 return dest_config