"""把稳定的产品配置映射为 Pipecat 用户轮次策略。"""

from __future__ import annotations

from typing import Any

from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.turns.user_start import (
    TranscriptionUserTurnStartStrategy,
    VADUserTurnStartStrategy,
)
from pipecat.turns.user_stop import (
    SpeechTimeoutUserTurnStopStrategy,
    TurnAnalyzerUserTurnStopStrategy,
)
from pipecat.turns.user_turn_strategies import UserTurnStrategies


DEFAULT_VAD = {
    "confidence": 0.7,
    "start_secs": 0.2,
    "stop_secs": 0.2,
    "min_volume": 0.6,
}
DEFAULT_TURN_DETECTION = {
    "strategy": "silence",
    "silence_timeout_secs": 0.6,
}


def _section(config: dict[str, Any], snake: str, camel: str) -> dict[str, Any]:
    value = config.get(snake, config.get(camel, {}))
    return value if isinstance(value, dict) else {}


def _value(config: dict[str, Any], snake: str, camel: str, default: Any) -> Any:
    return config.get(snake, config.get(camel, default))


def create_vad_analyzer(turn_config: dict[str, Any]) -> SileroVADAnalyzer:
    vad = _section(turn_config, "vad", "vad")
    return SileroVADAnalyzer(
        params=VADParams(
            confidence=float(vad.get("confidence", DEFAULT_VAD["confidence"])),
            start_secs=float(_value(vad, "start_secs", "startSecs", 0.2)),
            stop_secs=float(_value(vad, "stop_secs", "stopSecs", 0.2)),
            min_volume=float(_value(vad, "min_volume", "minVolume", 0.6)),
        )
    )


def create_user_turn_strategies(
    turn_config: dict[str, Any], *, enable_interruptions: bool
) -> UserTurnStrategies:
    barge_in = _section(turn_config, "barge_in", "bargeIn")
    start = []
    strategy = barge_in.get("strategy", "vad")
    if strategy == "vad":
        start.append(VADUserTurnStartStrategy(enable_interruptions=enable_interruptions))
    else:
        start.append(
            TranscriptionUserTurnStartStrategy(enable_interruptions=enable_interruptions)
        )

    detection = _section(turn_config, "turn_detection", "turnDetection")
    if detection.get("strategy", DEFAULT_TURN_DETECTION["strategy"]) == "smart_turn":
        stop = [
            TurnAnalyzerUserTurnStopStrategy(
                turn_analyzer=LocalSmartTurnAnalyzerV3(),
                wait_for_transcript=True,
            )
        ]
    else:
        stop = [
            SpeechTimeoutUserTurnStopStrategy(
                user_speech_timeout=float(
                    _value(
                        detection,
                        "silence_timeout_secs",
                        "silenceTimeoutSecs",
                        0.6,
                    )
                ),
                wait_for_transcript=True,
            )
        ]
    return UserTurnStrategies(start=start, stop=stop)