Coverage for custom_components/appliance_autonotify/discovery.py: 96%

1"""Find appliances with an identifiable cycle, and set each up or offer it as a discovered config entry""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING 

7 

8from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_INTEGRATION_DISCOVERY 

9from homeassistant.const import CONF_DEVICE_ID, CONF_NAME, CONF_TYPE 

10from homeassistant.core import Event, HomeAssistant, callback 

11from homeassistant.helpers import device_registry as dr 

12from homeassistant.helpers import discovery_flow 

13from homeassistant.helpers import entity_registry as er 

14 

15from .const import CONF_AUTO_DISCOVER, CYCLE_KEYS, DOMAIN, TYPE_DISCOVERY 

16 

17if TYPE_CHECKING: 

18 from homeassistant.helpers.entity_registry import EventEntityRegistryUpdatedData 

19 

20 

21@dataclass(frozen=True) 

22class Appliance: 

23 device_id: str 

24 name: str 

25 

26 

27@callback 

28def find_appliances(hass: HomeAssistant) -> list[Appliance]: 

29 """Devices of the known platforms that have a cycle state entity""" 

30 entity_registry = er.async_get(hass) 

31 device_registry = dr.async_get(hass) 

32 found: dict[str, Appliance] = {} 

33 for entity in entity_registry.entities.values(): 

34 if entity.device_id is None or entity.disabled or entity.translation_key is None: 

35 continue 

36 if CYCLE_KEYS.get(entity.platform) != entity.translation_key: 

37 continue 

38 device = device_registry.async_get(entity.device_id) 

39 if device is not None: 

40 found[device.id] = Appliance(device.id, device.name_by_user or device.name or entity.entity_id) 

41 return list(found.values()) 

42 

43 

44@callback 

45def auto_discovery(hass: HomeAssistant) -> bool: 

46 """Whether appliances found are set up without asking, so not while the discovery entry is disabled or deleted""" 

47 return any( 

48 entry.data.get(CONF_TYPE) == TYPE_DISCOVERY and entry.options.get(CONF_AUTO_DISCOVER) 

49 for entry in hass.config_entries.async_entries(DOMAIN, include_disabled=False) 

50 ) 

51 

52 

53@callback 

54def async_discover(hass: HomeAssistant) -> None: 

55 """Set up every appliance found, or with automatic discovery off, offer each to be added""" 

56 source = SOURCE_IMPORT if auto_discovery(hass) else SOURCE_INTEGRATION_DISCOVERY 

57 for appliance in find_appliances(hass): 

58 # appliances already configured, ignored or in progress are dropped by the flow's unique id 

59 discovery_flow.async_create_flow( 

60 hass, 

61 DOMAIN, 

62 context={"source": source}, 

63 data={CONF_DEVICE_ID: appliance.device_id, CONF_NAME: appliance.name}, 

64 ) 

65 

66 

67@callback 

68def async_start_discovery(hass: HomeAssistant) -> None: 

69 """Look for appliances now, and again whenever a known platform adds an entity""" 

70 

71 @callback 

72 def entity_registry_updated(event: Event[EventEntityRegistryUpdatedData]) -> None: 

73 if event.data["action"] != "create": 

74 return 

75 entity = er.async_get(hass).async_get(event.data["entity_id"]) 

76 if entity is not None and entity.platform in CYCLE_KEYS: 

77 async_discover(hass) 

78 

79 hass.bus.async_listen(er.EVENT_ENTITY_REGISTRY_UPDATED, entity_registry_updated) 

80 async_discover(hass)