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

1from __future__ import annotations 

2 

3import logging 

4from collections.abc import Mapping 

5from dataclasses import dataclass, field 

6from functools import partial 

7from typing import TYPE_CHECKING, Any 

8 

9import voluptuous as vol 

10from homeassistant.components.person import ATTR_USER_ID 

11from homeassistant.const import ( 

12 ATTR_AREA_ID, 

13 ATTR_ENTITY_ID, 

14 ATTR_FLOOR_ID, 

15 ATTR_LABEL_ID, 

16 CONF_ACTION, 

17 CONF_DEVICE_ID, 

18) 

19from homeassistant.helpers.aiohttp_client import async_get_clientsession 

20from homeassistant.helpers.entity_registry import RegistryEntry 

21from homeassistant.helpers.event import async_track_state_change_event, async_track_time_change, async_track_time_interval 

22from homeassistant.helpers.storage import Store 

23from homeassistant.helpers.target import TargetSelection, async_extract_referenced_entity_ids 

24from homeassistant.util import slugify 

25 

26if TYPE_CHECKING: 

27 import asyncio 

28 from collections.abc import Callable, Iterable, Iterator 

29 

30 import aiohttp 

31 from anyio import Path 

32 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, Service, ServiceResponse, State 

33 from homeassistant.helpers.entity_registry import EntityRegistry 

34 from homeassistant.helpers.typing import ConfigType 

35 from homeassistant.util.event_type import EventType 

36 

37 from .schema import ConditionsFunc 

38 

39import socket 

40import threading 

41from contextlib import contextmanager 

42from datetime import timedelta 

43from typing import TYPE_CHECKING, cast 

44 

45import homeassistant.components.camera as ha_camera 

46import homeassistant.components.image as ha_image 

47import homeassistant.components.trace 

48from homeassistant.components.group import DOMAIN as GROUP_DOMAIN 

49from homeassistant.components.group import expand_entity_ids 

50from homeassistant.components.trace.const import DATA_TRACE 

51from homeassistant.components.trace.models import ActionTrace 

52from homeassistant.components.trace.util import async_store_trace 

53from homeassistant.core import Context as HomeAssistantContext 

54from homeassistant.core import HomeAssistant, SupportsResponse 

55from homeassistant.exceptions import ConditionError, ConditionErrorContainer, HomeAssistantError, IntegrationError 

56from homeassistant.helpers import condition as condition_helper 

57from homeassistant.helpers import device_registry as dr 

58from homeassistant.helpers import entity_registry as er 

59from homeassistant.helpers import issue_registry as ir 

60from homeassistant.helpers.json import json_dumps 

61from homeassistant.helpers.network import get_url 

62from homeassistant.helpers.template import Template 

63from homeassistant.helpers.trace import trace_get, trace_path 

64from homeassistant.helpers.typing import ConfigType 

65 

66from . import DOMAIN 

67from .const import CONF_DEVICE_LABELS, CONF_DEVICE_TRACKER, CONF_MOBILE_APP_ID 

68from .model import ConditionVariables, SelectionRule 

69 

70if TYPE_CHECKING: 

71 from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry 

72 

73# avoid importing from homeassistant.components.mobile_app.const and triggering dependency chain 

74 

75CONF_USER_ID = "user_id" 

76ATTR_OS_NAME = "os_name" 

77ATTR_OS_VERSION = "os_version" 

78ATTR_APP_VERSION = "app_version" 

79ATTR_DEVICE_NAME = "device_name" 

80ATTR_MANUFACTURER = "manufacturer" 

81ATTR_MODEL = "model" 

82 

83_LOGGER = logging.getLogger(__name__) 

84 

85 

86@dataclass 

87class TrackedDeviceDetails: 

88 device_id: str 

89 device_labels: list[str] | None = None 

90 mobile_app_id: str | None = None 

91 device_name: str | None = None 

92 device_tracker: str | None = None 

93 action: str | None = None 

94 user_id: str | None = None 

95 area_id: str | None = None 

96 manufacturer: str | None = None 

97 model: str | None = None 

98 os_name: str | None = None 

99 os_version: str | None = None 

