Coverage for custom_components/camera_autonotify/discovery.py: 95%

1"""Find Frigate cameras, and zones if wanted, 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 ( 

16 CONF_AUTO_DISCOVER, 

17 CONF_DISCOVER_ZONES, 

18 CONF_SOURCE, 

19 CONF_SOURCE_ENTRY_ID, 

20 CONF_SOURCE_NAME, 

21 DOMAIN, 

22 FRIGATE_DOMAIN, 

23 SOURCE_FRIGATE, 

24 TYPE_CAMERA, 

25 TYPE_DISCOVERY, 

26 TYPE_ZONE, 

27) 

28 

29if TYPE_CHECKING: 

30 from homeassistant.helpers.entity_registry import EventEntityRegistryUpdatedData 

31 

32CAMERA_DOMAIN = "camera" 

33# the unique id the Frigate integration gives a camera's entity, `{entry_id}:camera:{name}`, the name as in Frigate 

34CAMERA_UNIQUE_ID = ":camera:" 

35 

36 

37@dataclass(frozen=True) 

38class Found: 

39 """A Frigate camera or zone, each a device of the Frigate integration""" 

40 

41 device_id: str 

42 name: str 

43 type: str 

44 source_entry_id: str 

45 # the camera's name as Frigate has it, as reviews give it, or a zone's as slugified for its device, since reviews 

46 # give a zone as written in the Frigate config, with whatever case it has there 

47 source_name: str 

48 

49 

50def _device_name(device: dr.DeviceEntry) -> str: 

51 return device.name_by_user or device.name or device.id 

52 

53 

54@callback 

55def find_cameras(hass: HomeAssistant) -> list[Found]: 

56 """The device of each Frigate camera, known by its camera entity""" 

57 device_registry = dr.async_get(hass) 

58 found: list[Found] = [] 

59 for entity in er.async_get(hass).entities.values(): 

60 if entity.platform != FRIGATE_DOMAIN or entity.domain != CAMERA_DOMAIN or entity.disabled: 

61 continue 

62 source_entry_id, sep, camera = entity.unique_id.partition(CAMERA_UNIQUE_ID) 

63 device = device_registry.async_get(entity.device_id) if entity.device_id else None 

64 if not sep or not isinstance(device, dr.DeviceEntry): 

65 continue 

66 found.append(Found(device.id, _device_name(device), TYPE_CAMERA, source_entry_id, camera)) 

67 return found 

68 

69 

70@callback 

71def find_zones(hass: HomeAssistant) -> list[Found]: 

72 """The device of each Frigate zone, the devices of a Frigate server that are neither the server nor a camera""" 

73 device_registry = dr.async_get(hass) 

74 cameras = {camera.device_id for camera in find_cameras(hass)} 

75 found: list[Found] = [] 

76 for entry in hass.config_entries.async_entries(FRIGATE_DOMAIN): 

77 prefix = f"{entry.entry_id}:" 

78 for device in dr.async_entries_for_config_entry(device_registry, entry.entry_id): 

79 if device.id in cameras or device.disabled: 

80 continue 

81 zone = next( 

82 ( 

83 value[len(prefix) :] 

84 for domain, value in device.identifiers 

85 if domain == FRIGATE_DOMAIN and value.startswith(prefix) 

86 ), 

87 None, 

88 ) 

89 if zone: 

90 found.append(Found(device.id, _device_name(device), TYPE_ZONE, entry.entry_id, zone)) 

91 return found 

92 

93 

94@callback 

95def _discovery_options(hass: HomeAssistant) -> dict[str, bool]: 

96 """The choices of the discovery entry, nothing found while it's disabled or deleted""" 

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

98 if entry.data.get(CONF_TYPE) == TYPE_DISCOVERY: 

99 return { 

100 CONF_AUTO_DISCOVER: bool(entry.options.get(CONF_AUTO_DISCOVER, True)), 

101 CONF_DISCOVER_ZONES: bool(entry.options.get(CONF_DISCOVER_ZONES, False)), 

102 } 

103 return {} 

104 

105 

106@callback 

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

108 """Set up every camera found, or with automatic discovery off, offer each to be added, and zones only if wanted""" 

109 options = _discovery_options(hass) 

110 if not options: 

111 return 

112 source = SOURCE_IMPORT if options[CONF_AUTO_DISCOVER] else SOURCE_INTEGRATION_DISCOVERY 

113 found = find_cameras(hass) + (find_zones(hass) if options[CONF_DISCOVER_ZONES] else []) 

114 for item in found: 

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

116 discovery_flow.async_create_flow( 

117 hass, 

118 DOMAIN, 

119 context={"source": source}, 

120 data={ 

121 CONF_DEVICE_ID: item.device_id, 

122 CONF_NAME: item.name, 

123 CONF_TYPE: item.type, 

124 CONF_SOURCE: SOURCE_FRIGATE, 

125 CONF_SOURCE_ENTRY_ID: item.source_entry_id, 

126 CONF_SOURCE_NAME: item.source_name, 

127 }, 

128 ) 

129 

130 

131@callback 

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

133 """Look for cameras now, and again whenever Frigate adds an entity, as it does for a new camera or zone""" 

134 

135 @callback 

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

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

138 return 

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

140 if entity is not None and entity.platform == FRIGATE_DOMAIN: 

141 async_discover(hass) 

142 

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

144 async_discover(hass)