Coverage for custom_components/supernotify/people.py: 91%

1import logging 

2from typing import TYPE_CHECKING, Any 

3 

4from homeassistant.const import ATTR_STATE, CONF_DEVICE_ID, STATE_HOME, STATE_NOT_HOME 

5from homeassistant.helpers import device_registry, entity_registry 

6from homeassistant.util import slugify 

7 

8from . import ( 

9 ATTR_EMAIL, 

10 ATTR_MOBILE_APP_ID, 

11 ATTR_PHONE, 

12 ATTR_USER_ID, 

13 CONF_ALIAS, 

14 CONF_DATA, 

15 CONF_DELIVERY, 

16 CONF_DEVICE_NAME, 

17 CONF_DEVICE_TRACKER, 

18 CONF_EMAIL, 

19 CONF_MANUFACTURER, 

20 CONF_MOBILE_APP_ID, 

21 CONF_MOBILE_DEVICES, 

22 CONF_MOBILE_DISCOVERY, 

23 CONF_MODEL, 

24 CONF_PERSON, 

25 CONF_PHONE_NUMBER, 

26 CONF_TARGET, 

27) 

28from .common import ensure_list 

29from .hass_api import HomeAssistantAPI 

30from .model import Target 

31 

32if TYPE_CHECKING: 

33 from homeassistant.core import State 

34 from homeassistant.helpers.device_registry import DeviceRegistry 

35 from homeassistant.helpers.entity_registry import EntityRegistry 

36 

37_LOGGER = logging.getLogger(__name__) 

38 

39 

40class PeopleRegistry: 

41 def __init__(self, recipients: list[dict[str, Any]], hass_api: HomeAssistantAPI) -> None: 

42 self.hass_api = hass_api 

43 self.people: dict[str, dict[str, Any]] = {} 

44 self._recipients: list[dict[str, Any]] = ensure_list(recipients) 

45 self.entity_registry = entity_registry 

46 self.device_registry = device_registry 

47 

48 def initialize(self) -> None: 

49 for r in self._recipients: 

50 if CONF_PERSON not in r or not r[CONF_PERSON]: 

51 _LOGGER.warning("SUPERNOTIFY Skipping invalid recipient with no 'person' key:%s", r) 

52 continue 

53 

54 if r.get(CONF_TARGET): 

55 r[CONF_TARGET] = Target(r[CONF_TARGET], target_data=r.get(CONF_DATA)) 

56 else: 

57 r[CONF_TARGET] = Target() 

58 if r.get(CONF_EMAIL): 

59 r[CONF_TARGET].extend(ATTR_EMAIL, r[CONF_EMAIL]) 

60 if r.get(CONF_PHONE_NUMBER): 

61 r[CONF_TARGET].extend(ATTR_PHONE, r[CONF_PHONE_NUMBER]) 

62 

63 if r.get(CONF_MOBILE_DISCOVERY): 

64 r[CONF_MOBILE_DEVICES].extend(self.mobile_devices_for_person(r[CONF_PERSON])) 

65 if r.get(CONF_MOBILE_DEVICES): 

66 _LOGGER.info("SUPERNOTIFY Auto configured %s for mobile devices %s", r[CONF_PERSON], r[CONF_MOBILE_DEVICES]) 

67 else: 

68 _LOGGER.warning("SUPERNOTIFY Unable to find mobile devices for %s", r[CONF_PERSON]) 

69 if r.get(CONF_MOBILE_DEVICES): 

70 r[CONF_TARGET].extend(ATTR_MOBILE_APP_ID, [d[CONF_MOBILE_APP_ID] for d in r[CONF_MOBILE_DEVICES]]) 

71 

72 state: State | None = self.hass_api.get_state(r[CONF_PERSON]) 

73 if state is not None: 

74 r[ATTR_USER_ID] = state.attributes.get(ATTR_USER_ID) 

75 

76 # TODO: replace dicts with typed classes and remove inconsistency of Target for target and {} for delivery 

