Coverage for custom_components/supernotify/transports/chime.py: 97%
264 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import logging
4from abc import abstractmethod
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING, Any, ClassVar, cast
8from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
9from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues
10 ATTR_DEVICE_ID,
11 ATTR_ENTITY_ID,
12 CONF_ALIAS,
13 CONF_DOMAIN,
14 CONF_TARGET,
15)
16from homeassistant.exceptions import NoEntitySpecifiedError
17from homeassistant.helpers import config_validation as cv
19from custom_components.supernotify.compat import humanize, vol
20from custom_components.supernotify.const import (
21 ATTR_MEDIA,
22 ATTR_PRIORITY,
23 CONF_DATA,
24 CONF_DURATION,
25 CONF_INCLUSION,
26 CONF_TUNE,
27 CONF_VOLUME,
28 INCLUSION_EXPLICIT,
29 OPTIONS_CHIME_DOMAINS,
30 RE_DEVICE_ID,
31 TRANSPORT_CHIME,
32)
33from custom_components.supernotify.model import (
34 DebugTrace,
35 SelectionRule,
36 TargetRequired,
37 TransportConfig,
38 TransportFeature,
39)
40from custom_components.supernotify.options import (
41 OPTION_DEVICE_DISCOVERY,
42 OPTION_DEVICE_DOMAIN,
43 OPTION_DEVICE_MODEL_SELECT,
44 OPTION_TARGET_SELECT,
45 SELECT_EXCLUDE,
46 DeliveryOption,
47)
48from custom_components.supernotify.schema import DATA_SCHEMA, TARGET_SCHEMA
49from custom_components.supernotify.target import Target, TargetEntityCategory
50from custom_components.supernotify.transport import Transport
52if TYPE_CHECKING:
53 from homeassistant.helpers.typing import ConfigType
55 from custom_components.supernotify.envelope import Envelope
56 from custom_components.supernotify.hass_api import HomeAssistantAPI
58# kept in sync with RE_VALID_CHIME below and this transport's target_categories property
59CHIME_ENTITY_DOMAINS = ["switch", "script", "group", "rest_command", "siren", "media_player"]
60RE_VALID_CHIME = r"(switch|script|group|rest_command|siren|media_player)\.[A-Za-z0-9_]+"
62# extra standard delivery grouping every siren - see build_standard_deliveries() below
63STANDARD_DELIVERY_SIREN_ALL = f"{TRANSPORT_CHIME}_siren_all"
65_LOGGER = logging.getLogger(__name__)
67# device-registry scan (see HomeAssistantAPI.discover_devices), independent of whether
68# AlexaDevicesTransport itself is currently viable/loaded - HA doesn't reliably prune
69# device registry entries when a config entry is unloaded, so chime can keep finding and
70# targeting alexa_devices devices after that transport has gone away
71DEVICE_DOMAINS = ["alexa_devices"]
73OPTION_CHIME_ALIASES = "chime_aliases"
74CHIME_ALIASES_SCHEMA = vol.Schema({
75 vol.Required(OPTION_CHIME_ALIASES, default=dict): vol.Schema({
76 cv.string: vol.Schema({
77 cv.string: vol.Any(
78 vol.Any(None, cv.string),
79 vol.Schema({
80 vol.Optional(CONF_ALIAS): cv.string,
81 vol.Optional(CONF_DOMAIN): vol.In(OPTIONS_CHIME_DOMAINS),
82 vol.Optional(CONF_TUNE): cv.string,
83 vol.Optional(CONF_DATA): DATA_SCHEMA,
84 vol.Optional(CONF_VOLUME): float,
85 vol.Optional(CONF_TARGET): TARGET_SCHEMA,
86 vol.Optional(CONF_DURATION): cv.positive_int,
87 }),
88 )
89 })
90 })
91})
94@dataclass
95class ActionCall:
96 domain: str
97 service: str
98 action_data: dict[str, Any] | None = field(default_factory=dict)
99 target_data: dict[str, Any] | None = field(default_factory=dict)
102class ChimeTargetConfig:
103 def __init__(
104 self,
105 entity_id: str | None = None,
106 device_id: str | None = None,
107 tune: str | None = None,
108 duration: int | None = None,
109 volume: float | None = None,
110 data: dict[str, Any] | None = None,
111 domain: str | None = None,
112 **kwargs: Any,
113 ) -> None:
114 self.entity_id: str | None = entity_id
115 self.device_id: str | None = device_id
116 self.domain: str | None = None
117 self.entity_name: str | None = None
118 if self.entity_id and "." in self.entity_id:
119 self.domain, self.entity_name = self.entity_id.split(".", 1)
120 elif self.device_id:
121 self.domain = domain
122 else:
123 _LOGGER.warning(
124 "SUPERNOTIFY Invalid chime target, entity_id: %s, device_id %s, tune:%s", entity_id, device_id, tune
125 )
126 raise NoEntitySpecifiedError("ChimeTargetConfig target must be entity_id or device_id")
127 if kwargs:
128 _LOGGER.warning("SUPERNOTIFY ChimeTargetConfig ignoring unexpected args: %s", kwargs)
129 self.volume: float | None = volume
130 self.tune: str | None = tune
131 self.duration: int | None = duration
132 self.data: dict[str, Any] | None = data or {}
134 def as_dict(self, **kwargs: Any) -> dict[str, Any]:
135 return {
136 "entity_id": self.entity_id,
137 "device_id": self.device_id,
138 "domain": self.domain,
139 "tune": self.tune,
140 "duration": self.duration,
141 "volume": self.volume,
142 "data": self.data,
143 }
145 def __repr__(self) -> str:
146 """Return a developer-oriented string representation of this ChimeTargetConfig"""
147 if self.device_id is not None:
148 return f"ChimeTargetConfig(device_id={self.device_id})"
149 return f"ChimeTargetConfig(entity_id={self.entity_id})"
152class MiniChimeTransport:
153 domain: str
155 @abstractmethod
156 def build(
157 self,
158 target_config: ChimeTargetConfig,
159 action_data: dict[str, Any] | None = None,
160 entity_name: str | None = None,
161 envelope: Envelope | None = None,
162 **_kwargs: Any,
163 ) -> ActionCall | None:
164 raise NotImplementedError()
167class RestCommandChimeTransport(MiniChimeTransport):
168 domain = "rest_command"
170 def build(
171 self,
172 target_config: ChimeTargetConfig,
173 action_data: dict[str, Any] | None = None,
174 entity_name: str | None = None,
175 envelope: Envelope | None = None,
176 **_kwargs: Any,
177 ) -> ActionCall | None:
178 if entity_name is None:
179 _LOGGER.warning("SUPERNOTIFY rest_command chime target requires entity")
180 return None
181 output_data = dict(target_config.data) if target_config.data else {}
182 return ActionCall(self.domain, entity_name, action_data=output_data)
185class SwitchChimeTransport(MiniChimeTransport):
186 domain = "switch"
188 def build(
189 self,
190 target_config: ChimeTargetConfig,
191 action_data: dict[str, Any] | None = None,
192 entity_name: str | None = None,
193 envelope: Envelope | None = None,
194 **_kwargs: Any,
195 ) -> ActionCall | None:
196 return ActionCall(self.domain, "turn_on", target_data={ATTR_ENTITY_ID: target_config.entity_id})
199class SirenChimeTransport(MiniChimeTransport):
200 domain = "siren"
202 def build(
203 self,
204 target_config: ChimeTargetConfig,
205 action_data: dict[str, Any] | None = None,
206 entity_name: str | None = None,
207 envelope: Envelope | None = None,
208 **_kwargs: Any,
209 ) -> ActionCall | None:
210 output_data: dict[str, Any] = {}
211 if target_config.tune:
212 output_data["tone"] = target_config.tune
213 if target_config.duration is not None:
214 output_data["duration"] = target_config.duration
215 if target_config.volume is not None:
216 output_data["volume_level"] = target_config.volume
217 return ActionCall(
218 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
219 )
222class ScriptChimeTransport(MiniChimeTransport):
223 domain = "script"
225 def build(
226 self,
227 target_config: ChimeTargetConfig,
228 action_data: dict[str, Any] | None = None,
229 entity_name: str | None = None,
230 envelope: Envelope | None = None,
231 **_kwargs: Any,
232 ) -> ActionCall | None:
233 if entity_name is None:
234 _LOGGER.warning("SUPERNOTIFY Script chime target requires entity")
235 return None
236 if envelope is None:
237 _LOGGER.warning("SUPERNOTIFY Script chime target requires envelope")
238 return None
239 variables: dict[str, Any] = dict(target_config.data or {})
240 variables[ATTR_MESSAGE] = envelope.message
241 variables[ATTR_TITLE] = envelope.title
242 variables[ATTR_PRIORITY] = envelope.priority
243 variables["chime_tune"] = target_config.tune
244 variables["chime_volume"] = target_config.volume
245 variables["chime_duration"] = target_config.duration
246 output_data: dict[str, Any] = {"variables": variables}
247 # use `turn_on` rather than direct call to run script in background
248 return ActionCall(
249 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
250 )
253class AlexaDevicesChimeTransport(MiniChimeTransport):
254 domain = "alexa_devices"
256 def build(
257 self,
258 target_config: ChimeTargetConfig,
259 action_data: dict[str, Any] | None = None,
260 entity_name: str | None = None,
261 envelope: Envelope | None = None,
262 **_kwargs: Any,
263 ) -> ActionCall | None:
264 output_data: dict[str, Any] = {
265 "device_id": target_config.device_id,
266 "sound": target_config.tune,
267 }
268 return ActionCall(self.domain, "send_sound", action_data=output_data)
271class MediaPlayerChimeTransport(MiniChimeTransport):
272 domain = "media_player"
274 def build(
275 self,
276 target_config: ChimeTargetConfig,
277 action_data: dict[str, Any] | None = None,
278 entity_name: str | None = None,
279 envelope: Envelope | None = None,
280 **_kwargs: Any,
281 ) -> ActionCall | None:
282 input_data = dict(target_config.data or {})
283 if action_data:
284 input_data.update(action_data)
285 output_data: dict[str, Any] = {
286 "media": {
287 "media_content_type": input_data.get(ATTR_MEDIA, {"media_content_type": "sound"}).get(
288 "media_content_type", "sound"
289 ),
290 "media_content_id": target_config.tune,
291 }
292 }
293 if input_data.get("enqueue") is not None:
294 output_data["enqueue"] = input_data.get("enqueue")
295 if input_data.get("announce") is not None:
296 output_data["announce"] = input_data.get("announce")
298 return ActionCall(
299 self.domain, "play_media", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
300 )
303class ChimeTransport(Transport):
304 name = TRANSPORT_CHIME
305 declared_options: ClassVar[list[DeliveryOption]] = [
306 DeliveryOption(OPTION_CHIME_ALIASES, "Custom chime device aliases and their per-domain tuning"),
307 ]
309 def __init__(self, *args: Any, **kwargs: Any) -> None:
310 super().__init__(*args, **kwargs)
311 self.mini_transports: dict[str, MiniChimeTransport] = {
312 t.domain: t
313 for t in [
314 RestCommandChimeTransport(),
315 SwitchChimeTransport(),
316 SirenChimeTransport(),
317 ScriptChimeTransport(),
318 AlexaDevicesChimeTransport(),
319 MediaPlayerChimeTransport(),
320 ]
321 }
323 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]:
324 if OPTION_CHIME_ALIASES in options:
325 chime_aliases: ConfigType = build_aliases(options[OPTION_CHIME_ALIASES])
326 if chime_aliases:
327 _LOGGER.info("SUPERNOTIFY Set up %s chime aliases for %s", len(chime_aliases), delivery_name)
328 else:
329 _LOGGER.warning("SUPERNOTIFY Chime aliases for %s configured but not recognized", delivery_name)
330 else:
331 chime_aliases = {}
332 _LOGGER.debug("SUPERNOTIFY No chime aliases configured for %s", delivery_name)
333 return {"chime_aliases": chime_aliases}
335 @property
336 def supported_features(self) -> TransportFeature:
337 return TransportFeature.SOUND
339 def extra_attributes(self) -> dict[str, Any]:
340 return {"mini_transports": [t.domain for t in self.mini_transports.values()]}
342 @property
343 def default_config(self) -> TransportConfig:
344 config = super().default_config
345 # usually has its own fixed default target configured (chime_aliases), so doesn't
346 # need one resolved from the call - one of the few transports where this needs to
347 # stay an explicit override rather than the inferred default
348 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
349 config.delivery_defaults.options = {
350 OPTION_TARGET_SELECT: [RE_VALID_CHIME, RE_DEVICE_ID],
351 OPTION_DEVICE_DISCOVERY: True,
352 OPTION_DEVICE_DOMAIN: DEVICE_DOMAINS,
353 OPTION_DEVICE_MODEL_SELECT: {SELECT_EXCLUDE: ["Speaker Group"]},
354 }
355 return config
357 @property
358 def other_target_categories(self) -> list[str | TargetEntityCategory]:
359 # borrows several other domains' shapes directly rather than a category of its own -
360 # see the roadmap note on making chime aliases into targets in their own right
361 return [TargetEntityCategory(domain=CHIME_ENTITY_DOMAINS), ATTR_DEVICE_ID]
363 def validate_action(self, action: str | None) -> bool:
364 return action is None
366 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
367 # an explicit delivery can supply its own chime_aliases regardless of whether the
368 # transport-level default is configured - is_viable() can't see delivery-level
369 # config, so it can't rule that out; DeliveryRegistry prunes this transport
370 # entirely once it's confirmed no delivery (explicit or auto) actually uses it
371 return True
373 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
374 """Its own default (only if chime_aliases is configured - with none, there's
375 nothing to map a tune/priority to a target), plus "..._siren_all" - all siren.*
376 entities, regardless of chime_aliases: unlike the other chime domains,
377 SirenChimeTransport.build() doesn't need a tune/alias mapping to call
378 siren.turn_on, so this extra doesn't share the alias requirement."""
379 deliveries: dict[str, ConfigType] = {}
380 if OPTION_CHIME_ALIASES in self.delivery_defaults.options:
381 deliveries[self.name] = {}
382 siren_entity_ids = hass_api.entity_ids_for_domain("siren")
383 if siren_entity_ids:
384 deliveries[STANDARD_DELIVERY_SIREN_ALL] = {
385 CONF_TARGET: {ATTR_ENTITY_ID: siren_entity_ids},
386 CONF_INCLUSION: [INCLUSION_EXPLICIT],
387 }
389 return deliveries
391 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
392 data: dict[str, Any] = {}
393 data.update(envelope.delivery.data)
394 data.update(envelope.data or {})
395 target: Target = envelope.target
397 # chime_repeat = data.pop("chime_repeat", 1)
398 chime_tune: str | None = data.pop("chime_tune", None)
399 chime_volume: float | None = data.pop("chime_volume", None)
400 chime_duration: int | None = data.pop("chime_duration", None)
402 _LOGGER.debug(
403 "SUPERNOTIFY notify_chime: %s -> %s (delivery: %s, env_data:%s, dlv_data:%s)",
404 chime_tune,
405 target.entity_ids,
406 envelope.delivery_name,
407 envelope.data,
408 envelope.delivery.data,
409 )
410 # expand groups
411 expanded_targets = {
412 e: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, entity_id=e)
413 for e in self.hass_api.expand_group(target.entity_ids)
414 }
415 model_filter = SelectionRule(envelope.delivery.options.get(OPTION_DEVICE_MODEL_SELECT))
417 expanded_targets.update({
418 d: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, device_id=d)
419 for d in target.device_ids
420 if not (dev_entry := self.hass_api.find_device(d))
421 or model_filter.match(dev_entry.model if isinstance(dev_entry.model, str) else None)
422 })
423 # resolve and include chime aliases
424 expanded_targets.update(
425 self.resolve_tune(chime_tune, envelope.delivery.transport_data.get("chime_aliases", {}), target)
426 ) # overwrite and extend
428 chimes = 0
429 if not expanded_targets:
430 _LOGGER.info("SUPERNOTIFY Skipping chime, no targets")
431 return False
432 if debug_trace:
433 debug_trace.record_delivery_artefact(envelope.delivery.name, "expanded_targets", expanded_targets)
435 for chime_entity_config in expanded_targets.values():
436 _LOGGER.debug("SUPERNOTIFY Chime %s: %s", chime_entity_config.entity_id, chime_entity_config.tune)
437 try:
438 action_call: ActionCall | None = self.analyze_target(chime_entity_config, data, envelope)
439 if action_call is not None:
440 if await self.call_action(
441 envelope,
442 qualified_action=f"{action_call.domain}.{action_call.service}",
443 action_data=action_call.action_data,
444 target_data=action_call.target_data,
445 ):
446 chimes += 1
447 else:
448 _LOGGER.debug("SUPERNOTIFY Chime skipping incomplete service for %s", chime_entity_config.entity_id)
449 except Exception as e:
450 _LOGGER.error(
451 "SUPERNOTIFY Failed to chime %s: %s",
452 chime_entity_config.entity_id,
453 e,
454 )
455 if debug_trace:
456 debug_trace.record_delivery_exception(envelope.delivery.name, "chime_target", e)
457 return chimes > 0
459 def analyze_target(self, target_config: ChimeTargetConfig, data: dict[str, Any], envelope: Envelope) -> ActionCall | None:
461 if not target_config.entity_id and not target_config.device_id:
462 _LOGGER.warning("SUPERNOTIFY Empty chime target")
463 return None
465 domain: str | None = None
466 name: str | None = None
468 # Alexa Devices use device_id not entity_id for sounds
469 # TODO: use method or delivery config vs fixed local constant for domains
470 if target_config.device_id is not None and DEVICE_DOMAINS:
471 if target_config.domain is not None and target_config.domain in DEVICE_DOMAINS:
472 _LOGGER.debug(f"SUPERNOTIFY Chime selected target {target_config.domain} for {target_config.device_id}")
473 domain = target_config.domain
474 else:
475 domain = self.hass_api.domain_for_device(target_config.device_id, DEVICE_DOMAINS)
476 _LOGGER.debug(f"SUPERNOTIFY Chime selected device {domain} for {target_config.device_id}")
478 elif target_config.entity_id and "." in target_config.entity_id:
479 domain, name = target_config.entity_id.split(".", 1)
480 if not domain:
481 _LOGGER.warning("SUPERNOTIFY Unknown domain: %s", target_config)
482 return None
483 mini_transport: MiniChimeTransport | None = self.mini_transports.get(domain)
484 if mini_transport is None:
485 _LOGGER.warning(
486 "SUPERNOTIFY No matching chime domain/tune: %s, target: %s, tune: %s",
487 domain,
488 target_config.entity_id,
489 target_config.tune,
490 )
491 return None
493 action_call: ActionCall | None = mini_transport.build(
494 envelope=envelope, entity_name=name, action_data=data, target_config=target_config
495 )
496 _LOGGER.debug("SUPERNOTIFY analyze_chime->%s", action_call)
498 return action_call
500 def resolve_tune(
501 self, tune_or_alias: str | None, chime_config: dict[str, Any], target: Target | None = None
502 ) -> dict[str, ChimeTargetConfig]:
503 target_configs: dict[str, ChimeTargetConfig] = {}
504 if tune_or_alias is not None:
505 for alias_config in chime_config.get(tune_or_alias, {}).values():
506 alias_target: Target | None = alias_config.get(CONF_TARGET, None)
507 alias_kwargs: dict[str, Any] = {k: v for k, v in alias_config.items() if k != CONF_TARGET}
508 # pass through variables or data if present
509 if alias_target is not None:
510 target_configs.update({
511 t: ChimeTargetConfig(entity_id=t, **alias_kwargs)
512 for t in self.hass_api.expand_group(alias_target.entity_ids)
513 })
514 target_configs.update({t: ChimeTargetConfig(device_id=t, **alias_kwargs) for t in alias_target.device_ids})
515 elif alias_config[CONF_DOMAIN] in DEVICE_DOMAINS and target is not None:
516 # bulk apply to all known target devices of this domain
517 bulk_apply = {
518 dev: ChimeTargetConfig(device_id=dev, **alias_kwargs)
519 for dev in target.device_ids
520 if dev not in target_configs # don't overwrite existing specific targets
521 }
522 # TODO: Constrain to device domain
523 target_configs.update(bulk_apply)
524 elif target is not None:
525 # bulk apply to all known target entities of this domain
526 bulk_apply = {
527 ent: ChimeTargetConfig(entity_id=ent, **alias_kwargs)
528 for ent in target.entity_ids
529 if ent.startswith(f"{alias_config[CONF_DOMAIN]}.")
530 and ent not in target_configs # don't overwrite existing specific targets
531 }
532 target_configs.update(bulk_apply)
533 _LOGGER.debug("SUPERNOTIFY transport_chime: Resolved tune %s to %s", tune_or_alias, target_configs)
534 return target_configs
537def build_aliases(src_config: ConfigType) -> ConfigType:
538 dest_config: dict[str, Any] = {}
539 try:
540 validated: ConfigType = CHIME_ALIASES_SCHEMA({OPTION_CHIME_ALIASES: src_config})
541 for alias, alias_config in validated[OPTION_CHIME_ALIASES].items():
542 alias_config = alias_config or {}
543 for domain_or_label, domain_config in alias_config.items():
544 domain_config = domain_config or {}
545 if isinstance(domain_config, str):
546 domain_config = {CONF_TUNE: domain_config}
547 domain_config = cast("dict[str, Any]", domain_config)
548 domain_config.setdefault(CONF_TUNE, alias)
549 if domain_or_label in OPTIONS_CHIME_DOMAINS:
550 domain_config.setdefault(CONF_DOMAIN, domain_or_label)
551 if CONF_DOMAIN not in domain_config and CONF_TARGET not in domain_config:
552 _LOGGER.warning("SUPERNOTIFY Chime alias %s ignoring %s, needs a domain or target", alias, domain_or_label)
553 continue
555 try:
556 if CONF_TARGET in domain_config:
557 domain_config[CONF_TARGET] = Target(domain_config[CONF_TARGET])
558 if not domain_config[CONF_TARGET].has_targets():
559 _LOGGER.warning("SUPERNOTIFY Chime alias %s has empty target", alias)
560 elif domain_config[CONF_TARGET].has_unknown_targets():
561 _LOGGER.warning("SUPERNOTIFY Chime alias %s has unknown targets", alias)
562 dest_config.setdefault(alias, {})
563 dest_config[alias][domain_or_label] = domain_config
564 except Exception:
565 _LOGGER.exception("SUPERNOTIFY Chime alias %s has invalid target", alias)
567 except vol.Invalid as ve:
568 _LOGGER.error("SUPERNOTIFY Chime alias configuration error: %s", ve)
569 _LOGGER.error("SUPERNOTIFY %s", humanize.humanize_error(src_config, ve))
570 except Exception:
571 _LOGGER.exception("SUPERNOTIFY Chime alias unexpected error")
572 return dest_config