100 app_version: str | None = None 

101 identifiers: set[tuple[str, str]] | None = None 

102 

103 def as_dict(self) -> dict[str, str | list[str] | None]: 

104 return { 

105 CONF_MOBILE_APP_ID: self.mobile_app_id, 

106 ATTR_DEVICE_NAME: self.device_name, 

107 CONF_DEVICE_ID: self.device_id, 

108 CONF_USER_ID: self.user_id, 

109 CONF_DEVICE_TRACKER: self.device_tracker, 

110 CONF_ACTION: self.action, 

111 ATTR_OS_NAME: self.os_name, 

112 ATTR_OS_VERSION: self.os_version, 

113 ATTR_APP_VERSION: self.app_version, 

114 ATTR_MANUFACTURER: self.manufacturer, 

115 ATTR_MODEL: self.model, 

116 CONF_DEVICE_LABELS: self.device_labels, 

117 } 

118 

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

120 """Test support""" 

121 as_dict = getattr(other, "as_dict", None) 

122 return other is not None and callable(as_dict) and as_dict() == self.as_dict() 

123 

124 

125@dataclass 

126class TargetSelectorResolution: 

127 """Outcome of resolving area/floor/label selectors to the entities they reference""" 

128 

129 entity_ids: list[str] = field(default_factory=list) 

130 missing_areas: list[str] = field(default_factory=list) 

131 missing_floors: list[str] = field(default_factory=list) 

132 missing_labels: list[str] = field(default_factory=list) 

133 

134 def has_missing(self) -> bool: 

135 return bool(self.missing_areas or self.missing_floors or self.missing_labels) 

136 

137 

138def ha_device_info(entry_id: str) -> dr.DeviceInfo: 

139 """Home Assistant device-registry DeviceInfo for the single 'SuperNotify' device. 

140 

141 Groups the platform entities in binary_sensor.py/sensor.py/switch.py (scenario/recipient 

142 state and control, notification/failure counters) under one device in the HA device 

143 registry. Unrelated to `TrackedDeviceDetails` above, which describes a discovered mobile_app 

144 device for targeting. 

145 """ 

146 return dr.DeviceInfo(identifiers={(DOMAIN, entry_id)}, name="SuperNotify", manufacturer="SuperNotify") 

147 

148 

149class HomeAssistantAPI: 

150 def __init__(self, hass: HomeAssistant) -> None: 

151 self._hass: HomeAssistant = hass 

152 self.internal_url: str = "" 

153 self.external_url: str = "" 

154 self.language: str = "" 

155 self.hass_name: str = "!UNDEFINED!" 

156 self.__entity_registry: er.EntityRegistry | None = None 

157 self.__device_registry: dr.DeviceRegistry | None = None 

158 self._service_info: dict[tuple[str, str], Any] = {} 

159 self.unsubscribes: list[CALLBACK_TYPE] = [] 

160 self._mobile_apps_by_tracker: dict[str, TrackedDeviceDetails] = {} 

161 self._mobile_apps_by_app_id: dict[str, TrackedDeviceDetails] = {} 

162 self._mobile_apps_by_device_id: dict[str, TrackedDeviceDetails] = {} 

163 self._mobile_apps_by_user_id: dict[str, list[TrackedDeviceDetails]] = {} 

164 

165 def initialize(self) -> None: 

166 self.hass_name = self._hass.config.location_name 

167 self.language = self._hass.config.language 

168 try: 

169 self.internal_url = get_url(self._hass, prefer_external=False) 

170 except Exception as e: 

171 self.internal_url = f"http://{socket.gethostname()}" 

172 _LOGGER.warning("SUPERNOTIFY Internal hass url not available, defaulting to %s: %s", self.internal_url, e) 

173 try: 

174 self.external_url = get_url(self._hass, prefer_external=True) 

175 except Exception as e: 

176 _LOGGER.warning("SUPERNOTIFY External hass url not available, defaulting to internal url: %s", e) 

177 self.external_url = self.internal_url 

178 

179 self.build_mobile_app_cache() 

180 

