mobile_push: Android commands and TTS sent as calls of their own

custom_components/supernotify/transports/mobile_push.py

@@ -27,15 +27,22 @@
     mobile_push_notification_tag    str   Notification tag for replacement (iOS) / grouping (Android)
     mobile_push_clear_notification  bool  Send clear_notification to dismiss previous same-tag notification.
                                       Requires push_notification_tag to be set.
-    mobile_push_tts_text            str   Android TTS text read aloud on device (Android 8+).
-                                      If omitted, push TTS is not activated.
+    mobile_push_tts_text            str   Android: text read aloud by the phone (Android 8+), sent as its own
+                                      `message: TTS` call after the notification, on the alarm stream at
+                                      full volume for critical. If omitted, push TTS is not activated.
+    mobile_push_tts_delay           int   Seconds between the notification and its TTS, so the notification's
+                                      own sound isn't cut off (default 5, 0 for straight away).
     mobile_push_tts_locale          str   BCP-47 language for TTS (e.g. "it-IT", "en-US").
                                       Only used when push_tts_text is set.
     mobile_push_tts_engine          str   TTS engine package (e.g. "com.google.android.tts").
                                       Only used when push_tts_text is set.
-    mobile_push_command_screen_on   bool  Android: turn on device screen on delivery (Android 8+)
-    mobile_push_command_dnd         str   Android: change Do Not Disturb ("toggle","off","on")
-    mobile_push_command_ringer_mode str   Android: change ringer mode ("silent","vibrate","normal")
+    mobile_push_command_screen_on   bool  Android: turn on device screen on delivery (Android 8+),
+                                      or "keep_screen_on"
+    mobile_push_command_dnd         str   Android: change Do Not Disturb ("alarms_only","priority_only",
+                                      "total_silence","off")
+    mobile_push_command_ringer_mode str   Android: change ringer mode ("normal","silent","vibrate")
+                                      The command_* keys are each sent as their own `message: command_...`
+                                      call before the notification, as the companion app takes them.
     mobile_push_channel_override    str   Android notification channel override (e.g. "alarm","general")
     mobile_push_alarm_stream        bool  Android: use the alarm_stream channel, which sounds through
                                       Do Not Disturb and silent mode
@@ -46,6 +53,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import time
 from collections.abc import Collection
@@ -117,6 +125,9 @@
 # documents for critical notifications, alarm_stream being what gets through Do Not Disturb
 ANDROID_CRITICAL_PRIORITY = "high"
 ANDROID_ALARM_STREAM_CHANNEL = "alarm_stream"
+# seconds between a notification and its TTS on Android - started together, the TTS cuts off the
+# notification's own sound
+ANDROID_TTS_DELAY = 5
 
 
 class MobilePushTransport(Transport):
@@ -197,6 +208,7 @@ def _extract_push_data(self, raw_data: dict[str, Any]) -> dict[str, Any]:
             "tts_text": raw_data.pop("mobile_push_tts_text", None),
             "tts_locale": raw_data.pop("mobile_push_tts_locale", None),
             "tts_engine": raw_data.pop("mobile_push_tts_engine", None),
+            "tts_delay": raw_data.pop("mobile_push_tts_delay", ANDROID_TTS_DELAY),
             # Android Notification Commands
             "command_screen_on": raw_data.pop("mobile_push_command_screen_on", None),
             "command_dnd": raw_data.pop("mobile_push_command_dnd", None),
