Coverage for custom_components/supernotify/transports/alexa_media_player.py: 98%
132 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1"""Alexa Media Player transport adaptor for Supernotify.
3Volume management: Amazon Alexa API does not expose a per-announcement
4volume parameter in notify.alexa_media. This adaptor handles it natively:
61. Snapshot - reads current volume_level of every target media_player.
7 If None (AMP startup bug, issue #1394), uses volume_fallback.
82. Pause/Stop- If pause_music=True and playing: media_pause only (preserves
9 streaming session for resume). No media_stop — calling it after
10 media_pause kills Spotify/streaming and prevents resume.
11 - If pause_music=False and playing: media_stop only (suppresses
12 Alexa beep before volume_set; no resume expected).
13 - If idle: neither (media_stop on idle Alexa triggers a beep).
143. Set vol - media_player.volume_set on every target.
154. Announce - notify.alexa_media without volume in payload.
165. Wait - estimates TTS duration, SSML-aware (energywave/multinotify).
17 Skipped when wait_for_tts=False (default) and no volume/music
18 restore is needed (fire-and-forget mode).
19 Duration calibrated per-language via tts_char_speed.
206. Resume - media_player.media_play after 2s delay if was playing.
217. Restore - media_player.volume_set back to previous level.
22 No media_stop in post-announce: Alexa is already idle after
23 TTS, calling media_stop would produce another unwanted beep.
25Data keys (all optional):
26 volume float 0-1 desired announcement volume
27 restore_volume bool restore previous volume (default True)
28 pause_music bool pause music if playing (default True)
29 volume_fallback float 0-1 fallback when volume_level is None (default 0.5)
30 wait_for_tts bool block until TTS finishes before returning.
31 Default False (fire-and-forget).
32 Set True to sequence automation actions after
33 the announcement (e.g. "open blinds only after
34 Alexa has finished speaking").
35 When volume/music restore is active this wait
36 happens implicitly; wait_for_tts=True only adds
37 extra blocking in pure fire-and-forget deliveries.
38 tts_char_speed float s/ch seconds per character for TTS duration estimate.
39 Default 0.06 (Italian/English calibration).
40 Suggested values by language family:
41 Italian / English / French : 0.060
42 Spanish / Portuguese : 0.058
43 German : 0.065
44 Russian / Polish : 0.062
45 Japanese / Chinese / Korean : 0.180
46 Arabic : 0.075
49References:
50- energywave/multinotify https://github.com/energywave/multinotify
51- ago19800/centralino https://github.com/ago19800/centralino
52- jumping2000/universal_notifier https://github.com/jumping2000/universal_notifier
53- AMP issue #1394 https://github.com/alandtse/alexa_media_player/issues/1394
54- AMP discussion #2782 https://github.com/alandtse/alexa_media_player/discussions/2782
55- multinotify issue #6 https://github.com/energywave/multinotify/issues/6
57"""
59from __future__ import annotations
61import asyncio
62import logging
63import re
64from typing import TYPE_CHECKING, Any, cast
66from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
67from homeassistant.const import ATTR_ENTITY_ID
69from custom_components.supernotify.common import boolify
70from custom_components.supernotify.const import (
71 OPTION_MEDIA_AUTO_PAUSE,
72 OPTION_MESSAGE_USAGE,
73 OPTION_SIMPLIFY_TEXT,
74 OPTION_STRIP_URLS,
75 OPTION_TARGET_CATEGORIES,
76 OPTION_TARGET_SELECT,
77 OPTION_UNIQUE_TARGETS,
78 TRANSPORT_ALEXA_MEDIA_PLAYER,
79)
80from custom_components.supernotify.model import (
81 DebugTrace,
82 MessageOnlyPolicy,
83 TargetRequired,
84 TransportConfig,
85 TransportFeature,
86)
87from custom_components.supernotify.transport import Transport
89if TYPE_CHECKING:
90 from custom_components.supernotify.envelope import Envelope
92RE_VALID_ALEXA = r"media_player\.[A-Za-z0-9_]+"
95RE_SSML_TAG = re.compile(r"<[^>]+>")
96PAUSE_CHARS = (", ", ". ", "! ", "? ", ": ", "; ")
98# ref: https://github.com/alandtse/alexa_media_player/wiki/Configuration%3A-Notification-Component
99SERVICE_DATA_KEYS = [ATTR_MESSAGE, ATTR_TITLE, ATTR_DATA, ATTR_TARGET]
100SERVICE_DATA_DATA_KEYS = ["type", "method"]
102_PAUSE_WEIGHT = 0.35
103_CHAR_WEIGHT = 0.06
104_BASE_DURATION = 5.0
105_MUSIC_RESUME_DELAY = 2.0
107_LOGGER = logging.getLogger(__name__)
110def _estimate_tts_duration(message: str, char_weight: float = _CHAR_WEIGHT) -> float:
111 """Estimate pronunciation duration in seconds, stripping SSML first.
113 Formula from energywave/multinotify:
114 duration = BASE + pause_chars x PAUSE_WEIGHT + chars x char_weight
116 Args:
117 message: The TTS message (SSML tags are stripped before counting).
118 char_weight: Seconds per plain-text character. Override via the
119 ``tts_char_speed`` data key to calibrate for the TTS
120 language (default 0.06 s/ch — Italian/English).
122 """
123 plain = RE_SSML_TAG.sub("", message)
124 pause_count = sum(plain.count(p) for p in PAUSE_CHARS)
125 return _BASE_DURATION + pause_count * _PAUSE_WEIGHT + len(plain) * char_weight
128class AlexaMediaPlayerTransport(Transport):
129 """Notify via Amazon Alexa announcements with full volume management.
131 options:
132 message_usage: standard | use_title | combine_title
133 media_auto_pause: bool, sets the default for pausing music/restoring volume
134 """
136 name = TRANSPORT_ALEXA_MEDIA_PLAYER
138 def __init__(self, *args: Any, **kwargs: Any) -> None:
139 super().__init__(*args, **kwargs)
141 @property
142 def supported_features(self) -> TransportFeature:
143 return TransportFeature.MESSAGE | TransportFeature.SPOKEN
145 @property
146 def default_config(self) -> TransportConfig:
147 config = TransportConfig()
148 config.delivery_defaults.action = "notify.alexa_media"
149 config.delivery_defaults.target_required = TargetRequired.ALWAYS
150 config.delivery_defaults.options = {
151 OPTION_SIMPLIFY_TEXT: True,
152 OPTION_STRIP_URLS: True,
153 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
154 OPTION_UNIQUE_TARGETS: True,
155 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID],
156 OPTION_TARGET_SELECT: [RE_VALID_ALEXA],
157 OPTION_MEDIA_AUTO_PAUSE: True,
158 }
159 return config
161 def validate_action(self, action: str | None) -> bool:
162 return action is not None
164 async def _safe_service(self, domain: str, service: str, service_data: dict[str, Any]) -> bool:
165 """Call a HA service via hass_api, catching exceptions so offline devices never block overall delivery."""
166 try:
167 await self.hass_api.call_service(domain, service, service_data=service_data)
168 return True
169 except Exception as exc:
170 _LOGGER.debug(
171 "SUPERNOTIFY alexa_media_player: %s.%s failed for %s: %s",
172 domain,
173 service,
174 service_data.get(ATTR_ENTITY_ID, "unknown"),
175 exc,
176 )
177 return False
179 async def _snapshot_states(self, media_players: list[str], volume_fallback: float) -> dict[str, dict[str, Any]]:
180 """Read volume and playback state for every target.
182 Uses volume_fallback when volume_level is None (AMP issue #1394).
183 """
184 states: dict[str, dict[str, Any]] = {}
185 for mp in media_players:
186 state = self.hass_api.get_state(mp)
187 if state is None:
188 _LOGGER.debug("SUPERNOTIFY alexa_media_player: %s not found", mp)
189 continue
190 vol = state.attributes.get("volume_level")
191 if vol is None:
192 _LOGGER.debug(
193 "SUPERNOTIFY alexa_media_player: %s volume_level None, using fallback %.2f (AMP issue #1394)",
194 mp,
195 volume_fallback,
196 )
197 vol = volume_fallback
198 states[mp] = {"volume": float(vol), "playing": state.state == "playing"}
199 return states
201 async def _pre_announce(
202 self,
203 states: dict[str, dict[str, Any]],
204 requested_volume: float,
205 pause_music: bool,
206 ) -> set[str]:
207 """Pause music, stop beep, set announcement volume. Runs per-device concurrently since
208 each Alexa cloud round trip can take seconds, and serializing across targets stacks
209 that latency instead of overlapping it."""
211 async def handle(mp: str, prev: dict[str, Any]) -> tuple[str, bool]:
212 if prev["playing"]:
213 if pause_music:
214 # Pause only — do NOT also call media_stop.
215 # media_stop after media_pause kills streaming sessions
216 # (Spotify, etc.) making them impossible to resume later.
217 # media_pause leaves the session alive for media_play resume.
218 await self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp})
219 else:
220 # Not pausing: use media_stop to suppress the Alexa
221 # confirmation beep before volume_set (no resume expected).
222 await self._safe_service("media_player", "media_stop", {ATTR_ENTITY_ID: mp})
223 ok = await self._safe_service(
224 "media_player",
225 "volume_set",
226 {ATTR_ENTITY_ID: mp, "volume_level": requested_volume},
227 )
228 return mp, ok
230 results = await asyncio.gather(*(handle(mp, prev) for mp, prev in states.items()))
231 return {mp for mp, ok in results if not ok}
233 async def _post_announce(
234 self,
235 states: dict[str, dict[str, Any]],
236 restore_volume: bool,
237 pause_music: bool,
238 ) -> None:
239 """Restore volume and resume music after announcement, per-device concurrently."""
240 music_devices = [mp for mp, s in states.items() if pause_music and s["playing"]]
241 if restore_volume:
242 # Do NOT call media_stop here: after TTS finishes Alexa is already
243 # idle, so media_stop would produce an unwanted confirmation beep.
244 await asyncio.gather(
245 *(
246 self._safe_service(
247 "media_player",
248 "volume_set",
249 {ATTR_ENTITY_ID: mp, "volume_level": prev["volume"]},
250 )
251 for mp, prev in states.items()
252 )
253 )
254 if music_devices:
255 await asyncio.sleep(_MUSIC_RESUME_DELAY)
256 await asyncio.gather(
257 *(self._safe_service("media_player", "media_play", {ATTR_ENTITY_ID: mp}) for mp in music_devices)
258 )
260 async def deliver(
261 self,
262 envelope: Envelope,
263 debug_trace: DebugTrace | None = None,
264 ) -> bool:
265 _LOGGER.debug("SUPERNOTIFY notify_alexa_media %s", envelope.message)
267 media_players = envelope.target.entity_ids or []
268 if not media_players:
269 _LOGGER.debug("SUPERNOTIFY Skipping alexa media player, no targets")
270 return False
272 # envelope.data is a flat dict — keys like volume, type, method
273 # are at the top level, not nested under a "data" key.
274 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
276 volume_raw = raw_data.pop("volume", None)
277 restore_volume: bool = boolify(raw_data.pop("restore_volume", True), default=True)
278 pause_music: bool = boolify(raw_data.pop("pause_music", True), default=True)
279 volume_fallback: float = float(raw_data.pop("volume_fallback", 0.5))
280 wait_for_tts: bool = boolify(raw_data.pop("wait_for_tts", False), default=False)
281 tts_char_speed: float = float(raw_data.pop("tts_char_speed", _CHAR_WEIGHT))
283 auto_pause = envelope.delivery.option_bool(OPTION_MEDIA_AUTO_PAUSE, True)
285 # Resolve Jinja2 template if volume is still a raw template string
286 # (scenarios store volume as a template; _resolve_data_templates only
287 # runs for archiving, not for delivery).
288 requested_volume: float | None = None
289 if isinstance(volume_raw, str) and "{{" in volume_raw:
290 try:
291 context_vars = (
292 cast("dict[str, Any]", envelope.condition_variables.as_dict()) if envelope.condition_variables else {}
293 )
294 rendered = self.hass_api.template(volume_raw).async_render(variables=context_vars)
295 requested_volume = float(rendered)
296 _LOGGER.debug("SUPERNOTIFY alexa_media_player: resolved volume template to %.2f", requested_volume)
297 except Exception as exc:
298 _LOGGER.warning("SUPERNOTIFY alexa_media_player: failed to resolve volume template %r: %s", volume_raw, exc)
299 elif volume_raw is not None:
300 try:
301 requested_volume = float(volume_raw)
302 except (TypeError, ValueError) as e: # py3.13 compat
303 _LOGGER.warning("SUPERNOTIFY alexa_media_player: invalid volume value %r, ignoring: %s", volume_raw, e)
305 if auto_pause:
306 # Pre-announce
307 states: dict[str, dict[str, Any]] = {}
308 needs_restore = requested_volume is not None or pause_music
310 if needs_restore:
311 states = await self._snapshot_states(media_players, volume_fallback)
313 volume_set_failed: set[str] = set()
314 if requested_volume is not None and states:
315 volume_set_failed = await self._pre_announce(states, requested_volume, pause_music)
316 elif pause_music and states:
317 for mp, prev in states.items():
318 if prev["playing"]:
319 await self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp})
321 # Announce
322 call_type: str = raw_data.pop("type", "announce")
323 action_data: dict[str, Any] = {
324 "message": envelope.message,
325 ATTR_DATA: {"type": call_type},
326 ATTR_TARGET: media_players,
327 }
328 if requested_volume is not None and volume_set_failed:
329 # Fallback path: if pre-announce volume_set fails for one or more
330 # players, pass volume through notify.alexa_media too.
331 action_data[ATTR_DATA]["volume"] = requested_volume
333 result = await self.call_action(envelope, action_data=action_data)
335 if auto_pause:
336 # Post-announce: optionally wait for TTS, then restore volume / resume music.
337 # needs_post_announce is True whenever there is something to undo (volume change
338 # or music was paused); in that case the TTS wait is always performed so the
339 # restore/resume happens after Alexa finishes speaking.
340 # wait_for_tts additionally blocks even in pure fire-and-forget deliveries,
341 # allowing automation sequences to run only after the announcement ends.
342 needs_post_announce = needs_restore and bool(states)
343 if (needs_post_announce or wait_for_tts) and envelope.message:
344 tts_duration = _estimate_tts_duration(envelope.message, tts_char_speed)
345 _LOGGER.debug(
346 "SUPERNOTIFY alexa_media_player: waiting %.1f s for TTS (%d chars, %.3f s/ch)",
347 tts_duration,
348 len(RE_SSML_TAG.sub("", envelope.message)),
349 tts_char_speed,
350 )
351 await asyncio.sleep(tts_duration)
352 if needs_post_announce:
353 await self._post_announce(states, restore_volume and requested_volume is not None, pause_music)
355 return result