Coverage for custom_components/supernotify/model.py: 97%

1from __future__ import annotations 

2 

3import logging 

4import re 

5from dataclasses import dataclass, field 

6from enum import IntFlag, StrEnum, auto 

7from traceback import format_exception 

8from typing import TYPE_CHECKING, Any, ClassVar 

9 

10import voluptuous as vol 

11from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN 

12 

13# This import brings in a bunch of other dependency noises, make it manual until py3.14/lazy import/HA updated 

14# from homeassistant.components.mobile_app import DOMAIN as MOBILE_APP_DOMAIN 

15from homeassistant.const import ( 

16 ATTR_AREA_ID, 

17 ATTR_DEVICE_ID, 

18 ATTR_ENTITY_ID, 

19 ATTR_FLOOR_ID, 

20 ATTR_LABEL_ID, 

21 CONF_ACTION, 

22 CONF_ALIAS, 

23 CONF_DEBUG, 

24 CONF_ENABLED, 

25 CONF_OPTIONS, 

26 CONF_TARGET, 

27 STATE_HOME, 

28 STATE_NOT_HOME, 

29) 

30from homeassistant.core import valid_entity_id 

31 

32from .common import ensure_list 

33from .const import ( 

34 ATTR_EMAIL, 

35 ATTR_MOBILE_APP_ID, 

36 ATTR_PERSON_ID, 

37 ATTR_PHONE, 

38 CONF_DATA, 

39 CONF_DELIVERY_DEFAULTS, 

40 CONF_DEVICE_DISCOVERY, 

41 CONF_DEVICE_DOMAIN, 

42 CONF_DEVICE_MODEL_EXCLUDE, 

43 CONF_DEVICE_MODEL_INCLUDE, 

44 CONF_PRIORITY, 

45 CONF_SELECTION, 

46 CONF_SELECTION_RANK, 

47 CONF_TARGET_REQUIRED, 

48 CONF_TARGET_USAGE, 

49 OPTION_DEVICE_DISCOVERY, 

50 OPTION_DEVICE_DOMAIN, 

51 OPTION_DEVICE_MODEL_SELECT, 

52 PRIORITY_MEDIUM, 

53 PRIORITY_VALUES, 

54 RE_DEVICE_ID, 

55 SELECT_EXCLUDE, 

56 SELECT_INCLUDE, 

57 SELECTION_DEFAULT, 

58 TARGET_USE_ON_NO_ACTION_TARGETS, 

59) 

60from .schema import SelectionRank, phone 

61 

62if TYPE_CHECKING: 

63 from collections.abc import Iterable, Sequence 

64 

65 from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

66 

67_LOGGER = logging.getLogger(__name__) 

68 

69# See note on import of homeassistant.components.mobile_app 

70MOBILE_APP_DOMAIN = "mobile_app" 

71 

72 

73class TransportFeature(IntFlag): 

74 MESSAGE = 1 

75 TITLE = 2 

76 IMAGES = 4 

77 VIDEO = 8 

78 ACTIONS = 16 

79 TEMPLATE_FILE = 32 

80 SNAPSHOT_IMAGE = 64 # transports will be deferred if a camera PTZ is defined 

81 SPOKEN = 128 

82 

83 

84class Target: 

85 # actual targets, that can positively identified with a validator 

86 DIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_ENTITY_ID, ATTR_DEVICE_ID, ATTR_EMAIL, ATTR_PHONE, ATTR_MOBILE_APP_ID] 

87 # references that lead to targets, that can positively identified with a validator 

88 AUTO_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_PERSON_ID] 

89 # references that lead to targets, that can't be positively identified with a validator 

90 EXPLICIT_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_AREA_ID, ATTR_FLOOR_ID, ATTR_LABEL_ID] 

91 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

92 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

93 

94 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES 

95 

96 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_" 

97 

98 def __init__( 

99 self, 

100 target: str 

101 | list[str] 

102 | dict[str, str] 

103 | dict[str, Sequence[str]] 

104 | dict[str, list[str]] 

105 | dict[str, str | list[str]] 

106 | None = None, 

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

108 target_specific_data: bool = False, 

109 ) -> None: 

