Coverage for custom_components/supernotify/transports/media_player.py: 100%

1from __future__ import annotations 

2 

3import logging 

4import urllib.parse 

5from typing import TYPE_CHECKING, Any, ClassVar 

6 

7from homeassistant.const import ( 

8 ATTR_ENTITY_ID, 

9) 

10from homeassistant.helpers.typing import ConfigType 

11 

12from custom_components.supernotify.const import ( 

13 ATTR_MEDIA, 

14 ATTR_MEDIA_SNAPSHOT_URL, 

15 RE_MEDIA_PLAYER_ENTITY_ID, 

16 TRANSPORT_MEDIA, 

17) 

18from custom_components.supernotify.model import DebugTrace, TransportConfig, TransportFeature 

19from custom_components.supernotify.options import MEDIA_OPTIONS, OPTION_TARGET_SELECT, DeliveryOption 

20from custom_components.supernotify.target import TargetEntityCategory 

21from custom_components.supernotify.transport import Transport 

22 

23if TYPE_CHECKING: 

24 from custom_components.supernotify.envelope import Envelope 

25 from custom_components.supernotify.hass_api import HomeAssistantAPI 

26 

27_LOGGER = logging.getLogger(__name__) 

28 

29 

30class MediaPlayerTransport(Transport): 

31 name = TRANSPORT_MEDIA 

32 integration_domain = "media_player" 

33 declared_options: ClassVar[list[DeliveryOption]] = [*MEDIA_OPTIONS] 

34 

35 def __init__(self, *args: Any, **kwargs: Any) -> None: 

36 super().__init__(*args, **kwargs) 

37 

38 @property 

39 def supported_features(self) -> TransportFeature: 

40 return TransportFeature.IMAGES | TransportFeature.VIDEO | TransportFeature.SOUND | TransportFeature.SNAPSHOT_IMAGE 

41 

42 @property 

43 def default_config(self) -> TransportConfig: 

44 config = super().default_config 

45 config.delivery_defaults.action = "media_player.play_media" 

46 config.delivery_defaults.options = { 

47 OPTION_TARGET_SELECT: [RE_MEDIA_PLAYER_ENTITY_ID], 

48 } 

49 return config 

50 

51 @property 

52 def other_target_categories(self) -> list[str | TargetEntityCategory]: 

53 # a bare media_player entity is too ambiguous to auto-select on - kodi, tts, this 

54 # transport and alexa_media_player all match the same domain, with no way to tell 

55 # from the entity alone which operation (playback, TTS, a chime) was meant 

56 return [TargetEntityCategory(domain="media_player")] 

57 

58 def is_viable(self, hass_api: HomeAssistantAPI) -> bool: 

59 return bool(hass_api.entity_ids_for_domain("media_player")) 

60 

61 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]: 

62 return {self.name: {}} 

63 

64 async def _resolve_snapshot_url(self, envelope: Envelope, data: dict[str, Any]) -> str | None: 

65 """Resolve a snapshot URL for the media player to play. 

66 

67 Order of resolution: 

68 1. An explicit snapshot_url in the media block (absolutised), unprocessed. 

69 2. envelope.grab_image() - respects delivery-specific `jpeg_opts`/`png_opts`/ 

70 `reprocess` options - converted to a shareable URL via media_storage.object_url(). 

71 3. None (caller skips the delivery). 

72 """ 

73 snapshot_url = data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL) 

74 if snapshot_url is None: 

75 # fallback to older idiosyncratic way for backward compatibility 

76 snapshot_url = data.get(ATTR_MEDIA_SNAPSHOT_URL) 

77 if snapshot_url is not None: 

78 return self.absolute_url(snapshot_url) 

79 return await self.image_url(envelope) 

80 

81 def absolute_url(self, url: str) -> str: 

82 # the external URL, probably preferred by Alexa Show etc - urljoin, unlike hass_api.abs_url(), 

83 # leaves other schemes like media-source:// alone 

84 return urllib.parse.urljoin(self.hass_api.external_url, url) 

85 

86 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: 

87 _LOGGER.debug("SUPERNOTIFY notify_media: %s", envelope.data) 

88 

89 data: dict[str, Any] = envelope.data or {} 

90 media_players: list[str] = envelope.target.entity_ids or [] 

91 if not media_players: 

92 _LOGGER.debug("SUPERNOTIFY Skipping media show, no targets") 

93 return False 

94 

95 content_id: str | None 

96 media_type: str 

97 explicit_content_id: str | None = data.get("media_content_id") 

98 if explicit_content_id: 

99 # Generic content (audio clip, video, stream, media-source:// id) - passed through 

100 # as-is, apart from relative URLs like /local/sounds/bell.mp3 which are absolutised 

101 # so remote players (Cast, Sonos etc) can fetch them. Defaults to `music` since 

102 # that is what most media players expect for an audio file. 

103 content_id = self.absolute_url(explicit_content_id) 

104 media_type = data.get("media_content_type", "music") 

105 else: 

106 content_id = await self._resolve_snapshot_url(envelope, data) 

107 media_type = data.get("media_content_type", "image") 

108 if content_id is None: 

109 _LOGGER.debug("SUPERNOTIFY Skipping media player, no media_content_id or snapshot url") 

110 return False 

111 

112 action_data: dict[str, Any] = {"media": {"media_content_id": content_id, "media_content_type": media_type}} 

113 if data and data.get("announce"): 

114 action_data["announce"] = data.get("announce") 

115 if data and data.get("enqueue"): 

116 action_data["enqueue"] = data.get("enqueue") 

117 

118 return await self.call_action(envelope, action_data=action_data, target_data={ATTR_ENTITY_ID: media_players})