Modernize picoTTS

homeassistant/components/picotts/__init__.py

@@ -1 +1,41 @@
 """Support for pico integration."""
+
+from __future__ import annotations
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DOMAIN
+
+CONFIG_SCHEMA = vol.Schema({vol.Optional(DOMAIN): {}}, extra=vol.ALLOW_EXTRA)
+
+PLATFORMS: list[Platform] = [Platform.TTS]
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up picoTTS from YAML (if present)."""
+    if DOMAIN in config:
+        hass.async_create_task(
+            hass.config_entries.flow.async_init(
+                DOMAIN,
+                context={"source": config_entries.SOURCE_IMPORT},
+                data={},
+            )
+        )
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up picoTTS from a config entry."""
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload picoTTS."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/picotts/config_flow.py

@@ -0,0 +1,54 @@
+"""Config flow for picoTTS integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant import config_entries
+from homeassistant.config_entries import ConfigEntry, ConfigFlowResult
+from homeassistant.core import callback
+
+from .const import DOMAIN
+
+
+class PicoTTSConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for picoTTS."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        # Prevent multiple instances
+        if self._async_current_entries():
+            return self.async_abort(reason="already_configured")
+
+        return self.async_create_entry(title="picoTTS", data={})
+
+    async def async_step_import(
+        self, import_config: dict[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle configuration import from YAML."""
+        if self._async_current_entries():
+            return self.async_abort(reason="already_configured")
+
+        return self.async_create_entry(title="picoTTS", data={})
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry: ConfigEntry) -> PicoTTSOptionsFlow:
+        """Return the options flow handler."""
+        return PicoTTSOptionsFlow(config_entry)
+
+
+class PicoTTSOptionsFlow(config_entries.OptionsFlow):
+    """Options flow for picoTTS (none required)."""
+
+    def __init__(self, entry: config_entries.ConfigEntry) -> None:
+        """Initialize options flow."""
+        self.entry = entry
+
+    async def async_step_init(self, user_input=None) -> ConfigFlowResult:
+        """No options available."""
+        return self.async_create_entry(title="", data={})

homeassistant/components/picotts/const.py

@@ -0,0 +1,3 @@
+"""Constants for picoTTS integration."""
+
+DOMAIN = "picotts"

homeassistant/components/picotts/manifest.json

@@ -2,7 +2,9 @@
   "domain": "picotts",
   "name": "Pico TTS",
   "codeowners": [],
+  "config_flow": true,
   "documentation": "https://www.home-assistant.io/integrations/picotts",
   "iot_class": "local_push",
-  "quality_scale": "legacy"
+  "quality_scale": "legacy",
+  "requirements": ["py-nanotts==0.1.1"]
 }

homeassistant/components/picotts/strings.json

@@ -0,0 +1,17 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    },
+    "step": {
+      "import": {
+        "description": "Imported configuration.",
+        "title": "picoTTS"
+      },
+      "user": {
+        "description": "Set up picoTTS.",
+        "title": "picoTTS"
+      }
+    }
+  }
+}

homeassistant/components/picotts/tts.py

@@ -1,82 +1,96 @@
 """Support for the Pico TTS speech service."""
 
+from __future__ import annotations
+
+import io
 import logging
-import os
-import shutil
-import subprocess
-import tempfile
 from typing import Any
+import wave
 
-import voluptuous as vol
+from py_nanotts import NanoTTS
 
-from homeassistant.components.tts import (
-    CONF_LANG,
-    PLATFORM_SCHEMA as TTS_PLATFORM_SCHEMA,
-    Provider,
-    TtsAudioType,
-)
+from homeassistant.components.tts import TextToSpeechEntity, TtsAudioType
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 _LOGGER = logging.getLogger(__name__)
 
 SUPPORT_LANGUAGES = ["en-US", "en-GB", "de-DE", "es-ES", "fr-FR", "it-IT"]
-
 DEFAULT_LANG = "en-US"
 