110 self.target_data: dict[str, Any] | None = None 

111 self.target_specific_data: dict[tuple[str, str], dict[str, Any]] | None = None 

112 self.targets: dict[str, list[str]] = {} 

113 

114 matched: list[str] 

115 

116 if isinstance(target, str): 

117 target = [target] 

118 

119 if target is None: 

120 pass # empty constructor is valid case for target building 

121 elif isinstance(target, list): 

122 # simplified and legacy way of assuming list of entities that can be discriminated by validator 

123 targets = list(target) 

124 for category in self.AUTO_CATEGORIES: 

125 matched = self._filter_by_category(category, targets) 

126 if matched: 

127 self.targets.setdefault(category, []) 

128 self.targets[category].extend([t for t in matched if t not in self.targets[category]]) 

129 targets = [t for t in targets if t not in matched] 

130 if not targets: 

131 break 

132 if targets: 

133 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets 

134 

135 elif isinstance(target, dict): 

136 for category in target: 

137 targets = ensure_list(target[category]) 

138 if not targets: 

139 continue 

140 if category in self.AUTO_CATEGORIES: 

141 matched = self._filter_by_category(category, targets) 

142 if matched: 

143 self.targets.setdefault(category, []) 

144 self.targets[category].extend([t for t in matched if t not in self.targets[category]]) 

145 

146 elif category in self.CATEGORIES: 

147 # categories that can't be automatically detected, like label_id 

148 self.targets[category] = targets 

149 else: 

150 # custom categories 

151 self.targets[category] = targets 

152 else: 

153 _LOGGER.warning("SUPERNOTIFY Target created with no valid targets: %s", target) 

154 

155 if target_data and target_specific_data: 

156 self.target_specific_data = {} 

157 for category, targets in self.targets.items(): 

158 for t in targets: 

159 self.target_specific_data[category, t] = target_data 

160 if target_data and not target_specific_data: 

161 self.target_data = target_data 

162 

163 def _filter_by_category(self, category: str, candidates: list[str]) -> list[str]: 

164 matched: list[str] = [] 

165 validator = getattr(self, f"is_{category}", None) 

166 if validator is not None: 

167 for t in candidates: 

168 if t not in matched and validator(t): 

169 matched.append(t) 

170 else: 

171 _LOGGER.debug("SUPERNOTIFY Missing validator for selective target category %s", category) 

172 return matched 

173 

174 # Targets by category 

175 

176 @property 

177 def email(self) -> list[str]: 

178 return self.targets.get(ATTR_EMAIL, []) 

179 

180 @property 

181 def entity_ids(self) -> list[str]: 

182 return self.targets.get(ATTR_ENTITY_ID, []) 

183 

184 @property 

185 def person_ids(self) -> list[str]: 

186 return self.targets.get(ATTR_PERSON_ID, []) 

187 

188 @property 

189 def device_ids(self) -> list[str]: 

190 return self.targets.get(ATTR_DEVICE_ID, []) 

191 

192 @property 

193 def phone(self) -> list[str]: 

194 return self.targets.get(ATTR_PHONE, []) 

195 

196 @property 

197 def mobile_app_ids(self) -> list[str]: 

198 return self.targets.get(ATTR_MOBILE_APP_ID, []) 

199 

200 def domain_entity_ids(self, domain: str | None) -> list[str]: 

201 return [t for t in self.targets.get(ATTR_ENTITY_ID, []) if domain is not None and t and t.startswith(f"{domain}.")] 

202 

203 def custom_ids(self, category: str) -> list[str]: 

204 return self.targets.get(category, []) if category not in self.CATEGORIES else [] 

205 

206 @property 

207 def area_ids(self) -> list[str]: 

208 return self.targets.get(ATTR_AREA_ID, []) 

209 

210 @property 

211 def floor_ids(self) -> list[str]: 

212 return self.targets.get(ATTR_FLOOR_ID, []) 

213 

214 @property 

215 def label_ids(self) -> list[str]: 

216 return self.targets.get(ATTR_LABEL_ID, []) 

217 

218 # Selectors / validators 

219 

220 @classmethod 

