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

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 TargetRequired, 

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.schema import SelectionRank 

37from custom_components.supernotify.target import Target, TargetEntityCategory 

38from custom_components.supernotify.transport import Transport 

39 

40if TYPE_CHECKING: 

41 from custom_components.supernotify.envelope import Envelope 

42 from custom_components.supernotify.hass_api import HomeAssistantAPI, TrackedDeviceDetails 

43 

44_LOGGER = logging.getLogger(__name__) 

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

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

47OPTION_TTS_ENTITY_ID = "tts_entity_id" 

48 

49 

50class TTSTransport(Transport): 

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

52 

53 options: 

54 message_usage: standard | use_title | combine_title 

55 

56 """ 

57 

58 name = TRANSPORT_TTS 

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

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

61 ] 

62 

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

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

65 

66 @property 

67 def supported_features(self) -> TransportFeature: 

68 return TransportFeature.MESSAGE | TransportFeature.SPOKEN 

69 

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

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

72 return action is not None 

73 

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

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

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

77 return False 

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

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

80 return False 

81 return True 

82 

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

84 return {self.name: {}} 

85 

86 @property 

87 def default_config(self) -> TransportConfig: 

88 config = TransportConfig() 

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

90 config.delivery_defaults.target_required = TargetRequired.ALWAYS 

91 config.delivery_defaults.selection_rank = SelectionRank.FIRST 

92 config.delivery_defaults.inclusion = self.inclusion_mode 

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_TTS_ENTITY_ID: "tts.home_assistant_cloud", 

99 OPTION_DEVICE_DISCOVERY: False, 

100 OPTION_DEVICE_DOMAIN: ["mobile_app"], 

101 OPTION_DEVICE_MANUFACTURER_SELECT: {SELECT_EXCLUDE: [MANUFACTURER_APPLE]}, 

102 } 

103 return config 

104 

105 @property 

106 def target_categories(self) -> list[str | TargetEntityCategory]: 

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

108 

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

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

111 

112 delivered: bool = False 

113 

114 media_player_targets = envelope.target.entity_ids or [] 

115 if media_player_targets: 

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

117 

118 mobile_targets = envelope.target.mobile_app_ids or [] 

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

120 delivered = True 

121 return delivered 

122 

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

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

125 if ATTR_LANGUAGE in envelope.data: 

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

127 if ATTR_CACHE in envelope.data: 

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

129 if ATTR_OPTIONS in envelope.data: 

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

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

132 

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

134 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets[0] 

135 else: 

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

137 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets 

138 

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

140 

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

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

143 if "media_stream" in envelope.data: 

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

145 

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

147 at_least_one: bool = False 

148 for target in targets: 

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

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

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

152 else: 

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

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

155 at_least_one = True 

156 return at_least_one