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_left = list(target) 

124 for category in self.AUTO_CATEGORIES: 

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

126 if validator is not None: 

127 matched = [] 

128 for t in targets_left: 

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

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

131 self.targets[category].append(t) 

132 matched.append(t) 

133 targets_left = [t for t in targets_left if t not in matched] 

134 else: 

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

136 if not targets_left: 

137 break 

138 if targets_left: 

139 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets_left 

140 

141 elif isinstance(target, dict): 

142 for category in target: 

143 targets = ensure_list(target[category]) 

144 if not targets: 

145 continue 

146 if category in self.AUTO_CATEGORIES: 

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

148 if validator is not None: 

149 for t in targets: 

150 if validator(t): 

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

152 if t not in self.targets[category]: 

153 self.targets[category].append(t) 

154 else: 

155 _LOGGER.warning("SUPERNOTIFY Target skipped invalid %s target: %s", category, t) 

156 else: 

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

158 

159 elif category in self.CATEGORIES: 

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

161 self.targets[category] = targets 

162 else: 

163 # custom categories 

164 self.targets[category] = targets 

165 else: 

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

167 

168 if target_data and target_specific_data: 

169 self.target_specific_data = {} 

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

171 for t in targets: 

172 self.target_specific_data[category, t] = target_data 

173 if target_data and not target_specific_data: 

174 self.target_data = target_data 

175 

176 # Targets by category 

177 

178 @property 

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

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

181 

182 @property 

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

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

185 

186 @property 

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

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

189 

190 @property 

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

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

193 

194 @property 

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

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

197 

198 @property 

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

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

201 

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

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

204 

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

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

207 

208 @property 

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

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

211 

212 @property 

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

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

215 

216 @property 

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

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

219 

220 # Selectors / validators 

221 

222 @classmethod 

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

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

225 

226 @classmethod 

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

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

229 

230 @classmethod 

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

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

233 

234 @classmethod 

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

236 try: 

237 return phone(target) is not None 

238 except vol.Invalid: 

239 return False 

240 

241 @classmethod 

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

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

244 

245 @classmethod 

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

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

248 

249 @classmethod 

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

251 try: 

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

253 except vol.Invalid: 

254 return False 

255 

256 def has_targets(self) -> bool: 

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

258 

259 def has_resolved_target(self) -> bool: 

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

261 

262 def has_unknown_targets(self) -> bool: 

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

264 

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

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

267 

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

269 result: list[str] = [] 

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

271 if category not in self.INDIRECT_CATEGORIES: 

272 result.extend(targets) 

273 return result 

274 

275 def hash_resolved(self) -> int: 

276 targets = [] 

277 for category in self.targets: 

278 if category not in self.INDIRECT_CATEGORIES: 

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

280 return hash(tuple(targets)) 

281 

282 @property 

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

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

285 

286 def direct(self) -> Target: 

287 t = Target( 

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

289 target_data=self.target_data, 

290 ) 

291 if self.target_specific_data: 

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

293 return t 

294 

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

296 targets = ensure_list(targets) 

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

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

299 

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

301 targets = ensure_list(targets) 

302 if category in self.targets: 

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

304 

305 def safe_copy(self) -> Target: 

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

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

308 return t 

309 

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

311 if not self.target_specific_data: 

312 result = self.safe_copy() 

313 result.target_specific_data = None 

314 return [result] 

315 results: list[Target] = [] 

316 default: Target = self.safe_copy() 

317 default.target_specific_data = None 

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

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

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

321 if last_found is None: 

322 last_found = data 

323 collected = {category: [target]} 

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

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

326 results.append(new_target) 

327 default -= new_target 

328 last_found = data 

329 collected = {category: [target]} 

330 else: 

331 collected.setdefault(category, []) 

332 collected[category].append(target) 

333 new_target = Target(collected, target_data=last_found) 

334 results.append(new_target) 

335 default -= new_target 

336 if default.has_targets(): 

337 results.append(default) 

338 return results 

339 

340 def __len__(self) -> int: 

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

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

343 

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

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

346 new = Target() 

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

348 for category in categories: 

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

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

351 

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

353 if other.target_data: 

354 if new.target_data is None: 

355 new.target_data = dict(other.target_data) 

356 else: 

357 new.target_data.update(other.target_data) 

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

359 if other.target_specific_data: 

360 if new.target_specific_data is None: 

361 new.target_specific_data = dict(other.target_specific_data) 

362 else: 

363 new.target_specific_data.update(other.target_specific_data) 

364 return new 

365 

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

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

368 new = Target() 

369 new.target_data = self.target_data 

370 if self.target_specific_data: 

371 new.target_specific_data = { 

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

373 } 

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

375 for category in categories: 

376 new.targets[category] = [] 

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

378 

379 return new 

380 

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

382 """Compare two targets""" 

383 if other is self: 

384 return True 

385 if other is None: 

386 return False 

387 if not isinstance(other, Target): 

388 return NotImplemented 

389 if self.target_data != other.target_data: 

390 return False 

391 if self.target_specific_data != other.target_specific_data: 

392 return False 

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

394 

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

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

397 

398 

399class TransportConfig: 

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

401 conf = conf or {} 

402 if class_config is not None: 

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

404 self.alias = conf.get(CONF_ALIAS) 