77 self.people[r[CONF_PERSON]] = { 

78 k: v 

79 for k, v in r.items() 

80 if k 

81 in ( 

82 CONF_PERSON, 

83 CONF_ALIAS, 

84 CONF_TARGET, 

85 CONF_EMAIL, 

86 CONF_TARGET, 

87 CONF_PHONE_NUMBER, 

88 ATTR_USER_ID, 

89 CONF_MOBILE_DISCOVERY, 

90 CONF_MOBILE_DEVICES, 

91 CONF_DELIVERY, 

92 ) 

93 } 

94 

95 def refresh_tracker_state(self) -> None: 

96 for person, person_config in self.people.items(): 

97 # TODO: possibly rate limit this 

98 try: 

99 tracker: State | None = self.hass_api.get_state(person) 

100 if tracker is None: 

101 person_config[ATTR_STATE] = None 

102 else: 

103 person_config[ATTR_STATE] = tracker.state 

104 except Exception as e: 

105 _LOGGER.warning("SUPERNOTIFY Unable to determine occupied status for %s: %s", person, e) 

106 

107 def determine_occupancy(self) -> dict[str, list[dict[str, Any]]]: 

108 results: dict[str, list[dict[str, Any]]] = {STATE_HOME: [], STATE_NOT_HOME: []} 

109 self.refresh_tracker_state() 

110 for person_config in self.people.values(): 

111 if person_config.get(ATTR_STATE) in (None, STATE_HOME): 

112 # default to at home if unknown tracker 

113 results[STATE_HOME].append(person_config) 

114 else: 

115 results[STATE_NOT_HOME].append(person_config) 

116 return results 

117 

118 def mobile_devices_for_person(self, person_entity_id: str, validate_targets: bool = False) -> list[dict[str, Any]]: 

119 """Auto detect mobile_app targets for a person. 

120 

121 Targets not currently validated as async registration may not be complete at this stage 

122 

123 Args: 

124 ---- 

125 person_entity_id (str): _description_ 

126 validate_targets (bool, optional): _description_. Defaults to False. 

127 

128 Returns: 

129 ------- 

130 list: mobile target actions for this person 

131 

132 """ 

133 mobile_devices = [] 

134 person_state = self.hass_api.get_state(person_entity_id) 

135 if not person_state: 

136 _LOGGER.warning("SUPERNOTIFY Unable to resolve %s", person_entity_id) 

137 else: 

138 ent_reg: EntityRegistry | None = self.hass_api.entity_registry() 

139 dev_reg: DeviceRegistry | None = self.hass_api.device_registry() 

140 if not ent_reg or not dev_reg: 

141 _LOGGER.warning("SUPERNOTIFY Unable to access entity or device registries for %s", person_entity_id) 

142 else: 

143 for d_t in person_state.attributes.get("device_trackers", ()): 

144 entity = ent_reg.async_get(d_t) 

145 if entity and entity.platform == "mobile_app" and entity.device_id: 

146 device = dev_reg.async_get(entity.device_id) 

147 if not device: 

148 _LOGGER.warning("SUPERNOTIFY Unable to find device %s", entity.device_id) 

149 else: 

150 mobile_app_id = f"mobile_app_{slugify(device.name)}" 

151 if validate_targets and not self.hass_api.has_service("notify", mobile_app_id): 

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

153 else: 

154 mobile_devices.append({ 

155 CONF_MANUFACTURER: device.manufacturer, 

156 CONF_MODEL: device.model, 

157 CONF_MOBILE_APP_ID: mobile_app_id, 

158 CONF_DEVICE_TRACKER: d_t, 

159 CONF_DEVICE_ID: device.id, 

160 CONF_DEVICE_NAME: device.name, 

161 # CONF_DEVICE_LABELS: device.labels, 

162 }) 

163 else: 

164 _LOGGER.debug("SUPERNOTIFY Ignoring device tracker %s", d_t) 

165 

166 return mobile_devices