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

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 

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 import mqtt 

36from homeassistant.components.group import expand_entity_ids 

37from homeassistant.components.trace.const import DATA_TRACE 

38from homeassistant.components.trace.models import ActionTrace 

39from homeassistant.components.trace.util import async_store_trace 

40from homeassistant.core import Context as HomeAssistantContext 

41from homeassistant.core import HomeAssistant, SupportsResponse 

42from homeassistant.exceptions import ConditionError, ConditionErrorContainer, IntegrationError 

43from homeassistant.helpers import condition as condition 

44from homeassistant.helpers import device_registry as dr 

45from homeassistant.helpers import entity_registry as er 

46from homeassistant.helpers import issue_registry as ir 

47from homeassistant.helpers.json import json_dumps 

48from homeassistant.helpers.network import get_url 

49from homeassistant.helpers.template import Template 

50from homeassistant.helpers.trace import trace_get, trace_path 

51from homeassistant.helpers.typing import ConfigType 

52 

53from . import DOMAIN 

54from .const import CONF_DEVICE_LABELS, CONF_DEVICE_TRACKER, CONF_MOBILE_APP_ID 

55from .model import ConditionVariables, SelectionRule 

56 

57if TYPE_CHECKING: 

58 from homeassistant.core import HomeAssistant 

59 from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry 

60 from homeassistant.helpers.typing import ConfigType 

61 

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

63 

64CONF_USER_ID = "user_id" 

65ATTR_OS_NAME = "os_name" 

66ATTR_OS_VERSION = "os_version" 

67ATTR_APP_VERSION = "app_version" 

68ATTR_DEVICE_NAME = "device_name" 

69ATTR_MANUFACTURER = "manufacturer" 

70ATTR_MODEL = "model" 

71 

72_LOGGER = logging.getLogger(__name__) 

73 

74 

75@dataclass 

76class DeviceInfo: 

77 device_id: str 

78 device_labels: list[str] | None = None 

79 mobile_app_id: str | None = None 

80 device_name: str | None = None 

81 device_tracker: str | None = None 

82 action: str | None = None 

83 user_id: str | None = None 

84 area_id: str | None = None 

85 manufacturer: str | None = None 

86 model: str | None = None 

87 os_name: str | None = None 

88 os_version: str | None = None 

89 app_version: str | None = None 

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

91 

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

93 return { 

94 CONF_MOBILE_APP_ID: self.mobile_app_id, 

95 ATTR_DEVICE_NAME: self.device_name, 

96 CONF_DEVICE_ID: self.device_id, 

97 CONF_USER_ID: self.user_id, 

98 CONF_DEVICE_TRACKER: self.device_tracker, 

99 CONF_ACTION: self.action, 

100 ATTR_OS_NAME: self.os_name, 

101 ATTR_OS_VERSION: self.os_version, 

102 ATTR_APP_VERSION: self.app_version, 

103 ATTR_MANUFACTURER: self.manufacturer, 

104 ATTR_MODEL: self.model, 

105 CONF_DEVICE_LABELS: self.device_labels, 

106 } 

107 

108 def __eq__(self, other: Any) -> bool: 

109 """Test support""" 

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

111 

112 

113class HomeAssistantAPI: 

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

115 self._hass: HomeAssistant = hass 

116 self.internal_url: str = "" 

117 self.external_url: str = "" 

118 self.language: str = "" 

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

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

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

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

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

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

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

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

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

128 

129 def initialize(self) -> None: 

130 self.hass_name = self._hass.config.location_name 

131 self.language = self._hass.config.language 

132 try: 

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

134 except Exception as e: 

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

136 _LOGGER.warning("SUPERNOTIFY could not get internal hass url, defaulting to %s: %s", self.internal_url, e) 

137 try: 

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

139 except Exception as e: 

140 _LOGGER.warning("SUPERNOTIFY could not get external hass url, defaulting to internal url: %s", e) 

141 self.external_url = self.internal_url 

142 

143 self.build_mobile_app_cache() 

144 

145 _LOGGER.debug( 

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

147 self.hass_name, 

148 self.internal_url, 

149 self.external_url, 

150 ) 

151 

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

153 _LOGGER.warning("SUPERNOTIFY invalid internal hass url %s", self.internal_url) 

154 

155 def disconnect(self) -> None: 

