Coverage for custom_components/supernotify/people.py: 95%
179 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-01 15:06 +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 = config[CONF_PERSON]
71 self.name: str = self.entity_id.replace("person.", "")
72 self.alias: str | None = config.get(CONF_ALIAS)
73 self.email: str | None = config.get(CONF_EMAIL)
74 self.phone_number: str | None = config.get(CONF_PHONE_NUMBER)
75 # test support only
76 self.user_id: str | None = config.get(ATTR_USER_ID)
78 self._target: Target = Target(config.get(CONF_TARGET, {}), target_data=config.get(CONF_DATA))
79 self.delivery_overrides: dict[str, DeliveryCustomization] = {
80 k: DeliveryCustomization(v, target_specific=True) for k, v in config.get(CONF_DELIVERY, {}).items()
81 }
82 self.enabled: bool = config.get(CONF_ENABLED, True)
83 self.mobile_discovery: bool = config.get(CONF_MOBILE_DISCOVERY, default_mobile_discovery)
84 self.mobile_devices: dict[str, dict[str, str | list[str] | None]] = {
85 c[CONF_MOBILE_APP_ID]: c for c in config.get(CONF_MOBILE_DEVICES, [])
86 }
87 self.disabled_mobile_app_ids: list[str] = [k for k, v in self.mobile_devices.items() if not v.get(CONF_ENABLED, True)]
88 _LOGGER.debug("SUPERNOTIFY Recipient config %s -> %s", config, self.as_dict())
90 def initialize(self, people_registry: PeopleRegistry) -> None:
92 self._target.extend(ATTR_PERSON_ID, [self.entity_id])
93 if self.email:
94 self._target.extend(ATTR_EMAIL, self.email)
95 if self.phone_number:
96 self._target.extend(ATTR_PHONE, self.phone_number)
97 if self.mobile_discovery:
98 discovered_devices: list[DeviceInfo] = people_registry.mobile_devices_for_person(self.entity_id)
99 if discovered_devices:
100 new_ids = []
101 for d in discovered_devices:
102 if d.mobile_app_id in self.mobile_devices:
103 # merge with manual registrations, with priority to manually overridden values
104 merged = d.as_dict()
105 merged.update(self.mobile_devices[d.mobile_app_id])
106 self.mobile_devices[d.mobile_app_id] = merged
107 new_ids.append(d.mobile_app_id)
108 _LOGGER.debug("SUPERNOTIFY Updating %s mobile device %s from registry", self.entity_id, d.mobile_app_id)
109 elif d.mobile_app_id is not None:
110 self.mobile_devices[d.mobile_app_id] = d.as_dict()
111 new_ids.append(d.mobile_app_id)
112 _LOGGER.info(
113 "SUPERNOTIFY Auto configured %s for mobile devices %s",
114 self.entity_id,
115 ",".join(new_ids),
116 )
117 else:
118 _LOGGER.info("SUPERNOTIFY Unable to find mobile devices for %s", self.entity_id)
119 if self.mobile_devices:
120 self._target.extend(ATTR_MOBILE_APP_ID, list(self.enabled_mobile_devices.keys()))
121 if not self.user_id or not self.alias:
122 attrs: dict[str, Any] | None = people_registry.person_attributes(self.entity_id)
123 if attrs:
124 if attrs.get(ATTR_USER_ID) and isinstance(attrs.get(ATTR_USER_ID), str):
125 self.user_id = attrs.get(ATTR_USER_ID)
126 if attrs.get(ATTR_ALIAS) and isinstance(attrs.get(ATTR_ALIAS), str):
127 self.alias = attrs.get(ATTR_ALIAS)
128 if not self.alias and attrs.get(ATTR_FRIENDLY_NAME) and isinstance(attrs.get(ATTR_FRIENDLY_NAME), str):
129 self.alias = attrs.get(ATTR_FRIENDLY_NAME)
130 _LOGGER.debug("SUPERNOTIFY Person attrs found for %s: %s,%s", self.entity_id, self.alias, self.user_id)
131 else:
132 _LOGGER.debug("SUPERNOTIFY No person attrs found for %s", self.entity_id)
133 _LOGGER.debug("SUPERNOTIFY Recipient %s target: %s", self.entity_id, self._target.as_dict())
135 @property
136 def enabled_mobile_devices(self) -> dict[str, dict[str, str | list[str] | None]]:
137 return {k: v for k, v in self.mobile_devices.items() if v.get(CONF_ENABLED, True)}
139 def enabling_delivery_names(self) -> list[str]:
140 return [delname for delname, delconf in self.delivery_overrides.items() if delconf.enabled is True]
142 def target(self, delivery_name: str) -> Target:
143 recipient_target: Target = self._target
144 personal_delivery: DeliveryCustomization | None = self.delivery_overrides.get(delivery_name)
145 if personal_delivery and personal_delivery.enabled is not False:
146 if personal_delivery.target and personal_delivery.target.has_targets():
147 recipient_target += personal_delivery.target
148 if personal_delivery.data:
149 recipient_target += Target([], target_data=personal_delivery.data, target_specific_data=True)
150 return recipient_target
152 def as_dict(self, occupancy_only: bool = False, **_kwargs: Any) -> dict[str, Any]:
153 result = {CONF_PERSON: self.entity_id, CONF_ENABLED: self.enabled}
154 if not occupancy_only:
155 result.update({
156 CONF_ALIAS: self.alias,
157 CONF_EMAIL: self.email,
158 CONF_PHONE_NUMBER: self.phone_number,
159 ATTR_USER_ID: self.user_id,
160 CONF_MOBILE_DISCOVERY: self.mobile_discovery,
161 CONF_MOBILE_DEVICES: list(self.mobile_devices.values()),
162 CONF_TARGET: self._target.as_dict() if self._target else None,
163 CONF_DELIVERY: {d: c.as_dict() for d, c in self.delivery_overrides.items()}
164 if self.delivery_overrides
165 else None,
166 })
167 return result
169 def attributes(self) -> dict[str, Any]:
170 """For exposure as entity state"""
171 attrs: dict[str, Any] = {
172 ATTR_ENTITY_ID: self.entity_id,
173 ATTR_ENABLED: self.enabled,
174 CONF_EMAIL: self.email,
175 CONF_PHONE_NUMBER: self.phone_number,
176 ATTR_USER_ID: self.user_id,
177 CONF_MOBILE_DEVICES: list(self.mobile_devices.values()),
178 CONF_MOBILE_DISCOVERY: self.mobile_discovery,
179 CONF_TARGET: self._target,
180 CONF_DELIVERY: self.delivery_overrides,
181 }
182 if self.alias:
183 attrs[ATTR_FRIENDLY_NAME] = self.alias
184 return attrs
187class PeopleRegistry:
188 def __init__(
189 self,
190 recipients: list[dict[str, Any]],
191 hass_api: HomeAssistantAPI,
192 discover: bool = False,
193 mobile_discovery: bool = True,
194 ) -> None:
195 self.hass_api = hass_api
196 self.people: dict[str, Recipient] = {}
197 self._recipients: list[dict[str, Any]] = ensure_list(recipients)
198 self.entity_registry = entity_registry
199 self.device_registry = device_registry
200 self.mobile_discovery = mobile_discovery
201 self.discover = discover
203 def initialize(self) -> None:
204 recipients: dict[str, dict[str, Any]] = {}
205 if self.discover:
206 entity_ids = self.find_people()
207 if entity_ids:
208 recipients = {entity_id: {CONF_PERSON: entity_id} for entity_id in entity_ids}
209 _LOGGER.info("SUPERNOTIFY Auto-discovered people: %s", entity_ids)
211 for r in self._recipients:
212 if CONF_PERSON not in r or not r[CONF_PERSON]:
213 _LOGGER.warning("SUPERNOTIFY Skipping invalid recipient with no 'person' key:%s", r)
214 continue
215 person_id = r[CONF_PERSON]
216 if person_id in recipients:
217 _LOGGER.debug("SUPERNOTIFY Overriding %s entity defaults from recipient config", person_id)
218 recipients[person_id].update(r)
219 else:
220 recipients[person_id] = r
222 for r in recipients.values():
223 recipient: Recipient = Recipient(r, default_mobile_discovery=self.mobile_discovery)
224 recipient.initialize(self)
226 self.people[recipient.entity_id] = recipient
228 def person_attributes(self, entity_id: str) -> dict[str, Any] | None:
229 state: State | None = self.hass_api.get_state(entity_id)
230 if state is not None and state.attributes:
231 return state.attributes
232 return None
234 def find_people(self) -> list[str]:
235 return self.hass_api.entity_ids_for_domain(PERSON_DOMAIN)
237 def enabled_recipients(self) -> list[Recipient]:
238 return [p for p in self.people.values() if p.enabled]
240 def filter_recipients_by_occupancy(self, delivery_occupancy: str) -> list[Recipient]:
241 if delivery_occupancy == OCCUPANCY_NONE:
242 return []
244 people = [p for p in self.people.values() if p.enabled]
245 if delivery_occupancy == OCCUPANCY_ALL:
246 return people
248 occupancy = self.determine_occupancy()
250 away = occupancy[STATE_NOT_HOME]
251 at_home = occupancy[STATE_HOME]
252 if delivery_occupancy == OCCUPANCY_ALL_IN:
253 return people if len(away) == 0 else []
254 if delivery_occupancy == OCCUPANCY_ALL_OUT:
255 return people if len(at_home) == 0 else []
256 if delivery_occupancy == OCCUPANCY_ANY_IN:
257 return people if len(at_home) > 0 else []
258 if delivery_occupancy == OCCUPANCY_ANY_OUT:
259 return people if len(away) > 0 else []
260 if delivery_occupancy == OCCUPANCY_ONLY_IN:
261 return at_home
262 if delivery_occupancy == OCCUPANCY_ONLY_OUT:
263 return away
265 _LOGGER.warning("SUPERNOTIFY Unknown occupancy tested: %s", delivery_occupancy)
266 return []
268 def _fetch_person_entity_state(self, person_id: str) -> str | None:
269 try:
270 tracker: State | None = self.hass_api.get_state(person_id)
271 if tracker and isinstance(tracker.state, str):
272 return tracker.state
273 _LOGGER.warning("SUPERNOTIFY Unexpected state %s for %s", tracker, person_id)
274 except Exception as e:
275 _LOGGER.warning("SUPERNOTIFY Unable to determine occupied status for %s: %s", person_id, e)
276 return None
278 def determine_occupancy(self) -> dict[str, list[Recipient]]:
279 results: dict[str, list[Recipient]] = {STATE_HOME: [], STATE_NOT_HOME: []}
280 for person_id, person_config in self.people.items():
281 if person_config.enabled:
282 state: str | None = self._fetch_person_entity_state(person_id)
283 if state in (None, STATE_HOME):
284 # default to at home if unknown tracker
285 results[STATE_HOME].append(person_config)
286 else:
287 results[STATE_NOT_HOME].append(person_config)
288 return results
290 def mobile_devices_for_person(self, person_entity_id: str) -> list[DeviceInfo]:
291 """Auto detect mobile_app targets for a person.
293 Targets not currently validated as async registration may not be complete at this stage
295 Args:
296 ----
297 person_entity_id (str): _description_
299 Returns:
300 -------
301 list: mobile target actions for this person
303 """
304 person_state = self.hass_api.get_state(person_entity_id)
305 if not person_state:
306 _LOGGER.warning("SUPERNOTIFY Unable to resolve %s", person_entity_id)
307 else:
308 user_id = person_state.attributes.get(ATTR_USER_ID)
309 if user_id:
310 return self.hass_api.mobile_app_by_user_id(user_id) or []
311 _LOGGER.debug("SUPERNOTIFY Unable to link %s to a user_id", person_entity_id)
312 return []