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