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

1import logging 

2import urllib.parse 

3from typing import Any 

4 

5from homeassistant.const import ( 

6 ATTR_ENTITY_ID, 

7) 

8 

9from custom_components.supernotify import OPTION_TARGET_CATEGORIES, OPTION_TARGET_INCLUDE_RE, TRANSPORT_MEDIA 

10from custom_components.supernotify.envelope import Envelope 

11from custom_components.supernotify.model import TransportConfig 

12from custom_components.supernotify.transport import Transport 

13 

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

15 

16_LOGGER = logging.getLogger(__name__) 

17 

18 

19class MediaPlayerTransport(Transport): 

20 name = TRANSPORT_MEDIA 

21 

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

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

24 

25 @property 

26 def default_config(self) -> TransportConfig: 

27 config = TransportConfig() 

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

29 config.delivery_defaults.options = { 

30 OPTION_TARGET_INCLUDE_RE: [RE_VALID_MEDIA_PLAYER], 

31 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID], 

32 } 

33 return config 

34 

35 async def deliver(self, envelope: Envelope) -> bool: 

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

37 

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

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

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

41 if not media_players: 

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

43 return False 

44 

45 snapshot_url = data.get("snapshot_url") 

46 if snapshot_url is None: 

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

48 return False 

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

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

51 

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

53 if data and data.get("data"): 

54 action_data["extra"] = data.get("data", {}) 

55 

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