221 def is_device_id(cls, target: str) -> bool: 

222 return re.fullmatch(RE_DEVICE_ID, target) is not None 

223 

224 @classmethod 

225 def is_entity_id(cls, target: str) -> bool: 

226 return valid_entity_id(target) and not target.startswith("person.") 

227 

228 @classmethod 

229 def is_person_id(cls, target: str) -> bool: 

230 return target.startswith("person.") and valid_entity_id(target) 

231 

232 @classmethod 

233 def is_phone(cls, target: str) -> bool: 

234 try: 

235 return phone(target) is not None 

236 except vol.Invalid: 

237 return False 

238 

239 @classmethod 

240 def is_mobile_app_id(cls, target: str) -> bool: 

241 return not valid_entity_id(target) and target.startswith(f"{MOBILE_APP_DOMAIN}_") 

242 

243 @classmethod 

244 def is_notify_entity(cls, target: str) -> bool: 

245 return valid_entity_id(target) and target.startswith(f"{NOTIFY_DOMAIN}.") 

246 

247 @classmethod 

248 def is_email(cls, target: str) -> bool: 

249 try: 

250 return vol.Email()(target) is not None # type: ignore[call-arg] 

251 except vol.Invalid: 

252 return False 

253 

254 def has_targets(self) -> bool: 

255 return any(targets for targets in self.targets.values()) 

256 

257 def has_resolved_target(self) -> bool: 

258 return any(targets for category, targets in self.targets.items() if category not in self.INDIRECT_CATEGORIES) 

259 

260 def has_unknown_targets(self) -> bool: 

261 return len(self.targets.get(self.UNKNOWN_CUSTOM_CATEGORY, [])) > 0 

262 

263 def for_category(self, category: str) -> list[str]: 

264 return self.targets.get(category, []) 

265 

266 def resolved_targets(self) -> list[str]: 

267 result: list[str] = [] 

268 for category, targets in self.targets.items(): 

269 if category not in self.INDIRECT_CATEGORIES: 

270 result.extend(targets) 

271 return result 

272 

273 def hash_resolved(self) -> int: 

274 targets = [] 

275 for category in self.targets: 

276 if category not in self.INDIRECT_CATEGORIES: 

277 targets.extend(self.targets[category]) 

278 return hash(tuple(targets)) 

279 

280 @property 

281 def direct_categories(self) -> list[str]: 

282 return self.DIRECT_CATEGORIES + [cat for cat in self.targets if cat not in self.CATEGORIES] 

283 

284 def direct(self) -> Target: 

285 t = Target( 

286 {cat: targets for cat, targets in self.targets.items() if cat in self.direct_categories}, 

287 target_data=self.target_data, 

288 ) 

289 if self.target_specific_data: 

290 t.target_specific_data = {k: v for k, v in self.target_specific_data.items() if k[0] in self.direct_categories} 

291 return t 

292 

293 def extend(self, category: str, targets: list[str] | str) -> None: 

294 targets = ensure_list(targets) 

295 self.targets.setdefault(category, []) 

296 self.targets[category].extend(t for t in targets if t not in self.targets[category]) 

297 

298 def remove(self, category: str, targets: list[str] | str) -> None: 

299 targets = ensure_list(targets) 

300 if category in self.targets: 

301 self.targets[category] = [t for t in self.targets[category] if t not in targets] 

302 

303 def safe_copy(self) -> Target: 

304 t = Target(dict(self.targets), target_data=dict(self.target_data) if self.target_data else None) 

305 t.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None 

306 return t 

307 

308 def split_by_target_data(self) -> list[Target]: 

309 if not self.target_specific_data: 

310 result = self.safe_copy() 

311 result.target_specific_data = None 

312 return [result] 

313 results: list[Target] = [] 

314 default: Target = self.safe_copy() 

315 default.target_specific_data = None 

316 last_found: dict[str, Any] | None = None 

317 collected: dict[str, list[str]] = {} 

318 for (category, target), data in self.target_specific_data.items(): 

319 if last_found is None: 

320 last_found = data 

321 collected = {category: [target]} 

322 elif data != last_found and last_found is not None: 

