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

1import logging 

2import re 

3from collections.abc import Iterable, Sequence 

4from dataclasses import dataclass, field 

5from enum import IntFlag, StrEnum, auto 

6from traceback import format_exception 

7from typing import Any, ClassVar 

8 

9import voluptuous as vol 

10from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN 

11 

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

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

14from homeassistant.const import ( 

15 ATTR_AREA_ID, 

16 ATTR_DEVICE_ID, 

17 ATTR_ENTITY_ID, 

18 ATTR_FLOOR_ID, 

19 ATTR_LABEL_ID, 

20 CONF_ACTION, 

21 CONF_ALIAS, 

22 CONF_DEBUG, 

23 CONF_ENABLED, 

24 CONF_OPTIONS, 

25 CONF_TARGET, 

26 STATE_HOME, 

27 STATE_NOT_HOME, 

28) 

29from homeassistant.core import valid_entity_id 

30from homeassistant.helpers.typing import ConfigType, TemplateVarsType 

31 

32from . import ( 

33 ATTR_EMAIL, 

34 ATTR_MOBILE_APP_ID, 

35 ATTR_PERSON_ID, 

36 ATTR_PHONE, 

37 CONF_DATA, 

38 CONF_DELIVERY_DEFAULTS, 

39 CONF_DEVICE_DISCOVERY, 

40 CONF_DEVICE_DOMAIN, 

41 CONF_DEVICE_MODEL_EXCLUDE, 

42 CONF_DEVICE_MODEL_INCLUDE, 

43 CONF_PRIORITY, 

44 CONF_SELECTION, 

45 CONF_SELECTION_RANK, 

46 CONF_TARGET_REQUIRED, 

47 CONF_TARGET_USAGE, 

48 OPTION_DEVICE_DISCOVERY, 

49 OPTION_DEVICE_DOMAIN, 

50 OPTION_DEVICE_MODEL_SELECT, 

51 PRIORITY_MEDIUM, 

52 PRIORITY_VALUES, 

53 RE_DEVICE_ID, 

54 SELECT_EXCLUDE, 

55 SELECT_INCLUDE, 

56 SELECTION_DEFAULT, 

57 TARGET_USE_ON_NO_ACTION_TARGETS, 

58 SelectionRank, 

59 phone, 

60) 

61from .common import ensure_list 

62 

63_LOGGER = logging.getLogger(__name__) 

64 

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

66MOBILE_APP_DOMAIN = "mobile_app" 

67 

68 

69class TransportFeature(IntFlag): 

70 MESSAGE = 1 

71 TITLE = 2 

72 IMAGES = 4 

73 VIDEO = 8 

74 ACTIONS = 16 

75 TEMPLATE_FILE = 32 

76 SNAPSHOT_IMAGE = 64 

77 

78 

79class Target: 

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

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

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

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

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

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

86 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

87 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

88 

89 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES 

90 

91 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_" 

92 

93 def __init__( 

94 self, 

95 target: str 

96 | list[str] 

97 | dict[str, str] 

98 | dict[str, Sequence[str]] 

99 | dict[str, list[str]] 

100 | dict[str, str | list[str]] 

101 | None = None, 

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

103 target_specific_data: bool = False, 

104 ) -> None: 

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

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

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

108 

109 matched: list[str] 

110 

111 if isinstance(target, str): 

112 target = [target] 

113 

114 if target is None: 

115 pass # empty constructor is valid case for target building 

116 elif isinstance(target, list): 

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

118 targets_left = list(target) 

119 for category in self.AUTO_CATEGORIES: 

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

121 if validator is not None: 

122 matched = [] 

123 for t in targets_left: 

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

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

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

127 matched.append(t) 

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

129 else: 

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

131 if not targets_left: 

132 break 

133 if targets_left: 

134 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets_left 

135 

136 elif isinstance(target, dict): 

137 for category in target: 

138 targets = ensure_list(target[category]) 

139 if not targets: 

140 continue 

141 if category in self.AUTO_CATEGORIES: 

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

143 if validator is not None: 

144 for t in targets: 

145 if validator(t): 

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

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

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

149 else: 

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

151 else: 

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

153 

154 elif category in self.CATEGORIES: 

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

156 self.targets[category] = targets 

157 else: 

158 # custom categories 

159 self.targets[category] = targets 

160 else: 

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

162 

163 if target_data and target_specific_data: 

164 self.target_specific_data = {} 

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

166 for t in targets: 

167 self.target_specific_data[category, t] = target_data 

168 if target_data and not target_specific_data: 

169 self.target_data = target_data 

170 

171 # Targets by category 

172 

173 @property 

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

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

176 

177 @property 

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

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

180 

181 @property 

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

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

184 

185 @property 

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

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

188 

189 @property 

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

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

192 

193 @property 

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

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

196 

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

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

199 

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

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

202 

203 @property 

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

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

206 

207 @property 

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

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

210 

211 @property 

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

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

214 

215 # Selectors / validators 

216 

217 @classmethod 

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

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

220 

221 @classmethod 

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

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

224 

225 @classmethod 

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

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

228 

229 @classmethod 

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

231 try: 

232 return phone(target) is not None 

233 except vol.Invalid: 

234 return False 

235 

236 @classmethod 

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

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

239 

240 @classmethod 

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

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

243 

244 @classmethod 

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

246 try: 

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

248 except vol.Invalid: 

249 return False 

250 

251 def has_targets(self) -> bool: 

252 return any(targets for category, targets in self.targets.items()) 

253 

254 def has_resolved_target(self) -> bool: 

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

256 

257 def has_unknown_targets(self) -> bool: 

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

259 

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

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

262 

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

264 result: list[str] = [] 

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

266 if category not in self.INDIRECT_CATEGORIES: 

267 result.extend(targets) 

268 return result 

269 

270 def hash_resolved(self) -> int: 

271 targets = [] 

272 for category in self.targets: 

273 if category not in self.INDIRECT_CATEGORIES: 

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

275 return hash(tuple(targets)) 

276 

277 @property 

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

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

280 

281 def direct(self) -> "Target": 

282 t = Target( 

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

284 target_data=self.target_data, 

285 ) 

286 if self.target_specific_data: 

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

288 return t 

289 

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

291 targets = ensure_list(targets) 

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

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

294 

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

296 targets = ensure_list(targets) 

297 if category in self.targets: 

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

299 

300 def safe_copy(self) -> "Target": 

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

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

303 return t 

304 

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

306 if not self.target_specific_data: 

307 result = self.safe_copy() 

308 result.target_specific_data = None 

309 return [result] 

310 results: list[Target] = [] 

311 default: Target = self.safe_copy() 

312 default.target_specific_data = None 

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

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

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

316 if last_found is None: 

317 last_found = data 

318 collected = {category: [target]} 

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

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

321 results.append(new_target) 

322 default -= new_target 

323 last_found = data 

324 collected = {category: [target]} 

325 else: 

326 collected.setdefault(category, []) 

327 collected[category].append(target) 

328 new_target = Target(collected, target_data=last_found) 

329 results.append(new_target) 

330 default -= new_target 

331 if default.has_targets(): 

332 results.append(default) 

333 return results 

334 

335 def __len__(self) -> int: 

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

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

338 

339 def __add__(self, other: "Target") -> "Target": 

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

341 new = Target() 

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

343 for category in categories: 

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

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

346 

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

348 if other.target_data: 

349 if new.target_data is None: 

350 new.target_data = dict(other.target_data) 

351 else: 

352 new.target_data.update(other.target_data) 

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

354 if other.target_specific_data: 

355 if new.target_specific_data is None: 

356 new.target_specific_data = dict(other.target_specific_data) 

357 else: 

358 new.target_specific_data.update(other.target_specific_data) 

359 return new 

360 

361 def __sub__(self, other: "Target") -> "Target": 

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

363 new = Target() 

364 new.target_data = self.target_data 

365 if self.target_specific_data: 

366 new.target_specific_data = { 

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

368 } 

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

370 for category in categories: 

371 new.targets[category] = [] 

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

373 

374 return new 

375 

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

377 """Compare two targets""" 

378 if other is self: 

379 return True 

380 if other is None: 

381 return False 

382 if not isinstance(other, Target): 

383 return NotImplemented 

384 if self.target_data != other.target_data: 

385 return False 

386 if self.target_specific_data != other.target_specific_data: 

387 return False 

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

389 

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

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

392 

393 

394class TransportConfig: 

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

396 conf = conf or {} 

397 if class_config is not None: 

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

399 self.alias = conf.get(CONF_ALIAS) 

400 self.delivery_defaults: DeliveryConfig = DeliveryConfig( 

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

402 ) 

403 else: 

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

405 self.alias = conf.get(CONF_ALIAS) 

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

407 

408 # deprecation support 

409 device_domain = conf.get(CONF_DEVICE_DOMAIN) 

410 if device_domain is not None: 

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

412 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain 

413 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE) 

414 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE) 

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

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

417 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = { 

418 SELECT_INCLUDE: device_model_include, 

419 SELECT_EXCLUDE: device_model_exclude, 

420 } 

421 device_discovery = conf.get(CONF_DEVICE_DISCOVERY) 

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

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

424 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery 

425 

426 

427class DeliveryCustomization: 

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

429 config = config or {} 

430 # perhaps should be false for wildcards 

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

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

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

434 self.target: Target | None 

435 

436 if config.get(CONF_TARGET): 

437 if self.data: 

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

439 else: 

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

441 else: 

442 self.target = None 

443 

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

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

446 

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

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

449 

450 

451class SelectionRule: 

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

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

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

455 if config is None: 

456 return 

457 if isinstance(config, SelectionRule): 

458 self.include = config.include 

459 self.exclude = config.exclude 

460 elif isinstance(config, str): 

461 self.include = [config] 

462 elif isinstance(config, list): 

463 self.include = config 

464 else: 

465 if config.get(SELECT_INCLUDE): 

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

467 if config.get(SELECT_EXCLUDE): 

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

469 

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

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

472 return True 

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

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

475 return False 

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

477 return False 

478 else: 

479 if self.exclude is not None: 

480 for vv in v: 

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

482 return False 

483 if self.include is not None: 

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

485 return True 

486 

487 

488class DeliveryConfig: 

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

490 

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

492 

493 if delivery_defaults is not None: 

494 # use transport defaults where no delivery level override 

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

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

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

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

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

500 

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

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

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

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

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

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

507 # only override options not set in config 

508 if isinstance(delivery_defaults.options, dict): 

509 for opt in delivery_defaults.options: 

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

511 else: 

512 # construct the transport defaults 

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

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

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

516 self.action = conf.get(CONF_ACTION) 

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

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

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

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

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

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

523 

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

525 return { 

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

527 CONF_ACTION: self.action, 

528 CONF_OPTIONS: self.options, 

529 CONF_DATA: self.data, 

530 CONF_SELECTION: self.selection, 

531 CONF_PRIORITY: self.priority, 

532 CONF_SELECTION_RANK: str(self.selection_rank), 

533 CONF_TARGET_REQUIRED: str(self.target_required), 

534 CONF_TARGET_USAGE: self.target_usage, 

535 } 

536 

537 def __repr__(self) -> str: 

538 """Log friendly representation""" 

539 return str(self.as_dict()) 

540 

541 

542@dataclass 

543class ConditionVariables: 

544 """Variables presented to all condition evaluations 

545 

546 Attributes 

547 ---------- 

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

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

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

551 notification_priority (str): Priority of the notification 

552 notification_message (str): Message of the notification 

553 notification_title (str): Title of the notification 

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

555 

556 """ 

557 

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

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

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

561 notification_priority: str = PRIORITY_MEDIUM 

562 notification_message: str | None = "" 

563 notification_title: str | None = "" 

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

565 

566 def __init__( 

567 self, 

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

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

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

571 delivery_priority: str | None = PRIORITY_MEDIUM, 

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

573 message: str | None = None, 

574 title: str | None = None, 

575 ) -> None: 

576 occupiers = occupiers or {} 

577 self.occupancy = [] 

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

579 self.occupancy.append("ALL_HOME") 

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

581 self.occupancy.append("ALL_AWAY") 

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

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

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

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

586 self.applied_scenarios = applied_scenarios or [] 

587 self.required_scenarios = required_scenarios or [] 

588 self.constrain_scenarios = constrain_scenarios or [] 

589 self.notification_priority = delivery_priority or PRIORITY_MEDIUM 

590 self.notification_message = message 

591 self.notification_title = title 

592 

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

594 return { 

595 "applied_scenarios": self.applied_scenarios, 

596 "required_scenarios": self.required_scenarios, 

597 "constrain_scenarios": self.constrain_scenarios, 

598 "notification_message": self.notification_message, 

599 "notification_title": self.notification_title, 

600 "notification_priority": self.notification_priority, 

601 "occupancy": self.occupancy, 

602 } 

603 

604 

605class SuppressionReason(StrEnum): 

606 SNOOZED = "SNOOZED" 

607 DUPE = "DUPE" 

608 NO_SCENARIO = "NO_SCENARIO" 

609 NO_ACTION = "NO_ACTION" 

610 NO_TARGET = "NO_TARGET" 

611 TRANSPORT_DISABLED = "TRANSPORT_DISABLED" 

612 PRIORITY = "PRIORITY" 

613 DELIVERY_CONDITION = "DELIVERY_CONDITION" 

614 UNKNOWN = "UNKNOWN" 

615 

616 

617class TargetRequired(StrEnum): 

618 ALWAYS = auto() 

619 NEVER = auto() 

620 OPTIONAL = auto() 

621 

622 @classmethod 

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

624 """Backward compatibility for binary values""" 

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

626 return cls.ALWAYS 

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

628 return cls.OPTIONAL 

629 return None 

630 

631 

632class TargetType(StrEnum): 

633 pass 

634 

635 

636class GlobalTargetType(TargetType): 

637 NONCRITICAL = "NONCRITICAL" 

638 EVERYTHING = "EVERYTHING" 

639 

640 

641class RecipientType(StrEnum): 

642 USER = "USER" 

643 EVERYONE = "EVERYONE" 

644 

645 

646class QualifiedTargetType(TargetType): 

647 TRANSPORT = "TRANSPORT" 

648 DELIVERY = "DELIVERY" 

649 CAMERA = "CAMERA" 

650 PRIORITY = "PRIORITY" 

651 MOBILE = "MOBILE" 

652 

653 

654class CommandType(StrEnum): 

655 SNOOZE = "SNOOZE" 

656 SILENCE = "SILENCE" 

657 NORMAL = "NORMAL" 

658 

659 

660class MessageOnlyPolicy(StrEnum): 

661 STANDARD = "STANDARD" # independent title and message 

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

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

664 COMBINE_TITLE = "COMBINE_TITLE" 

665 

666 

667class DebugTrace: 

668 def __init__( 

669 self, 

670 message: str | None, 

671 title: str | None, 

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

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

674 ) -> None: 

675 self.message: str | None = message 

676 self.title: str | None = title 

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

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

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

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