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

473 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2026-04-01 15:06 +0000

1import logging 

2import re 

3from dataclasses import dataclass, field 

4from enum import IntFlag, StrEnum, auto 

5from traceback import format_exception 

6from typing import TYPE_CHECKING, Any, ClassVar 

7 

8import voluptuous as vol 

9from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN 

10 

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

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

13from homeassistant.const import ( 

14 ATTR_AREA_ID, 

15 ATTR_DEVICE_ID, 

16 ATTR_ENTITY_ID, 

17 ATTR_FLOOR_ID, 

18 ATTR_LABEL_ID, 

19 CONF_ACTION, 

20 CONF_ALIAS, 

21 CONF_DEBUG, 

22 CONF_ENABLED, 

23 CONF_OPTIONS, 

24 CONF_TARGET, 

25 STATE_HOME, 

26 STATE_NOT_HOME, 

27) 

28from homeassistant.core import valid_entity_id 

29 

30from .common import ensure_list 

31from .const import ( 

32 ATTR_EMAIL, 

33 ATTR_MOBILE_APP_ID, 

34 ATTR_PERSON_ID, 

35 ATTR_PHONE, 

36 CONF_DATA, 

37 CONF_DELIVERY_DEFAULTS, 

38 CONF_DEVICE_DISCOVERY, 

39 CONF_DEVICE_DOMAIN, 

40 CONF_DEVICE_MODEL_EXCLUDE, 

41 CONF_DEVICE_MODEL_INCLUDE, 

42 CONF_PRIORITY, 

43 CONF_SELECTION, 

44 CONF_SELECTION_RANK, 

45 CONF_TARGET_REQUIRED, 

46 CONF_TARGET_USAGE, 

47 OPTION_DEVICE_DISCOVERY, 

48 OPTION_DEVICE_DOMAIN, 

49 OPTION_DEVICE_MODEL_SELECT, 

50 PRIORITY_MEDIUM, 

51 PRIORITY_VALUES, 

52 RE_DEVICE_ID, 

53 SELECT_EXCLUDE, 

54 SELECT_INCLUDE, 

55 SELECTION_DEFAULT, 

56 TARGET_USE_ON_NO_ACTION_TARGETS, 

57) 

58from .schema import SelectionRank, phone 

59 

60if TYPE_CHECKING: 

61 from collections.abc import Iterable, Sequence 

62 

63 from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

64 

65_LOGGER = logging.getLogger(__name__) 

66 

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

68MOBILE_APP_DOMAIN = "mobile_app" 

69 

70 

71class TransportFeature(IntFlag): 

72 MESSAGE = 1 

73 TITLE = 2 

74 IMAGES = 4 

75 VIDEO = 8 

76 ACTIONS = 16 

77 TEMPLATE_FILE = 32 

78 SNAPSHOT_IMAGE = 64 

79 SPOKEN = 128 

80 

81 

82class Target: 

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

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

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

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

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

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

89 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

90 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

91 

92 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES 

93 

94 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_" 

95 

96 def __init__( 

97 self, 

98 target: str 

99 | list[str] 

100 | dict[str, str] 

101 | dict[str, Sequence[str]] 

102 | dict[str, list[str]] 

103 | dict[str, str | list[str]] 

104 | None = None, 

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

106 target_specific_data: bool = False, 

107 ) -> None: 

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

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

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

111 

112 matched: list[str] 

113 

114 if isinstance(target, str): 

115 target = [target] 

116 

117 if target is None: 

118 pass # empty constructor is valid case for target building 

119 elif isinstance(target, list): 

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

121 targets_left = list(target) 

122 for category in self.AUTO_CATEGORIES: 

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

124 if validator is not None: 

125 matched = [] 

126 for t in targets_left: 

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

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

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

130 matched.append(t) 

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

132 else: 

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

134 if not targets_left: 

135 break 

136 if targets_left: 

137 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets_left 

138 

139 elif isinstance(target, dict): 

140 for category in target: 

141 targets = ensure_list(target[category]) 

142 if not targets: 

143 continue 

144 if category in self.AUTO_CATEGORIES: 

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

146 if validator is not None: 

147 for t in targets: 

148 if validator(t): 

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

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

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

152 else: 

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

154 else: 

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

156 

157 elif category in self.CATEGORIES: 

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

159 self.targets[category] = targets 

160 else: 

161 # custom categories 

162 self.targets[category] = targets 

163 else: 

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

165 

166 if target_data and target_specific_data: 

167 self.target_specific_data = {} 

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

169 for t in targets: 

170 self.target_specific_data[category, t] = target_data 

171 if target_data and not target_specific_data: 

172 self.target_data = target_data 

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__(self, config: ConfigType | None, target_specific: bool = False) -> None: 

432 config = config or {} 

433 # perhaps should be false for wildcards 

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

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

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

437 self.target: Target | None 

438 

439 if config.get(CONF_TARGET): 

440 if self.data: 

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

442 else: 

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

444 else: 

445 self.target = None 

446 

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

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

449 

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

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

452 

453 