323 new_target: Target = Target(collected, target_data=last_found) 

324 results.append(new_target) 

325 default -= new_target 

326 last_found = data 

327 collected = {category: [target]} 

328 else: 

329 collected.setdefault(category, []) 

330 collected[category].append(target) 

331 new_target = Target(collected, target_data=last_found) 

332 results.append(new_target) 

333 default -= new_target 

334 if default.has_targets(): 

335 results.append(default) 

336 return results 

337 

338 def __len__(self) -> int: 

339 """How many targets, whether direct or indirect""" 

340 return sum(len(targets) for targets in self.targets.values()) 

341 

342 def __add__(self, other: Target) -> Target: 

343 """Create a new target by adding another to this one""" 

344 new = Target() 

345 categories = set(list(self.targets.keys()) + list(other.targets.keys())) 

346 for category in categories: 

347 new.targets[category] = list(self.targets.get(category, [])) 

348 new.targets[category].extend(t for t in other.targets.get(category, []) if t not in new.targets[category]) 

349 

350 new.target_data = dict(self.target_data) if self.target_data else None 

351 if other.target_data: 

352 if new.target_data is None: 

353 new.target_data = dict(other.target_data) 

354 else: 

355 new.target_data.update(other.target_data) 

356 new.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None 

357 if other.target_specific_data: 

358 if new.target_specific_data is None: 

359 new.target_specific_data = dict(other.target_specific_data) 

360 else: 

361 new.target_specific_data.update(other.target_specific_data) 

362 return new 

363 

364 def __sub__(self, other: Target) -> Target: 

365 """Create a new target by removing another from this one, ignoring target_data""" 

366 new = Target() 

367 new.target_data = self.target_data 

368 if self.target_specific_data: 

369 new.target_specific_data = { 

370 k: v for k, v in self.target_specific_data.items() if k[1] not in other.targets.get(k[0], ()) 

371 } 

372 categories = set(list(self.targets.keys()) + list(other.targets.keys())) 

373 for category in categories: 

374 new.targets[category] = [] 

375 new.targets[category].extend(t for t in self.targets.get(category, []) if t not in other.targets.get(category, [])) 

376 

377 return new 

378 

379 def __eq__(self, other: object) -> bool: 

380 """Compare two targets""" 

381 if other is self: 

382 return True 

383 if other is None: 

384 return False 

385 if not isinstance(other, Target): 

386 return NotImplemented 

387 if self.target_data != other.target_data: 

388 return False 

389 if self.target_specific_data != other.target_specific_data: 

390 return False 

391 return all(self.targets.get(category, []) == other.targets.get(category, []) for category in self.CATEGORIES) 

392 

393 def as_dict(self, **_kwargs: Any) -> dict[str, list[str]]: 

394 return {k: v for k, v in self.targets.items() if v} 

395 

396 

397class TransportConfig: 

398 def __init__(self, conf: ConfigType | None = None, class_config: TransportConfig | None = None) -> None: 

399 conf = conf or {} 

400 if class_config is not None: 

401 self.enabled: bool = conf.get(CONF_ENABLED, class_config.enabled) 

402 self.alias = conf.get(CONF_ALIAS) 

403 self.delivery_defaults: DeliveryConfig = DeliveryConfig( 

404 conf.get(CONF_DELIVERY_DEFAULTS, {}), class_config.delivery_defaults or None 

405 ) 

406 else: 

407 self.enabled = conf.get(CONF_ENABLED, True) 

408 self.alias = conf.get(CONF_ALIAS) 

409 self.delivery_defaults = DeliveryConfig(conf.get(CONF_DELIVERY_DEFAULTS) or {}) 

410 

411 # deprecation support 

412 device_domain = conf.get(CONF_DEVICE_DOMAIN) 

413 if device_domain is not None: 

414 _LOGGER.warning("SUPERNOTIFY device_domain on transport deprecated, use options instead") 

415 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain 

416 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE) 

417 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE) 

418 if device_model_include is not None or device_model_exclude is not None: 

419 _LOGGER.warning("SUPERNOTIFY device_model_include/exclude on transport deprecated, use options instead") 