156 while self.unsubscribes: 

157 unsub = self.unsubscribes.pop() 

158 try: 

159 _LOGGER.debug("SUPERNOTIFY unsubscribing: %s", unsub) 

160 unsub() 

161 except Exception as e: 

162 _LOGGER.error("SUPERNOTIFY failed to unsubscribe: %s", e) 

163 _LOGGER.debug("SUPERNOTIFY disconnection complete") 

164 

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

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

167 

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

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

170 

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

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

173 

174 def in_hass_loop(self) -> bool: 

175 return self._hass is not None and self._hass.loop_thread_id == threading.get_ident() 

176 

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

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

179 

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

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

182 

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

184 if self.in_hass_loop(): 

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

186 else: 

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

188 

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

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

191 

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

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

194 

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

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

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

198 

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

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

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

202 

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

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

205 

206 async def call_service( 

207 self, 

208 domain: str, 

209 service: str, 

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

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

212 return_response: bool | None = None, 

213 blocking: bool | None = None, 

214 debug: bool = False, 

215 ) -> ServiceResponse | None: 

216 

217 if return_response is None or blocking is None: 

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

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

220 if supports_response == SupportsResponse.NONE: 

221 return_response = False 

222 elif supports_response == SupportsResponse.ONLY: 

223 return_response = True 

224 else: 

225 return_response = debug 

226 blocking = return_response or debug 

227 

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

229 domain, 

230 service, 

231 service_data=service_data, 

232 blocking=blocking, 

233 context=None, 

234 target=target, 

235 return_response=return_response, 

236 ) 

237 if response is not None and debug: 

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

239 return response 

240 

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

242 if not data: 

243 return data 

244 try: 

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

246 self.service_info(domain, service) 

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

248 if not service_info: 

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

250 return data 

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

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

253 return data 

254 service_schema = service_info["schema"] 

255 

256 while service_schema is not None and not isinstance(service_schema, vol.Schema): 

257 # e.g. entity services get schema wrapped in an vol.All 

258 if hasattr(service_schema, "validators") and hasattr(service_schema.validators, "__iter__"): 

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

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

261 service_schema = next( 

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

263 ) 

264 else: 

265 service_schema = None 

266 if not isinstance(service_schema, vol.Schema): 

267 service_schema = None 

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

269 

270 if service_schema: 

271 coercing_schema = service_schema.extend( 

272 {}, 

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

274 required=service_schema.required, 

275 ) 

276 cleaned = coercing_schema(data) 

277 else: 

278 return data 

279 if cleaned != data: 

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

281 return cleaned 

282 except Exception: 

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

284 return data 

285 

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

287 supports_response: SupportsResponse | None = None 

288 try: 

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

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

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

292 if service_obj: 

293 self._service_info[domain, service] = { 

294 "supports_response": service_obj.supports_response, 

295 "schema": service_obj.schema, 

296 } 

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

298 supports_response = service_info.get("supports_response") 

299 if supports_response is None: 

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

301 

302 except Exception as e: 

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

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

305 

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

307 try: 

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

309 if service_objs: 

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

311 if domain_obj.job and domain_obj.job.target: 

312 target_module: str | None = ( 

313 domain_obj.job.target.__self__.__module__ 

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

315 else domain_obj.job.target.__module__ 

316 ) 

317 if target_module == module: 

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

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

320 

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

322 except Exception as e: 

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

324 return None 

325 

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

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

328 return async_get_clientsession(self._hass) 

329 

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

331 return expand_entity_ids(self._hass, entity_ids) 

332 

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

334 return Template(template_format, self._hass) 

335 

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

337 if media_web_path is None: 

338 return False 

339 try: 

340 from homeassistant.components.http import StaticPathConfig 

341 

342 await self._hass.http.async_register_static_paths([ 

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

344 ]) 

345 return True 

346 except Exception as e: 

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

348 return False 

349 

350 async def trace_conditions( 

351 self, 

352 conditions: ConditionsFunc, 

353 condition_variables: ConditionVariables, 

354 trace_name: str | None = None, 

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

356 

357 result: bool | None = None 

358 this_trace: ActionTrace | None = None 

359 if DATA_TRACE not in self._hass.data: 

360 _LOGGER.warning("SUPERNOTIFY tracing not configured, attempting to set up") 

361 

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

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

364 cond_trace.set_trace(trace_get()) 

365 this_trace = cond_trace 

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

367 result = self.evaluate_conditions(conditions, condition_variables) 

368 _LOGGER.debug(cond_trace.as_dict()) 

369 return result, this_trace 

370 

371 async def build_conditions( 

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

373 ) -> ConditionsFunc | None: 

374 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER) 

