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 .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 

62_LOGGER = logging.getLogger(__name__) 

63 

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

65MOBILE_APP_DOMAIN = "mobile_app" 

66 

67 

68class TransportFeature(IntFlag): 

69 MESSAGE = 1 

70 TITLE = 2 

71 IMAGES = 4 

72 VIDEO = 8 

73 ACTIONS = 16 

74 TEMPLATE_FILE = 32 

75 SNAPSHOT_IMAGE = 64 

76 

77 

78class Target: 

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

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

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

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

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

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

85 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

86 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES 

87 

88 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES 

89 

90 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_" 

91 

92 def __init__( 

93 self, 

94 target: str 

95 | list[str] 

96 | dict[str, str] 

97 | dict[str, Sequence[str]] 

98 | dict[str, list[str]] 

99 | dict[str, str | list[str]] 

100 | None = None, 

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

102 target_specific_data: bool = False, 

103 ) -> None: 

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

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

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

107 

108 matched: list[str] 

109 

110 if isinstance(target, str): 

111 target = [target] 

112 

113 if target is None: 

114 pass # empty constructor is valid case for target building 

115 elif isinstance(target, list): 

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

117 targets_left = list(target) 

118 for category in self.AUTO_CATEGORIES: 

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

120 if validator is not None: 

121 matched = [] 

122 for t in targets_left: 

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

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

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

126 matched.append(t) 

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

128 else: 

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

130 if not targets_left: 

131 break 

132 if targets_left: 

133 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets_left 

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 validator = getattr(self, f"is_{category}", None) 

142 if validator is not None: 

143 for t in targets: 

144 if validator(t): 

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

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

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

148 else: 

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

150 else: 

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

152 

153 elif category in self.CATEGORIES: 

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

155 self.targets[category] = targets 

156 else: 

157 # custom categories 

158 self.targets[category] = targets 

159 else: 

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

161 

162 if target_data and target_specific_data: 

163 self.target_specific_data = {} 

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

165 for t in targets: 

166 self.target_specific_data[category, t] = target_data 

167 if target_data and not target_specific_data: 

168 self.target_data = target_data 

169 

170 # Targets by category 

171 

172 @property 

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

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

175 

176 @property 

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

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

179 

180 @property 

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

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

183 

184 @property 

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

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

187 

188 @property 

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

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

191 

192 @property 

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

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

195 

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

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

198 

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

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

201 

202 @property 

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

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

205 

206 @property 

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

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

209 

210 @property 

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

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

213 

214 # Selectors / validators 

215 

216 @classmethod 

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

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

219 

220 @classmethod 

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

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

223 

224 @classmethod 

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

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

227 

228 @classmethod 

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

230 try: 

231 return phone(target) is not None 

232 except vol.Invalid: 

233 return False 

234 

235 @classmethod 

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

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

238 

239 @classmethod 

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

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

242 

243 @classmethod 

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

245 try: 

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

247 except vol.Invalid: 

248 return False 

249 

250 def has_targets(self) -> bool: 

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

252 

253 def has_resolved_target(self) -> bool: 

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

255 

256 def has_unknown_targets(self) -> bool: 

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

258 

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

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

261 

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

263 result: list[str] = [] 

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

265 if category not in self.INDIRECT_CATEGORIES: 

266 result.extend(targets) 

267 return result 

268 

269 def hash_resolved(self) -> int: 

270 targets = [] 

271 for category in self.targets: 

272 if category not in self.INDIRECT_CATEGORIES: 

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

274 return hash(tuple(targets)) 

275 

276 @property 

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

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

279 

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

281 t = Target( 

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

283 target_data=self.target_data, 

284 ) 

285 if self.target_specific_data: 

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

287 return t 

288 

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

290 targets = ensure_list(targets) 

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

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

293 

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

295 targets = ensure_list(targets) 

296 if category in self.targets: 

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

298 

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

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

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

302 return t 

303 

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

305 if not self.target_specific_data: 

306 result = self.safe_copy() 

307 result.target_specific_data = None 

308 return [result] 

309 results: list[Target] = [] 

310 default: Target = self.safe_copy() 

311 default.target_specific_data = None 

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

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

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

315 if last_found is None: 

316 last_found = data 

317 collected = {category: [target]} 

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

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

320 results.append(new_target) 

321 default -= new_target 

322 last_found = data 

323 collected = {category: [target]} 

324 else: 

325 collected.setdefault(category, []) 

326 collected[category].append(target) 

327 new_target = Target(collected, target_data=last_found) 

328 results.append(new_target) 

329 default -= new_target 

330 if default.has_targets(): 

331 results.append(default) 

332 return results 

333 

334 def __len__(self) -> int: 

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

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

337 

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

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

340 new = Target() 

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

342 for category in categories: 

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

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

345 

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

347 if other.target_data: 

348 if new.target_data is None: 

349 new.target_data = dict(other.target_data) 

350 else: 

351 new.target_data.update(other.target_data) 

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

353 if other.target_specific_data: 

354 if new.target_specific_data is None: 

355 new.target_specific_data = dict(other.target_specific_data) 

356 else: 

357 new.target_specific_data.update(other.target_specific_data) 

358 return new 

359 

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

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

362 new = Target() 

363 new.target_data = self.target_data 

364 if self.target_specific_data: 

365 new.target_specific_data = { 

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

367 } 

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

369 for category in categories: 

