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

1from __future__ import annotations 

2 

3import logging 

4from dataclasses import dataclass 

5from functools import partial 

6from typing import TYPE_CHECKING, Any 

7 

8import voluptuous as vol 

9from homeassistant.components.person import ATTR_USER_ID 

10from homeassistant.const import CONF_ACTION, CONF_DEVICE_ID 

11from homeassistant.helpers.aiohttp_client import async_get_clientsession 

12from homeassistant.helpers.event import async_track_state_change_event, async_track_time_change 

13from homeassistant.util import slugify 

14 

15if TYPE_CHECKING: 

16 import asyncio 

17 from collections.abc import Callable, Iterable, Iterator, Mapping 

18 

19 import aiohttp 

20 from anyio import Path 

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

22 from homeassistant.helpers.entity import Entity 

23 from homeassistant.helpers.entity_registry import EntityRegistry 

24 from homeassistant.helpers.typing import ConfigType 

25 from homeassistant.util.event_type import EventType 

26 

27 from .schema import ConditionsFunc 

28 

29import socket 

30import threading 

31from contextlib import contextmanager 

32from typing import TYPE_CHECKING, cast 

33 

34import homeassistant.components.trace 

35from homeassistant.components.group import expand_entity_ids 

36from homeassistant.components.trace.const import DATA_TRACE 

37from homeassistant.components.trace.models import ActionTrace 

38from homeassistant.components.trace.util import async_store_trace 

39from homeassistant.core import Context as HomeAssistantContext 

40from homeassistant.core import HomeAssistant, SupportsResponse 

41from homeassistant.exceptions import ConditionError, ConditionErrorContainer, IntegrationError 

42from homeassistant.helpers import condition as condition_helper 

43from homeassistant.helpers import device_registry as dr 

44from homeassistant.helpers import entity_registry as er 

45from homeassistant.helpers import issue_registry as ir 

46from homeassistant.helpers.json import json_dumps 

47from homeassistant.helpers.network import get_url 

48from homeassistant.helpers.template import Template 

49from homeassistant.helpers.trace import trace_get, trace_path 

50from homeassistant.helpers.typing import ConfigType 

51 

52from . import DOMAIN 

53from .const import CONF_DEVICE_LABELS, CONF_DEVICE_TRACKER, CONF_MOBILE_APP_ID 

54from .model import ConditionVariables, SelectionRule 

55 

56if TYPE_CHECKING: 

57 from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry 

58 

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

60 

61CONF_USER_ID = "user_id" 

62ATTR_OS_NAME = "os_name" 

63ATTR_OS_VERSION = "os_version" 

64ATTR_APP_VERSION = "app_version" 

65ATTR_DEVICE_NAME = "device_name" 

66ATTR_MANUFACTURER = "manufacturer" 

67ATTR_MODEL = "model" 

68 

69_LOGGER = logging.getLogger(__name__) 

70 

71 

72@dataclass 

73class DeviceInfo: 

74 device_id: str 

75 device_labels: list[str] | None = None 

76 mobile_app_id: str | None = None 

77 device_name: str | None = None 

78 device_tracker: str | None = None 

79 action: str | None = None 

80 user_id: str | None = None 

81 area_id: str | None = None 

82 manufacturer: str | None = None 

83 model: str | None = None 

84 os_name: str | None = None 

85 os_version: str | None = None 

86 app_version: str | None = None 

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

88 

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

90 return { 

91 CONF_MOBILE_APP_ID: self.mobile_app_id, 

92 ATTR_DEVICE_NAME: self.device_name, 

93 CONF_DEVICE_ID: self.device_id, 

94 CONF_USER_ID: self.user_id, 

95 CONF_DEVICE_TRACKER: self.device_tracker, 

96 CONF_ACTION: self.action, 

97 ATTR_OS_NAME: self.os_name, 

98 ATTR_OS_VERSION: self.os_version, 

99 ATTR_APP_VERSION: self.app_version, 

100 ATTR_MANUFACTURER: self.manufacturer, 

101 ATTR_MODEL: self.model, 

102 CONF_DEVICE_LABELS: self.device_labels, 

103 } 

104 

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

106 """Test support""" 

107 return other is not None and hasattr(other, "as_dict") and other.as_dict() == self.as_dict() 

108 

109 

110class HomeAssistantAPI: 

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

112 self._hass: HomeAssistant = hass 

113 self.internal_url: str = "" 

114 self.external_url: str = "" 

115 self.language: str = "" 

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

117 self._entity_registry: er.EntityRegistry | None = None 

