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

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING, Any 

5 

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

7from homeassistant.components.tts.const import ATTR_CACHE, ATTR_LANGUAGE, ATTR_OPTIONS 

8from homeassistant.const import ATTR_ENTITY_ID 

9 

10from custom_components.supernotify.const import ( 

11 ATTR_MOBILE_APP_ID, 

12 MANUFACTURER_APPLE, 

13 OPTION_DEVICE_DISCOVERY, 

14 OPTION_DEVICE_DOMAIN, 

15 OPTION_DEVICE_MANUFACTURER_SELECT, 

16 OPTION_MESSAGE_USAGE, 

17 OPTION_SIMPLIFY_TEXT, 

18 OPTION_STRIP_URLS, 

19 OPTION_TARGET_CATEGORIES, 

20 OPTION_TARGET_SELECT, 

21 OPTION_TTS_ENTITY_ID, 

22 SELECT_EXCLUDE, 

23 TRANSPORT_TTS, 

24) 

25from custom_components.supernotify.model import ( 

26 DebugTrace, 

27 MessageOnlyPolicy, 

28 SelectionRule, 

29 Target, 

30 TargetRequired, 

31 TransportConfig, 

32 TransportFeature, 

33) 

34from custom_components.supernotify.schema import SelectionRank 

35from custom_components.supernotify.transport import Transport 

36 

37if TYPE_CHECKING: 

38 from custom_components.supernotify.envelope import Envelope 

39 from custom_components.supernotify.hass_api import DeviceInfo 

40 

41_LOGGER = logging.getLogger(__name__) 

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

43RE_MOBILE_APP = r"(notify\.)?mobile_app_[a-z0-9_]+" 

44ATTR_MEDIA_PLAYER_ENTITY_ID = "media_player_entity_id" # mypy flags up import from tts 

45 

46 

47class TTSTransport(Transport): 

48 """Notify via Home Assistant's built-in tts.speak action 

49 

50 options: 

51 message_usage: standard | use_title | combine_title 

52 

53 """ 

54 

55 name = TRANSPORT_TTS 

56 

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

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

59 

60 @property 

61 def supported_features(self) -> TransportFeature: 

62 return TransportFeature.MESSAGE | TransportFeature.SPOKEN 

63 

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

65 """Allow default action to be overridden, such as tts.say or tts.cloud_speak""" 

66 return action is not None 

67 

68 @property 

69 def default_config(self) -> TransportConfig: 

70 config = TransportConfig() 

71 config.delivery_defaults.action = "tts.speak" 

72 config.delivery_defaults.target_required = TargetRequired.ALWAYS 

73 config.delivery_defaults.selection_rank = SelectionRank.FIRST 

74 config.delivery_defaults.options = { 

75 OPTION_SIMPLIFY_TEXT: True, 

76 OPTION_STRIP_URLS: True, 

77 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

78 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID, ATTR_MOBILE_APP_ID], 

79 OPTION_TARGET_SELECT: [RE_VALID_MEDIA_PLAYER, RE_MOBILE_APP], 

80 OPTION_TTS_ENTITY_ID: "tts.home_assistant_cloud", 

81 OPTION_DEVICE_DISCOVERY: False, 

82 OPTION_DEVICE_DOMAIN: ["mobile_app"], 

83 OPTION_DEVICE_MANUFACTURER_SELECT: {SELECT_EXCLUDE: [MANUFACTURER_APPLE]}, 

84 } 

85 return config 

86 

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

88 _LOGGER.debug("SUPERNOTIFY tts: %s", envelope.message) 

89 

90 delivered: bool = False 

91 

92 media_player_targets = envelope.target.entity_ids or [] 

93 if media_player_targets: 

94 delivered = await self.call_media_players(envelope, media_player_targets) 

95 

96 mobile_targets = envelope.target.mobile_app_ids or [] 

97 if mobile_targets: 

98 if await self.call_mobile_apps(envelope, mobile_targets): 

99 delivered = True 

100 return delivered 

101 

102 async def call_media_players(self, envelope: Envelope, targets: list[str]) -> bool: 

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

104 if ATTR_LANGUAGE in envelope.data: 

105 action_data[ATTR_LANGUAGE] = envelope.data[ATTR_LANGUAGE] 

106 if ATTR_CACHE in envelope.data: 

107 action_data[ATTR_CACHE] = envelope.data[ATTR_CACHE] 

108 if ATTR_OPTIONS in envelope.data: 

109 action_data[ATTR_OPTIONS] = envelope.data[ATTR_OPTIONS] 

110 target_data: dict[str, Any] = {ATTR_ENTITY_ID: envelope.delivery.options.get(OPTION_TTS_ENTITY_ID)} 

111 

112 if targets and len(targets) == 1: 

113 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets[0] 

114 else: 

115 # despite the docs, the tts code accepts a list of media_player entity ids 

116 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets 

117 

118 return await self.call_action(envelope, action_data=action_data, target_data=target_data) 

119 

120 async def call_mobile_apps(self, envelope: Envelope, targets: list[str]) -> bool: 

121 action_data: dict[str, Any] = {ATTR_MESSAGE: "TTS", ATTR_DATA: {"tts_text": envelope.message or ""}} 

122 if "media_stream" in envelope.data: 

123 action_data["media_stream"] = envelope.data["media_stream"] 

124 

125 manufacturer_filter = SelectionRule(envelope.delivery.options.get(OPTION_DEVICE_MANUFACTURER_SELECT)) 

126 at_least_one: bool = False 

127 for target in targets: 

128 mobile_info: DeviceInfo | None = self.context.hass_api.mobile_app_by_id(target) 

129 if not mobile_info or not manufacturer_filter.match(mobile_info.manufacturer): 

130 _LOGGER.debug("SUPERNOTIFY Skipping tts target excluded by manufacturer filter: %s", mobile_info) 

131 else: 

132 full_target = target if Target.is_notify_entity(target) else f"notify.{target}" 

133 if await self.call_action(envelope, qualified_action=full_target, action_data=action_data, implied_target=True): 

134 at_least_one = True 

135 return at_least_one