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

1import logging 

2from typing import TYPE_CHECKING, Any 

3 

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

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

6from homeassistant.const import ATTR_ENTITY_ID 

7 

8from custom_components.supernotify import ( 

9 ATTR_MOBILE_APP_ID, 

10 OPTION_DEVICE_DISCOVERY, 

11 OPTION_DEVICE_DOMAIN, 

12 OPTION_DEVICE_MANUFACTURER_SELECT, 

13 OPTION_MESSAGE_USAGE, 

14 OPTION_SIMPLIFY_TEXT, 

15 OPTION_STRIP_URLS, 

16 OPTION_TARGET_CATEGORIES, 

17 OPTION_TARGET_SELECT, 

18 OPTION_TTS_ENTITY_ID, 

19 SELECT_EXCLUDE, 

20 TRANSPORT_TTS, 

21 SelectionRank, 

22) 

23from custom_components.supernotify.envelope import Envelope 

24from custom_components.supernotify.model import ( 

25 DebugTrace, 

26 MessageOnlyPolicy, 

27 Target, 

28 TargetRequired, 

29 TransportConfig, 

30 TransportFeature, 

31) 

32from custom_components.supernotify.transport import Transport 

33 

34if TYPE_CHECKING: 

35 from custom_components.supernotify.hass_api import DeviceInfo 

36 

37_LOGGER = logging.getLogger(__name__) 

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

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

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

41 

42 

43class TTSTransport(Transport): 

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

45 

46 options: 

47 message_usage: standard | use_title | combine_title 

48 

49 """ 

50 

51 name = TRANSPORT_TTS 

52 

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

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

55 

56 @property 

57 def supported_features(self) -> TransportFeature: 

58 return TransportFeature.MESSAGE 

59 

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

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

62 return action is not None 

63 

64 @property 

65 def default_config(self) -> TransportConfig: 

66 config = TransportConfig() 

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

68 config.delivery_defaults.target_required = TargetRequired.ALWAYS 

69 config.delivery_defaults.selection_rank = SelectionRank.FIRST 

70 config.delivery_defaults.options = { 

71 OPTION_SIMPLIFY_TEXT: True, 

72 OPTION_STRIP_URLS: True, 

73 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

74 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID, ATTR_MOBILE_APP_ID], 

75 OPTION_TARGET_SELECT: [RE_VALID_MEDIA_PLAYER, RE_MOBILE_APP], 

76 OPTION_TTS_ENTITY_ID: "tts.home_assistant_cloud", 

77 OPTION_DEVICE_DISCOVERY: False, 

78 OPTION_DEVICE_DOMAIN: ["mobile_app"], 

79 OPTION_DEVICE_MANUFACTURER_SELECT: {SELECT_EXCLUDE: ["Apple"]}, 

80 } 

81 return config 

82 

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

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

85 

86 delivered: bool = False 

87 

88 media_player_targets = envelope.target.entity_ids or [] 

89 if media_player_targets: 

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

91 

92 mobile_targets = envelope.target.mobile_app_ids or [] 

93 if mobile_targets: 

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

95 delivered = True 

96 return delivered 

97 

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

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

100 if ATTR_LANGUAGE in envelope.data: 

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

102 if ATTR_CACHE in envelope.data: 

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

104 if ATTR_OPTIONS in envelope.data: 

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

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

107 

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

109 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets[0] 

110 else: 

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

112 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = targets 

113 

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

115 

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

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

118 if "media_stream" in envelope.data: 

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

120 

121 at_least_one: bool = False 

122 for target in targets: 

123 bare_target = target.replace("notify.", "", 1) if target.startswith("notify.") else target 

124 mobile_info: DeviceInfo | None = self.context.hass_api.mobile_app_by_id(bare_target) 

125 if not mobile_info or mobile_info.manufacturer == "Apple": 

126 _LOGGER.debug("SUPERNOTIFY Skipping tts target that isn't confirmed as android: %s", mobile_info) 

127 else: 

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

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

130 at_least_one = True 

131 return at_least_one