181 _LOGGER.debug( 

182 "SUPERNOTIFY Configured for HomeAssistant instance %s at %s , %s", 

183 self.hass_name, 

184 self.internal_url, 

185 self.external_url, 

186 ) 

187 

188 if not self.internal_url or not self.internal_url.startswith("http"): 

189 _LOGGER.warning("SUPERNOTIFY Invalid internal hass url %s", self.internal_url) 

190 

191 def disconnect(self) -> None: 

192 while self.unsubscribes: 

193 unsub = self.unsubscribes.pop() 

194 try: 

195 _LOGGER.debug("SUPERNOTIFY Unsubscribing: %.100s", unsub) 

196 unsub() 

197 except Exception as e: 

198 _LOGGER.error("SUPERNOTIFY Failed to unsubscribe: %s", e) 

199 _LOGGER.debug("SUPERNOTIFY Disconnection complete") 

200 

201 def subscribe_event(self, event: EventType | str, callback: Callable) -> None: 

202 self.unsubscribes.append(self._hass.bus.async_listen(event, callback)) 

203 

204 def subscribe_state(self, entity_ids: str | Iterable[str], callback: Callable) -> None: 

205 self.unsubscribes.append(async_track_state_change_event(self._hass, entity_ids, callback)) 

206 

207 def subscribe_time(self, hour: int, minute: int, second: int, callback: Callable) -> None: 

208 self.unsubscribes.append(async_track_time_change(self._hass, callback, hour=hour, minute=minute, second=second)) 

209 

210 def subscribe_interval(self, seconds: int, callback: Callable) -> None: 

211 self.unsubscribes.append(async_track_time_interval(self._hass, callback, timedelta(seconds=seconds))) 

212 

213 def in_hass_loop(self) -> bool: 

214 return self.hass_avail("loop_thread_id") and self._hass.loop_thread_id == threading.get_ident() 

215 

216 def get_state(self, entity_id: str) -> State | None: 

217 return self._hass.states.get(entity_id) 

218 

219 def is_state(self, entity_id: str, state: str) -> bool: 

220 return self._hass.states.is_state(entity_id, state) 

221 

222 def has_service(self, domain: str, service: str) -> bool: 

223 return self._hass.services.has_service(domain, service) 

224 

225 def entity_ids_for_domain(self, domain: str) -> list[str]: 

226 return self._hass.states.async_entity_ids(domain) 

227 

228 async def async_real_user_ids(self) -> dict[str, str]: 

229 """Active, non-system HA user ids mapped to their login username - for discovering 

230 recipients that have no Person record (see CONF_USER_ID in const.py), excluding 

231 internal accounts like Supervisor/Home Assistant Content that aren't real people.""" 

232 users = await self._hass.auth.async_get_users() 

233 result: dict[str, str] = {} 

234 for user in users: 

235 if not user.is_active or user.system_generated: 

236 continue 

237 # same lookup HA's own user-management API uses (homeassistant/components/config/auth.py) 

238 username = next( 

239 (cred.data.get("username") for cred in user.credentials if cred.auth_provider_type == "homeassistant"), 

240 None, 

241 ) 

242 result[user.id] = username or user.name or user.id 

243 return result 

244 

245 def platform_for_entity(self, entity_id: str) -> str | None: 

246 """The integration that registered this entity (RegistryEntry.platform), if any.""" 

247 entity_registry: EntityRegistry | None = self._entity_registry() 

248 if entity_registry: 

249 reg_entry: RegistryEntry | None = entity_registry.async_get(entity_id) if entity_registry else None 

250 if reg_entry: 

251 return reg_entry.platform 

252 return None 

253 

254 def entity_ids_for_platform( 

255 self, domain: str, platform: str, device_model_select: str | list[str] | dict | SelectionRule | None = None 

256 ) -> list[str]: 

257 """entity_ids in `domain` (e.g. "notify") registered by a specific integration. 

258 

259 Reads the entity registry directly (not the state machine), so a freshly 

260 registered entity counts even before it has reported a first state. 

261 

262 `device_model_select` optionally filters by the backing device's model (e.g. 

263 `{"exclude": ["Speaker Group"]}`), same include/exclude rule shape used elsewhere. 

264 """ 

265 entity_registry = self._entity_registry() 

266 if not entity_registry: 

267 return [] 

268 entries: list[RegistryEntry] = [ 

269 e for e in entity_registry.entities.values() if e.domain == domain and e.platform == platform 

270 ] 

271 if device_model_select is not None: 

272 model_filter = SelectionRule(device_model_select) 

273 entries = [e for e in entries if model_filter.match(self._device_model(e.device_id))] 

274 return [e.entity_id for e in entries] 

275 

276 def _device_model(self, device_id: str | None) -> str | None: 

277 if device_id is None: 

278 return None 

279 dev_entry = self.find_device(device_id) 

280 return dev_entry.model if dev_entry else None 

281 

282 async def async_get_camera_image(self, entity_id: str, timeout: int = 10) -> ha_camera.Image | None: 

283 """Fetch a still image directly from a camera entity, via HA's own camera component API, 

284 rather than triggering the camera.snapshot service and polling the filesystem for the 

285 resulting file to appear.""" 

286 try: 

287 return await ha_camera.async_get_image(self._hass, entity_id, timeout=timeout) 

288 except HomeAssistantError as e: 

289 _LOGGER.warning("SUPERNOTIFY Unable to get camera image for %s: %s", entity_id, e) 

290 return None 

291 

292 async def async_get_image_entity_image(self, entity_id: str, timeout: int = 10) -> ha_image.Image | None: 

293 """Fetch a still image directly from an image entity, via HA's own image component API.""" 

294 try: 

295 return await ha_image.async_get_image(self._hass, entity_id, timeout=timeout) 

296 except HomeAssistantError as e: 

297 _LOGGER.warning("SUPERNOTIFY Unable to get image from entity %s: %s", entity_id, e) 

298 return None 

299 

300 async def load_storage(self, key: str, version: int = 1) -> Any: # ruff: ignore[any-type] 

301 """Load integration state previously persisted to Home Assistant's .storage/ area, 

302 via HA's own Store helper, or None if nothing has been persisted yet for this key.""" 

303 try: 

304 return await Store[Any](self._hass, version, key).async_load() 

305 except Exception as e: 

306 _LOGGER.warning("SUPERNOTIFY Unable to load storage %s: %s", key, e) 

307 return None 

308 

309 def save_storage(self, key: str, data: Any, version: int = 1) -> None: # ruff: ignore[any-type] 

310 """Persist integration state to Home Assistant's .storage/ area, via HA's own Store 

311 helper. Fire-and-forget: the write happens in a tracked background task rather than 

312 blocking the caller, since this is called from both sync and async contexts.""" 

313 store: Store[Any] = Store(self._hass, version, key) 

314 self._hass.async_create_task(store.async_save(data), f"supernotify_save_{key}") 

315 

316 def create_job(self, func: Callable, *args: Any) -> asyncio.Future[Any]: 

317 """Wrap a blocking function call in a HomeAssistant awaitable job""" 

318 return self._hass.async_add_executor_job(func, *args) 

319 

320 def fire_event( 

321 self, event_name: str, event_data: dict[str, Any] | None = None, context: HomeAssistantContext | None = None 

322 ) -> None: 

323 self._hass.bus.async_fire(event_name, event_data, context=context) 

324 

325 async def call_service( 

326 self, 

327 domain: str, 

328 service: str, 

329 service_data: dict[str, Any] | None = None, 

330 target: dict[str, Any] | None = None, 

331 return_response: bool | None = None, 

332 blocking: bool | None = None, 

333 debug: bool = False, 

334 context: HomeAssistantContext | None = None, 

335 ) -> ServiceResponse | None: 

336 

337 if return_response is None or blocking is None: 

338 # unknown service, for example defined in generic action, check if it supports response 

339 supports_response: SupportsResponse = self.service_info(domain, service) 

340 if supports_response == SupportsResponse.NONE: 

341 return_response = False 

342 elif supports_response == SupportsResponse.ONLY: 

343 return_response = True 

344 else: 

345 return_response = debug 

346 blocking = return_response or debug 

347 

