Trixy7/trixy_core/scheduler/action/tts.py

# -*- coding: utf-8 -*-
"""
TTS Action - Spricht Text auf Satellites via TTS.

Unterstuetzt Template-Platzhalter im Text ({date.now.hour}, {sys.hostname}, etc.)
und kann an einen bestimmten Satellite oder alle verbundenen Satellites senden.
"""

from __future__ import annotations

import uuid
from typing import Any

from trixy_core.scheduler.action.base import Action
from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError
from trixy_core.utils.template_formatter import format_template
from trixy_core.utils.debug import pdebug, perror


# Sonderwert fuer "alle Satellites"
_ALL_SATELLITES = "__all__"


class TTSSpeakAction(Action):
    """
    Action die Text via TTS auf Satellites ausgibt.

    Der Text wird vor der Ausgabe durch den Template-Formatter verarbeitet,
    sodass Platzhalter wie {date.now.hour} oder {sys.hostname} aufgeloest werden.

    Beispiel:
        TTSSpeakAction(
            text="Es ist {date.now.hour} Uhr {date.now.minute}.",
            target="__all__",
        )
    """

    def __init__(
        self,
        text: str,
        target: str = _ALL_SATELLITES,
        voice: str = "default",
        speed: float = 1.0,
        volume: float = 1.0,
        name: str | None = None,
        timeout_seconds: float = 30.0,
    ):
        """
        Initialisiert die TTS-Action.

        Args:
            text: Text-Template mit optionalen {placeholder}-Variablen.
            target: Satellite-ID oder "__all__" fuer alle.
            voice: TTS-Stimme/Profil.
            speed: Sprechgeschwindigkeit (1.0 = normal).
            volume: Lautstaerke (1.0 = normal).
            name: Action-Name.
            timeout_seconds: Timeout.
        """
        super().__init__(name or "tts_speak", timeout_seconds)
        self._text = text
        self._target = target
        self._voice = voice
        self._speed = speed
        self._volume = volume

    @property
    def text(self) -> str:
        """Text-Template."""
        return self._text

    @property
    def target(self) -> str:
        """Ziel-Satellite-ID oder '__all__'."""
        return self._target

    async def _execute(self, context: dict) -> dict:
        """Fuehrt die TTS-Ausgabe durch."""
        # Template-Platzhalter aufloesen
        # Hinweis: satellite.speak() formatiert ebenfalls, aber der Standalone-Pfad
        # geht direkt ueber event_manager.trigger() — daher hier formatieren.
        resolved_text = format_template(
            self._text, context=context, application=context.get("application")
        )
        pdebug(f"TTSSpeakAction: Text='{resolved_text}', Ziel={self._target}")

        # Satellite-Manager aus Kontext holen
        satellite_manager = context.get("satellite_manager")
        event_manager = context.get("event_manager")

        spoken_to: list[str] = []

        if satellite_manager:
            # Server-Modus: Direkt ueber Satellite.speak()
            if self._target == _ALL_SATELLITES:
                satellites = satellite_manager.get_connected()
            else:
                sat = satellite_manager.get(self._target)
                satellites = [sat] if sat and sat.is_connected else []

            pdebug(f"TTSSpeakAction: {len(satellites)} Satellite(s) gefunden")

            for satellite in satellites:
                try:
                    await satellite.speak(
                        resolved_text,
                        voice=self._voice,
                        speed=self._speed,
                        volume=self._volume,
                    )
                    spoken_to.append(satellite.satellite_id)
                    pdebug(f"TTSSpeakAction: Gesendet an {satellite.satellite_id}")
                except Exception as e:
                    perror(f"TTSSpeakAction: Fehler bei {satellite.satellite_id}: {e}")

        elif event_manager:
            # Standalone-Modus: Event ueber trigger() mit TTSRequest senden
            from trixy_core.events.event_data.basic import TTSRequest

            request_id = f"sched-tts-{uuid.uuid4().hex[:8]}"
            event_data = TTSRequest(
                request_id=request_id,
                satellite_id=self._target if self._target != _ALL_SATELLITES else "",
                text=resolved_text,
                voice=self._voice,
                speed=self._speed,
                volume=self._volume,
                source="TTSSpeakAction",
            )
            await event_manager.trigger("tts_request", event_data)
            spoken_to.append("event:" + request_id)
            pdebug(f"TTSSpeakAction: Event getriggert (request_id={request_id})")

        else:
            perror("TTSSpeakAction: Weder satellite_manager noch event_manager im Kontext")

        return {
            "text": resolved_text,
            "template": self._text,
            "target": self._target,
            "spoken_to": spoken_to,
        }

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "text": self._text,
            "target": self._target,
            "voice": self._voice,
            "speed": self._speed,
            "volume": self._volume,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "TTSSpeakAction":
        """Erstellt Action aus Dictionary."""
        return cls(
            text=data.get("text", ""),
            target=data.get("target", _ALL_SATELLITES),
            voice=data.get("voice", "default"),
            speed=data.get("speed", 1.0),
            volume=data.get("volume", 1.0),
            name=data.get("name"),
            timeout_seconds=data.get("timeout_seconds", 30.0),
        )

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "TTS-Sprachausgabe"

    @classmethod
    def form_fields(cls) -> list[FormField]:
        """Gibt die Formular-Felder fuer die TUI zurueck."""
        return [
            FormField(
                name="text",
                label="Text",
                field_type=FieldType.TEXTAREA,
                required=True,
                lines=4,
                placeholder="Es ist {date.now.hour} Uhr {date.now.minute}.",
                help_text=(
                    "Platzhalter: {date.now.hour}, {date.now.weekday}, "
                    "{sys.hostname}, {env.USER}, {ctx.job_name} ..."
                ),
            ),
            FormField(
                name="target",
                label="Ziel-Satellite",
                field_type=FieldType.TEXT,
                default=_ALL_SATELLITES,
                placeholder="__all__",
                help_text="Satellite-ID oder '__all__' fuer alle verbundenen Satellites",
            ),
            FormField(
                name="voice",
                label="Stimme",
                field_type=FieldType.TEXT,
                default="default",
                placeholder="default",
                help_text="TTS-Stimmprofil",
            ),
            FormField(
                name="speed",
                label="Geschwindigkeit",
                field_type=FieldType.NUMBER,
                default=1.0,
                min_value=0.1,
                max_value=3.0,
                is_float=True,
                help_text="1.0 = normal, 0.5 = langsam, 2.0 = schnell",
            ),
            FormField(
                name="volume",
                label="Lautstaerke",
                field_type=FieldType.NUMBER,
                default=1.0,
                min_value=0.0,
                max_value=2.0,
                is_float=True,
                help_text="1.0 = normal",
            ),
            FormField(
                name="timeout_seconds",
                label="Timeout (Sek.)",
                field_type=FieldType.NUMBER,
                default=30.0,
                min_value=1.0,
                is_float=True,
            ),
        ]

    @classmethod
    def validate_form(cls, data: dict) -> list[FormValidationError]:
        """Validiert Formular-Daten."""
        errors: list[FormValidationError] = []
        if not data.get("text", "").strip():
            errors.append(FormValidationError("text", "Text ist erforderlich"))
        target = data.get("target", "").strip()
        if not target:
            errors.append(FormValidationError("target", "Ziel ist erforderlich"))
        return errors