Coverage for custom_components/supernotify/people.py: 19%
179 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-01-07 15:35 +0000
1import logging
2from typing import TYPE_CHECKING, Any
4from homeassistant.components.binary_sensor import (
5 BinarySensorDeviceClass,
6)
7from homeassistant.components.person.const import DOMAIN as PERSON_DOMAIN
8from homeassistant.const import (
9 ATTR_ENTITY_ID,
10 ATTR_FRIENDLY_NAME,
11 CONF_ALIAS,
12 CONF_ENABLED,
13 STATE_HOME,
14 STATE_NOT_HOME,
15 EntityCategory,
16)
17from homeassistant.helpers import device_registry, entity_registry
19from . import (
20 ATTR_ALIAS,
21 ATTR_EMAIL,
22 ATTR_ENABLED,
23 ATTR_MOBILE_APP_ID,
24 ATTR_PERSON_ID,
25 ATTR_PHONE,
26 ATTR_USER_ID,
27 CONF_DATA,
28 CONF_DELIVERY,
29 CONF_EMAIL,
30 CONF_MOBILE_APP_ID,
31 CONF_MOBILE_DEVICES,
32 CONF_MOBILE_DISCOVERY,
33 CONF_PERSON,
34 CONF_PHONE_NUMBER,
35 CONF_TARGET,
36 OCCUPANCY_ALL,
37 OCCUPANCY_ALL_IN,
38 OCCUPANCY_ALL_OUT,
39 OCCUPANCY_ANY_IN,
40 OCCUPANCY_ANY_OUT,
41 OCCUPANCY_NONE,
42 OCCUPANCY_ONLY_IN,
43 OCCUPANCY_ONLY_OUT,
44)
45from .common import ensure_list
46from .hass_api import DeviceInfo, HomeAssistantAPI
47from .model import DeliveryCustomization, Target
49if TYPE_CHECKING:
50 from homeassistant.core import State
52_LOGGER = logging.getLogger(__name__)
55class Recipient:
56 """Recipient to distinguish from the native HA Person"""
58 # for future native entity use
59 _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
60 _attr_entity_category = EntityCategory.DIAGNOSTIC
61 _attr_name = "Recipient"
62 _attr_icon = "mdi:inbox_text_person"
64 def __init__(self, config: dict[str, Any] | None, default_mobile_discovery: bool = True) -> None:
65 config = config or {}
66 self.entity_id = config[CONF_PERSON]
67 self.name: str = self.entity_id.replace("person.", "")
68 self.alias: str | None = config.get(CONF_ALIAS)
69 self.email: str | None = config.get(CONF_EMAIL)
70 self.phone_number: str | None = config.get(CONF_PHONE_NUMBER)
71 # test support only
72 self.user_id: str | None = config.get(ATTR_USER_ID)
74 self._target: Target = Target(config.get(CONF_TARGET, {}), target_data=config.get(CONF_DATA))
75 self.delivery_overrides: dict[str, DeliveryCustomization] = {
76 k: DeliveryCustomization(v, target_specific=True) for k, v in config.get(CONF_DELIVERY, {}).items()
77 }
78 self.enabled: bool = config.get(CONF_ENABLED, True)
79 self.mobile_discovery: bool = config.get(CONF_MOBILE_DISCOVERY, default_mobile_discovery)
80 self.mobile_devices: dict[str, dict[str, str | list[str] | None]] = {
81 c[CONF_MOBILE_APP_ID]: c for c in config.get(CONF_MOBILE_DEVICES, [])
82 }
83 self.disabled_mobile_app_ids: list[str] = [k for k, v in self.mobile_devices.items() if not v.get(CONF_ENABLED, True)]
84 _LOGGER.debug("SUPERNOTIFY Recipient config %s -> %s", config, self.as_dict())
86 def initialize(self, people_registry: "PeopleRegistry") -> None:
88 self._target.extend(ATTR_PERSON_ID, [self.entity_id])
89 if self.email:
90 self._target.extend(ATTR_EMAIL, self.email)
91 if self.phone_number:
92 self._target.extend(ATTR_PHONE, self.phone_number)
93 if self.mobile_discovery:
94 discovered_devices: list[DeviceInfo] = people_registry.mobile_devices_for_person(self.entity_id)
95 if discovered_devices:
96 new_ids = []
97 for d in discovered_devices:
98 if d.mobile_app_id in self.mobile_devices:
99 # merge with manual registrations, with priority to manually overridden values
100 merged = d.as_dict()
101 merged.update(self.mobile_devices[d.mobile_app_id])
102 self.mobile_devices[d.mobile_app_id] = merged
103 new_ids.append(d.mobile_app_id)
104 _LOGGER.debug("SUPERNOTIFY Updating %s mobile device %s from registry", self.entity_id, d.mobile_app_id)
105 elif d.mobile_app_id is not None:
106 self.mobile_devices[d.mobile_app_id] = d.as_dict()
107 new_ids.append(d.mobile_app_id)
108 _LOGGER.info(
109 "SUPERNOTIFY Auto configured %s for mobile devices %s",
110 self.entity_id,
111 ",".join(new_ids),
112 )
113 else:
114 _LOGGER.info("SUPERNOTIFY Unable to find mobile devices for %s", self.entity_id)
115 if self.mobile_devices:
116 self._target.extend(ATTR_MOBILE_APP_ID, list(self.enabled_mobile_devices.keys()))
117 if not self.user_id or not self.alias:
118 attrs: dict[str, Any] | None = people_registry.person_attributes(self.entity_id)
119 if attrs:
120 if attrs.get(ATTR_USER_ID) and isinstance(attrs.get(ATTR_USER_ID), str):
121 self.user_id = attrs.get(ATTR_USER_ID)
122 if attrs.get(ATTR_ALIAS) and isinstance(attrs.get(ATTR_ALIAS), str):
123 self.alias = attrs.get(ATTR_ALIAS)
124 if not self.alias and attrs.get(ATTR_FRIENDLY_NAME) and isinstance(attrs.get(ATTR_FRIENDLY_NAME), str):
125 self.alias = attrs.get(ATTR_FRIENDLY_NAME)
126 _LOGGER.debug("SUPERNOTIFY Person attrs found for %s: %s,%s", self.entity_id, self.alias, self.user_id)
127 else:
128 _LOGGER.debug("SUPERNOTIFY No person attrs found for %s", self.entity_id)
129 _LOGGER.debug("SUPERNOTIFY Recipient %s target: %s", self.entity_id, self._target.as_dict())
131 @property
132 def enabled_mobile_devices(self) -> dict[str, dict[str, str | list[str] | None]]:
133 return {k: v for k, v in self.mobile_devices.items() if v.get(CONF_ENABLED, True)}
135 def enabling_delivery_names(self) -> list[str]:
136 return [delname for delname, delconf in self.delivery_overrides.items() if delconf.enabled is True]
138 def target(self, delivery_name: str) -> Target:
139 recipient_target: Target = self._target
140 personal_delivery: DeliveryCustomization | None = self.delivery_overrides.get(delivery_name)
141 if personal_delivery and personal_delivery.enabled is not False:
142 if personal_delivery.target and personal_delivery.target.has_targets():
143 recipient_target += personal_delivery.target
144 if personal_delivery.data:
145 recipient_target += Target([], target_data=personal_delivery.data, target_specific_data=True)
146 return recipient_target
148 def as_dict(self, occupancy_only: bool = False, **_kwargs: Any) -> dict[str, Any]:
149 result = {CONF_PERSON: self.entity_id, CONF_ENABLED: self.enabled}
150 if not occupancy_only:
151 result.update({
152 CONF_ALIAS: self.alias,
153 CONF_EMAIL: self.email,
154 CONF_PHONE_NUMBER: self.phone_number,
155 ATTR_USER_ID: self.user_id,
156 CONF_MOBILE_DISCOVERY: self.mobile_discovery,
157 CONF_MOBILE_DEVICES: list(self.mobile_devices.values()),
158 CONF_TARGET: self._target.as_dict() if self._target else None,
159 CONF_DELIVERY: {d: c.as_dict() for d, c in self.delivery_overrides.items()}
160 if self.delivery_overrides
161 else None,
162 })
163 return result
165 def attributes(self) -> dict[str, Any]:
166 """For exposure as entity state"""
167 attrs: dict[str, Any] = {
168 ATTR_ENTITY_ID: self.entity_id,
169 ATTR_ENABLED: self.enabled,
170 CONF_EMAIL: self.email,
171 CONF_PHONE_NUMBER: self.phone_number,
172 ATTR_USER_ID: self.user_id,
173 CONF_MOBILE_DEVICES: list(self.mobile_devices.values()),
174 CONF_MOBILE_DISCOVERY: self.mobile_discovery,
175 CONF_TARGET: self._target,
176 CONF_DELIVERY: self.delivery_overrides,
177 }
178 if self.alias:
179 attrs[ATTR_FRIENDLY_NAME] = self.alias
180 return attrs
183class PeopleRegistry:
184 def __init__(
185 self,
186 recipients: list[dict[str, Any]],
187 hass_api: HomeAssistantAPI,
188 discover: bool = False,
189 mobile_discovery: bool = True,
190 ) -> None:
191 self.hass_api = hass_api
192 self.people: dict[str, Recipient] = {}
193 self._recipients: list[dict[str, Any]] = ensure_list(recipients)
194 self.entity_registry = entity_registry
195 self.device_registry = device_registry
196 self.mobile_discovery = mobile_discovery
197 self.discover = discover
199 def initialize(self) -> None:
200 recipients: dict[str, dict[str, Any]] = {}
201 if self.discover:
202 entity_ids = self.find_people()
203 if entity_ids:
204 recipients = {entity_id: {CONF_PERSON: entity_id} for entity_id in entity_ids}
205 _LOGGER.info("SUPERNOTIFY Auto-discovered people: %s", entity_ids)
207 for r in self._recipients:
208 if CONF_PERSON not in r or not r[CONF_PERSON]:
209 _LOGGER.warning("SUPERNOTIFY Skipping invalid recipient with no 'person' key:%s", r)
210 continue
211 person_id = r[CONF_PERSON]
212 if person_id in recipients:
213 _LOGGER.debug("SUPERNOTIFY Overriding %s entity defaults from recipient config", person_id)
214 recipients[person_id].update(r)
215 else:
216 recipients[person_id] = r
218 for r in recipients.values():
219 recipient: Recipient = Recipient(r, default_mobile_discovery=self.mobile_discovery)
220 recipient.initialize(self)
222 self.people[recipient.entity_id] = recipient
224 def person_attributes(self, entity_id: str) -> dict[str, Any] | None:
225 state: State | None = self.hass_api.get_state(entity_id)
226 if state is not None and state.attributes:
227 return state.attributes
228 return None
230 def find_people(self) -> list[str]:
231 return self.hass_api.entity_ids_for_domain(PERSON_DOMAIN)
233 def enabled_recipients(self) -> list[Recipient]:
234 return [p for p in self.people.values() if p.enabled]
236 def filter_recipients_by_occupancy(self, delivery_occupancy: str) -> list[Recipient]:
237 if delivery_occupancy == OCCUPANCY_NONE:
238 return []
240 people = [p for p in self.people.values() if p.enabled]
241 if delivery_occupancy == OCCUPANCY_ALL:
242 return people
244 occupancy = self.determine_occupancy()
246 away = occupancy[STATE_NOT_HOME]
247 at_home = occupancy[STATE_HOME]
248 if delivery_occupancy == OCCUPANCY_ALL_IN:
249 return people if len(away) == 0 else []
250 if delivery_occupancy == OCCUPANCY_ALL_OUT:
251 return people if len(at_home) == 0 else []
252 if delivery_occupancy == OCCUPANCY_ANY_IN:
253 return people if len(at_home) > 0 else []
254 if delivery_occupancy == OCCUPANCY_ANY_OUT:
255 return people if len(away) > 0 else []
256 if delivery_occupancy == OCCUPANCY_ONLY_IN:
257 return at_home
258 if delivery_occupancy == OCCUPANCY_ONLY_OUT:
259 return away
261 _LOGGER.warning("SUPERNOTIFY Unknown occupancy tested: %s", delivery_occupancy)
262 return []
264 def _fetch_person_entity_state(self, person_id: str) -> str | None:
265 try:
266 tracker: State | None = self.hass_api.get_state(person_id)
267 if tracker and isinstance(tracker.state, str):
268 return tracker.state
269 _LOGGER.warning("SUPERNOTIFY Unexpected state %s for %s", tracker, person_id)
270 except Exception as e:
271 _LOGGER.warning("SUPERNOTIFY Unable to determine occupied status for %s: %s", person_id, e)
272 return None
274 def determine_occupancy(self) -> dict[str, list[Recipient]]:
275 results: dict[str, list[Recipient]] = {STATE_HOME: [], STATE_NOT_HOME: []}
276 for person_id, person_config in self.people.items():
277 if person_config.enabled:
278 state: str | None = self._fetch_person_entity_state(person_id)
279 if state in (None, STATE_HOME):
280 # default to at home if unknown tracker
281 results[STATE_HOME].append(person_config)
282 else:
283 results[STATE_NOT_HOME].append(person_config)
284 return results
286 def mobile_devices_for_person(self, person_entity_id: str) -> list[DeviceInfo]:
287 """Auto detect mobile_app targets for a person.
289 Targets not currently validated as async registration may not be complete at this stage
291 Args:
292 ----
293 person_entity_id (str): _description_
295 Returns:
296 -------
297 list: mobile target actions for this person
299 """
300 person_state = self.hass_api.get_state(person_entity_id)
301 if not person_state:
302 _LOGGER.warning("SUPERNOTIFY Unable to resolve %s", person_entity_id)
303 else:
304 user_id = person_state.attributes.get(ATTR_USER_ID)
305 if user_id:
306 return self.hass_api.mobile_app_by_user_id(user_id) or []
307 _LOGGER.debug("SUPERNOTIFY Unable to link %s to a user_id", person_entity_id)
308 return []