Coverage for custom_components/supernotify/transports/mobile_push.py: 100%
196 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1"""Mobile App Companion transport for SuperNotify.
3Sends push notifications to HA Companion App on iOS and Android devices.
4Supports per-device delivery with automatic snooze on failure.
6Priority mapping (auto, overridable via push_critical_level_ios):
7 critical → iOS: interruption_level=critical + Android: ttl=0
8 high → iOS: interruption_level=time-sensitive
9 medium → iOS: interruption_level=active (default)
10 low → iOS: interruption_level=passive
11 minimum → iOS: interruption_level=passive
13New data keys (all optional):
14 mobile_push_critical_level str iOS interruption_level override
15 ("passive","active","time-sensitive","critical")
16 If omitted, auto-mapped from SuperNotify priority.
17 mobile_push_critical_ttl int Android FCM TTL in ms (0=no caching/instant).
18 Auto-set to 0 for critical priority if not set.
19 mobile_push_critical_priority int Android FCM priority override (1=min, 5=max).
20 mobile_push_subtitle str iOS subtitle (line between title and message, iOS 10+)
21 mobile_push_group str Notification group for visual stacking (iOS thread-id / Android group).
22 Falls back to the camera entity id if there's a camera image,
23 otherwise left unset (notification appears individually).
24 mobile_push_notification_tag str Notification tag for replacement (iOS) / grouping (Android)
25 mobile_push_clear_notification bool Send clear_notification to dismiss previous same-tag notification.
26 Requires push_notification_tag to be set.
27 mobile_push_tts_text str Android TTS text read aloud on device (Android 8+).
28 If omitted, push TTS is not activated.
29 mobile_push_tts_locale str BCP-47 language for TTS (e.g. "it-IT", "en-US").
30 Only used when push_tts_text is set.
31 mobile_push_tts_engine str TTS engine package (e.g. "com.google.android.tts").
32 Only used when push_tts_text is set.
33 mobile_push_command_screen_on bool Android: turn on device screen on delivery (Android 8+)
34 mobile_push_command_dnd str Android: change Do Not Disturb ("toggle","off","on")
35 mobile_push_command_ringer_mode str Android: change ringer mode ("silent","vibrate","normal")
36 mobile_push_channel_override str Android notification channel override (e.g. "alarm","general")
37 mobile_push_alarm_stream bool Android: route audio through alarm stream (interrupts DND/silent)
38 mobile_push_alarm_stream_max bool Android: alarm stream at maximum device volume
40"""
42from __future__ import annotations
44import logging
45import time
46from datetime import timedelta
47from typing import TYPE_CHECKING, Any, ClassVar
49from aiohttp import ClientResponse, ClientSession, ClientTimeout
50from bs4 import BeautifulSoup
51from homeassistant.components.notify.const import ATTR_DATA
52from homeassistant.helpers.typing import ConfigType
54from custom_components.supernotify import const
55from custom_components.supernotify.const import (
56 ATTR_ACTION_URL,
57 ATTR_ACTION_URL_TITLE,
58 ATTR_DEFAULT,
59 ATTR_IMAGE,
60 ATTR_MEDIA_CAMERA_ENTITY_ID,
61 ATTR_MEDIA_CLIP_URL,
62 ATTR_MEDIA_SNAPSHOT_URL,
63 ATTR_MOBILE_APP_ID,
64 ATTR_VIDEO,
65 INCLUSION_DEFAULT,
66 MANUFACTURER_APPLE,
67 TRANSPORT_MOBILE_PUSH,
68)
69from custom_components.supernotify.media_grab import select_avail_camera
70from custom_components.supernotify.model import (
71 CommandType,
72 DebugTrace,
73 MessageOnlyPolicy,
74 QualifiedTargetType,
75 RecipientType,
76 SelectionRule,
77 TargetRequired,
78 TransportConfig,
79 TransportFeature,
80)
81from custom_components.supernotify.options import (
82 MEDIA_OPTIONS,
83 OPTION_DATA_KEYS_SELECT,
84 OPTION_DEVICE_DISCOVERY,
85 OPTION_DEVICE_DOMAIN,
86 OPTION_DEVICE_MODEL_SELECT,
87 OPTION_MESSAGE_USAGE,
88 OPTION_SIMPLIFY_TEXT,
89 OPTION_STRIP_URLS,
90 OPTION_UNIQUE_TARGETS,
91 DeliveryOption,
92)
93from custom_components.supernotify.target import Target, TargetEntityCategory
94from custom_components.supernotify.transport import Transport
96if TYPE_CHECKING:
97 from custom_components.supernotify.envelope import Envelope
98 from custom_components.supernotify.hass_api import HomeAssistantAPI, TrackedDeviceDetails
100_LOGGER = logging.getLogger(__name__)
102# iOS interruption_level mapping from SuperNotify priority
103IOS_INTERRUPTION_MAP: dict[str, str] = {
104 const.PRIORITY_CRITICAL: "critical",
105 const.PRIORITY_HIGH: "time-sensitive",
106 const.PRIORITY_MEDIUM: "active",
107 const.PRIORITY_LOW: "passive",
108 const.PRIORITY_MINIMUM: "passive",
109}
111# Android FCM TTL auto-set for critical priority (0 = instant, no FCM caching)
112ANDROID_CRITICAL_TTL = 0
115class MobilePushTransport(Transport):
116 name = TRANSPORT_MOBILE_PUSH
117 declared_options: ClassVar[list[DeliveryOption]] = [
118 *MEDIA_OPTIONS,
119 DeliveryOption(
120 OPTION_DATA_KEYS_SELECT,
121 "Prune the data block by including/excluding values or by regex pattern",
122 value_type=SelectionRule,
123 ),
124 ]
126 def __init__(self, *args: Any, **kwargs: Any) -> None:
127 super().__init__(*args, **kwargs)
128 self.action_titles: dict[str, str] = {}
129 self.action_title_failures: dict[str, float] = {}
131 @property
132 def supported_features(self) -> TransportFeature:
133 return (
134 TransportFeature.MESSAGE
135 | TransportFeature.TITLE
136 | TransportFeature.ACTIONS
137 | TransportFeature.IMAGES
138 | TransportFeature.VIDEO
139 | TransportFeature.SNAPSHOT_IMAGE
140 )
142 def extra_attributes(self) -> dict[str, Any]:
143 return {"action_titles": self.action_titles, "action_title_failures": self.action_title_failures}
145 @property
146 def inclusion_mode(self) -> list[str]:
147 # a mobile device maps cleanly to a recipient, so it's reasonable to fire on
148 # every notification by default
149 return [INCLUSION_DEFAULT]
151 @property
152 def default_config(self) -> TransportConfig:
153 config = TransportConfig()
154 config.delivery_defaults.target_required = TargetRequired.ALWAYS
155 config.delivery_defaults.inclusion = self.inclusion_mode
156 config.delivery_defaults.options = {
157 OPTION_SIMPLIFY_TEXT: False,
158 OPTION_STRIP_URLS: False,
159 OPTION_UNIQUE_TARGETS: True,
160 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
161 OPTION_DEVICE_DISCOVERY: False,
162 OPTION_DATA_KEYS_SELECT: None,
163 OPTION_DEVICE_DOMAIN: ["mobile_app"],
164 }
165 return config
167 @property
168 def target_categories(self) -> list[str | TargetEntityCategory]:
169 return [ATTR_MOBILE_APP_ID]
171 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
172 return hass_api.find_config_entry_data("mobile_app") is not None
174 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
175 return {self.name: {}}
177 def validate_action(self, action: str | None) -> bool:
178 return action is None
180 def _extract_push_data(self, raw_data: dict[str, Any]) -> dict[str, Any]:
181 """Extract and remove SuperNotify-specific push_* keys from raw_data.
183 Modifies raw_data in-place via pop().
184 After this call, raw_data contains only passthrough keys for the Companion App.
186 Returns a dict with all extracted push_* values (None if not provided).
187 """
188 return {
189 # iOS
190 "critical_level_ios": raw_data.pop("mobile_push_critical_level", None),
191 "subtitle": raw_data.pop("mobile_push_subtitle", None),
192 # Android critical
193 "critical_ttl": raw_data.pop("mobile_push_critical_ttl", None),
194 "critical_android_priority": raw_data.pop("mobile_push_critical_priority", None),
195 "channel_override": raw_data.pop("mobile_push_channel_override", None),
196 "alarm_stream": raw_data.pop("mobile_push_alarm_stream", False),
197 "alarm_stream_max": raw_data.pop("mobile_push_alarm_stream_max", False),
198 # Android TTS
199 "tts_text": raw_data.pop("mobile_push_tts_text", None),
200 "tts_locale": raw_data.pop("mobile_push_tts_locale", None),
201 "tts_engine": raw_data.pop("mobile_push_tts_engine", None),
202 # Android Notification Commands
203 "command_screen_on": raw_data.pop("mobile_push_command_screen_on", None),
204 "command_dnd": raw_data.pop("mobile_push_command_dnd", None),
205 "command_ringer_mode": raw_data.pop("mobile_push_command_ringer_mode", None),
206 # Cross-platform
207 "group": raw_data.pop("mobile_push_group", None),
208 "notification_tag": raw_data.pop("mobile_push_notification_tag", None),
209 "clear_notification": raw_data.pop("mobile_push_clear_notification", False),
210 }
212 def _android_payload(
213 self,
214 push_data: dict[str, Any],
215 priority: str | None,
216 ) -> dict[str, Any]:
217 """Apply Android-specific fields to the notification data dict.
219 Android fields live flat in data{}, not inside the push{} sub-dict.
220 """
221 android_data: dict[str, Any] = {}
222 # Channel override (Android 8+, determines sound/vibration/LED)
223 if push_data["channel_override"]:
224 android_data["channel"] = push_data["channel_override"]
226 # Alarm stream: routes audio through alarm stream, interrupts DND/silent
227 if push_data["alarm_stream"]:
228 android_data["alarm_stream"] = True
229 if push_data["alarm_stream_max"]:
230 android_data["alarm_stream_max"] = True
232 # FCM TTL: auto-set to 0 for critical (instant delivery, no FCM caching)
233 if push_data["critical_ttl"] is not None:
234 android_data["ttl"] = push_data["critical_ttl"]
235 elif priority == const.PRIORITY_CRITICAL:
236 android_data["ttl"] = ANDROID_CRITICAL_TTL
238 # FCM priority override
239 if push_data["critical_android_priority"] is not None:
240 android_data["priority"] = push_data["critical_android_priority"]
242 # Android TTS: read message aloud on device (Android 8+)
243 if push_data["tts_text"]:
244 android_data["tts_text"] = push_data["tts_text"]
245 if push_data["tts_locale"]:
246 android_data["tts_text_language"] = push_data["tts_locale"]
247 if push_data["tts_engine"]:
248 android_data["tts_engine"] = push_data["tts_engine"]
250 # Notification Commands (Android 8+)
251 if push_data["command_screen_on"]:
252 android_data["command_screen_on"] = True
253 if push_data["command_dnd"]:
254 android_data["command_dnd"] = push_data["command_dnd"]
255 if push_data["command_ringer_mode"]:
256 android_data["command_ringer_mode"] = push_data["command_ringer_mode"]
257 return android_data
259 async def action_title(self, url: str, retry_timeout: int = 900) -> str | None:
260 """Attempt to create a title for mobile action from the TITLE of the web page at the URL"""
261 if url in self.action_titles:
262 return self.action_titles[url]
263 if url in self.action_title_failures and time.time() - self.action_title_failures[url] < retry_timeout:
264 # don't retry too often
265 _LOGGER.debug("SUPERNOTIFY Skipping retry after previous failure to retrieve url title for %s", url)
266 return None
267 try:
268 websession: ClientSession = self.context.hass_api.http_session()
269 resp: ClientResponse = await websession.get(url, timeout=ClientTimeout(total=5.0))
270 body = await resp.content.read()
271 # wrap heavy bs4 parsing in a job to avoid blocking the event loop
272 html = await self.context.hass_api.create_job(BeautifulSoup, body, "html.parser")
273 if html.title and html.title.string:
274 self.action_titles[url] = html.title.string
275 return html.title.string
276 except Exception as e:
277 _LOGGER.warning("SUPERNOTIFY Failed to retrieve url title at %s: %s", url, e)
278 self.action_title_failures[url] = time.time()
279 return None
281 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
282 if not envelope.target.mobile_app_ids:
283 _LOGGER.warning("SUPERNOTIFY No targets provided for mobile_push")
284 return False
286 # 1. Extract SuperNotify push_* keys; raw_data becomes passthrough-only
287 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
288 push_data = self._extract_push_data(raw_data)
290 action_groups = envelope.action_groups
291 _LOGGER.debug("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.target.mobile_app_ids)
293 # 2. Build iOS interruption_level
294 ios_level = push_data["critical_level_ios"] or IOS_INTERRUPTION_MAP.get(
295 envelope.priority or const.PRIORITY_MEDIUM, "active"
296 )
298 # 3. Start with passthrough data, then layer SuperNotify fields
299 data: dict[str, Any] = dict(raw_data)
300 ios_data: dict[str, Any] = {}
302 ios_data.setdefault("push", {})
303 ios_data["push"]["interruption-level"] = ios_level
305 if ios_level == "critical":
306 ios_data["push"].setdefault("sound", {})
307 ios_data["push"]["sound"].setdefault("name", ATTR_DEFAULT)
308 ios_data["push"]["sound"]["critical"] = 1
309 ios_data["push"]["sound"].setdefault("volume", 1.0)
310 # critical notifications cannot be grouped on iOS
311 else:
312 media = envelope.media or {}
313 camera_entity_id_for_group = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
314 group = push_data["group"] or camera_entity_id_for_group
315 # unlike `tag`, an unset `group` leaves notifications ungrouped on the device
316 # (companion app default) rather than forcing them all into a shared bucket
317 if group:
318 data.setdefault("group", group)
320 # 4. iOS extra fields
322 if push_data["subtitle"]:
323 ios_data["subtitle"] = push_data["subtitle"]
325 # 5. Android-specific fields
326 android_data: dict[str, Any] = self._android_payload(push_data, envelope.priority)
328 # 6. Cross-platform: notification tag
329 notification_tag = push_data["notification_tag"]
330 if notification_tag:
331 data["tag"] = notification_tag
332 elif push_data["clear_notification"]:
333 _LOGGER.warning(
334 "SUPERNOTIFY mobile_push: push_clear_notification=True requires push_notification_tag to be set — ignoring"
335 )
337 # 7. Media: camera entity (grab processed image) + fallback URLs
338 media = envelope.media or {}
339 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
340 # Remove self.hass_api.abs_url for clip_url and snapshot_url
341 clip_url: str | None = media.get(ATTR_MEDIA_CLIP_URL)
342 snapshot_url: str | None = media.get(ATTR_MEDIA_SNAPSHOT_URL)
344 if camera_entity_id:
345 image_path = await envelope.grab_image()
346 if image_path:
347 image_url = await self.context.media_storage.share_path(image_path)
348 data[ATTR_IMAGE] = image_url or str(image_path)
349 else:
350 # fall back to letting device take the image, but only from a camera that's up,
351 # since one that's switched off or unavailable would only show a broken image.
352 # camera_entity_id itself already failed the grab above - exclude it here, since
353 # a camera disabled at the device (rather than truly unavailable) won't show that
354 # in its entity state, so re-offering it would just repeat the same failed fetch
355 available_camera_entity_id = select_avail_camera(
356 self.hass_api, self.context.cameras, camera_entity_id, exclude_primary=True
357 )
358 if available_camera_entity_id:
359 data["entity_id"] = available_camera_entity_id
360 else:
361 _LOGGER.info("SUPERNOTIFY mobile_push: no available camera for %s, sending without image", camera_entity_id)
362 if clip_url:
363 data[ATTR_VIDEO] = clip_url
365 if snapshot_url and ATTR_IMAGE not in data:
366 # Fallback: use pre-computed snapshot URL if grab_image() produced nothing
367 data[ATTR_IMAGE] = snapshot_url
369 # 8. Actions: URL-title fetching, snooze action, action groups (unchanged)
370 if "actions" in data and not isinstance(data["actions"], list):
371 _LOGGER.warning(
372 "SUPERNOTIFY mobile_push: data.actions must be a list of action objects, ignoring invalid value %s",
373 data["actions"],
374 )
375 data["actions"] = []
376 else:
377 data.setdefault("actions", [])
378 for action in envelope.actions:
379 app_url: str | None = self.hass_api.abs_url(action.get(ATTR_ACTION_URL))
380 if app_url:
381 app_url_title = action.get(ATTR_ACTION_URL_TITLE) or await self.action_title(app_url) or "Click for Action"
382 action[ATTR_ACTION_URL_TITLE] = app_url_title
383 data["actions"].append(action)
384 if camera_entity_id:
385 data["actions"].append({
386 "action": f"SUPERNOTIFY_SNOOZE_EVERYONE_CAMERA_{camera_entity_id}",
387 "title": f"Snooze camera notifications for {camera_entity_id}",
388 "behavior": "textInput",
389 "textInputButtonTitle": "Minutes to snooze",
390 "textInputPlaceholder": "60",
391 })
392 for group, actions in self.context.mobile_actions.items():
393 if action_groups is None or group in action_groups:
394 data["actions"].extend(actions)
395 if not data["actions"]:
396 del data["actions"]
398 # 9. Dispatch to each mobile target
399 clear_notification = bool(push_data["clear_notification"] and notification_tag)
400 model_filter = SelectionRule(envelope.delivery.options.get(OPTION_DEVICE_MODEL_SELECT))
401 hits = 0
403 for mobile_target in envelope.target.mobile_app_ids:
404 full_target = mobile_target if Target.is_notify_entity(mobile_target) else f"notify.{mobile_target}"
405 mobile_info: TrackedDeviceDetails | None = self.context.hass_api.mobile_app_by_id(mobile_target)
406 if mobile_info is not None and not model_filter.match(mobile_info.model):
407 _LOGGER.debug("SUPERNOTIFY Skipping %s, model %s excluded by delivery filter", mobile_target, mobile_info.model)
408 continue
410 # fresh copy per target - customize_data below may prune `data` down to nothing
411 # (e.g. an Android target with no android/ios fields to merge in), and that must
412 # not carry over and clobber the next target's action_data
413 target_data = dict(data)
414 if mobile_info is None:
415 target_data.update(android_data)
416 target_data.update(ios_data)
417 elif mobile_info.manufacturer != MANUFACTURER_APPLE:
418 target_data.update(android_data)
419 else:
420 target_data.update(ios_data)
422 action_data = envelope.core_action_data()
423 action_data[ATTR_DATA] = target_data
424 action_data = envelope.customize_data(action_data)
426 if clear_notification:
427 # Override message to "clear_notification" to dismiss same-tag notification on device
428 clear_action_data = dict(action_data)
429 clear_action_data["message"] = "clear_notification"
430 success = await self.call_action(
431 envelope, qualified_action=full_target, action_data=clear_action_data, implied_target=True
432 )
433 else:
434 success = await self.call_action(
435 envelope, qualified_action=full_target, action_data=action_data, implied_target=True
436 )
438 if success:
439 hits += 1
440 else:
441 simple_target = (
442 mobile_target if not Target.is_notify_entity(mobile_target) else mobile_target.replace("notify.", "")
443 )
444 _LOGGER.warning("SUPERNOTIFY Failed to send to %s, snoozing for a day", simple_target)
445 if self.people_registry:
446 # tie the mobile device back to a recipient for the snoozing API
447 for recipient in self.people_registry.enabled_recipients():
448 for md in recipient.mobile_devices:
449 if md in (simple_target, mobile_target):
450 self.context.snoozer.register_snooze(
451 CommandType.SNOOZE,
452 target_type=QualifiedTargetType.MOBILE,
453 target=simple_target,
454 recipient_type=RecipientType.USER,
455 recipient=recipient.entity_id,
456 snooze_for=timedelta(days=1),
457 reason="Action Failure",
458 )
459 return hits > 0