420 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = { 

421 SELECT_INCLUDE: device_model_include, 

422 SELECT_EXCLUDE: device_model_exclude, 

423 } 

424 device_discovery = conf.get(CONF_DEVICE_DISCOVERY) 

425 if device_discovery is not None and self.delivery_defaults.options.get(OPTION_DEVICE_DISCOVERY) is None: 

426 _LOGGER.warning("SUPERNOTIFY device_discovery on transport deprecated, use options instead") 

427 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery 

428 

429 

430class DeliveryCustomization: 

431 def __init__( 

432 self, config: ConfigType | None = None, target_specific: bool = False, default_enabled: bool | None = None 

433 ) -> None: 

434 config = config or {} 

435 # defining a customization doesn't imply that the delivery is always enabled - 

436 # default_enabled only fills in when the `enabled` key is omitted entirely (dict.get 

437 # default only applies when the key is absent), never overrides an explicit 

438 # `enabled: None`/`false`/`true` - e.g. Scenario.enabling_deliveries() relies on this to 

439 # treat a directly-named delivery with no `enabled` key as enabling it (matching the 

440 # list/string delivery config forms), but not one explicitly set to `enabled: None` 

441 # just to carry other data (e.g. a priority override). 

442 self.enabled: bool | None = config.get(CONF_ENABLED, default_enabled) 

443 self.data: dict[str, Any] | None = config.get(CONF_DATA) 

444 # TODO: only works for scenario or recipient, not action call 

445 self.target: Target | None 

446 

447 if config.get(CONF_TARGET): 

448 if self.data: 

449 self.target = Target(config.get(CONF_TARGET), target_data=self.data, target_specific_data=target_specific) 

450 else: 

451 self.target = Target(config.get(CONF_TARGET)) 

452 else: 

453 self.target = None 

454 

455 def data_value(self, key: str) -> Any: 

456 return self.data.get(key) if self.data else None 

457 

458 def as_dict(self, **_kwargs: Any) -> dict[str, Any]: 

459 return {CONF_TARGET: self.target.as_dict() if self.target else None, CONF_ENABLED: self.enabled, CONF_DATA: self.data} 

460 

461 

462class SelectionRule: 

463 def __init__(self, config: str | list[str] | dict | SelectionRule | None) -> None: 

464 self.include: list[str] | None = None 

465 self.exclude: list[str] | None = None 

466 if config is None: 

467 return 

468 if isinstance(config, SelectionRule): 

469 self.include = config.include 

470 self.exclude = config.exclude 

471 elif isinstance(config, str): 

472 self.include = [config] 

473 elif isinstance(config, list): 

474 self.include = config 

475 else: 

476 if config.get(SELECT_INCLUDE): 

477 self.include = ensure_list(config.get(SELECT_INCLUDE)) 

478 if config.get(SELECT_EXCLUDE): 

479 self.exclude = ensure_list(config.get(SELECT_EXCLUDE)) 

480 

481 def match(self, v: str | Iterable[str] | None) -> bool: 

482 if self.include is None and self.exclude is None: 

483 return True 

484 if isinstance(v, str) or v is None: 

485 if self.exclude is not None and v is not None and any(re.fullmatch(pat, v) for pat in self.exclude): 

486 return False 

487 if self.include is not None and (v is None or not any(re.fullmatch(pat, v) for pat in self.include)): 

488 return False 

489 else: 

490 if self.exclude is not None: 

491 for vv in v: 

492 if any(re.fullmatch(pat, vv) for pat in self.exclude): 

493 return False 

494 if self.include is not None: 

495 return any(any(re.fullmatch(pat, vv) for pat in self.include) for vv in v) 

496 return True 

497 

498 

499class DataFilter: 

500 """Accepts a dict structure and returns a filtered copy, with arbitrary-depth key filtering. 

501 

502 Config format (same structure applies recursively at each level): 

503 str | list -- shorthand: include only keys matching these patterns 

504 dict: 

505 include: list[str] -- include only keys matching these patterns 

506 exclude: list[str] -- exclude keys matching these patterns 

507 exclude: dict -- exclude tree: null value = exclude that key, 

508 dict value = keep key but apply tree recursively to its value 

509 <key>: sub-config -- any other key: sub-filter applied to that key's dict value 

510 

511 Patterns are matched with re.fullmatch. Sub-filter key lookup is exact (not regex). 

512 include and exclude can be combined; any non-reserved key adds a sub-filter. 

513 """ 

