Coverage for custom_components/supernotify/transports/chime.py: 97%
265 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import logging
4from abc import abstractmethod
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING, Any, ClassVar, cast
8import voluptuous as vol
9from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
10from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues
11 ATTR_DEVICE_ID,
12 ATTR_ENTITY_ID,
13 CONF_ALIAS,
14 CONF_DOMAIN,
15 CONF_TARGET,
16)
17from homeassistant.exceptions import NoEntitySpecifiedError
18from homeassistant.helpers import config_validation as cv
19from voluptuous.humanize import humanize_error
21from custom_components.supernotify.const import (
22 ATTR_DATA,
23 ATTR_MEDIA,
24 ATTR_PRIORITY,
25 CONF_DATA,
26 CONF_DURATION,
27 CONF_INCLUSION,
28 CONF_TUNE,
29 CONF_VOLUME,
30 INCLUSION_EXPLICIT,
31 OPTIONS_CHIME_DOMAINS,
32 RE_DEVICE_ID,
33 TRANSPORT_CHIME,
34)
35from custom_components.supernotify.model import (
36 DebugTrace,
37 SelectionRule,
38 TargetRequired,
39 TransportConfig,
40 TransportFeature,
41)
42from custom_components.supernotify.options import (
43 OPTION_DEVICE_DISCOVERY,
44 OPTION_DEVICE_DOMAIN,
45 OPTION_DEVICE_MODEL_SELECT,
46 OPTION_TARGET_SELECT,
47 SELECT_EXCLUDE,
48 DeliveryOption,
49)
50from custom_components.supernotify.schema import DATA_SCHEMA, TARGET_SCHEMA
51from custom_components.supernotify.target import Target, TargetEntityCategory
52from custom_components.supernotify.transport import Transport
54if TYPE_CHECKING:
55 from homeassistant.helpers.typing import ConfigType
57 from custom_components.supernotify.envelope import Envelope
58 from custom_components.supernotify.hass_api import HomeAssistantAPI
60# kept in sync with RE_VALID_CHIME below and this transport's target_categories property
61CHIME_ENTITY_DOMAINS = ["switch", "script", "group", "rest_command", "siren", "media_player"]
62RE_VALID_CHIME = r"(switch|script|group|rest_command|siren|media_player)\.[A-Za-z0-9_]+"
64# extra standard delivery grouping every siren - see build_standard_deliveries() below
65STANDARD_DELIVERY_SIREN_ALL = f"{TRANSPORT_CHIME}_siren_all"
67_LOGGER = logging.getLogger(__name__)
69# device-registry scan (see HomeAssistantAPI.discover_devices), independent of whether
70# AlexaDevicesTransport itself is currently viable/loaded - HA doesn't reliably prune
71# device registry entries when a config entry is unloaded, so chime can keep finding and
72# targeting alexa_devices devices after that transport has gone away
73DEVICE_DOMAINS = ["alexa_devices"]
75OPTION_CHIME_ALIASES = "chime_aliases"
76CHIME_ALIASES_SCHEMA = vol.Schema({
77 vol.Required(OPTION_CHIME_ALIASES, default=dict): vol.Schema({
78 cv.string: vol.Schema({
79 cv.string: vol.Any(
80 vol.Any(None, cv.string, vol.In(OPTIONS_CHIME_DOMAINS)),
81 vol.Schema({
82 vol.Optional(CONF_ALIAS): cv.string,
83 vol.Optional(CONF_DOMAIN): cv.string,
84 vol.Optional(CONF_TUNE): cv.string,
85 vol.Optional(CONF_DATA): DATA_SCHEMA,
86 vol.Optional(CONF_VOLUME): float,
87 vol.Optional(CONF_TARGET): TARGET_SCHEMA,
88 vol.Optional(CONF_DURATION): cv.positive_int,
89 }),
90 )
91 })
92 })
93})
96@dataclass
97class ActionCall:
98 domain: str
99 service: str
100 action_data: dict[str, Any] | None = field(default_factory=dict)
101 target_data: dict[str, Any] | None = field(default_factory=dict)
104class ChimeTargetConfig:
105 def __init__(
106 self,
107 entity_id: str | None = None,
108 device_id: str | None = None,
109 tune: str | None = None,
110 duration: int | None = None,
111 volume: float | None = None,
112 data: dict[str, Any] | None = None,
113 domain: str | None = None,
114 **kwargs: Any,
115 ) -> None:
116 self.entity_id: str | None = entity_id
117 self.device_id: str | None = device_id
118 self.domain: str | None = None
119 self.entity_name: str | None = None
120 if self.entity_id and "." in self.entity_id:
121 self.domain, self.entity_name = self.entity_id.split(".", 1)
122 elif self.device_id:
123 self.domain = domain
124 else:
125 _LOGGER.warning(
126 "SUPERNOTIFY Invalid chime target, entity_id: %s, device_id %s, tune:%s", entity_id, device_id, tune
127 )
128 raise NoEntitySpecifiedError("ChimeTargetConfig target must be entity_id or device_id")
129 if kwargs:
130 _LOGGER.warning("SUPERNOTIFY ChimeTargetConfig ignoring unexpected args: %s", kwargs)
131 self.volume: float | None = volume
132 self.tune: str | None = tune
133 self.duration: int | None = duration
134 self.data: dict[str, Any] | None = data or {}
136 def as_dict(self, **kwargs: Any) -> dict[str, Any]:
137 return {
138 "entity_id": self.entity_id,
139 "device_id": self.device_id,
140 "domain": self.domain,
141 "tune": self.tune,
142 "duration": self.duration,
143 "volume": self.volume,
144 "data": self.data,
145 }
147 def __repr__(self) -> str:
148 """Return a developer-oriented string representation of this ChimeTargetConfig"""
149 if self.device_id is not None:
150 return f"ChimeTargetConfig(device_id={self.device_id})"
151 return f"ChimeTargetConfig(entity_id={self.entity_id})"
154class MiniChimeTransport:
155 domain: str
157 @abstractmethod
158 def build(
159 self,
160 target_config: ChimeTargetConfig,
161 action_data: dict[str, Any] | None = None,
162 entity_name: str | None = None,
163 envelope: Envelope | None = None,
164 **_kwargs: Any,
165 ) -> ActionCall | None:
166 raise NotImplementedError()
169class RestCommandChimeTransport(MiniChimeTransport):
170 domain = "rest_command"
172 def build(
173 self,
174 target_config: ChimeTargetConfig,
175 action_data: dict[str, Any] | None = None,
176 entity_name: str | None = None,
177 envelope: Envelope | None = None,
178 **_kwargs: Any,
179 ) -> ActionCall | None:
180 if entity_name is None:
181 _LOGGER.warning("SUPERNOTIFY rest_command chime target requires entity")
182 return None
183 output_data = dict(target_config.data) if target_config.data else {}
184 return ActionCall(self.domain, entity_name, action_data=output_data)
187class SwitchChimeTransport(MiniChimeTransport):
188 domain = "switch"
190 def build(
191 self,
192 target_config: ChimeTargetConfig,
193 action_data: dict[str, Any] | None = None,
194 entity_name: str | None = None,
195 envelope: Envelope | None = None,
196 **_kwargs: Any,
197 ) -> ActionCall | None:
198 return ActionCall(self.domain, "turn_on", target_data={ATTR_ENTITY_ID: target_config.entity_id})
201class SirenChimeTransport(MiniChimeTransport):
202 domain = "siren"
204 def build(
205 self,
206 target_config: ChimeTargetConfig,
207 action_data: dict[str, Any] | None = None,
208 entity_name: str | None = None,
209 envelope: Envelope | None = None,
210 **_kwargs: Any,
211 ) -> ActionCall | None:
212 output_data: dict[str, Any] = {ATTR_DATA: {}}
213 if target_config.tune:
214 output_data[ATTR_DATA]["tone"] = target_config.tune
215 if target_config.duration is not None:
216 output_data[ATTR_DATA]["duration"] = target_config.duration
217 if target_config.volume is not None:
218 output_data[ATTR_DATA]["volume_level"] = target_config.volume
219 return ActionCall(
220 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
221 )
224class ScriptChimeTransport(MiniChimeTransport):
225 domain = "script"
227 def build(
228 self,
229 target_config: ChimeTargetConfig,
230 action_data: dict[str, Any] | None = None,
231 entity_name: str | None = None,
232 envelope: Envelope | None = None,
233 **_kwargs: Any,
234 ) -> ActionCall | None:
235 if entity_name is None:
236 _LOGGER.warning("SUPERNOTIFY Script chime target requires entity")
237 return None
238 if envelope is None:
239 _LOGGER.warning("SUPERNOTIFY Script chime target requires envelope")
240 return None
241 variables: dict[str, Any] = target_config.data or {}
242 variables[ATTR_MESSAGE] = envelope.message
243 variables[ATTR_TITLE] = envelope.title
244 variables[ATTR_PRIORITY] = envelope.priority
245 variables["chime_tune"] = target_config.tune
246 variables["chime_volume"] = target_config.volume
247 variables["chime_duration"] = target_config.duration
248 output_data: dict[str, Any] = {"variables": variables}
249 if envelope.delivery.debug:
250 output_data["wait"] = envelope.delivery.debug
251 # use `turn_on` rather than direct call to run script in background
252 return ActionCall(
253 self.domain, "turn_on", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
254 )
257class AlexaDevicesChimeTransport(MiniChimeTransport):
258 domain = "alexa_devices"
260 def build(
261 self,
262 target_config: ChimeTargetConfig,
263 action_data: dict[str, Any] | None = None,
264 entity_name: str | None = None,
265 envelope: Envelope | None = None,
266 **_kwargs: Any,
267 ) -> ActionCall | None:
268 output_data: dict[str, Any] = {
269 "device_id": target_config.device_id,
270 "sound": target_config.tune,
271 }
272 return ActionCall(self.domain, "send_sound", action_data=output_data)
275class MediaPlayerChimeTransport(MiniChimeTransport):
276 domain = "media_player"
278 def build(
279 self,
280 target_config: ChimeTargetConfig,
281 action_data: dict[str, Any] | None = None,
282 entity_name: str | None = None,
283 envelope: Envelope | None = None,
284 **_kwargs: Any,
285 ) -> ActionCall | None:
286 input_data = target_config.data or {}
287 if action_data:
288 input_data.update(action_data)
289 output_data: dict[str, Any] = {
290 "media": {
291 "media_content_type": input_data.get(ATTR_MEDIA, {"media_content_type": "sound"}).get(
292 "media_content_type", "sound"
293 ),
294 "media_content_id": target_config.tune,
295 }
296 }
297 if input_data.get("enqueue") is not None:
298 output_data["enqueue"] = input_data.get("enqueue")
299 if input_data.get("announce") is not None:
300 output_data["announce"] = input_data.get("announce")
302 return ActionCall(
303 self.domain, "play_media", action_data=output_data, target_data={ATTR_ENTITY_ID: target_config.entity_id}
304 )
307class ChimeTransport(Transport):
308 name = TRANSPORT_CHIME
309 declared_options: ClassVar[list[DeliveryOption]] = [
310 DeliveryOption(OPTION_CHIME_ALIASES, "Custom chime device aliases and their per-domain tuning"),
311 ]
313 def __init__(self, *args: Any, **kwargs: Any) -> None:
314 super().__init__(*args, **kwargs)
315 self.mini_transports: dict[str, MiniChimeTransport] = {
316 t.domain: t
317 for t in [
318 RestCommandChimeTransport(),
319 SwitchChimeTransport(),
320 SirenChimeTransport(),
321 ScriptChimeTransport(),
322 AlexaDevicesChimeTransport(),
323 MediaPlayerChimeTransport(),
324 ]
325 }
327 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]:
328 if OPTION_CHIME_ALIASES in options:
329 chime_aliases: ConfigType = build_aliases(options[OPTION_CHIME_ALIASES])
330 if chime_aliases:
331 _LOGGER.info("SUPERNOTIFY Set up %s chime aliases for %s", len(chime_aliases), delivery_name)
332 else:
333 _LOGGER.warning("SUPERNOTIFY Chime aliases for %s configured but not recognized", delivery_name)
334 else:
335 chime_aliases = {}
336 _LOGGER.debug("SUPERNOTIFY No chime aliases configured for %s", delivery_name)
337 return {"chime_aliases": chime_aliases}
339 @property
340 def supported_features(self) -> TransportFeature:
341 return TransportFeature.SOUND
343 def extra_attributes(self) -> dict[str, Any]:
344 return {"mini_transports": [t.domain for t in self.mini_transports.values()]}
346 @property
347 def default_config(self) -> TransportConfig:
348 config = TransportConfig()
349 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
350 config.delivery_defaults.inclusion = self.inclusion_mode
351 config.delivery_defaults.options = {
352 OPTION_TARGET_SELECT: [RE_VALID_CHIME, RE_DEVICE_ID],
353 OPTION_DEVICE_DISCOVERY: True,
354 OPTION_DEVICE_DOMAIN: DEVICE_DOMAINS,
355 OPTION_DEVICE_MODEL_SELECT: {SELECT_EXCLUDE: ["Speaker Group"]},
356 }
357 return config
359 @property
360 def target_categories(self) -> list[str | TargetEntityCategory]:
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 {domain} for {target_config.domain}")
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({t: ChimeTargetConfig(entity_id=t, **alias_kwargs) for t in alias_target.entity_ids})
511 target_configs.update({t: ChimeTargetConfig(device_id=t, **alias_kwargs) for t in alias_target.device_ids})
512 elif alias_config[CONF_DOMAIN] in DEVICE_DOMAINS and target is not None:
513 # bulk apply to all known target devices of this domain
514 bulk_apply = {
515 dev: ChimeTargetConfig(device_id=dev, **alias_kwargs)
516 for dev in target.device_ids
517 if dev not in target_configs # don't overwrite existing specific targets
518 and ATTR_DEVICE_ID not in alias_config
519 }
520 # TODO: Constrain to device domain
521 target_configs.update(bulk_apply)
522 elif target is not None:
523 # bulk apply to all known target entities of this domain
524 bulk_apply = {
525 ent: ChimeTargetConfig(entity_id=ent, **alias_kwargs)
526 for ent in target.entity_ids
527 if ent.startswith(f"{alias_config[CONF_DOMAIN]}.")
528 and ent not in target_configs # don't overwrite existing specific targets
529 and ATTR_ENTITY_ID not in alias_config
530 }
531 target_configs.update(bulk_apply)
532 _LOGGER.debug("SUPERNOTIFY transport_chime: Resolved tune %s to %s", tune_or_alias, target_configs)
533 return target_configs
536def build_aliases(src_config: ConfigType) -> ConfigType:
537 dest_config: dict[str, Any] = {}
538 try:
539 validated: ConfigType = CHIME_ALIASES_SCHEMA({OPTION_CHIME_ALIASES: src_config})
540 for alias, alias_config in validated[OPTION_CHIME_ALIASES].items():
541 alias_config = alias_config or {}
542 for domain_or_label, domain_config in alias_config.items():
543 domain_config = domain_config or {}
544 if isinstance(domain_config, str):
545 domain_config = {CONF_TUNE: domain_config}
546 domain_config = cast("dict[str, Any]", domain_config)
547 domain_config.setdefault(CONF_TUNE, alias)
548 if domain_or_label in OPTIONS_CHIME_DOMAINS:
549 domain_config.setdefault(CONF_DOMAIN, domain_or_label)
551 try:
552 if domain_config.get(CONF_TARGET):
553 domain_config[CONF_TARGET] = Target(domain_config[CONF_TARGET])
554 if not domain_config[CONF_TARGET].has_targets():
555 _LOGGER.warning("SUPERNOTIFY Chime alias %s has empty target", alias)
556 elif domain_config[CONF_TARGET].has_unknown_targets():
557 _LOGGER.warning("SUPERNOTIFY Chime alias %s has unknown targets", alias)
558 dest_config.setdefault(alias, {})
559 dest_config[alias][domain_or_label] = domain_config
560 except Exception:
561 _LOGGER.exception("SUPERNOTIFY Chime alias %s has invalid target", alias)
563 except vol.Invalid as ve:
564 _LOGGER.error("SUPERNOTIFY Chime alias configuration error: %s", ve)
565 _LOGGER.error("SUPERNOTIFY %s", humanize_error(src_config, ve))
566 except Exception:
567 _LOGGER.exception("SUPERNOTIFY Chime alias unexpected error")
568 return dest_config