Coverage for custom_components/supernotify/transports/alexa_media_player.py: 100%
152 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +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 there is no volume
18 to restore or music to resume (fire-and-forget mode).
19 Duration calibrated per-language via tts_char_speed.
206. Restore - media_player.volume_set back to previous level.
21 No media_stop in post-announce: Alexa is already idle after
22 TTS, calling media_stop would produce another unwanted beep.
237. Resume - media_player.media_play after 2s delay if was playing.
25Data keys (all optional):
26 type str announce (default) | tts | push, passed to notify.alexa_media
27 method str all | speak | show, passed to notify.alexa_media
28 volume float 0-1 desired announcement volume
29 restore_volume bool restore previous volume (default True)
30 pause_music bool pause music if playing (default True)
31 volume_fallback float 0-1 fallback when volume_level is None (default 0.5)
32 wait_for_tts bool block until TTS finishes before returning.
33 Default False (fire-and-forget).
34 Set True to sequence automation actions after
35 the announcement (e.g. "open blinds only after
36 Alexa has finished speaking").
37 When volume/music restore is active this wait
38 happens implicitly; wait_for_tts=True only adds
39 extra blocking in pure fire-and-forget deliveries.
40 audio_url str play an audio file (e.g. an mp3 chime) before the message,
41 via an SSML <audio> tag. Relative URLs like
42 /local/sounds/bell.mp3 are made absolute with the
43 external URL. Amazon fetches the file from its own
44 cloud, so it must be public https with a valid
45 certificate, MP3 at 48kbps and 16000/22050/24000 Hz.
46 Forces `type` to `tts` (Alexa is silent for SSML audio
47 in announce mode), and the message, if any, is spoken
48 after the audio as plain text.
49 audio_duration float s length of the audio_url clip, added to the TTS wait
50 so volume restore/music resume don't cut it short.
51 Default 0 (the 5s base wait covers short chimes).
52 tts_char_speed float s/ch seconds per character for TTS duration estimate.
53 Default 0.06 (Italian/English calibration).
54 Suggested values by language family:
55 Italian / English / French : 0.060
56 Spanish / Portuguese : 0.058
57 German : 0.065
58 Russian / Polish : 0.062
59 Japanese / Chinese / Korean : 0.180
60 Arabic : 0.075
63References:
64- energywave/multinotify https://github.com/energywave/multinotify
65- ago19800/centralino https://github.com/ago19800/centralino
66- jumping2000/universal_notifier https://github.com/jumping2000/universal_notifier
67- AMP issue #1394 https://github.com/alandtse/alexa_media_player/issues/1394
68- AMP discussion #2782 https://github.com/alandtse/alexa_media_player/discussions/2782
69- multinotify issue #6 https://github.com/energywave/multinotify/issues/6
71"""
73from __future__ import annotations
75import asyncio
76import html
77import logging
78import re
79import urllib.parse
80from asyncio import sleep
81from typing import TYPE_CHECKING, Any, ClassVar
83from homeassistant.components.notify.const import ATTR_DATA, ATTR_TARGET
84from homeassistant.const import (
85 ATTR_ENTITY_ID,
86)
87from homeassistant.helpers import config_validation as cv
89from custom_components.supernotify.common import boolify
90from custom_components.supernotify.const import (
91 RE_MEDIA_PLAYER_ENTITY_ID,
92 TRANSPORT_ALEXA_MEDIA_PLAYER,
93)
94from custom_components.supernotify.model import (
95 DebugTrace,
96 MessageOnlyPolicy,
97 TransportConfig,
98 TransportFeature,
99)
100from custom_components.supernotify.options import (
101 OPTION_MESSAGE_USAGE,
102 OPTION_SIMPLIFY_TEXT,
103 OPTION_STRIP_URLS,
104 OPTION_TARGET_SELECT,
105 DeliveryOption,
106)
107from custom_components.supernotify.target import TargetEntityCategory
108from custom_components.supernotify.transport import Transport
110if TYPE_CHECKING:
111 from homeassistant.core import Context as HAContext
113 from custom_components.supernotify.envelope import Envelope
115# alandtse/alexa_media_player HACS integration's notify platform module
116HA_ALEXA_MEDIA_PLAYER_MODULE = "custom_components.alexa_media.notify"
117# the entity registry platform for the media_player entities this integration creates
118HA_ALEXA_MEDIA_PLAYER_PLATFORM = "alexa_media"
121RE_SSML_TAG = re.compile(r"<[^>]+>")
122PAUSE_CHARS = (", ", ". ", "! ", "? ", ": ", "; ")
124_PAUSE_WEIGHT = 0.35
125_CHAR_WEIGHT = 0.06
126_BASE_DURATION = 5.0
127_MUSIC_RESUME_DELAY = 2.0
129_LOGGER = logging.getLogger(__name__)
131OPTION_MEDIA_AUTO_PAUSE = "media_auto_pause"
134def _estimate_tts_duration(message: str, char_weight: float = _CHAR_WEIGHT) -> float:
135 """Estimate pronunciation duration in seconds, stripping SSML first.
137 Formula from energywave/multinotify:
138 duration = BASE + pause_chars x PAUSE_WEIGHT + chars x char_weight
140 Args:
141 message: The TTS message (SSML tags are stripped before counting).
142 char_weight: Seconds per plain-text character. Override via the
143 ``tts_char_speed`` data key to calibrate for the TTS
144 language (default 0.06 s/ch — Italian/English).
146 """
147 plain = RE_SSML_TAG.sub("", message)
148 pause_count = sum(plain.count(p) for p in PAUSE_CHARS)
149 return _BASE_DURATION + pause_count * _PAUSE_WEIGHT + len(plain) * char_weight
152class AlexaMediaPlayerTransport(Transport):
153 """Notify via Amazon Alexa announcements with full volume management.
155 options:
156 message_usage: standard | use_title | combine_title | message_only
157 media_auto_pause: bool, sets the default for pausing music/restoring volume
158 """
160 name = TRANSPORT_ALEXA_MEDIA_PLAYER
161 integration_domain = HA_ALEXA_MEDIA_PLAYER_PLATFORM
162 action_configured = True
163 declared_options: ClassVar[list[DeliveryOption]] = [
164 DeliveryOption(
165 OPTION_MEDIA_AUTO_PAUSE,
166 "Pause (rather than stop) music if playing before announcing, and restore afterwards",
167 value_type=cv.boolean,
168 ),
169 ]
171 def __init__(self, *args: Any, **kwargs: Any) -> None:
172 super().__init__(*args, **kwargs)
174 @property
175 def supported_features(self) -> TransportFeature:
176 return TransportFeature.MESSAGE | TransportFeature.SPOKEN
178 @property
179 def default_config(self) -> TransportConfig:
180 config = super().default_config
181 config.delivery_defaults.action = self.hass_api.find_service("notify", HA_ALEXA_MEDIA_PLAYER_MODULE)
182 config.delivery_defaults.options = {
183 OPTION_SIMPLIFY_TEXT: True,
184 OPTION_STRIP_URLS: True,
185 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
186 OPTION_TARGET_SELECT: [RE_MEDIA_PLAYER_ENTITY_ID],
187 OPTION_MEDIA_AUTO_PAUSE: True,
188 }
189 return config
191 @property
192 def other_target_categories(self) -> list[str | TargetEntityCategory]:
193 # platform-scoped, but still just "a media_player entity" - doesn't say whether the
194 # intent was audio/TTS/a chime, so this stays `other`, not `unique`, like the rest of
195 # the media_player-domain transports (kodi/tts/the `media` transport)
196 return [TargetEntityCategory(domain="media_player", platform=HA_ALEXA_MEDIA_PLAYER_PLATFORM)]
198 def validate_action(self, action: str | None) -> bool:
199 return action is not None
201 async def _safe_service(
202 self, domain: str, service: str, service_data: dict[str, Any], context: HAContext | None = None
203 ) -> bool:
204 """Call a HA service via hass_api, catching exceptions so offline devices never block overall delivery."""
205 try:
206 await self.hass_api.call_service(domain, service, service_data=service_data, context=context)
207 return True
208 except Exception as exc:
209 _LOGGER.debug(
210 "SUPERNOTIFY alexa_media_player: %s.%s failed for %s: %s",
211 domain,
212 service,
213 service_data.get(ATTR_ENTITY_ID, "unknown"),
214 exc,
215 )
216 return False
218 async def _snapshot_states(self, media_players: list[str], volume_fallback: float) -> dict[str, dict[str, Any]]:
219 """Read volume and playback state for every target.
221 Uses volume_fallback when volume_level is None (AMP issue #1394).
222 """
223 states: dict[str, dict[str, Any]] = {}
224 for mp in media_players:
225 state = self.hass_api.get_state(mp)
226 if state is None:
227 _LOGGER.debug("SUPERNOTIFY alexa_media_player: %s not found", mp)
228 continue
229 vol = state.attributes.get("volume_level")
230 if vol is None:
231 _LOGGER.debug(
232 "SUPERNOTIFY alexa_media_player: %s volume_level None, using fallback %.2f (AMP issue #1394)",
233 mp,
234 volume_fallback,
235 )
236 vol = volume_fallback
237 states[mp] = {"volume": float(vol), "playing": state.state == "playing"}
238 return states
240 async def _pre_announce(
241 self,
242 states: dict[str, dict[str, Any]],
243 requested_volume: float,
244 pause_music: bool,
245 context: HAContext | None = None,
246 ) -> set[str]:
247 """Pause music, stop beep, set announcement volume. Runs per-device concurrently since
248 each Alexa cloud round trip can take seconds, and serializing across targets stacks
249 that latency instead of overlapping it."""
251 async def handle(mp: str, prev: dict[str, Any]) -> tuple[str, bool]:
252 if prev["playing"]:
253 if pause_music:
254 # Pause only — do NOT also call media_stop.
255 # media_stop after media_pause kills streaming sessions
256 # (Spotify, etc.) making them impossible to resume later.
257 # media_pause leaves the session alive for media_play resume.
258 await self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp}, context=context)
259 else:
260 # Not pausing: use media_stop to suppress the Alexa
261 # confirmation beep before volume_set (no resume expected).
262 await self._safe_service("media_player", "media_stop", {ATTR_ENTITY_ID: mp}, context=context)
263 ok = await self._safe_service(
264 "media_player",
265 "volume_set",
266 {ATTR_ENTITY_ID: mp, "volume_level": requested_volume},
267 context=context,
268 )
269 return mp, ok
271 results = await asyncio.gather(*(handle(mp, prev) for mp, prev in states.items()))
272 return {mp for mp, ok in results if not ok}
274 async def _post_announce(
275 self,
276 states: dict[str, dict[str, Any]],
277 restore_volume: bool,
278 pause_music: bool,
279 context: HAContext | None = None,
280 ) -> None:
281 """Restore volume and resume music after announcement, per-device concurrently."""
282 music_devices = [mp for mp, s in states.items() if pause_music and s["playing"]]
283 if restore_volume:
284 # Do NOT call media_stop here: after TTS finishes Alexa is already
285 # idle, so media_stop would produce an unwanted confirmation beep.
286 await asyncio.gather(
287 *(
288 self._safe_service(
289 "media_player",
290 "volume_set",
291 {ATTR_ENTITY_ID: mp, "volume_level": prev["volume"]},
292 context=context,
293 )
294 for mp, prev in states.items()
295 )
296 )
297 if music_devices:
298 await sleep(_MUSIC_RESUME_DELAY)
299 await asyncio.gather(
300 *(
301 self._safe_service("media_player", "media_play", {ATTR_ENTITY_ID: mp}, context=context)
302 for mp in music_devices
303 )
304 )
306 def _audio_ssml(self, audio_url: str, message: str | None) -> str:
307 """Wrap an audio clip, and the message spoken after it, in SSML for Alexa."""
308 url = urllib.parse.urljoin(self.hass_api.external_url or "", audio_url)
309 if not url.startswith("https://"):
310 _LOGGER.warning(
311 "SUPERNOTIFY alexa_media_player: audio_url %s is not https, Alexa will refuse to play it",
312 url,
313 )
314 spoken = html.escape(message, quote=False) if message else ""
315 return f'<speak><audio src="{html.escape(url, quote=True)}"/>{spoken}</speak>'
317 async def deliver(
318 self,
319 envelope: Envelope,
320 debug_trace: DebugTrace | None = None,
321 ) -> bool:
322 _LOGGER.debug("SUPERNOTIFY notify_alexa_media %s", envelope.message)
324 media_players = envelope.target.entity_ids or []
325 if not media_players:
326 _LOGGER.debug("SUPERNOTIFY Skipping alexa media player, no targets")
327 return False
329 # envelope.data is a flat dict — keys like volume, type, method
330 # are at the top level, not nested under a "data" key.
331 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
333 volume_raw = raw_data.pop("volume", None)
334 restore_volume: bool = boolify(raw_data.pop("restore_volume", True), default=True)
335 pause_music: bool = boolify(raw_data.pop("pause_music", True), default=True)
336 volume_fallback: float = float(raw_data.pop("volume_fallback", 0.5))
337 wait_for_tts: bool = boolify(raw_data.pop("wait_for_tts", False), default=False)
338 tts_char_speed: float = float(raw_data.pop("tts_char_speed", _CHAR_WEIGHT))
339 audio_url: str | None = raw_data.pop("audio_url", None)
340 audio_duration: float = float(raw_data.pop("audio_duration", 0) or 0)
342 message: str | None = envelope.message
343 if audio_url:
344 message = self._audio_ssml(audio_url, message)
346 auto_pause = envelope.delivery.option_bool(OPTION_MEDIA_AUTO_PAUSE, True)
348 # a template for volume has already been rendered by the envelope
349 requested_volume: float | None = None
350 if volume_raw is not None:
351 try:
352 requested_volume = float(volume_raw)
353 except (TypeError, ValueError) as e: # py3.13 compat
354 _LOGGER.warning("SUPERNOTIFY alexa_media_player: invalid volume value %r, ignoring: %s", volume_raw, e)
356 states: dict[str, dict[str, Any]] = {}
357 volume_set_failed: set[str] = set()
359 if auto_pause:
360 # Pre-announce
361 if requested_volume is not None or pause_music:
362 states = await self._snapshot_states(media_players, volume_fallback)
364 if requested_volume is not None and states:
365 volume_set_failed = await self._pre_announce(states, requested_volume, pause_music, context=envelope.ha_context)
366 elif pause_music and states:
367 await asyncio.gather(
368 *(
369 self._safe_service("media_player", "media_pause", {ATTR_ENTITY_ID: mp}, context=envelope.ha_context)
370 for mp, prev in states.items()
371 if prev["playing"]
372 )
373 )
374 elif requested_volume is not None:
375 _LOGGER.debug("SUPERNOTIFY alexa_media_player: volume ignored, %s is off", OPTION_MEDIA_AUTO_PAUSE)
377 # needs_post_announce is True only when there is something to undo (a volume to restore
378 # or music that was paused), so an announcement with neither doesn't wait for the TTS;
379 # computed here, before the announce, so it's available in the finally block below
380 # regardless of what happens during the announce/TTS wait.
381 restore_volume = restore_volume and requested_volume is not None
382 needs_post_announce = bool(states) and (
383 restore_volume or (pause_music and any(prev["playing"] for prev in states.values()))
384 )
386 # Announce
387 call_type: str = raw_data.pop("type", "announce")
388 if audio_url and call_type != "tts":
389 # Alexa stays silent for SSML <audio> in announce mode (checked on a real Echo),
390 # so a delivery-level `type: announce` default must not win here
391 _LOGGER.debug("SUPERNOTIFY alexa_media_player: audio_url forces type tts, was %s", call_type)
392 call_type = "tts"
393 action_data: dict[str, Any] = {
394 "message": message,
395 ATTR_DATA: {"type": call_type},
396 ATTR_TARGET: media_players,
397 }
398 if raw_data.get("method"):
399 action_data[ATTR_DATA]["method"] = raw_data.pop("method")
400 if requested_volume is not None and auto_pause and volume_set_failed:
401 # Fallback path: if pre-announce volume_set fails for one or more
402 # players, pass volume through notify.alexa_media too.
403 action_data[ATTR_DATA]["volume"] = requested_volume
405 result = False
406 try:
407 result = await self.call_action(envelope, action_data=action_data)
409 # Post-announce wait: optionally wait for TTS before restoring volume / resuming
410 # music. wait_for_tts additionally blocks even in pure fire-and-forget deliveries,
411 # allowing automation sequences to run only after the announcement ends, and does
412 # so regardless of auto_pause since it has nothing to do with restoring state.
413 if result and (needs_post_announce or wait_for_tts) and message:
414 tts_duration = _estimate_tts_duration(message, tts_char_speed) + audio_duration
415 _LOGGER.debug(
416 "SUPERNOTIFY alexa_media_player: waiting %.1f s for TTS (%d chars, %.3f s/ch, %.1f s audio)",
417 tts_duration,
418 len(RE_SSML_TAG.sub("", message)),
419 tts_char_speed,
420 audio_duration,
421 )
422 await sleep(tts_duration)
423 finally:
424 # Restore/resume must run even if the announce call or TTS wait raises or is
425 # cancelled (e.g. HA shutting down), otherwise a pre-announce pause/volume change
426 # made above is never undone and the device is stuck paused/at announce volume.
427 if needs_post_announce:
428 await self._post_announce(states, restore_volume, pause_music, context=envelope.ha_context)
430 return result