118 self._device_registry: dr.DeviceRegistry | None = None 

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

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

121 self.mobile_apps_by_tracker: dict[str, DeviceInfo] = {} 

122 self.mobile_apps_by_app_id: dict[str, DeviceInfo] = {} 

123 self.mobile_apps_by_device_id: dict[str, DeviceInfo] = {} 

124 self.mobile_apps_by_user_id: dict[str, list[DeviceInfo]] = {} 

125 

126 def initialize(self) -> None: 

127 self.hass_name = self._hass.config.location_name 

128 self.language = self._hass.config.language 

129 try: 

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

131 except Exception as e: 

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

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

134 try: 

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

136 except Exception as e: 

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

138 self.external_url = self.internal_url 

139 

140 self.build_mobile_app_cache() 

141 

142 _LOGGER.debug( 

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

144 self.hass_name, 

145 self.internal_url, 

146 self.external_url, 

147 ) 

148 

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

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

151 

152 def disconnect(self) -> None: 

153 while self.unsubscribes: 

154 unsub = self.unsubscribes.pop() 

155 try: 

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

157 unsub() 

158 except Exception as e: 

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

160 _LOGGER.debug("SUPERNOTIFY Disconnection complete") 

161 

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

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

164 

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

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

167 

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

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

170 

171 def in_hass_loop(self) -> bool: 

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

173 

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

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

176 

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

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

179 

180 def set_state(self, entity_id: str, state: str | int | bool, attributes: dict[str, Any] | None = None) -> None: 

181 if self.in_hass_loop(): 

182 self._hass.states.async_set(entity_id, str(state), attributes=attributes) 

183 else: 

184 self._hass.states.set(entity_id, str(state), attributes=attributes) 

185 

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

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

188 

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

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

191 

192 def domain_entity(self, domain: str, entity_id: str) -> Entity | None: 

193 # TODO: must be a better hass method than this 

194 return self._hass.data.get(domain, {}).get_entity(entity_id) 

195 

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

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

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

199 

200 def fire_event(self, event_name: str, event_data: dict[str, Any] | None = None) -> None: 

201 self._hass.bus.async_fire(event_name, event_data) 

202 

203 async def call_service( 

204 self, 

205 domain: str, 

206 service: str, 

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

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

209 return_response: bool | None = None, 

210 blocking: bool | None = None, 

211 debug: bool = False, 

212 ) -> ServiceResponse | None: 

213 

214 if return_response is None or blocking is None: 

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

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

217 if supports_response == SupportsResponse.NONE: 

218 return_response = False 

219 elif supports_response == SupportsResponse.ONLY: 

220 return_response = True 

221 else: 

222 return_response = debug 

223 blocking = return_response or debug 

224 

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

226 domain, 

227 service, 

228 service_data=service_data, 

229 blocking=blocking, 

230 context=None, 

231 target=target, 

232 return_response=return_response, 

233 ) 

234 if response is not None and debug: 

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

236 return response 

237 

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

239 if not data: 

240 return data 

241 try: 

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

243 self.service_info(domain, service) 

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

245 if not service_info: 

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

247 return data 

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

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

250 return data 

251 service_schema = service_info["schema"] 

252 

253 while service_schema is not None and not ( 

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

255 ): 

256 if isinstance(service_schema, vol.Schema): 

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

258 service_schema = service_schema.schema 

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

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

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

262 service_schema = next( 

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

264 ) 

265 else: 

266 service_schema = None 

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

268 service_schema = None 

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

270 

271 if service_schema: 

272 coercing_schema = service_schema.extend( 

273 {}, 

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

275 required=service_schema.required, 

276 ) 

277 cleaned = coercing_schema(data) 

278 else: 

279 return data 

280 if cleaned != data: 

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

282 return cleaned 

283 except Exception: 

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

285 return data 

286 

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

288 supports_response: SupportsResponse | None = None 

289 try: 

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

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

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

293 if service_obj: 

294 self._service_info[domain, service] = { 

295 "supports_response": service_obj.supports_response, 

296 "schema": service_obj.schema, 

297 } 

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

299 supports_response = service_info.get("supports_response") 

300 if supports_response is None: 

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

302 

303 except Exception as e: 

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

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

306 

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

308 try: 

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

310 if service_objs: 

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

312 if domain_obj.job and domain_obj.job.target: 

313 target_module: str | None = ( 

314 domain_obj.job.target.__self__.__module__ 

315 if hasattr(domain_obj.job.target, "__self__") 

316 else domain_obj.job.target.__module__ 

317 ) 