348 response: ServiceResponse | None = await self._hass.services.async_call( 

349 domain, 

350 service, 

351 service_data=service_data, 

352 blocking=blocking, 

353 context=context, 

354 target=target, 

355 return_response=return_response, 

356 ) 

357 if response is not None and debug: 

358 _LOGGER.info("SUPERNOTIFY Service %s.%s response: %s", domain, service, response) 

359 return response 

360 

361 def coerce_schema(self, domain: str, service: str, data: ConfigType) -> ConfigType: 

362 if not data: 

363 return data 

364 try: 

365 if (domain, service) not in self._service_info: 

366 self.service_info(domain, service) 

367 service_info = self._service_info.get((domain, service)) 

368 if not service_info: 

369 _LOGGER.info("SUPERNOTIFY No service found to pre-validate action data for %s.%s", domain, service) 

370 return data 

371 if not service_info.get("schema"): 

372 _LOGGER.info("SUPERNOTIFY No vol schema found to pre-validate action data for %s.%s", domain, service) 

373 return data 

374 service_schema = service_info["schema"] 

375 

376 while service_schema is not None and not ( 

377 isinstance(service_schema, vol.Schema) and isinstance(service_schema.schema, dict) 

378 ): 

379 if isinstance(service_schema, vol.Schema): 

380 # e.g. entity services get schema wrapped as vol.Schema(vol.All(...)) 

381 service_schema = service_schema.schema 

382 elif hasattr(service_schema, "validators") and hasattr(service_schema.validators, "__iter__"): 

383 # e.g. vol.All — strip extras using first dict Schema sub-validator only 

384 # (don't run the full chain; other validators may require target fields not in data) 

385 service_schema = next( 

386 (v for v in service_schema.validators if isinstance(v, vol.Schema) or hasattr(v, "validators")), None 

387 ) 

388 else: 

389 service_schema = None 

390 if not (isinstance(service_schema, vol.Schema) and isinstance(service_schema.schema, dict)): 

391 service_schema = None 

392 _LOGGER.info("SUPERNOTIFY Unable to find schema for %s.%s", domain, service) 

393 

394 if service_schema: 

395 coercing_schema = service_schema.extend( 

396 {}, 

397 extra=vol.REMOVE_EXTRA if service_schema.extra == vol.PREVENT_EXTRA else service_schema.extra, 

398 required=service_schema.required, 

399 ) 

400 cleaned = coercing_schema(data) 

401 else: 

402 return data 

403 if cleaned != data: 

404 _LOGGER.debug("SUPERNOTIFY Coerced data for %s.%s from %s->%s", domain, service, data, cleaned) 

405 return cleaned 

406 except Exception: 

407 _LOGGER.exception("SUPERNOTIFY Unable to coerce %s.%s schema for %s", domain, service, data) 

408 return data 

409 

410 def service_info(self, domain: str, service: str) -> SupportsResponse: 

411 supports_response: SupportsResponse | None = None 

412 try: 

413 if (domain, service) not in self._service_info: 

414 service_objs: dict[str, Service] = self._hass.services.async_services_for_domain(domain) 

415 service_obj: Service | None = service_objs.get(service) 

416 if service_obj: 

417 self._service_info[domain, service] = { 

418 "supports_response": service_obj.supports_response, 

419 "schema": service_obj.schema, 

420 } 

421 service_info: dict[str, Any] = self._service_info.get((domain, service), {}) 

422 supports_response = service_info.get("supports_response") 

423 if supports_response is None: 

424 _LOGGER.debug("SUPERNOTIFY Unable to find service info for %s.%s", domain, service) 

425 

426 except Exception as e: 

427 _LOGGER.warning("SUPERNOTIFY Unable to get service info for %s.%s: %s", domain, service, e) 

428 return supports_response or SupportsResponse.NONE # default to no response 

429 

430 def resolve_target_selectors( 

431 self, 

432 area_ids: list[str] | None = None, 

433 floor_ids: list[str] | None = None, 

434 label_ids: list[str] | None = None, 

435 ) -> TargetSelectorResolution: 