454class SelectionRule: 

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

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

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

458 if config is None: 

459 return 

460 if isinstance(config, SelectionRule): 

461 self.include = config.include 

462 self.exclude = config.exclude 

463 elif isinstance(config, str): 

464 self.include = [config] 

465 elif isinstance(config, list): 

466 self.include = config 

467 else: 

468 if config.get(SELECT_INCLUDE): 

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

470 if config.get(SELECT_EXCLUDE): 

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

472 

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

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

475 return True 

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

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

478 return False 

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

480 return False 

481 else: 

482 if self.exclude is not None: 

483 for vv in v: 

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

485 return False 

486 if self.include is not None: 

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

488 return True 

489 

490 

491class DeliveryConfig: 

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

493 

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

495 

496 if delivery_defaults is not None: 

497 # use transport defaults where no delivery level override 

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

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

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

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

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

503 

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

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

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

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

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

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

510 # only override options not set in config 

511 if isinstance(delivery_defaults.options, dict): 

512 for opt in delivery_defaults.options: 

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

514 else: 

515 # construct the transport defaults 

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

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

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

519 self.action = conf.get(CONF_ACTION) 

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

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

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

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

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

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

526 

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

528 return { 

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

530 CONF_ACTION: self.action, 

531 CONF_OPTIONS: self.options, 

532 CONF_DATA: self.data, 

533 CONF_SELECTION: self.selection, 

534 CONF_PRIORITY: self.priority, 

535 CONF_SELECTION_RANK: str(self.selection_rank), 

536 CONF_TARGET_REQUIRED: str(self.target_required), 

537 CONF_TARGET_USAGE: self.target_usage, 

538 } 

539 

540 def __repr__(self) -> str: 

541 """Log friendly representation""" 

542 return str(self.as_dict()) 

543 

544 

545@dataclass 

546class ConditionVariables: 

547 """Variables presented to all condition evaluations 

548 

549 Attributes 

550 ---------- 

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

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

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

554 notification_priority (str): Priority of the notification 

555 notification_message (str): Message of the notification 

556 notification_title (str): Title of the notification 

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

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

559 

560 """ 

561 

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

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

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

565 notification_priority: str = PRIORITY_MEDIUM 

566 notification_message: str | None = "" 

567 notification_title: str | None = "" 

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

569 

570 def __init__( 

571 self, 

572 applied_scenarios: list[str] | None = None, 

573 required_scenarios: list[str] | None = None, 

574 constrain_scenarios: list[str] | None = None, 

575 delivery_priority: str | None = PRIORITY_MEDIUM, 

576 occupiers: dict[str, list[Any]] | None = None, 

577 message: str | None = None, 

578 title: str | None = None, 

579 notification_data: dict[str, Any] | None = None, 

580 ) -> None: 

581 occupiers = occupiers or {} 

582 self.occupancy = [] 

583 if not occupiers.get(STATE_NOT_HOME) and occupiers.get(STATE_HOME): 

584 self.occupancy.append("ALL_HOME") 

585 elif occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME): 

586 self.occupancy.append("ALL_AWAY") 

587 if len(occupiers.get(STATE_HOME, [])) == 1: 

588 self.occupancy.extend(["LONE_HOME", "SOME_HOME"]) 

589 elif len(occupiers.get(STATE_HOME, [])) > 1 and occupiers.get(STATE_NOT_HOME): 

590 self.occupancy.extend(["MULTI_HOME", "SOME_HOME"]) 

591 self.applied_scenarios = applied_scenarios or [] 

592 self.required_scenarios = required_scenarios or [] 

593 self.constrain_scenarios = constrain_scenarios or [] 

594 self.notification_priority = delivery_priority or PRIORITY_MEDIUM 

595 self.notification_message = message 

596 self.notification_title = title 

597 self.notification_data: dict[str, Any] = notification_data or {} 

598 

599 def as_dict(self, **_kwargs: Any) -> TemplateVarsType: 

600 return { 

601 "applied_scenarios": self.applied_scenarios, 

602 "required_scenarios": self.required_scenarios, 

603 "constrain_scenarios": self.constrain_scenarios, 

604 "notification_message": self.notification_message, 

605 "notification_title": self.notification_title, 

606 "notification_priority": self.notification_priority, 

607 "occupancy": self.occupancy, 

608 "notification_data": self.notification_data, 

609 } 

610 

611 

612class SuppressionReason(StrEnum): 

613 SNOOZED = "SNOOZED" 

614 DUPE = "DUPE" 

615 NO_SCENARIO = "NO_SCENARIO" 

616 NO_ACTION = "NO_ACTION" 

617 NO_TARGET = "NO_TARGET" 

618 INVALID_ACTION_DATA = "INVALID_ACTION_DATA" 

619 TRANSPORT_DISABLED = "TRANSPORT_DISABLED" 

620 PRIORITY = "PRIORITY" 

621 DELIVERY_CONDITION = "DELIVERY_CONDITION" 

622 UNKNOWN = "UNKNOWN" 

623 

624 

625class TargetRequired(StrEnum): 

626 ALWAYS = auto() 

627 NEVER = auto() 

628 OPTIONAL = auto() 

629 

630 @classmethod 

631 def _missing_(cls, value: Any) -> TargetRequired | None: 

632 """Backward compatibility for binary values""" 

633 if value is True or (isinstance(value, str) and value.lower() in ("true", "on")): 

634 return cls.ALWAYS 

635 if value is False or (isinstance(value, str) and value.lower() in ("false", "off")): 

636 return cls.OPTIONAL 

637 return None 

638 

639 

640class TargetType(StrEnum): 

641 pass 

642 

643 

644class GlobalTargetType(TargetType): 

645 NONCRITICAL = "NONCRITICAL" 

646 EVERYTHING = "EVERYTHING" 

647 

648 

649class RecipientType(StrEnum): 

650 USER = "USER" 

651 EVERYONE = "EVERYONE" 

652 

653 

654class QualifiedTargetType(TargetType): 

655 TRANSPORT = "TRANSPORT" 

656 DELIVERY = "DELIVERY" 

657 CAMERA = "CAMERA" 

658 PRIORITY = "PRIORITY" 

659 MOBILE = "MOBILE" 

660 

661 

662class CommandType(StrEnum): 

663 SNOOZE = "SNOOZE" 

664 SILENCE = "SILENCE" 

665 NORMAL = "NORMAL" 

666 

667 

668class MessageOnlyPolicy(StrEnum): 

669 STANDARD = "STANDARD" # independent title and message 

670 USE_TITLE = "USE_TITLE" # use title in place of message, no title 

671 # use combined title and message as message, no title 

672 COMBINE_TITLE = "COMBINE_TITLE" 

673 

674 

675class DebugTrace: 

676 def __init__( 

677 self, 

678 message: str | None, 

679 title: str | None, 

680 data: dict[str, Any] | None, 

681 target: dict[str, list[str]] | list[str] | str | None, 

682 ) -> None: 

683 self.message: str | None = message 

684 self.title: str | None = title 

685 self.data: dict[str, Any] | None = dict(data) if data else data 

686 self.target: dict[str, list[str]] | list[str] | str | None = list(target) if target else target 

687 self.resolved: dict[str, dict[str, Any]] = {} 

688 self.delivery_selection: dict[str, list[str]] = {} 

689 self.delivery_artefacts: dict[str, Any] = {} 

690 self.delivery_exceptions: dict[str, Any] = {} 

691 self._last_stage: dict[str, str] = {} 

692 self._last_target: dict[str, Any] = {} 

693 

694 def contents(self, **_kwargs: Any) -> dict[str, Any]: 

695 results: dict[str, Any] = { 

696 "arguments": { 

697 "message": self.message, 

698 "title": self.title, 

699 "data": self.data, 

700 "target": self.target, 

701 }, 

702 "delivery_selection": self.delivery_selection, 

703 "resolved": self.resolved, 

704 } 

705 if self.delivery_artefacts: 

706 results["delivery_artefacts"] = self.delivery_artefacts 

707 if self.delivery_artefacts: 

708 results["delivery_exceptions"] = self.delivery_exceptions 

709 return results 

710 

711 def record_target(self, delivery_name: str, stage: str, computed: Target | list[Target]) -> None: 

712 """Debug support for recording detailed target resolution in archived notification""" 

713 self.resolved.setdefault(delivery_name, {}) 

714 self.resolved[delivery_name].setdefault(stage, {}) 

715 self._last_target.setdefault(delivery_name, {}) 

716 self._last_target[delivery_name].setdefault(stage, {}) 

717 if isinstance(computed, Target): 

718 combined = computed 

719 else: 

720 combined = Target() 

721 for target in ensure_list(computed): 

722 combined += target 

723 new_target: dict[str, Any] = combined.as_dict() 

724 result: str | dict[str, Any] = new_target 

725 if self._last_stage.get(delivery_name): 

726 last_target = self._last_target[delivery_name][self._last_stage[delivery_name]] 

727 if last_target is not None and last_target == result: 

728 result = "NO_CHANGE" 

729 

730 self.resolved[delivery_name][stage] = result 

731 self._last_stage[delivery_name] = stage 

732 self._last_target[delivery_name][stage] = new_target 

733 

734 def record_delivery_selection(self, stage: str, delivery_selection: list[str]) -> None: 

735 """Debug support for recording detailed target resolution in archived notification""" 

736 self.delivery_selection[stage] = delivery_selection 

737 

738 def record_delivery_artefact(self, delivery: str, artefact_name: str, artefact: Any) -> None: 

739 self.delivery_artefacts.setdefault(delivery, {}) 

740 self.delivery_artefacts[delivery][artefact_name] = artefact 

741 

742 def record_delivery_exception(self, delivery: str, context: str, exception: Exception) -> None: 

743 self.delivery_exceptions.setdefault(delivery, {}) 

744 self.delivery_exceptions[delivery].setdefault(context, []) 

745 self.delivery_exceptions[delivery][context].append(format_exception(exception))