Coverage for custom_components / supernotify / transports / mobile_push.py: 87%
175 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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_notification_tag str Notification tag for replacement (iOS) / grouping (Android)
22 mobile_push_clear_notification bool Send clear_notification to dismiss previous same-tag notification.
23 Requires push_notification_tag to be set.
24 mobile_push_tts_text str Android TTS text read aloud on device (Android 8+).
25 If omitted, push TTS is not activated.
26 mobile_push_tts_locale str BCP-47 language for TTS (e.g. "it-IT", "en-US").
27 Only used when push_tts_text is set.
28 mobile_push_tts_engine str TTS engine package (e.g. "com.google.android.tts").
29 Only used when push_tts_text is set.
30 mobile_push_command_screen_on bool Android: turn on device screen on delivery (Android 8+)
31 mobile_push_command_dnd str Android: change Do Not Disturb ("toggle","off","on")
32 mobile_push_command_ringer_mode str Android: change ringer mode ("silent","vibrate","normal")
33 mobile_push_channel_override str Android notification channel override (e.g. "alarm","general")
34 mobile_push_alarm_stream bool Android: route audio through alarm stream (interrupts DND/silent)
35 mobile_push_alarm_stream_max bool Android: alarm stream at maximum device volume
37"""
39from __future__ import annotations
41import logging
42import time
43from datetime import timedelta
44from typing import TYPE_CHECKING, Any
46from aiohttp import ClientResponse, ClientSession, ClientTimeout
47from bs4 import BeautifulSoup
48from homeassistant.components.notify.const import ATTR_DATA
50import custom_components.supernotify.const as const
51from custom_components.supernotify.const import (
52 ATTR_ACTION_CATEGORY,
53 ATTR_ACTION_URL,
54 ATTR_ACTION_URL_TITLE,
55 ATTR_DEFAULT,
56 ATTR_IMAGE,
57 ATTR_MEDIA_CAMERA_ENTITY_ID,
58 ATTR_MEDIA_CLIP_URL,
59 ATTR_MEDIA_SNAPSHOT_URL,
60 ATTR_MOBILE_APP_ID,
61 ATTR_VIDEO,
62 MANUFACTURER_APPLE,
63 OPTION_DATA_KEYS_SELECT,
64 OPTION_DEVICE_DISCOVERY,
65 OPTION_DEVICE_DOMAIN,
66 OPTION_DEVICE_MODEL_SELECT,
67 OPTION_MESSAGE_USAGE,
68 OPTION_SIMPLIFY_TEXT,
69 OPTION_STRIP_URLS,
70 OPTION_TARGET_CATEGORIES,
71 TRANSPORT_MOBILE_PUSH,
72)
73from custom_components.supernotify.model import (
74 CommandType,
75 DebugTrace,
76 DeliveryConfig,
77 MessageOnlyPolicy,
78 QualifiedTargetType,
79 RecipientType,
80 SelectionRule,
81 Target,
82 TargetRequired,
83 TransportConfig,
84 TransportFeature,
85)
86from custom_components.supernotify.transport import Transport
88if TYPE_CHECKING:
89 from custom_components.supernotify.envelope import Envelope
90 from custom_components.supernotify.hass_api import DeviceInfo, HomeAssistantAPI
92_LOGGER = logging.getLogger(__name__)
94# iOS interruption_level mapping from SuperNotify priority
95IOS_INTERRUPTION_MAP: dict[str, str] = {
96 const.PRIORITY_CRITICAL: "critical",
97 const.PRIORITY_HIGH: "time-sensitive",
98 const.PRIORITY_MEDIUM: "active",
99 const.PRIORITY_LOW: "passive",
100 const.PRIORITY_MINIMUM: "passive",
101}
103# Android FCM TTL auto-set for critical priority (0 = instant, no FCM caching)
104ANDROID_CRITICAL_TTL = 0
107class MobilePushTransport(Transport):
108 name = TRANSPORT_MOBILE_PUSH
110 def __init__(self, *args: Any, **kwargs: Any) -> None:
111 super().__init__(*args, **kwargs)
112 self.action_titles: dict[str, str] = {}
113 self.action_title_failures: dict[str, float] = {}
115 @property
116 def supported_features(self) -> TransportFeature:
117 return (
118 TransportFeature.MESSAGE
119 | TransportFeature.TITLE
120 | TransportFeature.ACTIONS
121 | TransportFeature.IMAGES
122 | TransportFeature.VIDEO
123 | TransportFeature.SNAPSHOT_IMAGE
124 )
126 def extra_attributes(self) -> dict[str, Any]:
127 return {"action_titles": self.action_titles, "action_title_failures": self.action_title_failures}
129 @property
130 def default_config(self) -> TransportConfig:
131 config = TransportConfig()
132 config.delivery_defaults.target_required = TargetRequired.ALWAYS
133 config.delivery_defaults.options = {
134 OPTION_SIMPLIFY_TEXT: False,
135 OPTION_STRIP_URLS: False,
136 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
137 OPTION_TARGET_CATEGORIES: [ATTR_MOBILE_APP_ID],
138 OPTION_DEVICE_DISCOVERY: False,
139 OPTION_DATA_KEYS_SELECT: None,
140 OPTION_DEVICE_DOMAIN: ["mobile_app"],
141 }
142 return config
144 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None: # noqa: ARG002
145 return self.delivery_defaults
147 def validate_action(self, action: str | None) -> bool:
148 return action is None
150 def _extract_push_data(self, raw_data: dict[str, Any]) -> dict[str, Any]:
151 """Extract and remove SuperNotify-specific push_* keys from raw_data.
153 Modifies raw_data in-place via pop().
154 After this call, raw_data contains only passthrough keys for the Companion App.
156 Returns a dict with all extracted push_* values (None if not provided).
157 """
158 return {
159 # iOS
160 "critical_level_ios": raw_data.pop("mobile_push_critical_level", None),
161 "subtitle": raw_data.pop("mobile_push_subtitle", None),
162 # Android critical
163 "critical_ttl": raw_data.pop("mobile_push_critical_ttl", None),
164 "critical_android_priority": raw_data.pop("mobile_push_critical_priority", None),
165 "channel_override": raw_data.pop("mobile_push_channel_override", None),
166 "alarm_stream": raw_data.pop("mobile_push_alarm_stream", False),
167 "alarm_stream_max": raw_data.pop("mobile_push_alarm_stream_max", False),
168 # Android TTS
169 "tts_text": raw_data.pop("mobile_push_tts_text", None),
170 "tts_locale": raw_data.pop("mobile_push_tts_locale", None),
171 "tts_engine": raw_data.pop("mobile_push_tts_engine", None),
172 # Android Notification Commands
173 "command_screen_on": raw_data.pop("mobile_push_command_screen_on", None),
174 "command_dnd": raw_data.pop("mobile_push_command_dnd", None),
175 "command_ringer_mode": raw_data.pop("mobile_push_command_ringer_mode", None),
176 # Cross-platform
177 "notification_tag": raw_data.pop("mobile_push_notification_tag", None),
178 "clear_notification": raw_data.pop("mobile_push_clear_notification", False),
179 }
181 def _android_payload(
182 self,
183 push_data: dict[str, Any],
184 priority: str | None,
185 ) -> dict[str, Any]:
186 """Apply Android-specific fields to the notification data dict.
188 Android fields live flat in data{}, not inside the push{} sub-dict.
189 """
190 android_data: dict[str, Any] = {}
191 # Channel override (Android 8+, determines sound/vibration/LED)
192 if push_data["channel_override"]:
193 android_data["channel"] = push_data["channel_override"]
195 # Alarm stream: routes audio through alarm stream, interrupts DND/silent
196 if push_data["alarm_stream"]:
197 android_data["alarm_stream"] = True
198 if push_data["alarm_stream_max"]:
199 android_data["alarm_stream_max"] = True
201 # FCM TTL: auto-set to 0 for critical (instant delivery, no FCM caching)
202 if push_data["critical_ttl"] is not None:
203 android_data["ttl"] = push_data["critical_ttl"]
204 elif priority == const.PRIORITY_CRITICAL:
205 android_data["ttl"] = ANDROID_CRITICAL_TTL
207 # FCM priority override
208 if push_data["critical_android_priority"] is not None:
209 android_data["priority"] = push_data["critical_android_priority"]
211 # Android TTS: read message aloud on device (Android 8+)
212 if push_data["tts_text"]:
213 android_data["tts_text"] = push_data["tts_text"]
214 if push_data["tts_locale"]:
215 android_data["tts_text_language"] = push_data["tts_locale"]
216 if push_data["tts_engine"]:
217 android_data["tts_engine"] = push_data["tts_engine"]
219 # Notification Commands (Android 8+)
220 if push_data["command_screen_on"]:
221 android_data["command_screen_on"] = True
222 if push_data["command_dnd"]:
223 android_data["command_dnd"] = push_data["command_dnd"]
224 if push_data["command_ringer_mode"]:
225 android_data["command_ringer_mode"] = push_data["command_ringer_mode"]
226 return android_data
228 async def action_title(self, url: str, retry_timeout: int = 900) -> str | None:
229 """Attempt to create a title for mobile action from the TITLE of the web page at the URL"""
230 if url in self.action_titles:
231 return self.action_titles[url]
232 if url in self.action_title_failures:
233 # don't retry too often
234 if time.time() - self.action_title_failures[url] < retry_timeout:
235 _LOGGER.debug("SUPERNOTIFY skipping retry after previous failure to retrieve url title for ", url)
236 return None
237 try:
238 websession: ClientSession = self.context.hass_api.http_session()
239 resp: ClientResponse = await websession.get(url, timeout=ClientTimeout(total=5.0))
240 body = await resp.content.read()
241 # wrap heavy bs4 parsing in a job to avoid blocking the event loop
242 html = await self.context.hass_api.create_job(BeautifulSoup, body, "html.parser")
243 if html.title and html.title.string:
244 self.action_titles[url] = html.title.string
245 return html.title.string
246 except Exception as e:
247 _LOGGER.warning("SUPERNOTIFY failed to retrieve url title at %s: %s", url, e)
248 self.action_title_failures[url] = time.time()
249 return None
251 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002
252 if not envelope.target.mobile_app_ids:
253 _LOGGER.warning("SUPERNOTIFY No targets provided for mobile_push")
254 return False
256 # 1. Extract SuperNotify push_* keys; raw_data becomes passthrough-only
257 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
258 push_data = self._extract_push_data(raw_data)
260 action_groups = envelope.action_groups
261 _LOGGER.debug("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.target.mobile_app_ids)
263 # 2. Build iOS interruption_level
264 ios_level = push_data["critical_level_ios"] or IOS_INTERRUPTION_MAP.get(
265 envelope.priority or const.PRIORITY_MEDIUM, "active"
266 )
268 # 3. Start with passthrough data, then layer SuperNotify fields
269 data: dict[str, Any] = dict(raw_data)
270 ios_data: dict[str, Any] = {}
272 category = data.get(ATTR_ACTION_CATEGORY, "general")
274 ios_data.setdefault("push", {})
275 ios_data["push"]["interruption-level"] = ios_level
277 if ios_level == "critical":
278 ios_data["push"].setdefault("sound", {})
279 ios_data["push"]["sound"].setdefault("name", ATTR_DEFAULT)
280 ios_data["push"]["sound"]["critical"] = 1
281 ios_data["push"]["sound"].setdefault("volume", 1.0)
282 # critical notifications cannot be grouped on iOS
283 else:
284 media = envelope.media or {}
285 camera_entity_id_for_group = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
286 data.setdefault("group", category or camera_entity_id_for_group or "appd")
288 # 4. iOS extra fields
290 if push_data["subtitle"]:
291 ios_data["subtitle"] = push_data["subtitle"]
293 # 5. Android-specific fields
294 android_data: dict[str, Any] = self._android_payload(push_data, envelope.priority)
296 # 6. Cross-platform: notification tag
297 notification_tag = push_data["notification_tag"]
298 if notification_tag:
299 data["tag"] = notification_tag
300 elif push_data["clear_notification"]:
301 _LOGGER.warning(
302 "SUPERNOTIFY mobile_push: push_clear_notification=True requires push_notification_tag to be set — ignoring"
303 )
305 # 7. Media: camera entity (grab processed image) + fallback URLs
306 media = envelope.media or {}
307 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
308 # Remove self.hass_api.abs_url for clip_url and snapshot_url
309 clip_url: str | None = media.get(ATTR_MEDIA_CLIP_URL)
310 snapshot_url: str | None = media.get(ATTR_MEDIA_SNAPSHOT_URL)
312 if camera_entity_id:
313 image_path = await envelope.grab_image()
314 if image_path:
315 image_url = await self.context.media_storage.share_path(image_path)
316 data[ATTR_IMAGE] = image_url or str(image_path)
317 else:
318 # fall back to letting device take the image
319 data["entity_id"] = camera_entity_id
320 if clip_url:
321 data[ATTR_VIDEO] = clip_url
323 if snapshot_url and ATTR_IMAGE not in data:
324 # Fallback: use pre-computed snapshot URL if grab_image() produced nothing
325 data[ATTR_IMAGE] = snapshot_url
327 # 8. Actions: URL-title fetching, snooze action, action groups (unchanged)
328 data.setdefault("actions", [])
329 for action in envelope.actions:
330 app_url: str | None = self.hass_api.abs_url(action.get(ATTR_ACTION_URL))
331 if app_url:
332 app_url_title = action.get(ATTR_ACTION_URL_TITLE) or await self.action_title(app_url) or "Click for Action"
333 action[ATTR_ACTION_URL_TITLE] = app_url_title
334 data["actions"].append(action)
335 if camera_entity_id:
336 data["actions"].append({
337 "action": f"SUPERNOTIFY_SNOOZE_EVERYONE_CAMERA_{camera_entity_id}",
338 "title": f"Snooze camera notifications for {camera_entity_id}",
339 "behavior": "textInput",
340 "textInputButtonTitle": "Minutes to snooze",
341 "textInputPlaceholder": "60",
342 })
343 for group, actions in self.context.mobile_actions.items():
344 if action_groups is None or group in action_groups:
345 data["actions"].extend(actions)
346 if not data["actions"]:
347 del data["actions"]
349 # 9. Dispatch to each mobile target
350 action_data = envelope.core_action_data()
351 action_data[ATTR_DATA] = data
352 clear_notification = bool(push_data["clear_notification"] and notification_tag)
353 model_filter = SelectionRule(envelope.delivery.options.get(OPTION_DEVICE_MODEL_SELECT))
354 hits = 0
356 for mobile_target in envelope.target.mobile_app_ids:
357 full_target = mobile_target if Target.is_notify_entity(mobile_target) else f"notify.{mobile_target}"
358 mobile_info: DeviceInfo | None = self.context.hass_api.mobile_app_by_id(mobile_target)
359 if mobile_info is not None and not model_filter.match(mobile_info.model):
360 _LOGGER.debug("SUPERNOTIFY Skipping %s, model %s excluded by delivery filter", mobile_target, mobile_info.model)
361 continue
362 if mobile_info is None:
363 action_data[ATTR_DATA].update(android_data)
364 action_data[ATTR_DATA].update(ios_data)
365 elif mobile_info.manufacturer != MANUFACTURER_APPLE:
366 action_data[ATTR_DATA].update(android_data)
367 else:
368 action_data[ATTR_DATA].update(ios_data)
370 action_data = envelope.customize_data(action_data)
372 if clear_notification:
373 # Override message to "clear_notification" to dismiss same-tag notification on device
374 clear_action_data = dict(action_data)
375 clear_action_data["message"] = "clear_notification"
376 success = await self.call_action(
377 envelope, qualified_action=full_target, action_data=clear_action_data, implied_target=True
378 )
379 else:
380 success = await self.call_action(
381 envelope, qualified_action=full_target, action_data=action_data, implied_target=True
382 )
384 if success:
385 hits += 1
386 else:
387 simple_target = (
388 mobile_target if not Target.is_notify_entity(mobile_target) else mobile_target.replace("notify.", "")
389 )
390 _LOGGER.warning("SUPERNOTIFY Failed to send to %s, snoozing for a day", simple_target)
391 if self.people_registry:
392 # tie the mobile device back to a recipient for the snoozing API
393 for recipient in self.people_registry.enabled_recipients():
394 for md in recipient.mobile_devices:
395 if md in (simple_target, mobile_target):
396 self.context.snoozer.register_snooze(
397 CommandType.SNOOZE,
398 target_type=QualifiedTargetType.MOBILE,
399 target=simple_target,
400 recipient_type=RecipientType.USER,
401 recipient=recipient.entity_id,
402 snooze_for=timedelta(days=1),
403 reason="Action Failure",
404 )
405 return hits > 0