370 new.targets[category] = [] 

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

372 

373 return new 

374 

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

376 """Compare two targets""" 

377 if other is self: 

378 return True 

379 if other is None: 

380 return False 

381 if not isinstance(other, Target): 

382 return NotImplemented 

383 if self.target_data != other.target_data: 

384 return False 

385 if self.target_specific_data != other.target_specific_data: 

386 return False 

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

388 

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

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

391 

392 

393class TransportConfig: 

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

395 conf = conf or {} 

396 if class_config is not None: 

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

398 self.alias = conf.get(CONF_ALIAS) 

399 self.delivery_defaults: DeliveryConfig = DeliveryConfig( 

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

401 ) 

402 else: 

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

404 self.alias = conf.get(CONF_ALIAS) 

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

406 

407 # deprecation support 

408 device_domain = conf.get(CONF_DEVICE_DOMAIN) 

409 if device_domain is not None: 

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

411 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain 

412 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE) 

413 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE) 

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

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

416 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = { 

417 SELECT_INCLUDE: device_model_include, 

418 SELECT_EXCLUDE: device_model_exclude, 

419 } 

420 device_discovery = conf.get(CONF_DEVICE_DISCOVERY) 

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

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

423 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery 

424 

425 

426class DeliveryCustomization: 

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

428 config = config or {} 

429 # perhaps should be false for wildcards 

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

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

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

433 self.target: Target | None 

434 

435 if config.get(CONF_TARGET): 

436 if self.data: 

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

438 else: 

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

440 else: 

441 self.target = None 

442 

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

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

445 

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

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

448 

449 

450class SelectionRule: 

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

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

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

454 if config is None: 

455 return 

456 if isinstance(config, SelectionRule): 

457 self.include = config.include 

458 self.exclude = config.exclude 

459 elif isinstance(config, str): 

460 self.include = [config] 

461 elif isinstance(config, list): 

462 self.include = config 

463 else: 

464 if config.get(SELECT_INCLUDE): 

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

466 if config.get(SELECT_EXCLUDE): 

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

468 

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

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

471 return True 

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

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

474 return False 

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

476 return False 

477 else: 

478 if self.exclude is not None: 

479 for vv in v: 

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

481 return False 

482 if self.include is not None: 

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

484 return True 

485 

486 

487class DeliveryConfig: 

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

489 

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

491 

492 if delivery_defaults is not None: 

493 # use transport defaults where no delivery level override 

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

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

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

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

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

499 

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

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

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

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

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

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

506 # only override options not set in config 

507 if isinstance(delivery_defaults.options, dict): 

508 for opt in delivery_defaults.options: 

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

510 else: 

511 # construct the transport defaults 

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

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

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

515 self.action = conf.get(CONF_ACTION) 

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

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

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

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

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

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

522 

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

524 return { 

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

526 CONF_ACTION: self.action, 

527 CONF_OPTIONS: self.options, 

528 CONF_DATA: self.data, 

529 CONF_SELECTION: self.selection, 

530 CONF_PRIORITY: self.priority, 

531 CONF_SELECTION_RANK: str(self.selection_rank), 

532 CONF_TARGET_REQUIRED: str(self.target_required), 

533 CONF_TARGET_USAGE: self.target_usage, 

534 } 

535 

536 def __repr__(self) -> str: 

537 """Log friendly representation""" 

538 return str(self.as_dict()) 

539 

540 

541@dataclass 

542class ConditionVariables: 

543 """Variables presented to all condition evaluations 

544 

545 Attributes 

546 ---------- 

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

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

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

550 notification_priority (str): Priority of the notification 

551 notification_message (str): Message of the notification 

552 notification_title (str): Title of the notification 

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

554 

555 """ 

556 

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

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

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

560 notification_priority: str = PRIORITY_MEDIUM 

561 notification_message: str | None = "" 

562 notification_title: str | None = "" 

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

564 

565 def __init__( 

566 self, 

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

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

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

570 delivery_priority: str | None = PRIORITY_MEDIUM, 

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

572 message: str | None = None, 

573 title: str | None = None, 

574 ) -> None: 

575 occupiers = occupiers or {} 

576 self.occupancy = [] 

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

578 self.occupancy.append("ALL_HOME") 

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

580 self.occupancy.append("ALL_AWAY") 

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

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

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

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

585 self.applied_scenarios = applied_scenarios or [] 

586 self.required_scenarios = required_scenarios or [] 

587 self.constrain_scenarios = constrain_scenarios or [] 

588 self.notification_priority = delivery_priority or PRIORITY_MEDIUM 

589 self.notification_message = message 

590 self.notification_title = title 

591 

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

593 return { 

594 "applied_scenarios": self.applied_scenarios, 

595 "required_scenarios": self.required_scenarios, 

596 "constrain_scenarios": self.constrain_scenarios, 

597 "notification_message": self.notification_message, 

598 "notification_title": self.notification_title, 

599 "notification_priority": self.notification_priority, 

600 "occupancy": self.occupancy, 

601 } 

602 

603 

604class SuppressionReason(StrEnum): 

605 SNOOZED = "SNOOZED" 

606 DUPE = "DUPE" 

607 NO_SCENARIO = "NO_SCENARIO" 

608 NO_ACTION = "NO_ACTION" 

609 NO_TARGET = "NO_TARGET" 

610 INVALID_ACTION_DATA = "INVALID_ACTION_DATA" 

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]] = {}