318 if target_module == module: 

319 _LOGGER.debug("SUPERNOTIFY Found service %s for domain %s", service, domain) 

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

321 

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

323 except Exception as e: 

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

325 return None 

326 

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

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

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

330 return None 

331 try: 

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

333 if entries: 

334 return entries[0].data 

335 except Exception as e: 

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

337 return None 

338 

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

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

341 return async_get_clientsession(self._hass) 

342 

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

344 return expand_entity_ids(self._hass, entity_ids) 

345 

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

347 return Template(template_format, self._hass) 

348 

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

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

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

352 

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

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

355 return False 

356 try: 

357 from homeassistant.components.http import StaticPathConfig 

358 

359 await self._hass.http.async_register_static_paths([ 

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

361 ]) 

362 return True 

363 except Exception as e: 

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

365 return False 

366 

367 async def trace_conditions( 

368 self, 

369 conditions: ConditionsFunc, 

370 condition_variables: ConditionVariables, 

371 trace_name: str | None = None, 

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

373 

374 result: bool | None = None 

375 this_trace: ActionTrace | None = None 

376 if DATA_TRACE not in self._hass.data: 

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

378 

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

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

381 cond_trace.set_trace(trace_get()) 

382 this_trace = cond_trace 

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

384 result = self.evaluate_conditions(conditions, condition_variables) 

385 _LOGGER.debug(cond_trace.as_dict()) 

386 return result, this_trace 

387 

388 async def build_conditions( 

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

390 ) -> ConditionsFunc | None: 

391 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

392 condition_variables: ConditionVariables = ConditionVariables() 

393 cond_list: list[ConfigType] 

394 try: 

395 if validate: 

396 cond_list = cast( 

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

398 ) 

399 else: 

400 cond_list = condition_config 

401 except Exception: 

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

403 raise 

404 try: 

405 if strict: 

406 force_strict_template_mode(cond_list, undo=False) 

407 

408 test: ConditionsFunc = await condition_helper.async_conditions_from_config( 

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

410 ) 

411 if test is None: 

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

413 test(condition_variables.as_dict()) 

414 return test 

415 except Exception: 

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

417 raise 

418 finally: 

419 if strict: 

420 force_strict_template_mode(condition_config, undo=True) 

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

422 for exception in capturing_logger.condition_errors: 

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

424 raise capturing_logger.condition_errors[0] 

425 

426 def evaluate_conditions( 

427 self, 

428 conditions: ConditionsFunc, 

429 condition_variables: ConditionVariables, 

430 ) -> bool | None: 

431 try: 

432 if not condition_variables: 

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

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

435 except Exception as e: 

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

437 raise 

438 

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

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

441 if fragment: 

442 if fragment.startswith("http"): 

443 return fragment 

444 if fragment.startswith("/"): 

445 return base_url + fragment 

446 return base_url + "/" + fragment 

447 return None 

448 

449 def raise_issue( 

450 self, 

451 issue_id: str, 

452 issue_key: str, 

453 issue_map: dict[str, str], 

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

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

456 is_fixable: bool = False, 

457 ) -> None: 

458 ir.async_create_issue( 

459 self._hass, 

460 DOMAIN, 

461 issue_id, 

462 translation_key=issue_key, 

463 translation_placeholders=issue_map, 

464 severity=severity, 

465 learn_more_url=learn_more_url, 

466 is_fixable=is_fixable, 

467 ) 

468 

469 def mobile_app_by_tracker(self, device_tracker: str) -> DeviceInfo | None: 

470 return self.mobile_apps_by_tracker.get(device_tracker) 

471 

472 def mobile_app_by_id(self, mobile_app_id: str) -> DeviceInfo | None: 

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

474 return self.mobile_apps_by_app_id.get(mobile_app_id) 

475 

476 def mobile_app_by_device_id(self, device_id: str) -> DeviceInfo | None: 

477 return self.mobile_apps_by_device_id.get(device_id) 

478 

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

480 return self.mobile_apps_by_user_id.get(user_id) 

481 

482 def build_mobile_app_cache(self) -> None: 

483 """All enabled mobile apps""" 

484 ent_reg: EntityRegistry | None = self.entity_registry() 

485 if not ent_reg: 

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

487 return 

488 

489 found: int = 0 

490 complete: int = 0 

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

492 try: 

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

494 device_tracker: str | None = None 

495 notify_action: str | None = None 

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

497 notify_action = f"notify.{mobile_app_id}" 

498 else: 

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

500 

501 registry_entries = ent_reg.entities.get_entries_for_device_id(mobile_app_info.device_id) 

502 for reg_entry in registry_entries: 

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

504 device_tracker = reg_entry.entity_id 

505 

506 if device_tracker and notify_action: 

507 complete += 1 

508 

509 mobile_app_info.mobile_app_id = mobile_app_id 

510 mobile_app_info.device_tracker = device_tracker 

511 mobile_app_info.action = notify_action 

512 

513 found += 1 

514 self.mobile_apps_by_app_id[mobile_app_id] = mobile_app_info 

515 self.mobile_apps_by_device_id[mobile_app_info.device_id] = mobile_app_info 

516 if device_tracker: 

517 self.mobile_apps_by_tracker[device_tracker] = mobile_app_info 

518 if mobile_app_info.user_id is not None: 

519 self.mobile_apps_by_user_id.setdefault(mobile_app_info.user_id, []) 

520 self.mobile_apps_by_user_id[mobile_app_info.user_id].append(mobile_app_info) 

521 

522 except Exception as e: 

523 _LOGGER.error("SUPERNOTIFY Failure examining device %s: %s", mobile_app_info, e) 

524 

525 _LOGGER.info(f"SUPERNOTIFY Found {found} enabled mobile app devices, {complete} complete config") 

526 

527 def device_config_info(self, device: DeviceEntry) -> dict[str, str | None]: 

528 results: dict[str, str | None] = {ATTR_OS_NAME: None, ATTR_OS_VERSION: None, CONF_USER_ID: None, ATTR_APP_VERSION: None} 

529 try: 

530 # HA 2026.8+ restricts devices to a single config entry 

531 config_entry_ids: Iterable[str] = (device.config_entry_id,) # type: ignore[attr-defined] 

532 except AttributeError: 

533 # pre-2026.8: config_entry_id doesn't exist yet, fall back to deprecated plural set 

534 config_entry_ids = device.config_entries 

535 for config_entry_id in config_entry_ids: 

536 config_entry = self._hass.config_entries.async_get_entry(config_entry_id) 

537 if config_entry and config_entry.data: 

538 for attr, value in results.items(): 

539 results[attr] = config_entry.data.get(attr) or value 

540 return results 

541 

542 def discover_devices( 

543 self, 

544 discover_domain: str, 

545 device_model_select: SelectionRule | None = None, 

546 device_manufacturer_select: SelectionRule | None = None, 

547 device_os_select: SelectionRule | None = None, 

548 device_area_select: SelectionRule | None = None, 

549 device_label_select: SelectionRule | None = None, 

550 ) -> list[DeviceInfo]: 

551 devices: list[DeviceInfo] = [] 

552 dev_reg: DeviceRegistry | None = self.device_registry() 

553 if dev_reg is None or not hasattr(dev_reg, "devices"): 

554 _LOGGER.warning(f"SUPERNOTIFY Unable to discover devices for {discover_domain} - no device registry found") 

555 return [] 

556 

557 all_devs = enabled_devs = found_devs = skipped_devs = 0 

558 for dev in dev_reg.devices.values(): 

559 all_devs += 1 

560 

561 if dev.disabled: 

562 _LOGGER.debug("SUPERNOTIFY Excluded disabled device %s", dev.name) 

563 else: 

564 enabled_devs += 1 

565 for identifier in dev.identifiers: 

566 if identifier and len(identifier) > 1 and identifier[0] == discover_domain: 

567 _LOGGER.debug("SUPERNOTIFY Discovered %s device %s for id %s", dev.model, dev.name, identifier) 

568 found_devs += 1 

569 if device_model_select is not None and not device_model_select.match(dev.model): 

570 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no model %s match", dev.name, dev.model) 

571 skipped_devs += 1 

572 continue 

573 if device_manufacturer_select is not None and not device_manufacturer_select.match(dev.manufacturer): 

574 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no manufacturer %s match", dev.name, dev.manufacturer) 

575 skipped_devs += 1 

576 continue 

577 device_config_info = self.device_config_info(dev) 

578 if device_os_select is not None and not device_os_select.match(device_config_info[ATTR_OS_NAME]): 

579 _LOGGER.debug( 

580 "SUPERNOTIFY Skipped dev %s, no OS %s match", dev.name, device_config_info[ATTR_OS_NAME] 

581 ) 

582 skipped_devs += 1 

583 continue 

584 if device_area_select is not None and not device_area_select.match(dev.area_id): 

585 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no area %s match", dev.name, dev.area_id) 

586 skipped_devs += 1 

587 continue 

588 if device_label_select is not None and not device_label_select.match(dev.labels): 

589 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no label %s match", dev.name, dev.labels) 

590 skipped_devs += 1 

591 continue 

592 devices.append( 

593 DeviceInfo( 

594 device_id=dev.id, 

595 device_name=dev.name, 

596 manufacturer=dev.manufacturer, 

597 model=dev.model, 

598 area_id=dev.area_id, 

599 user_id=device_config_info[ATTR_USER_ID], 

600 os_name=device_config_info[ATTR_OS_NAME], 

601 os_version=device_config_info[ATTR_OS_VERSION], 

602 app_version=device_config_info[ATTR_APP_VERSION], 

603 device_labels=list(dev.labels) if dev.labels else [], 

604 identifiers=dev.identifiers, 

605 ) 

606 ) 

607 

608 elif identifier: 

609 # HomeKit has triples for identifiers, other domains may behave similarly 

610 _LOGGER.debug("SUPERNOTIFY Ignoring device %s id: %s", dev.name, identifier) 

611 else: 

612 _LOGGER.debug( # type: ignore 

613 "SUPERNOTIFY Unexpected %s device %s without id", dev.model, dev.name 

614 ) 

615 

616 _LOGGER.debug(f"SUPERNOTIFY {discover_domain} device discovery, all={all_devs},enabled={enabled_devs} ") 

617 _LOGGER.debug(f"SUPERNOTIFY {discover_domain} skipped={skipped_devs}, found={found_devs}") 

618 

619 return devices 

620 

621 def domain_for_device(self, device_id: str, domains: list[str]) -> str | None: 

622 # discover domain from device registry 

623 verified_domain: str | None = None 

624 device_registry = self.device_registry() 

625 if device_registry: 

626 device: DeviceEntry | None = device_registry.async_get(device_id) 

627 if device: 

628 matching_domains = [d for d, _id in device.identifiers if d in domains] 

629 if matching_domains: 

630 # TODO: limited to first domain found, unlikely to be more 

631 return matching_domains[0] 

632 _LOGGER.warning( 

633 "SUPERNOTIFY A target that looks like a device_id can't be matched to supported integration: %s", 

634 device_id, 

635 ) 

636 return verified_domain 

637 

638 def entity_registry(self) -> er.EntityRegistry | None: 

639 """Hass entity registry is weird, every component ends up creating its own, with a store, subscribing 

640 to all entities, so do it once here 

641 """ 

642 if self._entity_registry is not None: 

643 return self._entity_registry 

644 try: 

645 self._entity_registry = er.async_get(self._hass) 

646 except Exception as e: 

647 _LOGGER.warning("SUPERNOTIFY Unable to get entity registry: %s", e) 

648 return self._entity_registry 

649 

650 def device_registry(self) -> dr.DeviceRegistry | None: 

651 """Hass device registry is weird, every component ends up creating its own, with a store, subscribing 

652 to all devices, so do it once here 

653 """ 

654 if self._device_registry is not None: 

655 return self._device_registry 

656 try: 

657 self._device_registry = dr.async_get(self._hass) 

658 except Exception as e: 

659 _LOGGER.warning("SUPERNOTIFY Unable to get device registry: %s", e) 

660 return self._device_registry 

661 

662 async def mqtt_available(self, raise_on_error: bool = True) -> bool: 

663 from homeassistant.components import mqtt 

664 

665 try: 

666 return await mqtt.async_wait_for_mqtt_client(self._hass) is True 

667 except Exception: 

668 _LOGGER.exception("SUPERNOTIFY MQTT integration failed on available check") 

669 if raise_on_error: 

670 raise 

671 return False 

672 

673 async def mqtt_publish( 

674 self, topic: str, payload: Any = None, qos: int = 0, retain: bool = False, raise_on_error: bool = True 

675 ) -> None: 

676 from homeassistant.components import mqtt 

677 

678 try: 

679 await mqtt.async_publish( 

680 self._hass, 

681 topic=topic, 

682 payload=json_dumps(payload), 

683 qos=qos, 

684 retain=retain, 

685 ) 

686 except Exception: 

687 _LOGGER.exception(f"SUPERNOTIFY MQTT publish failed to {topic}") 

688 if raise_on_error: 

689 raise 

690 

691 

692class ConditionErrorLoggingAdaptor(logging.LoggerAdapter): 

693 def __init__(self, *args: Any, **kwargs: Any) -> None: 

694 super().__init__(*args, **kwargs) 

695 self.condition_errors: list[ConditionError] = []