Coverage for custom_components/supernotify/transports/webostv.py: 98%

1"""LG webOS TV transport for SuperNotify. 

2 

3Shows a toast on LG webOS TVs set up with Home Assistant's `webostv` integration. That 

4integration has a legacy notify action for each TV, named after its config entry, such as 

5`notify.living_room_tv`, rather than a notify entity. So the TV's `media_player` entity is the 

6target, and each one is mapped to its notify action through its config entry. As an entity, 

7it can be reached by area, floor, label or device too. 

8 

9Only the message is shown: the title is used only when there is no message. 

10 

11TVs that are off, or unavailable, are skipped, since the integration can't reach them and 

12would otherwise raise an error. 

13 

14Supported data keys (optional): 

15 icon str path to an image file on the Home Assistant host, shown as the toast's icon 

16""" 

17 

18from __future__ import annotations 

19 

20import logging 

21from typing import TYPE_CHECKING, Any 

22 

23from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE 

24from homeassistant.const import ATTR_ICON, STATE_OFF, STATE_UNAVAILABLE 

25from homeassistant.helpers.typing import ConfigType 

26from homeassistant.util import slugify 

27 

28from custom_components.supernotify.const import TRANSPORT_WEBOSTV 

29from custom_components.supernotify.model import DebugTrace, SuppressionReason, TransportFeature 

30from custom_components.supernotify.target import TargetEntityCategory 

31from custom_components.supernotify.transport import Transport 

32 

33if TYPE_CHECKING: 

34 from custom_components.supernotify.envelope import Envelope 

35 from custom_components.supernotify.hass_api import HomeAssistantAPI 

36 

37_LOGGER = logging.getLogger(__name__) 

38 

39HA_WEBOSTV_DOMAIN = "webostv" 

40 

41 

42class WebOsTvTransport(Transport): 

43 """Notify via a toast on LG webOS TVs, using the webostv integration""" 

44 

45 name = TRANSPORT_WEBOSTV 

46 integration_domain = HA_WEBOSTV_DOMAIN 

47 

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

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

50 

51 @property 

52 def supported_features(self) -> TransportFeature: 

53 return TransportFeature.MESSAGE 

54 

55 @property 

56 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

57 return [TargetEntityCategory(domain="media_player", platform=HA_WEBOSTV_DOMAIN)] 

58 

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

60 return hass_api.find_config_entry_data(HA_WEBOSTV_DOMAIN) is not None 

61 

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

63 return {self.name: {}} 

64 

65 def validate_action(self, action: str | None) -> bool: 

66 """The action is found for each TV, not configured""" 

67 return action is None 

68 

69 def notify_action(self, entity_id: str) -> str | None: 

70 """The TV's legacy notify action, named by the integration after its config entry""" 

71 title: str | None = self.hass_api.config_entry_title_for_entity(entity_id) 

72 if not title: 

73 return None 

74 service: str = slugify(title) 

75 return f"notify.{service}" if self.hass_api.has_service("notify", service) else None 

76 

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

78 _LOGGER.debug("SUPERNOTIFY webostv %s", envelope.message) 

79 

80 action_data: dict[str, Any] = {ATTR_MESSAGE: envelope.message or ""} 

81 icon: str | None = (envelope.data or {}).get(ATTR_ICON) 

82 if icon: 

83 action_data[ATTR_DATA] = {ATTR_ICON: icon} 

84 

85 sent: int = 0 

86 for entity_id in envelope.target.entity_ids: 

87 state = self.hass_api.get_state(entity_id) 

88 if state is not None and state.state in (STATE_OFF, STATE_UNAVAILABLE): 

89 _LOGGER.debug("SUPERNOTIFY webostv skipping %s, which is %s", entity_id, state.state) 

90 continue 

91 action: str | None = self.notify_action(entity_id) 

92 if action is None: 

93 _LOGGER.warning("SUPERNOTIFY webostv no notify action found for %s", entity_id) 

94 continue 

95 if await self.call_action( 

96 envelope, qualified_action=action, action_data=dict(action_data), implied_target=True, single_target=True 

97 ): 

98 sent += 1 

99 

100 if not sent and not envelope.error_count: 

101 envelope.skipped = 1 

102 envelope.skip_reason = SuppressionReason.NO_TARGET 

103 return sent > 0