@@ -243,22 +255,39 @@ def _android_payload(
         elif critical:
             android_data["priority"] = ANDROID_CRITICAL_PRIORITY
 
-        # Android TTS: read message aloud on device (Android 8+)
+        return android_data
+
+    def _android_extra_calls(
+        self, push_data: dict[str, Any], priority: str | None
+    ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+        """Notification commands and TTS, which the companion app takes as the `message` of a call of
+        their own rather than as notification data - commands to go before the notification, so the
+        screen is on or Do Not Disturb off when it arrives, and TTS after it."""
+        # a call without high priority can wait until the phone is unlocked - a critical's TTS would then be
+        # spoken late, maybe twice - so for critical these go as fast as the notification itself
+        urgent: dict[str, Any] = (
+            {"ttl": ANDROID_CRITICAL_TTL, "priority": ANDROID_CRITICAL_PRIORITY} if priority == const.PRIORITY_CRITICAL else {}
+        )
+        before: list[dict[str, Any]] = []
+        for key in ("command_dnd", "command_ringer_mode", "command_screen_on"):
+            value = push_data[key]
+            if value:
+                before.append({"message": key, ATTR_DATA: ({} if value is True else {"command": value}) | urgent})
+        after: list[dict[str, Any]] = []
         if push_data["tts_text"]:
-            android_data["tts_text"] = push_data["tts_text"]
+            tts_data: dict[str, Any] = {"tts_text": push_data["tts_text"], **urgent}
+            if push_data["alarm_stream_max"] or priority == const.PRIORITY_CRITICAL:
+                # at full volume for critical - on a watch the alarm volume can be too low to hear,
+                # and the companion app puts the volume back afterwards
+                tts_data["media_stream"] = "alarm_stream_max"
+            elif push_data["alarm_stream"]:
+                tts_data["media_stream"] = ANDROID_ALARM_STREAM_CHANNEL
             if push_data["tts_locale"]:
-                android_data["tts_text_language"] = push_data["tts_locale"]
+                tts_data["tts_text_language"] = push_data["tts_locale"]
             if push_data["tts_engine"]:
-                android_data["tts_engine"] = push_data["tts_engine"]
-
-        # Notification Commands (Android 8+)
-        if push_data["command_screen_on"]:
-            android_data["command_screen_on"] = True
-        if push_data["command_dnd"]:
-            android_data["command_dnd"] = push_data["command_dnd"]
-        if push_data["command_ringer_mode"]:
-            android_data["command_ringer_mode"] = push_data["command_ringer_mode"]
-        return android_data
+                tts_data["tts_engine"] = push_data["tts_engine"]
+            after.append({"message": "TTS", ATTR_DATA: tts_data})
+        return before, after
 
     async def action_title(self, url: str, retry_timeout: int = 900) -> str | None:
         """Attempt to create a title for mobile action from the TITLE of the web page at the URL"""
@@ -328,6 +357,7 @@ async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = Non
 
         # 5. Android-specific fields
         android_data: dict[str, Any] = self._android_payload(push_data, envelope.priority, given=raw_data.keys())
+        android_before, android_after = self._android_extra_calls(push_data, envelope.priority)
 
         # 6. Cross-platform: notification tag
         notification_tag = push_data["notification_tag"]
@@ -403,6 +433,7 @@ async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = Non
         clear_notification = bool(push_data["clear_notification"] and notification_tag)
         model_filter = SelectionRule(envelope.delivery.options.get(OPTION_DEVICE_MODEL_SELECT))
         hits = 0
+        tts_pending: list[tuple[str, dict[str, Any]]] = []
 
         for mobile_target in envelope.target.mobile_app_ids:
             full_target = mobile_target if Target.is_notify_entity(mobile_target) else f"notify.{mobile_target}"
@@ -415,13 +446,24 @@ async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = Non
             # (e.g. an Android target with no android/ios fields to merge in), and that must
             # not carry over and clobber the next target's action_data
             target_data = dict(data)
+            is_android: bool = mobile_info is not None and mobile_info.manufacturer != MANUFACTURER_APPLE
             if mobile_info is None:
                 target_data.update(android_data)
                 target_data.update(ios_data)
-            elif mobile_info.manufacturer != MANUFACTURER_APPLE:  # TODO Make this os_name based
+                if android_before or android_after:
+                    # an iPhone would show a command or TTS call as a notification saying "command_dnd"
+                    _LOGGER.debug(
+                        "SUPERNOTIFY mobile_push: not sending Android commands/TTS to unknown device %s", mobile_target
+                    )
+            elif is_android:  # TODO Make this os_name based
                 target_data.update(android_data)
             else:
                 target_data.update(ios_data)
+            if is_android:
+                for extra in android_before:
+                    await self.call_action(
+                        envelope, qualified_action=full_target, action_data=dict(extra), implied_target=True, single_target=True
+                    )
 
             action_data = envelope.core_action_data()
             action_data[ATTR_DATA] = target_data
@@ -447,6 +489,9 @@ async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = Non
                     single_target=True,
                 )
 
+            if success and is_android:
+                tts_pending.extend((full_target, dict(extra)) for extra in android_after)
+
             if success:
                 hits += 1
             else:
@@ -478,4 +523,12 @@ async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = Non
                                     snooze_for=timedelta(days=1),
                                     reason="Action Failure",
                                 )