-PLATFORM_SCHEMA = TTS_PLATFORM_SCHEMA.extend(
-    {vol.Optional(CONF_LANG, default=DEFAULT_LANG): vol.In(SUPPORT_LANGUAGES)}
-)
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up picoTTS TTS entity from a config entry."""
+    async_add_entities([PicoTTSEntity()])
 
 
-def get_engine(hass, config, discovery_info=None):
-    """Set up Pico speech component."""
-    if shutil.which("pico2wave") is None:
-        _LOGGER.error("'pico2wave' was not found")
-        return False
-    return PicoProvider(config[CONF_LANG])
+class PicoTTSEntity(TextToSpeechEntity):
+    """picoTTS entity using NanoTTS."""
 
+    _attr_name = "PicoTTS"
+    _attr_unique_id = "picotts"
+    _attr_supported_languages = SUPPORT_LANGUAGES
+    _attr_default_language = DEFAULT_LANG
 
-class PicoProvider(Provider):
-    """The Pico TTS API provider."""
+    _attr_supported_options = ["pitch", "speed", "volume"]
 
-    def __init__(self, lang):
-        """Initialize Pico TTS provider."""
-        self._lang = lang
-        self.name = "PicoTTS"
+    def __init__(self) -> None:
+        """Initialize entity."""
+        self._engine = NanoTTS()
 
-    @property
-    def default_language(self) -> str:
-        """Return the default language."""
-        return self._lang
-
-    @property
-    def supported_languages(self) -> list[str]:
-        """Return list of supported languages."""
-        return SUPPORT_LANGUAGES
-
-    def get_tts_audio(
-        self, message: str, language: str, options: dict[str, Any]
+    async def async_get_tts_audio(
+        self,
+        message: str,
+        language: str,
+        options: dict[str, Any],
     ) -> TtsAudioType:
-        """Load TTS using pico2wave."""
-        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpf:
-            fname = tmpf.name
+        """Generate TTS audio.
+
+        Return (content_type, bytes) or (None, None) on failure.
+        """
+        pitch = _coerce_float(options.get("pitch"))
+        speed = _coerce_float(options.get("speed"))
+        volume = _coerce_float(options.get("volume"))
+
+        voice = language if language in SUPPORT_LANGUAGES else DEFAULT_LANG
+        if voice != language:
+            _LOGGER.debug(
+                "Unsupported language %r requested; using %r", language, voice
+            )
 
-        cmd = ["pico2wave", "--wave", fname, "-l", language]
-        result = subprocess.run(cmd, text=True, input=message, check=False)
-        data = None
         try:
-            if result.returncode != 0:
-                _LOGGER.error(
-                    "Error running pico2wave, return code: %s", result.returncode
-                )
-                return (None, None)
-            with open(fname, "rb") as voice:
-                data = voice.read()
-        except OSError:
-            _LOGGER.error("Error trying to read %s", fname)
+            pcm = self._engine.speak(
+                message,
+                voice=voice,
+                speed=speed,
+                pitch=pitch,
+                volume=volume,
+            )
+        except Exception:
+            _LOGGER.exception("NanoTTS failed generating audio")
             return (None, None)
-        finally:
-            os.remove(fname)
 
-        if data:
-            return ("wav", data)
-        return (None, None)
+        # Wrap returned PCM frames in a WAV container (mono, 16kHz, 16-bit)
+        with io.BytesIO() as wav_io:
+            with wave.open(wav_io, "wb") as wav_file:
+                wav_file.setframerate(16000)
+                wav_file.setsampwidth(2)
+                wav_file.setnchannels(1)
+                wav_file.writeframes(pcm)
+
+            return ("wav", wav_io.getvalue())
+
+
+def _coerce_float(value: Any) -> float | None:
+    """Convert service option values to float, returning None if unset/invalid."""
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except TypeError, ValueError:
+        return None

homeassistant/generated/config_flows.py

@@ -535,6 +535,7 @@ FLOWS = {
         "philips_js",
         "pi_hole",
         "picnic",
+        "picotts",
         "ping",
         "plaato",
         "playstation_network",

homeassistant/generated/integrations.json

@@ -5196,7 +5196,7 @@
     "picotts": {
       "name": "Pico TTS",
       "integration_type": "hub",
-      "config_flow": false,
+      "config_flow": true,
       "iot_class": "local_push"
     },
     "pilight": {

requirements_all.txt

@@ -1855,6 +1855,9 @@ py-madvr2==1.6.40
 # homeassistant.components.melissa
 py-melissa-climate==3.0.3
 
+# homeassistant.components.picotts
+py-nanotts==0.1.1
+
 # homeassistant.components.nextbus
 py-nextbusnext==2.3.0
 

requirements_test_all.txt

@@ -1604,6 +1604,9 @@ py-madvr2==1.6.40
 # homeassistant.components.melissa
 py-melissa-climate==3.0.3
 
+# homeassistant.components.picotts
+py-nanotts==0.1.1
+
 # homeassistant.components.nextbus
 py-nextbusnext==2.3.0
 

tests/components/picotts/__init__.py

@@ -0,0 +1 @@
+"""Test for picoTTS."""

tests/components/picotts/test_tts.py

@@ -0,0 +1,88 @@
+"""Test for picoTTS."""
+
+from __future__ import annotations
+
+import io
+from unittest.mock import MagicMock, patch
+import wave
+
+import pytest
+
+from homeassistant.components import picotts
+from homeassistant.components.picotts.tts import DEFAULT_LANG, PicoTTSEntity
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def _setup_integration(hass: HomeAssistant) -> MockConfigEntry:
+    """Set up the integration via a config entry."""
+    entry = MockConfigEntry(domain=picotts.DOMAIN, data={})
+    entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    return entry
+
+
+@pytest.mark.asyncio
+async def test_setup_creates_tts_entity(hass: HomeAssistant) -> None:
+    """Config entry setup should create the TTS entity."""
+    # Patch NanoTTS so entity construction doesn't touch real backend
+    with patch("homeassistant.components.picotts.tts.NanoTTS", autospec=True):
+        await _setup_integration(hass)
+
+    state = hass.states.get("tts.picotts")
+    assert state is not None, "Expected tts.picotts entity to exist"
+    assert state.name == "PicoTTS"
+
+
+@pytest.mark.asyncio
+async def test_async_get_tts_audio_returns_valid_wav(hass: HomeAssistant) -> None:
+    """Ensure async_get_tts_audio returns a valid WAV file."""
+    nano = MagicMock()
+    nano.speak.return_value = b"\x00\x01" * 200  # fake PCM16 mono samples
+
+    with patch("homeassistant.components.picotts.tts.NanoTTS", return_value=nano):
+        await _setup_integration(hass)
+        entity = PicoTTSEntity()
+
+        content_type, data = await entity.async_get_tts_audio(
+            message="hello",
+            language="en-US",
+            options={},
+        )
+
+    assert content_type == "wav"
+    assert isinstance(data, (bytes, bytearray))
+
+    # Validate WAV structure using Python's wave module
+    with wave.open(io.BytesIO(data), "rb") as wav_file:
+        assert wav_file.getnchannels() == 1
+        assert wav_file.getsampwidth() == 2
+        assert wav_file.getframerate() == 16000
+        assert wav_file.getnframes() == 200
+
+
+@pytest.mark.asyncio
+async def test_language_fallback_to_default(hass: HomeAssistant) -> None:
+    """Unsupported language should fall back to DEFAULT_LANG when calling NanoTTS."""
+    nano = MagicMock()
+    nano.speak.return_value = b"\x00\x01" * 50
+
+    with patch("homeassistant.components.picotts.tts.NanoTTS", return_value=nano):
+        await _setup_integration(hass)
+
+        entity = PicoTTSEntity()
+
+        await entity.async_get_tts_audio(
+            message="hello",
+            language="xx-YY",  # unsupported
+            options={"speed": "1.1", "pitch": "1.2", "volume": "1.3"},
+        )
+
+    # Ensure voice fell back to DEFAULT_LANG
+    kwargs = nano.speak.call_args.kwargs
+    assert kwargs["voice"] == DEFAULT_LANG
+    assert kwargs["speed"] == 1.1
+    assert kwargs["pitch"] == 1.2
+    assert kwargs["volume"] == 1.3