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 

6 

7from homeassistant.const import ( 

8 ATTR_ENTITY_ID, 

9) 

10 

11from custom_components.supernotify.const import ( 

12 ATTR_MEDIA, 

13 ATTR_MEDIA_SNAPSHOT_URL, 

14 OPTION_TARGET_CATEGORIES, 

15 OPTION_TARGET_SELECT, 

16 TRANSPORT_MEDIA, 

17) 

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

19from custom_components.supernotify.transport import Transport 

20 

21if TYPE_CHECKING: 

22 from custom_components.supernotify.envelope import Envelope 

23 

24RE_VALID_MEDIA_PLAYER = r"media_player\.[A-Za-z0-9_]+" 

25 

26_LOGGER = logging.getLogger(__name__) 

27 

28 

29class MediaPlayerTransport(Transport): 

30 name = TRANSPORT_MEDIA 

31 

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

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

34 

35 @property 

36 def supported_features(self) -> TransportFeature: 

37 return TransportFeature.IMAGES | TransportFeature.VIDEO 

38 

39 @property 

40 def default_config(self) -> TransportConfig: 

41 config = TransportConfig() 

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

43 config.delivery_defaults.options = { 

44 OPTION_TARGET_SELECT: [RE_VALID_MEDIA_PLAYER], 

45 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID], 

46 } 

47 return config 

48 

49 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002 

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

51 

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

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

54 media_type: str = data.get("media_content_type", "image") 

55 if not media_players: 

56 _LOGGER.debug("SUPERNOTIFY skipping media show, no targets") 

57 return False 

58 

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

60 if snapshot_url is None: 

61 # fallback to older idiosyncratic way for backward compatibility 

62 snapshot_url = data.get(ATTR_MEDIA_SNAPSHOT_URL) 

63 if snapshot_url is None: 

64 _LOGGER.debug("SUPERNOTIFY skipping media player, no snapshot url") 

65 return False 

66 # absolutize relative URL for external URl, probably preferred by Alexa Show etc 

67 snapshot_url = urllib.parse.urljoin(self.hass_api.external_url, snapshot_url) 

68 

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

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

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

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

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

74 

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