436 """Resolve HA area/floor/label selectors to the entities they reference, through the same 

437 core helper as HA entity actions, so groups are expanded and entities inherit the area of 

438 their device. 

439 """ 

440 resolution = TargetSelectorResolution() 

441 if not (area_ids or floor_ids or label_ids): 

442 return resolution 

443 selection: dict[str, list[str]] = {} 

444 if area_ids: 

445 selection[ATTR_AREA_ID] = list(area_ids) 

446 if floor_ids: 

447 selection[ATTR_FLOOR_ID] = list(floor_ids) 

448 if label_ids: 

449 selection[ATTR_LABEL_ID] = list(label_ids) 

450 try: 

451 selected = async_extract_referenced_entity_ids(self._hass, TargetSelection(selection), expand_group=True) 

452 resolution.entity_ids = sorted(selected.referenced | selected.indirectly_referenced) 

453 resolution.missing_areas = sorted(selected.missing_areas) 

454 resolution.missing_floors = sorted(selected.missing_floors) 

455 resolution.missing_labels = sorted(selected.missing_labels) 

456 except Exception as e: 

457 _LOGGER.warning("SUPERNOTIFY Unable to resolve target selectors %s: %s", selection, e) 

458 return resolution 

459 

460 def find_service(self, domain: str, module: str) -> str | None: 

461 try: 

462 service_objs: dict[str, Service] = self._hass.services.async_services_for_domain(domain) 

463 if service_objs: 

464 for service, domain_obj in service_objs.items(): 

465 if domain_obj.job and domain_obj.job.target: 

466 target = domain_obj.job.target 

467 bound_self = getattr(target, "__self__", None) 

468 target_module: str | None = bound_self.__module__ if bound_self is not None else target.__module__ 

469 if target_module == module: 

470 # Legacy notify platforms with a targets property (e.g. alexa_media_player) 

471 # register extra per-target services (notify.<platform>_<device>) that share 

472 # this same bound method but hard-code their own target, ignoring any target: 

473 # passed by the caller - skip those and hold out for the base platform service. 

474 registered_targets = getattr(bound_self, "registered_targets", None) 

475 if registered_targets is not None and service in registered_targets: 

476 continue 

477 _LOGGER.debug("SUPERNOTIFY Found service %s for domain %s in %s", service, domain, module) 

478 return f"{domain}.{service}" 

479 

480 _LOGGER.debug("SUPERNOTIFY Unable to find service for %s, module %s", domain, module) 

481 except Exception as e: 

482 _LOGGER.warning("SUPERNOTIFY Unable to find service for %s, module %s : %s", domain, module, e) 

483 return None 

484 

485 def find_config_entry_data(self, domain: str) -> Mapping[str, Any] | None: 

486 """Return the data of the first enabled, non-ignored config entry for domain, if any.""" 

487 if not self.hass_avail("config_entries"): 

488 return None 

489 try: 

490 entries = self._hass.config_entries.async_entries(domain, include_ignore=False, include_disabled=False) 

491 if entries: 

492 return entries[0].data 

493 except Exception as e: 

494 _LOGGER.warning("SUPERNOTIFY Unable to find config entry for %s: %s", domain, e) 

495 return None 

496 

497 def http_session(self) -> aiohttp.ClientSession: 

498 """Client aiohttp session for async web requests""" 

499 return async_get_clientsession(self._hass) 

500 

501 def expand_group(self, entity_ids: str | list[str]) -> list[str]: 

502 return expand_entity_ids(self._hass, entity_ids) 

503 

504 def group_members(self, entity_id: str, _seen: set[str] | None = None) -> list[str] | None: 

505 """Fully expanded members of a `group.*` helper or a platform group (media_player, light... groups 

506 created by the group integration expose members in an `entity_id` state attribute). None if not a group. 

507 

508 Other entities, e.g. `scene.*` or min/max `sensor.*`, also expose an `entity_id` attribute, so anything 

509 outside the `group` domain is only treated as a group if the entity registry says its platform is `group`. 

510 """ 

511 if not self.hass_avail("states"): 

512 return None 

513 state = self._hass.states.get(entity_id) 

