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