Trixy7/trixy_core/conversation/profiler.py

# -*- coding: utf-8 -*-
"""
Conversation Profiler — misst Phasen-Dauern im Debug-Modus.

Registriert sich mit SYSTEM-Prioritaet auf alle relevanten Events
der Conversation-Pipeline und zeichnet Timestamps auf.

WICHTIG: SYSTEM (0) statt MONITOR (100) ist notwendig, weil die Events
verschachtelt sind — ein Handler fuer raw_audio_received loest intern
stt_completed → speech_recognized → intent_received → ... → tts_completed aus.
Mit MONITOR wuerden die Timestamps erst NACH der gesamten inneren Kette
aufgezeichnet, was zu fehlenden Phasen fuehrt.

Tracking ueber satellite_id als primaeren Key, da dies das einzige
Feld ist das in allen Events konsistent vorhanden ist.

Ausgabe erfolgt bei tts_completed (letztes Event der Pipeline),
da conversation_ended im Server-Modus nicht als Event emittiert wird.

Null Overhead wenn nicht im Debug-Modus (Guard in jedem Handler).
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any, TYPE_CHECKING

from trixy_core.service.iservice import IService
from trixy_core.service.enums import ServicePriority, ServiceGroup
from trixy_core.events.decorators import TrixyEvent
from trixy_core.events.enums import EventPriority
from trixy_core.utils.debug import pinfo, pdebug

if TYPE_CHECKING:
    from trixy_core.application import IApplication


@dataclass
class SessionProfile:
    """Timing-Daten fuer eine Conversation-Session."""
    satellite_id: str = ""
    session_id: str = ""
    timestamps: dict[str, float] = field(default_factory=dict)
    providers: dict[str, str] = field(default_factory=dict)
    text: str = ""
    intent: str = ""
    confidence: float = 0.0
    is_keyboard: bool = False


def _get_field(data: Any, name: str, default: Any = "") -> Any:
    """Extrahiert ein Feld aus EventData (trigger) oder dict-metadata (emit)."""
    # Direktes Attribut (typisierte EventData via trigger())
    val = getattr(data, name, None)
    if val is not None and val != "":
        return val
    # metadata-Dict (emit()-Events: Felder in EventData.metadata)
    if hasattr(data, "get"):
        val = data.get(name)
        if val is not None and val != "":
            return val
    return default


# Phasen-Definition: (Name, Start-Event, End-Event, Default-Komponente)
_PHASES: list[tuple[str, str, str, str]] = [
    ("Wakeword",      "wakeword_detected",   "conversation_started",  "WakewordService"),
    ("Recording",     "conversation_started", "raw_audio_received",    "AudioAccumulator"),
    ("STT",           "raw_audio_received",   "stt_completed",         "STT"),
    ("STT-Korrektur", "stt_completed",        "speech_recognized",     "STTCorrector"),
    ("NLP",           "speech_recognized",     "intent_received",      "NLP"),
    ("Handler",       "intent_received",       "intent_handled",       "Handler"),
    ("LLM",           "create_output_text",    "output_text_created",  "LLM"),
    ("TTS",           "output_text_created",   "tts_completed",        "TTS"),
]


class ConversationProfiler(IService):
    """
    Conversation Profiler — misst Phasen-Dauern im Debug-Modus.

    Tracking ueber satellite_id als primaeren Key.
    Ausgabe bei tts_completed (letztes Event der Pipeline).
    Null Overhead wenn nicht im Debug-Modus.
    """

    PRIORITY = ServicePriority.OPTIONAL
    GROUP = ServiceGroup.CONVERSATION
    DEPENDENCIES: list[str] = []
    NAME = "ConversationProfiler"

    def __init__(self, application: "IApplication") -> None:
        super().__init__(application)
        # satellite_id → SessionProfile
        self._profiles: dict[str, SessionProfile] = {}

    async def start(self) -> None:
        """Startet den Profiler."""
        if self._application.debug:
            pinfo("ConversationProfiler gestartet (Debug-Modus)")

    async def stop(self) -> None:
        """Stoppt den Profiler."""
        self._profiles.clear()

    # === Hilfsmethoden ===

    def _get_sat_id(self, event_data: Any) -> str:
        """Extrahiert satellite_id aus Event-Daten."""
        return str(_get_field(event_data, "satellite_id", ""))

    def _get_profile(self, sat_id: str) -> SessionProfile | None:
        """Holt ein bestehendes Profil fuer den Satellite."""
        if not sat_id:
            return None
        return self._profiles.get(sat_id)

    def _ensure_profile(self, sat_id: str) -> SessionProfile | None:
        """Holt oder erstellt ein Profil fuer den Satellite."""
        if not sat_id:
            return None
        if sat_id not in self._profiles:
            self._profiles[sat_id] = SessionProfile(satellite_id=sat_id)
        return self._profiles[sat_id]

    # === Event-Handler (alle mit SYSTEM-Prioritaet) ===

    @TrixyEvent(["wakeword_detected"], priority=EventPriority.SYSTEM)
    async def on_wakeword_detected(self, event_name: str, event_data: Any) -> None:
        """Wakeword erkannt — Startpunkt der Pipeline."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        if not sat_id:
            return

        # Neues Profil starten (altes verwerfen falls vorhanden)
        self._profiles[sat_id] = SessionProfile(satellite_id=sat_id)
        profile = self._profiles[sat_id]
        profile.timestamps["wakeword_detected"] = time.monotonic()
        profile.session_id = str(_get_field(event_data, "session_id", ""))

    @TrixyEvent(["conversation_started"], priority=EventPriority.SYSTEM)
    async def on_conversation_started(self, event_name: str, event_data: Any) -> None:
        """Conversation gestartet (nach Arbitrierung)."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        if not sat_id:
            return

        profile = self._ensure_profile(sat_id)
        if profile:
            profile.timestamps["conversation_started"] = time.monotonic()
            # conversation_id als session_id uebernehmen (Server-seitige ID)
            conv_id = str(_get_field(event_data, "conversation_id", ""))
            if conv_id:
                profile.session_id = conv_id

    @TrixyEvent(["raw_audio_received"], priority=EventPriority.SYSTEM)
    async def on_raw_audio_received(self, event_name: str, event_data: Any) -> None:
        """Rohe Audio-Daten empfangen."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if profile and "raw_audio_received" not in profile.timestamps:
            profile.timestamps["raw_audio_received"] = time.monotonic()

    @TrixyEvent(["stt_completed"], priority=EventPriority.SYSTEM)
    async def on_stt_completed(self, event_name: str, event_data: Any) -> None:
        """STT abgeschlossen."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if profile:
            profile.timestamps["stt_completed"] = time.monotonic()
            provider = str(_get_field(event_data, "provider", ""))
            if provider:
                profile.providers["STT"] = provider

    @TrixyEvent(["speech_recognized"], priority=EventPriority.SYSTEM)
    async def on_speech_recognized(self, event_name: str, event_data: Any) -> None:
        """Sprache erkannt (nach STT-Korrektur oder Keyboard-Input)."""
        if not self._application.debug:
            return

        source = str(_get_field(event_data, "source", "stt"))
        sat_id = self._get_sat_id(event_data)
        text = str(_get_field(event_data, "text", ""))

        # Keyboard-Input: Neues Profil ab diesem Punkt
        if source == "keyboard":
            # Fuer Standalone: satellite_id kann leer sein
            key = sat_id or "standalone"
            self._profiles[key] = SessionProfile(
                satellite_id=key,
                is_keyboard=True,
                text=text,
            )
            self._profiles[key].timestamps["speech_recognized"] = time.monotonic()
            return

        # Normaler STT-Pfad
        profile = self._get_profile(sat_id)
        if profile:
            profile.timestamps["speech_recognized"] = time.monotonic()
            if text:
                profile.text = text

    @TrixyEvent(["intent_received"], priority=EventPriority.SYSTEM)
    async def on_intent_received(self, event_name: str, event_data: Any) -> None:
        """Intent erkannt."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if not profile:
            return

        profile.timestamps["intent_received"] = time.monotonic()
        intent = str(_get_field(event_data, "intent", ""))
        confidence = float(_get_field(event_data, "confidence", 0.0))
        if intent:
            profile.intent = intent
            profile.providers["Handler"] = intent
        if confidence:
            profile.confidence = confidence
        if not profile.text:
            profile.text = str(_get_field(event_data, "original_text", ""))

    @TrixyEvent(["intent_handled"], priority=EventPriority.SYSTEM)
    async def on_intent_handled(self, event_name: str, event_data: Any) -> None:
        """Intent verarbeitet."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if profile:
            profile.timestamps["intent_handled"] = time.monotonic()

    @TrixyEvent(["create_output_text"], priority=EventPriority.SYSTEM)
    async def on_create_output_text(self, event_name: str, event_data: Any) -> None:
        """LLM-Textgenerierung angefordert."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if profile:
            profile.timestamps["create_output_text"] = time.monotonic()

    @TrixyEvent(["output_text_created"], priority=EventPriority.SYSTEM)
    async def on_output_text_created(self, event_name: str, event_data: Any) -> None:
        """Antworttext generiert."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if profile:
            profile.timestamps["output_text_created"] = time.monotonic()

    @TrixyEvent(["tts_completed"], priority=EventPriority.SYSTEM)
    async def on_tts_completed(self, event_name: str, event_data: Any) -> None:
        """TTS abgeschlossen — Profil ausgeben (letztes Event der Pipeline)."""
        if not self._application.debug:
            return

        sat_id = self._get_sat_id(event_data)
        profile = self._get_profile(sat_id)
        if not profile:
            return

        profile.timestamps["tts_completed"] = time.monotonic()
        provider = str(_get_field(event_data, "provider", ""))
        if provider:
            profile.providers["TTS"] = provider

        # Profil ausgeben und aufraeumen
        self._print_profile(profile)
        self._profiles.pop(sat_id, None)

    # === Ausgabe ===

    def _print_profile(self, profile: SessionProfile) -> None:
        """Gibt das Profil als formatierte Tabelle aus."""
        ts = profile.timestamps

        # Phasen berechnen
        phases: list[tuple[str, str, float]] = []

        if profile.is_keyboard:
            # Keyboard-Input: Nur Phasen ab NLP (kein Wakeword/Recording/STT)
            phase_list = _PHASES[4:]
        else:
            phase_list = _PHASES

        for phase_name, start_evt, end_evt, default_component in phase_list:
            if start_evt in ts and end_evt in ts:
                duration = ts[end_evt] - ts[start_evt]
                component = profile.providers.get(phase_name, default_component)
                phases.append((phase_name, component, duration))

        if not phases:
            return

        # Gesamt-Dauer
        all_ts = sorted(ts.values())
        total_duration = all_ts[-1] - all_ts[0] if len(all_ts) >= 2 else 0.0

        # Tabelle formatieren
        w_phase = 16
        w_comp = 16
        w_dur = 19
        w_inner = w_phase + 1 + w_comp + 1 + w_dur

        # Session-ID kuerzen fuer Titel
        display_id = profile.session_id or profile.satellite_id
        short_id = display_id[:16] if len(display_id) > 16 else display_id

        lines: list[str] = []
        lines.append(f"╔{'═' * w_inner}╗")
        title = f"Conversation Profil — {short_id}"
        lines.append(f"║{title:^{w_inner}}║")
        lines.append(f"╠{'═' * w_inner}╣")

        # Header
        lines.append(
            f"║ {'Phase':<{w_phase - 1}}│ {'Komponente':<{w_comp - 1}}│ {'Dauer':<{w_dur - 1}}║"
        )
        lines.append(
            f"╠{'═' * w_phase}╪{'═' * w_comp}╪{'═' * w_dur}╣"
        )

        # Phasen-Zeilen
        for phase_name, component, duration in phases:
            dur_str = f"{duration:>7.3f} sec"
            lines.append(
                f"║ {phase_name:<{w_phase - 1}}│ {component:<{w_comp - 1}}│ {dur_str:<{w_dur - 1}}║"
            )

        # Gesamt
        lines.append(
            f"╠{'═' * w_phase}╪{'═' * w_comp}╪{'═' * w_dur}╣"
        )
        total_str = f"{total_duration:>7.3f} sec"
        lines.append(
            f"║ {'GESAMT':<{w_phase - 1}}│ {'':<{w_comp - 1}}│ {total_str:<{w_dur - 1}}║"
        )

        # Meta-Informationen
        lines.append(
            f"╠{'═' * w_phase}╧{'═' * w_comp}╧{'═' * w_dur}╣"
        )

        if profile.text:
            text_display = profile.text[:w_inner - 9]
            lines.append(f"║ Text: \"{text_display}\"{' ' * max(0, w_inner - 9 - len(text_display))}║")

        if profile.intent:
            intent_info = f"Intent: {profile.intent}"
            if profile.confidence:
                intent_info += f" (confidence: {profile.confidence:.2f})"
            intent_display = intent_info[:w_inner - 2]
            lines.append(f"║ {intent_display:<{w_inner - 2}} ║")

        if profile.satellite_id:
            sat_display = f"Satellite: {profile.satellite_id}"[:w_inner - 2]
            lines.append(f"║ {sat_display:<{w_inner - 2}} ║")

        if profile.is_keyboard:
            lines.append(f"║ {'Quelle: Keyboard-Input':<{w_inner - 2}} ║")

        lines.append(f"╚{'═' * w_inner}╝")

        # Ausgabe ueber pinfo()
        for line in lines:
            pinfo(line)