Coverage for custom_components/supernotify/transports/chime.py: 93%
166 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-21 23:31 +0000
1import logging
2from typing import Any
4from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
5from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues
6 ATTR_DEVICE_ID,
7 ATTR_ENTITY_ID,
8 CONF_VARIABLES,
9)
11from custom_components.supernotify import (
12 ATTR_DATA,
13 ATTR_PRIORITY,
14 OPTION_CHIME_ALIASES,
15 OPTION_TARGET_CATEGORIES,
16 OPTION_TARGET_INCLUDE_RE,
17 RE_DEVICE_ID,
18 TRANSPORT_CHIME,
19)
20from custom_components.supernotify.envelope import Envelope
21from custom_components.supernotify.model import Target, TargetRequired, TransportConfig
22from custom_components.supernotify.transport import Transport
24RE_VALID_CHIME = r"(switch|script|group|siren|media_player)\.[A-Za-z0-9_]+"
26_LOGGER = logging.getLogger(__name__)
28DATA_SCHEMA_RESTRICT: dict[str, list[str]] = {
29 "media_player": ["data", "entity_id", "media_content_id", "media_content_type", "enqueue", "announce"],
30 "switch": ["entity_id"],
31 "script": ["data", "variables", "context", "wait"],
32 "siren": ["data", "entity_id"],
33 "alexa_devices": ["sound", "device_id"],
34} # TODO: source directly from component schema
36DEVICE_DOMAINS = ["alexa_devices"]
39class ChimeTargetConfig:
40 def __init__(
41 self,
42 entity_id: str | None = None,
43 device_id: str | None = None,
44 tune: str | None = None,
45 duration: int | None = None,
46 volume: float | None = None,
47 data: dict[str, Any] | None = None,
48 domain: str | None = None,
49 **kwargs: Any,
50 ) -> None:
51 self.entity_id: str | None = entity_id
52 self.device_id: str | None = device_id
53 self.domain: str | None = None
54 self.entity_name: str | None = None
55 if self.entity_id:
56 self.domain, self.entity_name = self.entity_id.split(".", 1)
57 elif self.device_id:
58 self.domain = domain
59 else:
60 raise ValueError("ChimeTargetConfig target must be entity_id or device_id")
61 if kwargs:
62 _LOGGER.warning("SUPERNOTIFY ChimeTargetConfig ignoring unexpected args: %s", kwargs)
63 self.volume: float | None = volume
64 self.tune: str | None = tune
65 self.duration: int | None = duration
66 self.data: dict[str, Any] | None = data or {}
68 def __repr__(self) -> str:
69 """Return a developer-oriented string representation of this ChimeTargetConfig"""
70 if self.device_id is not None:
71 return f"ChimeTargetConfig(device_id={self.device_id})"
72 return f"ChimeTargetConfig(entity_id={self.entity_id})"
75class ChimeTransport(Transport):
76 name = TRANSPORT_CHIME
78 def __init__(self, *args: Any, **kwargs: Any) -> None:
79 super().__init__(*args, **kwargs)
81 @property
82 def default_config(self) -> TransportConfig:
83 config = TransportConfig()
84 config.delivery_defaults.options = {}
85 config.delivery_defaults.target_required = TargetRequired.OPTIONAL
86 config.device_domain = DEVICE_DOMAINS
87 config.delivery_defaults.options = {
88 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID, ATTR_DEVICE_ID],
89 OPTION_TARGET_INCLUDE_RE: [RE_VALID_CHIME, RE_DEVICE_ID],
90 }
91 return config
93 @property
94 def chime_aliases(self) -> dict[str, Any]:
95 return self.delivery_defaults.options.get(OPTION_CHIME_ALIASES) or {}
97 def validate_action(self, action: str | None) -> bool:
98 return action is None
100 async def deliver(self, envelope: Envelope) -> bool:
101 data: dict[str, Any] = {}
102 data.update(envelope.delivery.data)
103 data.update(envelope.data or {})
104 target: Target = envelope.target
106 # chime_repeat = data.pop("chime_repeat", 1)
107 chime_tune: str | None = data.pop("chime_tune", None)
108 chime_volume: float | None = data.pop("chime_volume", None)
109 chime_duration: int | None = data.pop("chime_duration", None)
111 _LOGGER.info(
112 "SUPERNOTIFY notify_chime: %s -> %s (delivery: %s, env_data:%s, dlv_data:%s)",
113 chime_tune,
114 target.entity_ids,
115 envelope.delivery_name,
116 envelope.data,
117 envelope.delivery.data,
118 )
119 # expand groups
120 expanded_targets = {
121 e: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, entity_id=e)
122 for e in self.hass_api.expand_group(target.entity_ids)
123 }
124 expanded_targets.update({
125 d: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, device_id=d)
126 for d in target.device_ids
127 })
128 # resolve and include chime aliases
129 expanded_targets.update(self.resolve_tune(chime_tune)) # overwrite and extend
131 chimes = 0
132 if not expanded_targets:
133 _LOGGER.info("SUPERNOTIFY skipping chime, no targets")
134 return False
135 for chime_entity_config in expanded_targets.values():
136 _LOGGER.debug("SUPERNOTIFY chime %s: %s", chime_entity_config.entity_id, chime_entity_config.tune)
137 action_data = None
138 try:
139 domain, service, action_data, target_data = self.analyze_target(chime_entity_config, data, envelope)
140 if domain is not None and service is not None:
141 action_data = self.prune_data(domain, action_data)
143 if await self.call_action(
144 envelope, qualified_action=f"{domain}.{service}", action_data=action_data, target_data=target_data
145 ):
146 chimes += 1
147 else:
148 _LOGGER.debug("SUPERNOTIFY Chime skipping incomplete service for %s", chime_entity_config.entity_id)
149 except Exception:
150 _LOGGER.exception("SUPERNOTIFY Failed to chime %s: %s [%s]", chime_entity_config.entity_id, action_data)
151 return chimes > 0
153 def prune_data(self, domain: str, data: dict[str, Any]) -> dict[str, Any]:
154 pruned: dict[str, Any] = {}
155 if data and domain in DATA_SCHEMA_RESTRICT:
156 restrict: list[str] = DATA_SCHEMA_RESTRICT.get(domain) or []
157 for key in list(data.keys()):
158 if key in restrict:
159 pruned[key] = data[key]
160 return pruned
162 def analyze_target(
163 self, target_config: ChimeTargetConfig, data: dict[str, Any], envelope: Envelope
164 ) -> tuple[str | None, str | None, dict[str, Any], dict[str, Any]]:
165 if not target_config.entity_id and not target_config.device_id:
166 _LOGGER.warning("SUPERNOTIFY Empty chime target")
167 return "", None, {}, {}
169 domain: str | None = None
170 name: str | None = None
172 # Alexa Devices use device_id not entity_id for sounds
173 # TODO: use method or delivery config vs fixed local constant for domains
174 if target_config.device_id is not None and DEVICE_DOMAINS:
175 if target_config.domain is not None and target_config.domain in DEVICE_DOMAINS:
176 domain = target_config.domain
177 else:
178 domain = self.hass_api.domain_for_device(target_config.device_id, DEVICE_DOMAINS)
180 elif target_config.entity_id and "." in target_config.entity_id:
181 domain, name = target_config.entity_id.split(".", 1)
183 action_data: dict[str, Any] = {}
184 target_data: dict[str, Any] = {}
185 action: str | None = None
187 if domain == "switch":
188 action = "turn_on"
189 target_data[ATTR_ENTITY_ID] = target_config.entity_id
191 elif domain == "siren":
192 action = "turn_on"
193 target_data[ATTR_ENTITY_ID] = target_config.entity_id
194 action_data[ATTR_DATA] = {}
195 if target_config.tune:
196 action_data[ATTR_DATA]["tone"] = target_config.tune
197 if target_config.duration is not None:
198 action_data[ATTR_DATA]["duration"] = target_config.duration
199 if target_config.volume is not None:
200 action_data[ATTR_DATA]["volume_level"] = target_config.volume
202 elif domain == "script":
203 action_data.setdefault(CONF_VARIABLES, {})
204 if target_config.data:
205 action_data[CONF_VARIABLES] = target_config.data.get(CONF_VARIABLES, {})
206 if data:
207 # override data sourced from chime alias with explicit variables in envelope/data
208 action_data[CONF_VARIABLES].update(data.get(CONF_VARIABLES, {}))
209 action = name
210 action_data[CONF_VARIABLES][ATTR_MESSAGE] = envelope.message
211 action_data[CONF_VARIABLES][ATTR_TITLE] = envelope.title
212 action_data[CONF_VARIABLES][ATTR_PRIORITY] = envelope.priority
213 action_data[CONF_VARIABLES]["chime_tune"] = target_config.tune
214 action_data[CONF_VARIABLES]["chime_volume"] = target_config.volume
215 action_data[CONF_VARIABLES]["chime_duration"] = target_config.duration
217 elif domain == "alexa_devices" and target_config.tune:
218 action = "send_sound"
219 action_data["device_id"] = target_config.device_id
220 action_data["sound"] = target_config.tune
222 elif domain == "media_player" and target_config.tune:
223 if target_config.data:
224 action_data.update(target_config.data)
225 if data:
226 action_data.update(data)
227 action = "play_media"
228 target_data[ATTR_ENTITY_ID] = target_config.entity_id
229 action_data["media_content_type"] = "sound"
230 action_data["media_content_id"] = target_config.tune
232 else:
233 _LOGGER.warning(
234 "SUPERNOTIFY No matching chime domain/tune: %s, target: %s, tune: %s",
235 domain,
236 target_config.entity_id,
237 target_config.tune,
238 )
240 return domain, action, action_data, target_data
242 def resolve_tune(self, tune_or_alias: str | None) -> dict[str, ChimeTargetConfig]:
243 target_configs: dict[str, ChimeTargetConfig] = {}
244 if tune_or_alias is not None:
245 for label, alias_config in self.chime_aliases.get(tune_or_alias, {}).items():
246 if isinstance(alias_config, str):
247 tune = alias_config
248 alias_config = {}
249 else:
250 tune = alias_config.get("tune", tune_or_alias)
252 alias_config["tune"] = tune
253 alias_config.setdefault("domain", label)
254 alias_config.setdefault("data", {})
255 raw_target = alias_config.pop("target", None)
257 # pass through variables or data if present
258 if raw_target is not None:
259 target = Target(raw_target)
260 target_configs.update({t: ChimeTargetConfig(entity_id=t, **alias_config) for t in target.entity_ids})
261 target_configs.update({t: ChimeTargetConfig(device_id=t, **alias_config) for t in target.device_ids})
262 elif alias_config["domain"] in DEVICE_DOMAINS:
263 # bulk apply to all known target devices of this domain
264 bulk_apply = {
265 dev: ChimeTargetConfig(device_id=dev, **alias_config)
266 for dev in self.targets.device_ids
267 if dev not in target_configs # don't overwrite existing specific targets
268 and ATTR_DEVICE_ID not in alias_config
269 }
270 # TODO: Constrain to device domain
271 target_configs.update(bulk_apply)
272 else:
273 # bulk apply to all known target entities of this domain
274 bulk_apply = {
275 ent: ChimeTargetConfig(entity_id=ent, **alias_config)
276 for ent in self.targets.entity_ids
277 if ent.startswith(f"{alias_config['domain']}.")
278 and ent not in target_configs # don't overwrite existing specific targets
279 and ATTR_ENTITY_ID not in alias_config
280 }
281 target_configs.update(bulk_apply)
282 _LOGGER.debug("SUPERNOTIFY transport_chime: Resolved tune %s to %s", tune_or_alias, target_configs)
283 return target_configs