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