Add notify platform to TTS (#40028)

homeassistant/components/tts/notify.py

@@ -0,0 +1,60 @@
+"""Support notifications through TTS service."""
+import logging
+
+import voluptuous as vol
+
+from homeassistant.components.notify import PLATFORM_SCHEMA, BaseNotificationService
+from homeassistant.const import CONF_NAME
+from homeassistant.core import split_entity_id
+import homeassistant.helpers.config_validation as cv
+
+from . import ATTR_ENTITY_ID, ATTR_LANGUAGE, ATTR_MESSAGE, DOMAIN
+
+CONF_MEDIA_PLAYER = "media_player"
+CONF_TTS_SERVICE = "tts_service"
+
+# mypy: allow-untyped-calls, allow-untyped-defs, no-check-untyped-defs
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_NAME): cv.string,
+        vol.Required(CONF_TTS_SERVICE): cv.entity_id,
+        vol.Required(CONF_MEDIA_PLAYER): cv.entity_id,
+        vol.Optional(ATTR_LANGUAGE): cv.string,
+    }
+)
+
+
+async def async_get_service(hass, config, discovery_info=None):
+    """Return the notify service."""
+
+    return TTSNotificationService(config)
+
+
+class TTSNotificationService(BaseNotificationService):
+    """The TTS Notification Service."""
+
+    def __init__(self, config):
+        """Initialize the service."""
+        _, self._tts_service = split_entity_id(config[CONF_TTS_SERVICE])
+        self._media_player = config[CONF_MEDIA_PLAYER]
+        self._language = config.get(ATTR_LANGUAGE)
+
+    async def async_send_message(self, message="", **kwargs):
+        """Call TTS service to speak the notification."""
+        _LOGGER.debug("%s '%s' on %s", self._tts_service, message, self._media_player)
+
+        data = {
+            ATTR_MESSAGE: message,
+            ATTR_ENTITY_ID: self._media_player,
+        }
+        if self._language:
+            data[ATTR_LANGUAGE] = self._language
+
+        await self.hass.services.async_call(
+            DOMAIN,
+            self._tts_service,
+            data,
+        )

tests/components/tts/test_notify.py

@@ -0,0 +1,92 @@
+"""The tests for the TTS component."""
+import pytest
+import yarl
+
+import homeassistant.components.media_player as media_player
+from homeassistant.components.media_player.const import (
+    DOMAIN as DOMAIN_MP,
+    SERVICE_PLAY_MEDIA,
+)
+import homeassistant.components.notify as notify
+import homeassistant.components.tts as tts
+from homeassistant.config import async_process_ha_core_config
+from homeassistant.setup import async_setup_component
+
+from tests.async_mock import patch
+from tests.common import assert_setup_component, async_mock_service
+
+
+def relative_url(url):
+    """Convert an absolute url to a relative one."""
+    return str(yarl.URL(url).relative())
+
+
+@pytest.fixture(autouse=True)
+def mutagen_mock():
+    """Mock writing tags."""
+    with patch(
+        "homeassistant.components.tts.SpeechManager.write_tags",
+        side_effect=lambda *args: args[1],
+    ):
+        yield
+
+
+@pytest.fixture(autouse=True)
+async def internal_url_mock(hass):
+    """Mock internal URL of the instance."""
+    await async_process_ha_core_config(
+        hass,
+        {"internal_url": "http://example.local:8123"},
+    )
+
+
+async def test_setup_platform(hass):
+    """Set up the tts platform ."""
+    config = {
+        notify.DOMAIN: {
+            "platform": "tts",
+            "name": "tts_test",
+            "tts_service": "tts.demo_say",
+            "media_player": "media_player.demo",
+        }
+    }
+    with assert_setup_component(1, notify.DOMAIN):
+        assert await async_setup_component(hass, notify.DOMAIN, config)
+
+    assert hass.services.has_service(notify.DOMAIN, "tts_test")
+
+
+async def test_setup_component_and_test_service(hass):
+    """Set up the demo platform and call service."""
+    calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
+
+    config = {
+        tts.DOMAIN: {"platform": "demo"},
+        media_player.DOMAIN: {"platform": "demo"},
+        notify.DOMAIN: {
+            "platform": "tts",
+            "name": "tts_test",
+            "tts_service": "tts.demo_say",
+            "media_player": "media_player.demo",
+            "language": "en",
+        },
+    }
+
+    with assert_setup_component(1, tts.DOMAIN):
+        assert await async_setup_component(hass, tts.DOMAIN, config)
+
+    with assert_setup_component(1, notify.DOMAIN):
+        assert await async_setup_component(hass, notify.DOMAIN, config)
+
+    await hass.services.async_call(
+        notify.DOMAIN,
+        "tts_test",
+        {
+            tts.ATTR_MESSAGE: "There is someone at the door.",
+        },
+        blocking=True,
+    )
+
+    await hass.async_block_till_done()
+
+    assert len(calls) == 1