405 self.delivery_defaults: DeliveryConfig = DeliveryConfig( 

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

407 ) 

408 else: 

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

410 self.alias = conf.get(CONF_ALIAS) 

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

412 

413 # deprecation support 

414 device_domain = conf.get(CONF_DEVICE_DOMAIN) 

415 if device_domain is not None: 

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

417 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain 

418 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE) 

419 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE) 

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

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

422 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = { 

423 SELECT_INCLUDE: device_model_include, 

424 SELECT_EXCLUDE: device_model_exclude, 

425 } 

426 device_discovery = conf.get(CONF_DEVICE_DISCOVERY) 

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

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

429 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery 

430 

431 

432class DeliveryCustomization: 

433 def __init__(self, config: ConfigType | None, target_specific: bool = False) -> None: 

434 config = config or {} 

435 # perhaps should be false for wildcards 

436 self.enabled: bool | None = config.get(CONF_ENABLED, True) 

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

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

439 self.target: Target | None 

440 

441 if config.get(CONF_TARGET): 

442 if self.data: 

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

444 else: 

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

446 else: 

447 self.target = None 

448 

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

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

451 

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

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

454 

455 

456class SelectionRule: 

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

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

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

460 if config is None: 

461 return 

462 if isinstance(config, SelectionRule): 

463 self.include = config.include 

464 self.exclude = config.exclude 

465 elif isinstance(config, str): 

466 self.include = [config] 

467 elif isinstance(config, list): 

468 self.include = config 

469 else: 

470 if config.get(SELECT_INCLUDE): 

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

472 if config.get(SELECT_EXCLUDE): 

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

474 

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

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

477 return True 

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

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

480 return False 

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

482 return False 

483 else: 

484 if self.exclude is not None: 

485 for vv in v: 

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

487 return False 

488 if self.include is not None: 

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

490 return True 

491 

492 

493class DataFilter: 

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

495 

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

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

498 dict: 

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

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

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

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

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

504 

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

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

507 """ 

508 

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

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

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

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

513 if config is None: 

514 return 

515 if isinstance(config, str): 

516 self._include = [config] 

517 elif isinstance(config, list): 

518 self._include = config 

519 else: 

520 self._init_from_dict(config) 

521 

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

523 include_val = config.get(SELECT_INCLUDE) 

524 exclude_val = config.get(SELECT_EXCLUDE) 

525 if isinstance(include_val, dict): 

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

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

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

529 if v is not None: 

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

531 elif include_val is not None: 

532 self._include = ensure_list(include_val) 

533 if isinstance(exclude_val, dict): 

534 excludes, subs = DataFilter._parse_exclude_tree(exclude_val) 

535 self._exclude = excludes or None 

536 self._sub.update(subs) 

537 elif exclude_val is not None: 

538 self._exclude = ensure_list(exclude_val) 

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

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

541 continue 

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

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

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

545 else: 

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

547 if self._include is None: 

548 self._include = [] 

549 self._include.append(k) 

550 if v is not None: 

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

552 

553 @staticmethod 

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

555 excludes: list[str] = [] 

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

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

558 if v is None: 

559 excludes.append(k) 

560 else: 

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

562 return excludes, subs 

563 

564 @staticmethod 

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

566 df = DataFilter(None) 

567 excludes, subs = DataFilter._parse_exclude_tree(tree) 

568 df._exclude = excludes or None 

569 df._sub = subs 

570 return df 

571 

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

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

574 return True 

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

576 return False 

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

578 

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

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

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

582 if not self._match(key): 

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

584 continue 

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

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

587 if prune_empty and value == {}: 

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

589 continue 

590 result[key] = value 

591 return result 

592 

593 

594class DeliveryConfig: 

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

596 

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

598 

599 if delivery_defaults is not None: 

600 # use transport defaults where no delivery level override 

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

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

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

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

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

606 

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

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

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

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

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

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

613 # only override options not set in config 

614 if isinstance(delivery_defaults.options, dict): 

615 for opt in delivery_defaults.options: 

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

617 else: 

618 # construct the transport defaults 

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

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

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

622 self.action = conf.get(CONF_ACTION) 

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

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

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

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

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

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

629 

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

631 return { 

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

633 CONF_ACTION: self.action, 

634 CONF_OPTIONS: self.options, 

635 CONF_DATA: self.data, 

636 CONF_SELECTION: self.selection, 

637 CONF_PRIORITY: self.priority, 

638 CONF_SELECTION_RANK: str(self.selection_rank), 

639 CONF_TARGET_REQUIRED: str(self.target_required), 

640 CONF_TARGET_USAGE: self.target_usage, 

641 } 

642 

643 def __repr__(self) -> str: 

644 """Log friendly representation""" 

645 return str(self.as_dict()) 

646 

647 

648@dataclass 

649class ConditionVariables: 

650 """Variables presented to all condition evaluations 

651 

652 Attributes 

653 ---------- 

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

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

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

657 notification_priority (str): Priority of the notification 

658 notification_message (str): Message of the notification 

659 notification_title (str): Title of the notification 

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

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

662 

663 """ 

664 

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

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

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

668 notification_priority: str = PRIORITY_MEDIUM 

669 notification_message: str | None = "" 

670 notification_title: str | None = "" 

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

672 

673 def __init__(