514 members = state.attributes.get(ATTR_ENTITY_ID) if state else None 

515 if not isinstance(members, (list, tuple)): 

516 return None 

517 if entity_id.partition(".")[0] != GROUP_DOMAIN and self.platform_for_entity(entity_id) != GROUP_DOMAIN: 

518 return None 

519 seen: set[str] = _seen if _seen is not None else set() 

520 seen.add(entity_id) 

521 expanded: list[str] = [] 

522 for member in members: 

523 if member in seen: 

524 continue 

525 nested = self.group_members(member, seen) 

526 for e in nested if nested is not None else [member]: 

527 if e not in expanded: 

528 expanded.append(e) 

529 return expanded 

530 

531 def template(self, template_format: str) -> Template: 

532 return Template(template_format, self._hass) 

533 

534 def hass_avail(self, property: str) -> bool: 

535 """Guard for HA functionality, largely for tests or docgen""" 

536 return self._hass is not None and getattr(self._hass, property, None) is not None 

537 

538 async def register_web_path(self, media_web_path: Path | None, url_prefix: str) -> bool: 

539 if media_web_path is None or not self.hass_avail("http"): 

540 return False 

541 try: 

542 from homeassistant.components.http import StaticPathConfig 

543 

544 await self._hass.http.async_register_static_paths([ 

545 StaticPathConfig(url_prefix, str(media_web_path), cache_headers=False) 

546 ]) 

547 return True 

548 except Exception as e: 

549 _LOGGER.error("SUPERNOTIFY Unable to register media web exposed path for %s: %s", media_web_path, e) 

550 return False 

551 

552 async def trace_conditions( 

553 self, 

554 conditions: ConditionsFunc, 

555 condition_variables: ConditionVariables, 

556 trace_name: str | None = None, 

557 ) -> tuple[bool | None, ActionTrace | None]: 

558 

559 result: bool | None = None 

560 this_trace: ActionTrace | None = None 

561 if DATA_TRACE not in self._hass.data: 

562 _LOGGER.warning("SUPERNOTIFY Tracing not configured, attempting to set up") 

563 

564 await homeassistant.components.trace.async_setup(self._hass, {}) # type: ignore 

565 with trace_action(self._hass, trace_name or "anon_condition") as cond_trace: 

566 cond_trace.set_trace(trace_get()) 

567 this_trace = cond_trace 

568 with trace_path(["condition", "conditions"]) as _tp: 

569 result = self.evaluate_conditions(conditions, condition_variables) 

570 _LOGGER.debug(cond_trace.as_dict()) 

571 return result, this_trace 

572 

573 async def build_conditions( 

574 self, condition_config: list[ConfigType], strict: bool = False, validate: bool = False, name: str = DOMAIN 

575 ) -> ConditionsFunc | None: 

576 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

577 condition_variables: ConditionVariables = ConditionVariables() 

578 cond_list: list[ConfigType] 

579 try: 

580 if validate: 

581 cond_list = cast( 

582 "list[ConfigType]", await condition_helper.async_validate_conditions_config(self._hass, condition_config) 

583 ) 

584 else: 

585 cond_list = condition_config 

586 except Exception: 

587 _LOGGER.exception("SUPERNOTIFY Conditions validation failed") 

588 raise 

589 try: 

590 if strict: 

591 force_strict_template_mode(cond_list, undo=False) 

592 

593 test: ConditionsFunc = await condition_helper.async_conditions_from_config( 

594 self._hass, cond_list, cast("logging.Logger", capturing_logger), name 

595 ) 

596 if test is None: 

597 raise IntegrationError(f"Invalid condition {condition_config}") 

598 test(condition_variables.as_dict()) 

599 return test 

600 except Exception: 

601 _LOGGER.exception("SUPERNOTIFY Conditions eval failed") 

602 raise 

603 finally: 

604 if strict: 

605 force_strict_template_mode(condition_config, undo=True) 

606 if strict and capturing_logger.condition_errors and len(capturing_logger.condition_errors) > 0: 

607 for exception in capturing_logger.condition_errors: 

608 _LOGGER.warning("SUPERNOTIFY Invalid condition %s:%s", condition_config, exception) 