514 

515 def __init__(self, config: str | list[str] | dict | None) -> None: 

516 self._include: list[str] | None = None 

517 self._exclude: list[str] | None = None 

518 self._sub: dict[str, DataFilter] = {} 

519 if config is None: 

520 return 

521 if isinstance(config, str): 

522 self._include = [config] 

523 elif isinstance(config, list): 

524 self._include = config 

525 else: 

526 self._init_from_dict(config) 

527 

528 def _init_from_dict(self, config: dict) -> None: 

529 include_val = config.get(SELECT_INCLUDE) 

530 exclude_val = config.get(SELECT_EXCLUDE) 

531 if isinstance(include_val, dict): 

532 # include as dict: keys = include patterns, non-null values = sub-filters 

533 self._include = list(include_val.keys()) 

534 for k, v in include_val.items(): 

535 if v is not None: 

536 self._sub[k] = DataFilter(v) 

537 elif include_val is not None: 

538 self._include = ensure_list(include_val) 

539 if isinstance(exclude_val, dict): 

540 excludes, subs = DataFilter._parse_exclude_tree(exclude_val) 

541 self._exclude = excludes or None 

542 self._sub.update(subs) 

543 elif exclude_val is not None: 

544 self._exclude = ensure_list(exclude_val) 

545 for k, v in config.items(): 

546 if k in (SELECT_INCLUDE, SELECT_EXCLUDE) or k in self._sub: 

547 continue 

548 if isinstance(v, dict) and (SELECT_INCLUDE in v or SELECT_EXCLUDE in v): 

549 # value is an explicit DataFilter config (has reserved keys) → sub-filter only, all keys pass 

550 self._sub[k] = DataFilter(v) 

551 else: 

552 # null or value without reserved keys → include pattern (+ sub-filter if non-null) 

553 if self._include is None: 

554 self._include = [] 

555 self._include.append(k) 

556 if v is not None: 

557 self._sub[k] = DataFilter(v) 

558 

559 @staticmethod 

560 def _parse_exclude_tree(tree: dict) -> tuple[list[str], dict[str, DataFilter]]: 

561 excludes: list[str] = [] 

562 subs: dict[str, DataFilter] = {} 

563 for k, v in tree.items(): 

564 if v is None: 

565 excludes.append(k) 

566 else: 

567 subs[k] = DataFilter._exclude_tree_to_filter(v) 

568 return excludes, subs 

569 

570 @staticmethod 

571 def _exclude_tree_to_filter(tree: dict) -> DataFilter: 

572 df = DataFilter(None) 

573 excludes, subs = DataFilter._parse_exclude_tree(tree) 

574 df._exclude = excludes or None 

575 df._sub = subs 

576 return df 

577 

578 def _match(self, key: str) -> bool: 

579 if self._exclude is None and self._include is None: 

580 return True 

581 if self._exclude is not None and any(re.fullmatch(p, key) for p in self._exclude): 

582 return False 

583 return self._include is None or any(re.fullmatch(p, key) for p in self._include) 

584 

585 def apply(self, data: dict[str, Any], *, prune_empty: bool = False) -> dict[str, Any]: 

586 result: dict[str, Any] = {} 

587 for key, value in data.items(): 

588 if not self._match(key): 

589 _LOGGER.debug("SUPERNOTIFY Pruning %s:%s", key, value) 

590 continue 

591 if key in self._sub and isinstance(value, dict): 

592 value = self._sub[key].apply(value, prune_empty=prune_empty) 

593 if prune_empty and value == {}: 

594 _LOGGER.debug("SUPERNOTIFY Pruning empty %s", key) 

595 continue 

596 result[key] = value 

597 return result 

598 

599 

600class DeliveryConfig: 

601 """Shared config for transport defaults and Delivery definitions""" 

602 

603 def __init__(self, conf: ConfigType, delivery_defaults: DeliveryConfig | None = None) -> None: 