+        if tts_pending:
+            # one wait for all the devices, after every notification has gone
+            if push_data["tts_delay"]:
+                await asyncio.sleep(push_data["tts_delay"])
+            for full_target, extra in tts_pending:
+                await self.call_action(
+                    envelope, qualified_action=full_target, action_data=extra, implied_target=True, single_target=True
+                )
         return hits > 0

docs/transports/mobile_push.md

@@ -80,6 +80,30 @@ changes the `high` priority):
     Some phones (Samsung One UI for one) also need the Home Assistant app allowed as an exception to
     Do Not Disturb, in the phone's own settings.
 
+## Android Commands and Spoken Text
+
+The companion app takes [notification commands](https://companion.home-assistant.io/docs/notifications/notification-commands)
+and TTS as the `message` of a call of their own, so Supernotify sends them that way, to Android phones only:
+`mobile_push_command_dnd` (`alarms_only`, `priority_only`, `total_silence`, `off`),
+`mobile_push_command_ringer_mode` (`normal`, `silent`, `vibrate`) and `mobile_push_command_screen_on` each
+go just before the notification, and `mobile_push_tts_text` 5 seconds after it, so the notification's own
+sound isn't cut off (`mobile_push_tts_delay` changes that, `0` for straight away) - for a `critical` notification
+on the alarm stream at full volume (`alarm_stream_max`, the app puts the volume back afterwards), since a
+watch's alarm volume can be too low to hear:
+
+```yaml
+  - action: supernotify.notify
+    data:
+      message: Water leak under the sink
+      priority: critical
+      extra_data:
+        mobile_push_command_screen_on: true
+        mobile_push_tts_text: Water leak under the sink
+```
+
+!!! note
+    `command_dnd` needs a permission the companion app asks for the first time it's used.
+
 ## Notification Grouping
 
 Set `mobile_push_group` in `extra_data:` to visually stack notifications together on the device

tests/components/supernotify/transports/test_transport_mobile_push.py

@@ -574,12 +574,9 @@ async def test_deliver_android_and_ios_extra_data_keys() -> None:
     assert "alarm_stream_max" not in data
     assert data["ttl"] == 30
     assert data["priority"] == "high"
-    assert data["tts_text"] == "speak this"
-    assert data["tts_text_language"] == "en-US"
-    assert data["tts_engine"] == "com.google.android.tts"
-    assert data["command_screen_on"] is True
-    assert data["command_dnd"] == "off"
-    assert data["command_ringer_mode"] == "silent"
+    # commands and TTS are calls of their own, and an unknown device might be an iPhone, so none here
+    assert not any(k.startswith(("tts_", "command_")) for k in data)
+    assert len(e.calls) == 1
     assert data["subtitle"] == "a subtitle"
     assert data["tag"] == "my-tag"
 
@@ -783,6 +780,117 @@ async def test_android_critical_sounds_through_do_not_disturb(
     assert not any(k.startswith("mobile_push_") for k in sent)
 
 
+async def _deliver_to(context: Context, manufacturer: str, priority: str, data: dict[str, Any]) -> Envelope:
+    uut = MobilePushTransport(context)
+    context.hass_api.mobile_app_by_id.return_value = TrackedDeviceDetails("id001", manufacturer=manufacturer, model="Phone")  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+    context.configure_for_tests([uut])
+    await context.initialize()
+    e = Envelope(
+        Delivery("default", {CONF_TRANSPORT: TRANSPORT_MOBILE_PUSH}, uut),
+        Notification(context, message="smoke", title="alarm", action_data={ATTR_PRIORITY: priority}),
+        target=Target({"mobile_app_id": ["mobile_app_phone"]}),
+        data=data,
+    )
+    await uut.deliver(e)
+    return e
+
+
+async def test_android_commands_go_before_the_notification_as_their_own_calls(unmocked_config: Context) -> None:
+    e = await _deliver_to(
+        unmocked_config,
+        "Samsung",
+        PRIORITY_MEDIUM,
+        {
+            "mobile_push_command_dnd": "off",
+            "mobile_push_command_ringer_mode": "normal",
+            "mobile_push_command_screen_on": True,
+        },
+    )
+
+    sent = [c.action_data for c in e.calls]
+    assert sent[:3] == [
+        {"message": "command_dnd", "data": {"command": "off"}},
+        {"message": "command_ringer_mode", "data": {"command": "normal"}},
+        {"message": "command_screen_on", "data": {}},
+    ]
+    assert sent[3]["message"] == "smoke"  # type: ignore[index]
+    assert len(sent) == 4
+    assert all(c.action == "mobile_app_phone" for c in e.calls)
+
+
+@pytest.mark.parametrize(
+    ("priority", "data", "media_stream"),
+    [
+        (PRIORITY_MEDIUM, {}, None),
+        (PRIORITY_CRITICAL, {}, "alarm_stream_max"),
+        (PRIORITY_MEDIUM, {"mobile_push_alarm_stream": True}, "alarm_stream"),
+        (PRIORITY_MEDIUM, {"mobile_push_alarm_stream_max": True}, "alarm_stream_max"),
+    ],
+)
+async def test_android_tts_is_its_own_call_after_the_notification(
+    unmocked_config: Context, priority: str, data: dict[str, Any], media_stream: str | None
+) -> None:
+    with patch("custom_components.supernotify.transports.mobile_push.asyncio.sleep") as sleep:
+        e = await _deliver_to(unmocked_config, "Samsung", priority, {"mobile_push_tts_text": "smoke in the kitchen", **data})
+    sleep.assert_awaited_once_with(5)  # so the notification's own sound is heard before the TTS
+
+    assert len(e.calls) == 2
+    assert e.calls[0].action_data["message"] == "smoke"  # type: ignore[index]
+    assert "tts_text" not in e.calls[0].action_data.get("data", {})  # type: ignore[union-attr]
+    tts = e.calls[1].action_data
+    assert tts is not None
+    assert tts["message"] == "TTS"
+    assert tts["data"]["tts_text"] == "smoke in the kitchen"
+    assert tts["data"].get("media_stream") == media_stream
+    # a critical's TTS goes as fast as the notification, or it waits for the phone to be unlocked
+    urgent = {"ttl": 0, "priority": "high"} if priority == PRIORITY_CRITICAL else {}
+    assert {k: tts["data"][k] for k in ("ttl", "priority") if k in tts["data"]} == urgent
+
+
+async def test_iphone_gets_no_android_commands_or_tts(unmocked_config: Context) -> None:
+    e = await _deliver_to(
+        unmocked_config,
+        "Apple",
+        PRIORITY_CRITICAL,
+        {"mobile_push_command_dnd": "off", "mobile_push_tts_text": "smoke in the kitchen"},
+    )
+
+    assert len(e.calls) == 1
+    assert e.calls[0].action_data["message"] == "smoke"  # type: ignore[index]
+
+
+async def test_android_commands_for_critical_go_with_high_priority(unmocked_config: Context) -> None:
+    e = await _deliver_to(unmocked_config, "Samsung", PRIORITY_CRITICAL, {"mobile_push_command_screen_on": True})
+
+    assert e.calls[0].action_data == {"message": "command_screen_on", "data": {"ttl": 0, "priority": "high"}}
+
+
+async def test_android_tts_waits_once_for_all_devices_and_can_go_straight_away(unmocked_config: Context) -> None:
+    context = unmocked_config
+    uut = MobilePushTransport(context)
+    context.hass_api.mobile_app_by_id.return_value = TrackedDeviceDetails("id001", manufacturer="Samsung", model="Phone")  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+    context.configure_for_tests([uut])
+    await context.initialize()
+
+    def envelope(data: dict[str, Any]) -> Envelope:
+        return Envelope(
+            Delivery("default", {CONF_TRANSPORT: TRANSPORT_MOBILE_PUSH}, uut),
+            Notification(context, message="smoke", title="alarm", action_data={ATTR_PRIORITY: PRIORITY_CRITICAL}),
+            target=Target({"mobile_app_id": ["mobile_app_phone", "mobile_app_watch"]}),
+            data=data,
+        )
+
+    with patch("custom_components.supernotify.transports.mobile_push.asyncio.sleep") as sleep:
+        e = envelope({"mobile_push_tts_text": "smoke"})
+        await uut.deliver(e)
+    sleep.assert_awaited_once_with(5)
+    assert [c.action_data["message"] for c in e.calls] == ["smoke", "smoke", "TTS", "TTS"]  # type: ignore[index]
+
+    with patch("custom_components.supernotify.transports.mobile_push.asyncio.sleep") as sleep:
+        await uut.deliver(envelope({"mobile_push_tts_text": "smoke", "mobile_push_tts_delay": 0}))
+    sleep.assert_not_awaited()
+
+
 async def test_missing_mobile_action_is_not_an_error(hass: HomeAssistant) -> None:
     from pytest_homeassistant_custom_component.common import async_mock_service