Coverage for custom_components / supernotify / transports / alexa_media_player.py: 98%
131 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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
48References:
49- energywave/multinotify https://github.com/energywave/multinotify
50- ago19800/centralino https://github.com/ago19800/centralino
51- jumping2000/universal_notifier https://github.com/jumping2000/universal_notifier
52- AMP issue #1394 https://github.com/alandtse/alexa_media_player/issues/1394
53- AMP discussion #2782 https://github.com/alandtse/alexa_media_player/discussions/2782
54- multinotify issue #6 https://github.com/energywave/multinotify/issues/6
56"""
58from __future__ import annotations
60import asyncio
61import logging
62import re
63from typing import TYPE_CHECKING, Any, cast
65from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
66from homeassistant.const import ATTR_ENTITY_ID
68from custom_components.supernotify.common import boolify
69from custom_components.supernotify.const import (
70 OPTION_MESSAGE_USAGE,
71 OPTION_SIMPLIFY_TEXT,
72 OPTION_STRIP_URLS,
73 OPTION_TARGET_CATEGORIES,
74 OPTION_TARGET_SELECT,
75 OPTION_UNIQUE_TARGETS,
76 TRANSPORT_ALEXA_MEDIA_PLAYER,
77)
78from custom_components.supernotify.model import (
79 DebugTrace,
80 MessageOnlyPolicy,
81 TargetRequired,
82 TransportConfig,
83 TransportFeature,
84)
85from custom_components.supernotify.transport import Transport
87if TYPE_CHECKING:
88 from custom_components.supernotify.envelope import Envelope
90RE_VALID_ALEXA = r"media_player\.[A-Za-z0-9_]+"
93RE_SSML_TAG = re.compile(r"<[^>]+>")
94PAUSE_CHARS = (", ", ". ", "! ", "? ", ": ", "; ")
96# ref: https://github.com/alandtse/alexa_media_player/wiki/Configuration%3A-Notification-Component
97SERVICE_DATA_KEYS = [ATTR_MESSAGE, ATTR_TITLE, ATTR_DATA, ATTR_TARGET]
98SERVICE_DATA_DATA_KEYS = ["type", "method"]
100_PAUSE_WEIGHT = 0.35
101_CHAR_WEIGHT = 0.06
102_BASE_DURATION = 5.0
103_MUSIC_RESUME_DELAY = 2.0
105_LOGGER = logging.getLogger(__name__)
108def _estimate_tts_duration(message: str, char_weight: float = _CHAR_WEIGHT) -> float:
109 """Estimate pronunciation duration in seconds, stripping SSML first.
111 Formula from energywave/multinotify:
112 duration = BASE + pause_chars x PAUSE_WEIGHT + chars x char_weight
114 Args:
115 message: The TTS message (SSML tags are stripped before counting).
116 char_weight: Seconds per plain-text character. Override via the
117 ``tts_char_speed`` data key to calibrate for the TTS
118 language (default 0.06 s/ch — Italian/English).
120 """
121 plain = RE_SSML_TAG.sub("", message)
122 pause_count = sum(plain.count(p) for p in PAUSE_CHARS)
123 return _BASE_DURATION + pause_count * _PAUSE_WEIGHT + len(plain) * char_weight
126class AlexaMediaPlayerTransport(Transport):
127 """Notify via Amazon Alexa announcements with full volume management.
129 options:
130 message_usage: standard | use_title | combine_title
131 """
133 name = TRANSPORT_ALEXA_MEDIA_PLAYER
135 def __init__(self, *args: Any, **kwargs: Any) -> None:
136 super().__init__(*args, **kwargs)
138 @property
139 def supported_features(self) -> TransportFeature:
140 return TransportFeature.MESSAGE | TransportFeature.SPOKEN
142 @property
143 def default_config(self) -> TransportConfig:
144 config = TransportConfig()
145 config.delivery_defaults.action = "notify.alexa_media"
146 config.delivery_defaults.target_required = TargetRequired.ALWAYS
147 config.delivery_defaults.options = {
148 OPTION_SIMPLIFY_TEXT: True,
149 OPTION_STRIP_URLS: True,
150 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
151 OPTION_UNIQUE_TARGETS: True,
152 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID],
153 OPTION_TARGET_SELECT: [RE_VALID_ALEXA],
154 }
155 return config
157 def validate_action(self, action: str | None) -> bool:
158 return action is not None
160 async def _safe_service(self, domain: str, service: str, service_data: dict[str, Any]) -> bool:
161 """Call a HA service via hass_api, catching exceptions so offline devices never block overall delivery."""
162 try:
163 await self.hass_api.call_service(domain, service, service_data=service_data)
164 return True
165 except Exception as exc:
166 _LOGGER.debug(
167 "SUPERNOTIFY alexa_media_player: %s.%s failed for %s: %s",
168 domain,
169 service,
170 service_data.get(ATTR_ENTITY_ID, "unknown"),
171 exc,
172 )
173 return False
175 async def _snapshot_states(self, media_players: list[str], volume_fallback: float) -> dict[str, dict[str, Any]]:
176 """Read volume and playback state for every target.
178 Uses volume_fallback when volume_level is None (AMP issue #1394).
179 """
180 states: dict[str, dict[str, Any]] = {}
181 for mp in media_players:
182 state = self.hass_api.get_state(mp)
183 if state is None:
184 _LOGGER.debug("SUPERNOTIFY alexa_media_player: %s not found", mp)
185 continue
186 vol = state.attributes.get("volume_level")
187 if vol is None:
188 _LOGGER.debug(
189 "SUPERNOTIFY alexa_media_player: %s volume_level None, using fallback %.2f (AMP issue #1394)",
190 mp,
191 volume_fallback,
192 )
193 vol = volume_fallback
194 states[mp] = {"volume": float(vol), "playing": state.state == "playing"}
195 return states
197 async def _pre_announce(
198 self,
199 states: dict[str, dict[str, Any]],
200 requested_volume: float,
201 pause_music: bool,
202 ) -> set[str]:
203 """Pause music, stop beep, set announcement volume."""
204 volume_set_failed: set[str] = set()
205 for mp, prev in states.items():
206 if prev["playing"]:
207 if pause_music:
208 # Pause only — do NOT also call media_stop.
209 # media_stop after media_pause kills streaming sessions
210 # (Spotify, etc.) making them impossible to resume later.
211 # media_pause leaves the session alive for media_play resume.
212 await self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp})
213 else:
214 # Not pausing: use media_stop to suppress the Alexa
215 # confirmation beep before volume_set (no resume expected).
216 await self._safe_service("media_player", "media_stop", {ATTR_ENTITY_ID: mp})
217 if not await self._safe_service(
218 "media_player",
219 "volume_set",
220 {ATTR_ENTITY_ID: mp, "volume_level": requested_volume},
221 ):
222 volume_set_failed.add(mp)
223 return volume_set_failed
225 async def _post_announce(
226 self,
227 states: dict[str, dict[str, Any]],
228 restore_volume: bool,
229 pause_music: bool,
230 ) -> None:
231 """Restore volume and resume music after announcement."""
232 music_devices = [mp for mp, s in states.items() if pause_music and s["playing"]]
233 for mp, prev in states.items():
234 # Do NOT call media_stop here: after TTS finishes Alexa is already
235 # idle, so media_stop would produce an unwanted confirmation beep.
236 if restore_volume:
237 await self._safe_service(
238 "media_player",
239 "volume_set",
240 {ATTR_ENTITY_ID: mp, "volume_level": prev["volume"]},
241 )
242 if music_devices:
243 await asyncio.sleep(_MUSIC_RESUME_DELAY)
244 for mp in music_devices:
245 await self._safe_service("media_player", "media_play", {ATTR_ENTITY_ID: mp})
247 async def deliver(
248 self,
249 envelope: Envelope,
250 debug_trace: DebugTrace | None = None, # noqa: ARG002
251 ) -> bool:
252 _LOGGER.debug("SUPERNOTIFY notify_alexa_media %s", envelope.message)
254 media_players = envelope.target.entity_ids or []
255 if not media_players:
256 _LOGGER.debug("SUPERNOTIFY skipping alexa media player, no targets")
257 return False
259 # envelope.data is a flat dict — keys like volume, type, method
260 # are at the top level, not nested under a "data" key.
261 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
263 volume_raw = raw_data.pop("volume", None)
264 restore_volume: bool = boolify(raw_data.pop("restore_volume", True), default=True)
265 pause_music: bool = boolify(raw_data.pop("pause_music", True), default=True)
266 volume_fallback: float = float(raw_data.pop("volume_fallback", 0.5))
267 wait_for_tts: bool = boolify(raw_data.pop("wait_for_tts", False), default=False)
268 tts_char_speed: float = float(raw_data.pop("tts_char_speed", _CHAR_WEIGHT))
270 # Resolve Jinja2 template if volume is still a raw template string
271 # (scenarios store volume as a template; _resolve_data_templates only
272 # runs for archiving, not for delivery).
273 requested_volume: float | None = None
274 if isinstance(volume_raw, str) and "{{" in volume_raw:
275 try:
276 context_vars = (
277 cast("dict[str, Any]", envelope.condition_variables.as_dict()) if envelope.condition_variables else {}
278 )
279 rendered = self.hass_api.template(volume_raw).async_render(variables=context_vars)
280 requested_volume = float(rendered)
281 _LOGGER.debug("SUPERNOTIFY alexa_media_player: resolved volume template to %.2f", requested_volume)
282 except Exception as exc:
283 _LOGGER.warning("SUPERNOTIFY alexa_media_player: failed to resolve volume template %r: %s", volume_raw, exc)
284 elif volume_raw is not None:
285 try:
286 requested_volume = float(volume_raw)
287 except (TypeError, ValueError) as e: # py3.13 compat
288 _LOGGER.warning("SUPERNOTIFY alexa_media_player: invalid volume value %r, ignoring: %s", volume_raw, e)
290 # Pre-announce
291 states: dict[str, dict[str, Any]] = {}
292 needs_restore = requested_volume is not None or pause_music
294 if needs_restore:
295 states = await self._snapshot_states(media_players, volume_fallback)
297 volume_set_failed: set[str] = set()
298 if requested_volume is not None and states:
299 volume_set_failed = await self._pre_announce(states, requested_volume, pause_music)
300 elif pause_music and states:
301 for mp, prev in states.items():
302 if prev["playing"]:
303 await self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp})
305 # Announce
306 call_type: str = raw_data.pop("type", "announce")
307 action_data: dict[str, Any] = {
308 "message": envelope.message,
309 ATTR_DATA: {"type": call_type},
310 ATTR_TARGET: media_players,
311 }
312 if requested_volume is not None and volume_set_failed:
313 # Fallback path: if pre-announce volume_set fails for one or more
314 # players, pass volume through notify.alexa_media too.
315 action_data[ATTR_DATA]["volume"] = requested_volume
317 result = await self.call_action(envelope, action_data=action_data)
319 # Post-announce: optionally wait for TTS, then restore volume / resume music.
320 # needs_post_announce is True whenever there is something to undo (volume change
321 # or music was paused); in that case the TTS wait is always performed so the
322 # restore/resume happens after Alexa finishes speaking.
323 # wait_for_tts additionally blocks even in pure fire-and-forget deliveries,
324 # allowing automation sequences to run only after the announcement ends.
325 needs_post_announce = needs_restore and bool(states)
326 if (needs_post_announce or wait_for_tts) and envelope.message:
327 tts_duration = _estimate_tts_duration(envelope.message, tts_char_speed)
328 _LOGGER.debug(
329 "SUPERNOTIFY alexa_media_player: waiting %.1f s for TTS (%d chars, %.3f s/ch)",
330 tts_duration,
331 len(RE_SSML_TAG.sub("", envelope.message)),
332 tts_char_speed,
333 )
334 await asyncio.sleep(tts_duration)
335 if needs_post_announce:
336 await self._post_announce(states, restore_volume and requested_volume is not None, pause_music)
338 return result