604 

605 if delivery_defaults is not None: 

606 # use transport defaults where no delivery level override 

607 self.target: Target | None = Target(conf.get(CONF_TARGET)) if CONF_TARGET in conf else delivery_defaults.target 

608 self.target_required: TargetRequired = conf.get(CONF_TARGET_REQUIRED, delivery_defaults.target_required) 

609 self.target_usage: str = conf.get(CONF_TARGET_USAGE) or delivery_defaults.target_usage 

610 self.action: str | None = conf.get(CONF_ACTION) or delivery_defaults.action 

611 self.debug: bool = conf.get(CONF_DEBUG, delivery_defaults.debug) 

612 

613 self.data: ConfigType = dict(delivery_defaults.data) if isinstance(delivery_defaults.data, dict) else {} 

614 self.data.update(conf.get(CONF_DATA, {})) 

615 self.selection: list[str] = conf.get(CONF_SELECTION, delivery_defaults.selection) 

616 self.priority: list[str] = conf.get(CONF_PRIORITY, delivery_defaults.priority) 

617 self.selection_rank: SelectionRank = conf.get(CONF_SELECTION_RANK, delivery_defaults.selection_rank) 

618 self.options: ConfigType = conf.get(CONF_OPTIONS, {}) 

619 # only override options not set in config 

620 if isinstance(delivery_defaults.options, dict): 

621 for opt in delivery_defaults.options: 

622 self.options.setdefault(opt, delivery_defaults.options[opt]) 

623 else: 

624 # construct the transport defaults 

625 self.target = Target(conf.get(CONF_TARGET)) if conf.get(CONF_TARGET) else None 

626 self.target_required = conf.get(CONF_TARGET_REQUIRED, TargetRequired.ALWAYS) 

627 self.target_usage = conf.get(CONF_TARGET_USAGE, TARGET_USE_ON_NO_ACTION_TARGETS) 

628 self.action = conf.get(CONF_ACTION) 

629 self.debug = conf.get(CONF_DEBUG, False) 

630 self.options = conf.get(CONF_OPTIONS, {}) 

631 self.data = conf.get(CONF_DATA, {}) 

632 self.selection = conf.get(CONF_SELECTION, [SELECTION_DEFAULT]) 

633 self.priority = conf.get(CONF_PRIORITY, list(PRIORITY_VALUES.keys())) 

634 self.selection_rank = conf.get(CONF_SELECTION_RANK, SelectionRank.ANY) 

635 

636 def as_dict(self, **_kwargs: Any) -> dict[str, Any]: 

637 return { 

638 CONF_TARGET: self.target.as_dict() if self.target else None, 

639 CONF_ACTION: self.action, 

640 CONF_OPTIONS: self.options, 

641 CONF_DATA: self.data, 

642 CONF_SELECTION: self.selection, 

643 CONF_PRIORITY: self.priority, 

644 CONF_SELECTION_RANK: str(self.selection_rank), 

645 CONF_TARGET_REQUIRED: str(self.target_required), 

646 CONF_TARGET_USAGE: self.target_usage, 

647 } 

648 

649 def __repr__(self) -> str: 

650 """Log friendly representation""" 

651 return str(self.as_dict()) 

652 

653 

654@dataclass 

655class ConditionVariables: 

656 """Variables presented to all condition evaluations 

657 

658 Attributes 

659 ---------- 

660 applied_scenarios (list[str]): Scenarios that have been applied 

661 required_scenarios (list[str]): Scenarios that must be applied 

662 constrain_scenarios (list[str]): Only scenarios in this list, or in explicit apply_scenarios, can be applied 

663 notification_priority (str): Priority of the notification 

664 notification_message (str): Message of the notification 

665 notification_title (str): Title of the notification 

666 occupancy (list[str]): List of occupancy scenarios 

667 notification_data (dict[str,Any]): Additional data passed on notify action call 

668 

669 """ 

670 

671 applied_scenarios: list[str] = field(default_factory=list) 

672 required_scenarios: list[str] = field(default_factory=list) 

673 constrain_scenarios: list[str] = field(default_factory=list)