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