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

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING, Any, ClassVar 

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 

9from homeassistant.helpers.typing import ConfigType 

10 

11from custom_components.supernotify.const import ( 

12 ATTR_MOBILE_APP_ID, 

13 MANUFACTURER_APPLE, 

14 RE_MEDIA_PLAYER_ENTITY_ID, 

15 TRANSPORT_TTS, 

16) 

17from custom_components.supernotify.model import ( 

18 DebugTrace, 

19 MessageOnlyPolicy, 

20 SelectionRule, 

21 SuppressionReason, 

22 TransportConfig, 

23 TransportFeature, 

24) 

25from custom_components.supernotify.options import ( 

26 OPTION_DEVICE_DISCOVERY, 

27 OPTION_DEVICE_DOMAIN, 

28 OPTION_DEVICE_MANUFACTURER_SELECT, 

29 OPTION_MESSAGE_USAGE, 

30 OPTION_SIMPLIFY_TEXT, 

31 OPTION_STRIP_URLS, 

32 OPTION_TARGET_SELECT, 

33 SELECT_EXCLUDE, 

34 DeliveryOption, 

35) 

36from custom_components.supernotify.target import Target, TargetEntityCategory 

37from custom_components.supernotify.transport import Transport 

38 

39if TYPE_CHECKING: 

40 from custom_components.supernotify.envelope import Envelope 

41 from custom_components.supernotify.hass_api import HomeAssistantAPI, TrackedDeviceDetails 

42 

43_LOGGER = logging.getLogger(__name__) 

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

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

46OPTION_TTS_ENTITY_ID = "tts_entity_id" 

47# used when no `tts_entity_id` is configured, if present, else the first tts entity found 

48DEFAULT_TTS_ENTITY_ID = "tts.home_assistant_cloud" 

49 

50 

51class TTSTransport(Transport): 

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

53 

54 options: 

55 message_usage: standard | use_title | combine_title | message_only 

56 

57 """ 

58 

59 name = TRANSPORT_TTS 

60 integration_domain = "tts" 

61 declared_options: ClassVar[list[DeliveryOption]] = [ 

62 DeliveryOption(OPTION_TTS_ENTITY_ID, "The tts entity used to generate speech"), 

63 ] 

64 

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

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

67 

68 @property 

69 def supported_features(self) -> TransportFeature: 

70 return TransportFeature.MESSAGE | TransportFeature.SPOKEN 

71 

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

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

74 return action is not None 

75 

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

77 if not hass_api.has_service("tts", "speak"): 

78 _LOGGER.debug("SUPERNOTIFY No tts.speak action available, `tts` transport not configured") 

79 return False 

80 if not hass_api.entity_ids_for_domain("media_player"): 

81 _LOGGER.debug("SUPERNOTIFY No media players available, `tts` transport not configured") 

82 return False 

83 return True 

84 

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

86 return {self.name: {}} 

87 

88 @property 

89 def default_config(self) -> TransportConfig: 

90 config = super().default_config 

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

92 # v1.8.1 added a FIRST selection rank here, remove for v2.11 since it seemed obsolete 

93 config.delivery_defaults.options = { 

94 OPTION_SIMPLIFY_TEXT: True, 

95 OPTION_STRIP_URLS: True, 

96 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

97 OPTION_TARGET_SELECT: [RE_MEDIA_PLAYER_ENTITY_ID, RE_MOBILE_APP], 

98 OPTION_DEVICE_DISCOVERY: False, 

99 OPTION_DEVICE_DOMAIN: ["mobile_app"], 

100 OPTION_DEVICE_MANUFACTURER_SELECT: {SELECT_EXCLUDE: [MANUFACTURER_APPLE]}, 

101 } 

102 return config 

103 

104 @property 

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

106 # media_player is ambiguous the same way as kodi/media/alexa_media_player's; mobile_app_id 

107 # is already mobile_push's unique category, but tts using it too (to speak via the app's 

108 # own TTS bridge, not to push-notify) doesn't drive auto-selection from `other` 

109 return [TargetEntityCategory(domain="media_player"), ATTR_MOBILE_APP_ID] 

110 

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

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

113 

114 delivered: bool = False 

115 

116 media_player_targets = envelope.target.entity_ids or [] 

117 if media_player_targets: 

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

119 

120 mobile_targets = envelope.target.mobile_app_ids or [] 

121 if mobile_targets and await self.call_mobile_apps(envelope, mobile_targets): 

122 delivered = True 

123 return delivered 

124 

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

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

127 if ATTR_LANGUAGE in envelope.data: 

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

129 if ATTR_CACHE in envelope.data: 

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

131 if ATTR_OPTIONS in envelope.data: 

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

133 if (envelope.delivery.action or "").partition(".")[2] != "speak": 

134 # legacy `<platform>_say` actions, like `tts.cloud_say`, target the media players directly 

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

136 

137 tts_entity_id: str | None = self.tts_entity_id(envelope) 

138 if tts_entity_id is None: 

139 _LOGGER.warning("SUPERNOTIFY tts: no tts entity found, set the `%s` option", OPTION_TTS_ENTITY_ID) 

140 envelope.skipped = 1 

141 envelope.skip_reason = SuppressionReason.NO_TARGET 

142 return False 

143 target_data: dict[str, Any] = {ATTR_ENTITY_ID: tts_entity_id} 

144 

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

146 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets[0] 

147 else: 

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

149 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets 

150 

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

152 

153 def tts_entity_id(self, envelope: Envelope) -> str | None: 

154 """The configured tts entity, or else Nabu Casa's if there is one, or else the first found""" 

155 configured: str | None = envelope.delivery.options.get(OPTION_TTS_ENTITY_ID) 

156 if configured: 

157 return configured 

158 available: list[str] = self.hass_api.entity_ids_for_domain("tts") 

159 if DEFAULT_TTS_ENTITY_ID in available: 

160 return DEFAULT_TTS_ENTITY_ID 

161 return available[0] if available else None 

162 

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

164 # "TTS" is the Companion App's trigger for speaking `tts_text`, not text to translate 

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

166 if "media_stream" in envelope.data: 

167 action_data[ATTR_DATA]["media_stream"] = envelope.data["media_stream"] 

168 

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

170 at_least_one: bool = False 

171 for target in targets: 

172 mobile_info: TrackedDeviceDetails | None = self.context.hass_api.mobile_app_by_id(target) 

173 if not mobile_info: 

174 _LOGGER.debug("SUPERNOTIFY Skipping unknown tts mobile app target: %s", target) 

175 elif not manufacturer_filter.match(mobile_info.manufacturer): 

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

177 else: 

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

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

180 at_least_one = True 

181 return at_least_one