375 condition_variables: ConditionVariables = ConditionVariables() 

376 cond_list: list[ConfigType] 

377 try: 

378 if validate: 

379 cond_list = cast( 

380 "list[ConfigType]", await condition.async_validate_conditions_config(self._hass, condition_config) 

381 ) 

382 else: 

383 cond_list = condition_config 

384 except Exception as e: 

385 _LOGGER.exception("SUPERNOTIFY Conditions validation failed: %s", e) 

386 raise 

387 try: 

388 if strict: 

389 force_strict_template_mode(cond_list, undo=False) 

390 

391 test: ConditionsFunc = await condition.async_conditions_from_config( 

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

393 ) 

394 if test is None: 

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

396 test(condition_variables.as_dict()) 

397 return test 

398 except Exception as e: 

399 _LOGGER.exception("SUPERNOTIFY Conditions eval failed: %s", e) 

400 raise 

401 finally: 

402 if strict: 

403 force_strict_template_mode(condition_config, undo=True) 

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

405 for exception in capturing_logger.condition_errors: 

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

407 raise capturing_logger.condition_errors[0] 

408 

409 def evaluate_conditions( 

410 self, 

411 conditions: ConditionsFunc, 

412 condition_variables: ConditionVariables, 

413 ) -> bool | None: 

414 try: 

415 if not condition_variables: 

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

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

418 except Exception as e: 

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

420 raise 

421 

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

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

424 if fragment: 

425 if fragment.startswith("http"): 

426 return fragment 

427 if fragment.startswith("/"): 

428 return base_url + fragment 

429 return base_url + "/" + fragment 

430 return None 

431 

432 def raise_issue( 

433 self, 

434 issue_id: str, 

435 issue_key: str, 

436 issue_map: dict[str, str], 

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

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

439 is_fixable: bool = False, 

440 ) -> None: 

441 ir.async_create_issue( 

442 self._hass, 

443 DOMAIN, 

444 issue_id, 

445 translation_key=issue_key, 

446 translation_placeholders=issue_map, 

447 severity=severity, 

448 learn_more_url=learn_more_url, 

449 is_fixable=is_fixable, 

450 ) 

451 

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

453 return self.mobile_apps_by_tracker.get(device_tracker) 

454 

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

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

457 return self.mobile_apps_by_app_id.get(mobile_app_id) 

458 

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

460 return self.mobile_apps_by_device_id.get(device_id) 

461 

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

463 return self.mobile_apps_by_user_id.get(user_id) 

464 

465 def build_mobile_app_cache(self) -> None: 

466 """All enabled mobile apps""" 

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

468 if not ent_reg: 

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

470 return 

471 

472 found: int = 0 

473 complete: int = 0 

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

475 try: 

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

477 device_tracker: str | None = None 

478 notify_action: str | None = None 

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

480 notify_action = f"notify.{mobile_app_id}" 

481 else: 

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

483 

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

485 for reg_entry in registry_entries: 

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

487 device_tracker = reg_entry.entity_id 

488 

489 if device_tracker and notify_action: 

490 complete += 1 

491 

492 mobile_app_info.mobile_app_id = mobile_app_id 

493 mobile_app_info.device_tracker = device_tracker 

494 mobile_app_info.action = notify_action 

495 

496 found += 1 

497 self.mobile_apps_by_app_id[mobile_app_id] = mobile_app_info 

498 self.mobile_apps_by_device_id[mobile_app_info.device_id] = mobile_app_info 

499 if device_tracker: 

500 self.mobile_apps_by_tracker[device_tracker] = mobile_app_info 

501 if mobile_app_info.user_id is not None: 

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

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

504 

505 except Exception as e: 

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

507 

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

509 

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

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

512 for config_entry_id in device.config_entries: 

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

514 if config_entry and config_entry.data: 

515 for attr in results: 

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

517 return results 

518 

