Coverage for custom_components/supernotify/methods/chime.py: 93%

1import logging 

2import re 

3from typing import TYPE_CHECKING, Any 

4 

5from homeassistant.components.group import expand_entity_ids 

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

7from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues 

8 ATTR_ENTITY_ID, 

9 CONF_VARIABLES, 

10) 

11 

12from custom_components.supernotify import ( 

13 ATTR_DATA, 

14 ATTR_PRIORITY, 

15 CONF_DATA, 

16 CONF_DEVICE_DOMAIN, 

17 CONF_TARGETS_REQUIRED, 

18 METHOD_CHIME, 

19) 

20from custom_components.supernotify.common import ensure_list 

21from custom_components.supernotify.delivery_method import DeliveryMethod 

22from custom_components.supernotify.envelope import Envelope 

23 

24if TYPE_CHECKING: 

25 from homeassistant.helpers.device_registry import DeviceEntry 

26 

27RE_VALID_CHIME = r"(switch|script|group|siren|media_player)\.[A-Za-z0-9_]+" 

28 

29_LOGGER = logging.getLogger(__name__) 

30 

31DATA_SCHEMA_RESTRICT: dict[str, list[str]] = { 

32 "media_player": ["data", "entity_id", "media_content_id", "media_content_type", "enqueue", "announce"], 

33 "switch": ["entity_id"], 

34 "script": ["data", "variables", "context", "wait"], 

35 "siren": ["data", "entity_id"], 

36 "alexa_devices": ["sound", "device_id"], 

37} # TODO: source directly from component schema 

38DEVICE_DOMAINS = ["alexa_devices"] 

39 

40 

41class ChimeTargetConfig: 

42 def __init__( 

43 self, 

44 target: str, 

45 tune: str | None = None, 

46 duration: int | None = None, 

47 volume: float | None = None, 

48 data: dict[str, Any] | None = None, 

49 domain: str | None = None, 

50 **kwargs: Any, 

51 ) -> None: 

52 self.entity_id: str | None = None 

53 self.device_id: str | None = None 

54 self.domain: str | None = None 

55 self.entity_name: str | None = None 

56 if "." in target: 

57 self.entity_id = target 

58 self.domain, self.entity_name = target.split(".", 1) 

59 else: 

60 if self.is_device(target): 

61 self.device_id = target 

62 self.domain = domain 

63 else: 

64 raise ValueError(f"ChimeTargetConfig target must be entity_id or device_id: {target}") 

65 if kwargs: 

66 _LOGGER.warning("SUPERNOTIFY ChimeTargetConfig ignoring unexpected args: %s", kwargs) 

67 self.volume: float | None = volume 

68 self.tune: str | None = tune 

69 self.duration: int | None = duration 

70 self.data: dict[str, Any] | None = data or {} 

71 

72 def __repr__(self) -> str: 

73 """Return a developer-oriented string representation of this ChimeTargetConfig""" 

74 if self.device_id is not None: 

75 return f"ChimeTargetConfig(device_id={self.device_id})" 

76 return f"ChimeTargetConfig(entity_id={self.entity_id})" 

77 

78 @classmethod 

79 def is_device(cls, target: str) -> bool: 

80 return re.match(r"^[0-9a-f]{32}$", target) is not None 

81 

82 

83class ChimeDeliveryMethod(DeliveryMethod): 

84 method = METHOD_CHIME 

85 

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

87 kwargs.setdefault(CONF_TARGETS_REQUIRED, False) 

88 # support optional auto discovery 

89 kwargs.setdefault(CONF_DEVICE_DOMAIN, DEVICE_DOMAINS) 

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

91 

92 @property 

93 def chime_aliases(self) -> dict[str, Any]: 

94 return self.default_options.get("chime_aliases") or {} 

95 

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

97 return action is None 

98 

99 def select_target(self, target: str) -> bool: 

100 return re.fullmatch(RE_VALID_CHIME, target) is not None or ChimeTargetConfig.is_device(target) 

101 

102 async def deliver(self, envelope: Envelope) -> bool: 

103 config = self.delivery_config(envelope.delivery_name) 

104 data: dict[str, Any] = {} 

105 data.update(config.get(CONF_DATA) or {}) 

106 data.update(envelope.data or {}) 

107 targets = envelope.targets or [] 

108 

109 # chime_repeat = data.pop("chime_repeat", 1) 

110 chime_tune: str | None = data.pop("chime_tune", None) 

111 chime_volume: float | None = data.pop("chime_volume", None) 

112 chime_duration: int | None = data.pop("chime_duration", None) 

113 

114 _LOGGER.info( 

115 "SUPERNOTIFY notify_chime: %s -> %s (delivery: %s, env_data:%s, dlv_data:%s)", 

116 chime_tune, 

117 targets, 

118 envelope.delivery_name, 

119 envelope.data, 

120 config.get(CONF_DATA), 

121 ) 

122 # expand groups 

123 expanded_targets = { 

124 e: ChimeTargetConfig(tune=chime_tune, volume=chime_volume, duration=chime_duration, target=e) 

125 for e in expand_entity_ids(self.hass, targets) 

126 } 

127 # resolve and include chime aliases 

128 expanded_targets.update(self.resolve_tune(chime_tune)) # overwrite and extend 

129 

130 chimes = 0 

131 for chime_entity_config in expanded_targets.values(): 

132 _LOGGER.debug("SUPERNOTIFY chime %s: %s", chime_entity_config.entity_id, chime_entity_config.tune) 

133 action_data = None 

134 try: 

135 domain, service, action_data = self.analyze_target(chime_entity_config, data, envelope) 

136 if domain is not None and service is not None: 

137 action_data = self.prune_data(domain, action_data) 

138 

139 if await self.call_action(envelope, qualified_action=f"{domain}.{service}", action_data=action_data): 

140 chimes += 1 

141 else: 

142 _LOGGER.debug("SUPERNOTIFY Chime skipping incomplete service for %s", chime_entity_config.entity_id) 

143 except Exception: 

144 _LOGGER.exception("SUPERNOTIFY Failed to chime %s: %s [%s]", chime_entity_config.entity_id, action_data) 

145 return chimes > 0 

146 

147 def prune_data(self, domain: str, data: dict[str, Any]) -> dict[str, Any]: 

148 pruned: dict[str, Any] = {} 

149 if data and domain in DATA_SCHEMA_RESTRICT: 

150 restrict: list[str] = DATA_SCHEMA_RESTRICT.get(domain) or [] 

151 for key in list(data.keys()): 

152 if key in restrict: 

153 pruned[key] = data[key] 

154 return pruned 

155 

156 def analyze_target( 

157 self, target_config: ChimeTargetConfig, data: dict[str, Any], envelope: Envelope 

158 ) -> tuple[str | None, str | None, dict[str, Any]]: 

159 if not target_config.entity_id and not target_config.device_id: 

160 _LOGGER.warning("SUPERNOTIFY Empty chime target") 

161 return "", None, {} 

162 

163 domain: str | None = None 

164 name: str | None = None 

165 

166 # Alexa Devices use device_id not entity_id for sounds 

167 if target_config.device_id is not None: 

168 if target_config.domain is not None: 

169 domain = target_config.domain 

170 else: 

171 # discover domain from device registry 

172 device_registry = self.context.device_registry() 

173 if device_registry: 

174 device: DeviceEntry | None = device_registry.async_get(target_config.device_id) 

175 if device and "alexa_devices" in [d for d, _id in device.identifiers]: 

176 domain = "alexa_devices" 

177 if domain is None: 

178 _LOGGER.warning( 

179 "SUPERNOTIFY A target that looks like a device_id can't be matched to supported integration: %s", 

180 target_config.device_id, 

181 ) 

182 elif target_config.entity_id and "." in target_config.entity_id: 

183 domain, name = target_config.entity_id.split(".", 1) 

184 

185 action_data: dict[str, Any] = {} 

186 action: str | None = None 

187 

188 if domain == "switch": 

189 action = "turn_on" 

190 action_data[ATTR_ENTITY_ID] = target_config.entity_id 

191 

192 elif domain == "siren": 

193 action = "turn_on" 

194 action_data[ATTR_ENTITY_ID] = target_config.entity_id 

195 action_data[ATTR_DATA] = {} 

196 if target_config.tune: 

197 action_data[ATTR_DATA]["tone"] = target_config.tune 

198 if target_config.duration is not None: 

199 action_data[ATTR_DATA]["duration"] = target_config.duration 

200 if target_config.volume is not None: 

201 action_data[ATTR_DATA]["volume_level"] = target_config.volume 

202 

203 elif domain == "script": 

204 action_data.setdefault(CONF_VARIABLES, {}) 

205 if target_config.data: 

206 action_data[CONF_VARIABLES] = target_config.data.get(CONF_VARIABLES, {}) 

207 if data: 

208 # override data sourced from chime alias with explicit variables in envelope/data 

209 action_data[CONF_VARIABLES].update(data.get(CONF_VARIABLES, {})) 

210 action = name 

211 action_data[CONF_VARIABLES][ATTR_MESSAGE] = envelope.message 

212 action_data[CONF_VARIABLES][ATTR_TITLE] = envelope.title 

213 action_data[CONF_VARIABLES][ATTR_PRIORITY] = envelope.priority 

214 action_data[CONF_VARIABLES]["chime_tune"] = target_config.tune 

215 action_data[CONF_VARIABLES]["chime_volume"] = target_config.volume 

216 action_data[CONF_VARIABLES]["chime_duration"] = target_config.duration 

217 

218 elif domain == "alexa_devices" and target_config.tune: 

219 action = "send_sound" 

220 action_data["device_id"] = target_config.device_id 

221 action_data["sound"] = target_config.tune 

222 

223 elif domain == "media_player" and target_config.tune: 

224 if target_config.data: 

225 action_data.update(target_config.data) 

226 if data: 

227 action_data.update(data) 

228 action = "play_media" 

229 action_data[ATTR_ENTITY_ID] = target_config.entity_id 

230 action_data["media_content_type"] = "sound" 

231 action_data["media_content_id"] = target_config.tune 

232 

233 else: 

234 _LOGGER.warning( 

235 "SUPERNOTIFY No matching chime domain/tune: %s, target: %s, tune: %s", 

236 domain, 

237 target_config.entity_id, 

238 target_config.tune, 

239 ) 

240 

241 return domain, action, action_data 

242 

243 def resolve_tune(self, tune_or_alias: str | None) -> dict[str, ChimeTargetConfig]: 

244 target_configs: dict[str, ChimeTargetConfig] = {} 

245 if tune_or_alias is not None: 

246 for domain, alias_config in self.chime_aliases.get(tune_or_alias, {}).items(): 

247 if isinstance(alias_config, str): 

248 tune = alias_config 

249 alias_config = {} 

250 else: 

251 tune = alias_config.get("tune", tune_or_alias) 

252 

253 alias_config["tune"] = tune 

254 alias_config.setdefault("domain", domain) 

255 alias_config.setdefault("data", {}) 

256 target = alias_config.pop("target", None) 

257 

258 # pass through variables or data if present 

259 if target is not None: 

260 target_configs.update({t: ChimeTargetConfig(target=t, **alias_config) for t in ensure_list(target)}) # type: ignore 

261 elif domain in DEVICE_DOMAINS: 

262 # bulk apply to all known target devices of this domain 

263 bulk_apply = { 

264 dev: ChimeTargetConfig(target=dev, **alias_config) # type: ignore 

265 for dev in self.targets 

266 if ChimeTargetConfig.is_device(dev) 

267 and dev not in target_configs # don't overwrite existing specific targets 

268 } 

269 target_configs.update(bulk_apply) 

270 else: 

271 # bulk apply to all known target entities of this domain 

272 bulk_apply = { 

273 ent: ChimeTargetConfig(target=ent, **alias_config) # type: ignore 

274 for ent in self.targets 

275 if ent.startswith(f"{alias_config['domain']}.") 

276 and ent not in target_configs # don't overwrite existing specific targets 

277 } 

278 target_configs.update(bulk_apply) 

279 _LOGGER.debug("SUPERNOTIFY method_chime: Resolved tune %s to %s", tune_or_alias, target_configs) 

280 return target_configs