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

1from __future__ import annotations

3import logging

4from dataclasses import dataclass, field

5from typing import TYPE_CHECKING, Any, ClassVar

7from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN

8from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE

10# ATTR_VARIABLES from script.const has import issues

11from homeassistant.const import ATTR_ENTITY_ID

12from homeassistant.helpers import config_validation as cv

13from homeassistant.helpers.typing import ConfigType

14from yarl import URL

16from custom_components.supernotify.common import ensure_list

17from custom_components.supernotify.const import (

18 ATTR_ACTION_URL,

19 ATTR_ACTIONS,

20 ATTR_MEDIA,

21 ATTR_MEDIA_SNAPSHOT_URL,

22 ATTR_PRIORITY,

23 PRIORITY_CRITICAL,

24 PRIORITY_HIGH,

25 PRIORITY_LOW,

26 PRIORITY_MEDIUM,

27 PRIORITY_MINIMUM,

28 PRIORITY_VALUES,

29 TRANSPORT_GENERIC,

30)

31from custom_components.supernotify.model import (

32 DataFilter,

33 DebugTrace,

34 MessageOnlyPolicy,

35 SuppressionReason,

36 Target,

37 TargetRequired,

38 TransportConfig,

39 TransportFeature,

40)

41from custom_components.supernotify.options import (

42 DATA_KEYS_SELECT_OPTION,

43 OPTION_DATA_KEYS_SELECT,

44 OPTION_MESSAGE_USAGE,

45 OPTION_SIMPLIFY_TEXT,

46 OPTION_STRIP_URLS,

47 OPTION_TARGET_CATEGORIES,

48 DeliveryOption,

49)

50from custom_components.supernotify.transport import (

51 Transport,

52)

54if TYPE_CHECKING:

55 from custom_components.supernotify.delivery import Delivery

56 from custom_components.supernotify.envelope import Envelope

57 from custom_components.supernotify.hass_api import HomeAssistantAPI

59_LOGGER = logging.getLogger(__name__)

61OPTION_RAW = "raw"

62ATTR_MEDIA_PLAYER_ENTITY_ID = "media_player_entity_id" # as in tts transport

63# actions handled here that now have their own transport, mapped to that transport

64DEPRECATED_ACTIONS: dict[str, str] = {"ntfy.publish": "ntfy"}

65OPTION_GENERIC_DOMAIN_STYLE = "handle_as_domain"

68class GenericTransport(Transport):

69 """Call any service, including non-notify ones, like switch.turn_on or mqtt.publish"""

71 name = TRANSPORT_GENERIC

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

73 DeliveryOption(OPTION_RAW, "Don't apply domain specific data handling and pruning rules", value_type=cv.boolean),

74 DeliveryOption(OPTION_GENERIC_DOMAIN_STYLE, "Treat the action call in the same way as a known domain"),

75 DATA_KEYS_SELECT_OPTION,

76 ]

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

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

81 @property

82 def supported_features(self) -> TransportFeature:

83 return TransportFeature.MESSAGE | TransportFeature.TITLE

85 @property

86 def default_config(self) -> TransportConfig:

87 config = super().default_config

88 # bring-your-own-categories: a delivery's own OPTION_TARGET_CATEGORIES is what it

89 # actually targets, invisible at the transport level, so this can't be inferred -

90 # one of the few transports where this needs to stay an explicit override

91 config.delivery_defaults.target_required = TargetRequired.OPTIONAL

92 config.delivery_defaults.options = {

93 OPTION_SIMPLIFY_TEXT: False,

94 OPTION_STRIP_URLS: False,

95 OPTION_RAW: False,

96 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,

97 OPTION_DATA_KEYS_SELECT: None,

98 OPTION_GENERIC_DOMAIN_STYLE: None,

99 }

100 return config

101

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

103 domain, _, service = (action or "").partition(".")

104 if domain and service:

105 replacement: str | None = DEPRECATED_ACTIONS.get(action or "")

106 if replacement:

107 _LOGGER.warning(

108 "SUPERNOTIFY Deprecated use of %s with the generic transport - use the %s transport", action, replacement

109 )

110 self.hass_api.raise_issue(

111 f"generic_{domain}_{service}_deprecated",

112 issue_key="generic_action_deprecated",

113 issue_map={"action": action or "", "transport": replacement},

114 learn_more_url=f"https://supernotify.rhizomatics.org.uk/transports/{replacement}/",

115 )

116 return True

117 _LOGGER.warning("SUPERNOTIFY Generic transport must have a qualified action name, e.g. notify.foo")

118 return False

119

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

121 # entirely delivery-driven (bring-your-own-action) - there's no transport-level

122 # prerequisite to check. A transport-level default action still gates whether

123 # build_standard_deliveries() below produces something usable; if not,

124 # DeliveryRegistry prunes this transport entirely once it's confirmed no delivery

125 # (explicit or auto) uses it

126 return True

127

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

129 # with no default action configured, there's nothing to auto-generate a delivery

130 # from - validate_action()/Delivery.initialize() reject it before it's ever used

131 action = self.delivery_defaults.action

132 if action is None or "." not in action:

133 return {}

134 return {self.name: {}}

135

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

137 # inputs - copied, since some domains rework it, and the envelope's own is archived

138 data: dict[str, Any] = dict(envelope.data or {})

139 core_action_data: dict[str, Any] = envelope.core_action_data(force_message=False)

140 raw_mode: bool = envelope.delivery.options.get(OPTION_RAW, False)

141 qualified_action: str | None = envelope.delivery.action

142 split_action = (

143 qualified_action.split(".", 1) if qualified_action and "." in qualified_action else [None, qualified_action]

144 )

145 domain: str | None = split_action[0]

146 service: str | None = split_action[1]

147

148 equiv_domain: str | None = domain

149 if envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE):

150 equiv_domain = envelope.delivery.options.get(OPTION_GENERIC_DOMAIN_STYLE)

151 _LOGGER.debug("SUPERNOTIFY Handling %s generic message as if it was %s", domain, equiv_domain)

152 # entities of the action's own domain, plus those of the domain it's handled as

153 entity_ids: list[str] = envelope.target.domain_entity_ids(domain)

154 if equiv_domain != domain:

155 entity_ids += envelope.target.domain_entity_ids(equiv_domain)

156

157 # outputs

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

159 target_data: dict[str, Any] | None = None

160 build_targets: bool = False

161 prune_data: bool = True

162 mini_envelopes: list[MiniEnvelope] = [] # only used for script and ntfy

163

164 if raw_mode:

165 action_data = core_action_data

166 action_data.update(data)

167 build_targets = True

168 elif equiv_domain == "notify":

169 action_data = core_action_data

170 if qualified_action == "notify.send_message":

171 # amongst the wild west of notifty handling, at least care for the modern core one

172 action_data = core_action_data

173 target_data = {ATTR_ENTITY_ID: envelope.target.domain_entity_ids(domain)}

174 prune_data = False

175 else:

176 action_data = core_action_data

177 action_data[ATTR_DATA] = data

178 build_targets = True

179 elif equiv_domain == "input_text":

180 target_data = {ATTR_ENTITY_ID: entity_ids}

181 if "value" in data:

182 action_data = {"value": data["value"]}

183 else:

184 action_data = {"value": core_action_data[ATTR_MESSAGE]}

185 elif equiv_domain == "switch":

186 target_data = {ATTR_ENTITY_ID: entity_ids}

187 elif equiv_domain == "mqtt":

188 action_data = data

189 if "payload" not in action_data:

190 action_data["payload"] = envelope.message

191 elif equiv_domain == "tts":

192 action_data = core_action_data

193 action_data.update(data)

194 media_players: list[str] = envelope.target.domain_entity_ids("media_player")

195 if service == "speak":

196 # tts entity as target, media players as data

197 target_data = {ATTR_ENTITY_ID: entity_ids}

198 if media_players and ATTR_MEDIA_PLAYER_ENTITY_ID not in action_data:

199 action_data[ATTR_MEDIA_PLAYER_ENTITY_ID] = media_players[0] if len(media_players) == 1 else media_players

200 else:

201 # legacy `<platform>_say` actions, targeting the media players directly

202 target_data = {ATTR_ENTITY_ID: media_players}

203 elif equiv_domain == "notify_events":

204 mini_envelopes.extend(

205 notify_events(envelope.message, envelope.title, core_action_data, data, envelope.delivery, envelope.priority)

206 )

207 elif qualified_action == "ntfy.publish":

208 mini_envelopes.extend(

209 ntfy(core_action_data, data, envelope.target, envelope.delivery, envelope.priority, self.hass_api)

210 )

211 elif equiv_domain in ("siren", "light"):

212 target_data = {ATTR_ENTITY_ID: entity_ids}

213 action_data = data

214 elif equiv_domain == "rest_command":

215 action_data = data

216 elif equiv_domain == "script":

217 mini_envelopes.extend(

218 script(qualified_action, core_action_data, data, envelope.target, envelope.delivery, self.hass_api)

219 )

220 else:

221 action_data = core_action_data

222 action_data.update(data)

223 build_targets = True

224

225 if mini_envelopes:

226 mini_envelopes = [m for m in mini_envelopes if not _missing_entities(m.target_data)]

227 if not mini_envelopes:

228 return self._skip_no_target(envelope, qualified_action)

229 results: list[bool] = [

230 await self.call_action(

231 envelope, qualified_action, action_data=mini_envelope.action_data, target_data=mini_envelope.target_data

232 )

233 for mini_envelope in mini_envelopes

234 ]

235 return all(results)

236

237 if _missing_entities(target_data):

238 return self._skip_no_target(envelope, qualified_action)

239

240 if build_targets:

241 all_targets: list[str] = []

242 if OPTION_TARGET_CATEGORIES in envelope.delivery.options:

243 for category in ensure_list(envelope.delivery.options.get(OPTION_TARGET_CATEGORIES, [])):

244 all_targets.extend(envelope.target.for_category(category))

245 else:

246 all_targets = envelope.target.resolved_targets()

247 if len(all_targets) == 1:

248 action_data[ATTR_TARGET] = all_targets[0]

249 elif len(all_targets) >= 1:

250 action_data[ATTR_TARGET] = all_targets

251

252 if prune_data and action_data:

253 action_data = envelope.customize_data(action_data)

254 if not raw_mode and domain and service:

255 # use the service schema to remove unsupported fields or force type

256 action_data = self.context.hass_api.coerce_schema(domain, service, action_data)

257

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

259

260 def _skip_no_target(self, envelope: Envelope, qualified_action: str | None) -> bool:

261 _LOGGER.debug(

262 "SUPERNOTIFY Skipping %s call to %s, no matching entity targets", envelope.delivery.name, qualified_action

263 )

264 envelope.skipped = 1

265 envelope.skip_reason = SuppressionReason.NO_TARGET

266 return False

267

268

269def _missing_entities(target_data: dict[str, Any] | None) -> bool:

270 """An entity-targeted call, with no entities - which Home Assistant would quietly do nothing with"""

271 return target_data is not None and ATTR_ENTITY_ID in target_data and not target_data[ATTR_ENTITY_ID]

272

273

274@dataclass

275class MiniEnvelope:

276 action_data: dict[str, Any] = field(default_factory=dict)

277 target_data: dict[str, Any] | None = None

278

279

280def script(

281 qualified_action: str | None,

282 core_action_data: dict[str, Any],

283 data: dict[str, Any],

284 target: Target,

285 delivery: Delivery,

286 hass_api: HomeAssistantAPI,

287) -> list[MiniEnvelope]:

288 """Customize `data` for script integration"""

289 results: list[MiniEnvelope] = []

290 if qualified_action in ("script.turn_on", "script.turn_off"):

291 action_data = {}

292 action_data["variables"] = core_action_data

293 if "variables" in data:

294 action_data["variables"].update(data.pop("variables"))

295 action_data["variables"].update(data)

296 action_data = hass_api.coerce_schema("script", qualified_action.replace("script.", ""), action_data)

297 results.append(MiniEnvelope(action_data=action_data, target_data={ATTR_ENTITY_ID: target.domain_entity_ids("script")}))

298 else:

299 action_data = core_action_data

300 action_data.update(data)

301 results.append(MiniEnvelope(action_data=action_data))

302

303 return results

304

305

306def ntfy(

307 core_action_data: dict[str, Any],

308 data: dict[str, Any],

309 target: Target,

310 delivery: Delivery,

311 priority: str | None,

312 hass_api: HomeAssistantAPI,

313) -> list[MiniEnvelope]:

314 """Customize `data` for ntfy integration

315

316 Deprecated, use the ntfy transport

317 """

318 results: list[MiniEnvelope] = []

319 action_data: dict[str, Any] = dict(core_action_data)

320 action_data.update(data)

321 action_data = hass_api.coerce_schema("ntfy", "publish", action_data)

322 # the schema coerces URLs to objects, which the action itself only accepts as strings

323 action_data = {k: str(v) if isinstance(v, URL) else v for k, v in action_data.items()}

324

325 if priority and priority in PRIORITY_VALUES:

326 action_data[ATTR_PRIORITY] = PRIORITY_VALUES.get(priority, 3)

327

328 media = action_data.pop(ATTR_MEDIA, {})

329 if media and media.get(ATTR_MEDIA_SNAPSHOT_URL) and "attach" not in action_data:

330 action_data["attach"] = media.get(ATTR_MEDIA_SNAPSHOT_URL)

331 actions = action_data.pop(ATTR_ACTIONS, [])

332 if len(actions) > 0:

333 first_action = actions[0]

334 if first_action.get(ATTR_ACTION_URL) and "click" not in action_data:

335 action_data["click"] = first_action.get(ATTR_ACTION_URL)

336

337 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)

338 action_data = DataFilter(rules).apply(action_data)

339

340 if target.email and "email" not in action_data:

341 for email in target.email:

342 call_data: dict[str, Any] = dict(action_data)

343 if len(results) == 1 and len(target.email) == 1:

344 results[0].action_data["email"] = email

345 else:

346 call_data["email"] = email

347 results.append(MiniEnvelope(action_data=call_data))

348 if target.phone and "call" not in action_data:

349 for phone in target.phone:

350 call_data = dict(action_data)

351 if len(results) == 1 and len(target.phone) == 1:

352 results[0].action_data["call"] = phone

353 else:

354 call_data["call"] = phone

355 results.append(MiniEnvelope(action_data=call_data))

356 if not results:

357 results.append(MiniEnvelope(action_data=dict(action_data)))

358 # every publish goes through the ntfy topic entities, including those adding an email or call

359 notify_entities = target.domain_entity_ids(NOTIFY_DOMAIN)

360 for result in results:

361 result.target_data = {ATTR_ENTITY_ID: notify_entities}

362

363 return results

364

365

366def notify_events(

367 message: str | None,

368 title: str | None,

369 core_action_data: dict[str, Any],

370 data: dict[str, Any],

371 delivery: Delivery,

372 priority: str | None,

373) -> list[MiniEnvelope]:

374 """Customize `data` for notify_events integration

375

376 This is a vendor notification aggregation service (https://notify.events)

377 with an old little used integration in Home Assistant

378 """

379 results: list[MiniEnvelope] = []

380 input_data: dict[str, Any] = dict(core_action_data)

381 input_data.update(data)

382

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

384 action_data[ATTR_MESSAGE] = message

385 priority_mapping: dict[str, str] = {

386 PRIORITY_MINIMUM: "lowest",

387 PRIORITY_LOW: "low",

388 PRIORITY_MEDIUM: "normal",

389 PRIORITY_HIGH: "high",

390 PRIORITY_CRITICAL: "highest",

391 }

392 if title:

393 action_data.setdefault(ATTR_DATA, {})

394 action_data[ATTR_DATA][ATTR_TITLE] = title

395

396 if ATTR_DATA in input_data:

397 # notify_events is schema-less for action

398 action_data[ATTR_DATA] = input_data[ATTR_DATA]

399

400 if priority and priority in PRIORITY_VALUES:

401 action_data.setdefault(ATTR_DATA, {})

402 action_data[ATTR_DATA][ATTR_PRIORITY] = priority_mapping.get(priority)

403 elif priority and priority in priority_mapping.values():

404 action_data.setdefault(ATTR_DATA, {})

405 action_data[ATTR_DATA][ATTR_PRIORITY] = priority

406

407 if "token" in input_data:

408 action_data.setdefault(ATTR_DATA, {})

409 action_data[ATTR_DATA]["token"] = input_data["token"]

410 if "level" in input_data:

411 action_data.setdefault(ATTR_DATA, {})

412 action_data[ATTR_DATA]["level"] = input_data["level"]

413

414 if input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL) and "images" not in input_data:

415 action_data.setdefault(ATTR_DATA, {})

416 action_data[ATTR_DATA].setdefault("images", [])

417 action_data[ATTR_DATA]["images"].append({"url": input_data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)})

418

419 rules = delivery.options.get(OPTION_DATA_KEYS_SELECT)

420 action_data = DataFilter(rules).apply(action_data)

421 results.append(MiniEnvelope(action_data=dict(action_data)))

422

423 return results