609 raise capturing_logger.condition_errors[0] 

610 

611 def evaluate_conditions( 

612 self, 

613 conditions: ConditionsFunc, 

614 condition_variables: ConditionVariables, 

615 ) -> bool | None: 

616 try: 

617 if not condition_variables: 

618 _LOGGER.warning("SUPERNOTIFY No cond vars provided for condition") 

619 return conditions(condition_variables.as_dict() if condition_variables is not None else None) 

620 except Exception as e: 

621 _LOGGER.error("SUPERNOTIFY Condition eval failed: %s", e) 

622 raise 

623 

624 def abs_url(self, fragment: str | None, prefer_external: bool = True) -> str | None: 

625 base_url = self.external_url if prefer_external else self.internal_url 

626 if fragment: 

627 if fragment.startswith("http"): 

628 return fragment 

629 if fragment.startswith("/"): 

630 return base_url + fragment 

631 return base_url + "/" + fragment 

632 return None 

633 

634 def raise_issue( 

635 self, 

636 issue_id: str, 

637 issue_key: str, 

638 issue_map: dict[str, str], 

639 severity: ir.IssueSeverity = ir.IssueSeverity.WARNING, 

640 learn_more_url: str = "https://supernotify.rhizomatics.org.uk", 

641 is_fixable: bool = False, 

642 ) -> None: 

643 ir.async_create_issue( 

644 self._hass, 

645 DOMAIN, 

646 issue_id, 

647 translation_key=issue_key, 

648 translation_placeholders=issue_map, 

649 severity=severity, 

650 learn_more_url=learn_more_url, 

651 is_fixable=is_fixable, 

652 ) 

653 

654 def mobile_app_by_tracker(self, device_tracker: str) -> TrackedDeviceDetails | None: 

655 return self._mobile_apps_by_tracker.get(device_tracker) 

656 

657 def mobile_app_by_id(self, mobile_app_id: str) -> TrackedDeviceDetails | None: 

658 mobile_app_id = mobile_app_id.replace("notify.", "", 1) if mobile_app_id.startswith("notify.") else mobile_app_id 

659 return self._mobile_apps_by_app_id.get(mobile_app_id) 

660 

661 def mobile_app_by_device_id(self, device_id: str) -> TrackedDeviceDetails | None: 

662 return self._mobile_apps_by_device_id.get(device_id) 

663 

664 def mobile_app_by_user_id(self, user_id: str) -> list[TrackedDeviceDetails] | None: 

665 return self._mobile_apps_by_user_id.get(user_id) 

666 

667 def build_mobile_app_cache(self) -> None: 

668 """All enabled mobile apps""" 

669 entity_registry: EntityRegistry | None = self._entity_registry() 

670 if not entity_registry: 

671 _LOGGER.warning("SUPERNOTIFY Unable to discover devices for - no entity registry found") 

672 return 

673 

674 found: int = 0 

675 complete: int = 0 

676 for mobile_app_info in self.discover_devices("mobile_app"): 

677 try: 

678 mobile_app_id: str = f"mobile_app_{slugify(mobile_app_info.device_name)}" 

679 device_tracker: str | None = None 

680 notify_action: str | None = None 

681 if self.has_service("notify", mobile_app_id): 

682 notify_action = f"notify.{mobile_app_id}" 

683 else: 

684 _LOGGER.warning("SUPERNOTIFY Unable to find notify action <%s>", mobile_app_id) 

685 

686 registry_entries = entity_registry.entities.get_entries_for_device_id(mobile_app_info.device_id) 

687 for reg_entry in registry_entries: 

688 if reg_entry.platform == "mobile_app" and reg_entry.domain == "device_tracker": 

689 device_tracker = reg_entry.entity_id 

690 

691 if device_tracker and notify_action: 

692 complete += 1 

693 

694 mobile_app_info.mobile_app_id = mobile_app_id 

695 mobile_app_info.device_tracker = device_tracker 

696 mobile_app_info.action = notify_action 

697 

698 found += 1 

699 self._mobile_apps_by_app_id[mobile_app_id] = mobile_app_info