519 def discover_devices( 

520 self, 

521 discover_domain: str, 

522 device_model_select: SelectionRule | None = None, 

523 device_manufacturer_select: SelectionRule | None = None, 

524 device_os_select: SelectionRule | None = None, 

525 device_area_select: SelectionRule | None = None, 

526 device_label_select: SelectionRule | None = None, 

527 ) -> list[DeviceInfo]: 

528 devices: list[DeviceInfo] = [] 

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

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

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

532 return [] 

533 

534 all_devs = enabled_devs = found_devs = skipped_devs = 0 

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

536 all_devs += 1 

537 

538 if dev.disabled: 

539 _LOGGER.debug("SUPERNOTIFY excluded disabled device %s", dev.name) 

540 else: 

541 enabled_devs += 1 

542 for identifier in dev.identifiers: 

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

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

545 found_devs += 1 

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

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

548 skipped_devs += 1 

549 continue 

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

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

552 skipped_devs += 1 

553 continue 

554 device_config_info = self.device_config_info(dev) 

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

556 _LOGGER.debug( 

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

558 ) 

559 skipped_devs += 1 

560 continue 

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

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

563 skipped_devs += 1 

564 continue 

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

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

567 skipped_devs += 1 

568 continue 

569 devices.append( 

570 DeviceInfo( 

571 device_id=dev.id, 

572 device_name=dev.name, 

573 manufacturer=dev.manufacturer, 

574 model=dev.model, 

575 area_id=dev.area_id, 

576 user_id=device_config_info[ATTR_USER_ID], 

577 os_name=device_config_info[ATTR_OS_NAME], 

578 os_version=device_config_info[ATTR_OS_VERSION], 

579 app_version=device_config_info[ATTR_APP_VERSION], 

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

581 identifiers=dev.identifiers, 

582 ) 

583 ) 

584 

585 elif identifier: 

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

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

588 else: 

589 _LOGGER.debug( # type: ignore 

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

591 ) 

592 

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

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

595 

596 return devices 

597 

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

599 # discover domain from device registry 

600 verified_domain: str | None = None 

601 device_registry = self.device_registry() 

602 if device_registry: 

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

604 if device: 

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

606 if matching_domains: 

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

608 return matching_domains[0] 

609 _LOGGER.warning( 

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

611 device_id, 

612 ) 

613 return verified_domain 

614 

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

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

617 to all entities, so do it once here 

618 """ # noqa: D205 

619 if self._entity_registry is not None: 

620 return self._entity_registry 

621 try: 

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

623 except Exception as e: 

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

625 return self._entity_registry 

626 

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

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

629 to all devices, so do it once here 

630 """ # noqa: D205 

631 if self._device_registry is not None: 

632 return self._device_registry 

633 try: 

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

635 except Exception as e: 

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

637 return self._device_registry 

638 

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

640 try: 

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

642 except Exception: 

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

644 if raise_on_error: 

645 raise 

646 return False 

647 

648 async def mqtt_publish( 

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

650 ) -> None: 

651 try: 

652 await mqtt.async_publish( 

653 self._hass, 

654 topic=topic, 

655 payload=json_dumps(payload), 

656 qos=qos, 

657 retain=retain, 

658 ) 

659 except Exception: 

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

661 if raise_on_error: 

662 raise 

663 

664 

665class ConditionErrorLoggingAdaptor(logging.LoggerAdapter): 

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

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

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

669 

670 def capture(self, args: Any) -> None: 

671 if args and isinstance(args, list | tuple): 

672 for arg in args: 

673 if isinstance(arg, ConditionErrorContainer): 

674 self.condition_errors.extend(arg.errors) 

675 elif isinstance(arg, ConditionError): 

676 self.condition_errors.append(arg) 

677 

678 def error(self, msg: Any, *args: object, **kwargs: Any) -> None: 

679 self.capture(args) 

680 self.logger.error(msg, args, kwargs) 

681 

682 def warning(self, msg: Any, *args: Any, **kwargs: Any) -> None: 

683 self.capture(args) 

684 self.logger.warning(msg, args, kwargs) 

685 

686 

687def force_strict_template_mode(conditions: list[ConfigType], undo: bool = False) -> None: 

688 class TemplateWrapper: 

689 def __init__(self, obj: Template) -> None: 

690 self._obj = obj 

691 

692 def __getattr__(self, name: str) -> Any: