main.py

import asyncio
import os
import sys
import time
import logging
from pipecat.frames.frames import (
    TextFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame,
    TTSStartedFrame, TTSStoppedFrame
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams
from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat.vad.silero import SileroVADAnalyzer
from azure_openai import AzureOpenAILLMService
from elevenlabs import ElevenLabs

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Configuration constants
SILENCE_TIMEOUT_SECONDS = float(os.environ.get("SILENCE_TIMEOUT_SECONDS", 10))
MAX_SILENCE_PROMPTS = int(os.environ.get("MAX_SILENCE_PROMPTS", 3))
SILENCE_PROMPT_TEXT = "Are you still there?"
GOODBYE_PROMPT_TEXT = "It seems you're no longer there. I'm hanging up now. Goodbye."

class SilenceAndCallLogicProcessor(FrameProcessor):
    def __init__(self, tts_service, pipeline, app):
        super().__init__()
        self.tts_service = tts_service
        self.pipeline = pipeline
        self.app = app
        self.last_activity_ts = time.time()
        self.silence_prompts_count = 0
        self._bot_is_speaking = False
        self._silence_check_task = None
        self.app.current_call_stats["silence_events"] = 0

    async def start(self):
        self.last_activity_ts = time.time()
        self.silence_prompts_count = 0
        self._bot_is_speaking = False
        if self._silence_check_task:
            self._silence_check_task.cancel()
        self._silence_check_task = asyncio.create_task(self._check_silence_loop())

    async def stop(self):
        if self._silence_check_task:
            self._silence_check_task.cancel()
            try:
                await self._silence_check_task
            except asyncio.CancelledError:
                pass
        await self.tts_service.stop()

    def _reset_activity_timer(self):
        self.last_activity_ts = time.time()
        self.silence_prompts_count = 0

    async def process_frame(self, frame, direction: FrameDirection):
        if isinstance(frame, (UserStartedSpeakingFrame, TextFrame)) and direction == FrameDirection.UPSTREAM:
            self._reset_activity_timer()
        if isinstance(frame, TTSStartedFrame) and direction == FrameDirection.DOWNSTREAM:
            self._bot_is_speaking = True
        elif isinstance(frame, TTSStoppedFrame) and direction == FrameDirection.DOWNSTREAM:
            self._bot_is_speaking = False
            self.last_activity_ts = time.time()
        await self.push_frame(frame, direction)

    async def _check_silence_loop(self):
        while True:
            await asyncio.sleep(1)
            if self._bot_is_speaking:
                continue
            if time.time() - self.last_activity_ts > SILENCE_TIMEOUT_SECONDS:
                self.app.current_call_stats["silence_events"] += 1
                self.silence_prompts_count += 1
                self._bot_is_speaking = True
                if self.silence_prompts_count >= MAX_SILENCE_PROMPTS:
                    await self.push_frame(TextFrame(GOODBYE_PROMPT_TEXT), FrameDirection.DOWNSTREAM)
                    await asyncio.sleep(2)
                    await self.pipeline.stop_when_done()
                    break
                else:
                    await self.push_frame(TextFrame(SILENCE_PROMPT_TEXT), FrameDirection.DOWNSTREAM)
                    self.last_activity_ts = time.time()
                    self._bot_is_speaking = False

class PhoneChatbotApp:
    def __init__(self):
        self.daily_transport = None
        self.pipeline = None
        self.stt_service = None
        self.tts_service = None
        self.llm_service = None
        self.silence_processor = None
        self.call_start_time = None
        self.current_call_stats = {
            "duration_seconds": 0,
            "silence_events": 0,
            "start_time": None,
            "end_time": None,
            "ended_by_silence": False
        }

    def _reset_call_stats(self):
        self.call_start_time = time.time()
        self.current_call_stats = {
            "duration_seconds": 0,
            "silence_events": 0,
            "start_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.call_start_time)),
            "end_time": None,
            "ended_by_silence": False
        }

    async def _log_call_summary(self):
        if self.call_start_time:
            call_end_time = time.time()
            self.current_call_stats["duration_seconds"] = round(call_end_time - self.call_start_time, 2)
            self.current_call_stats["end_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(call_end_time))
            if self.silence_processor and self.silence_processor.silence_prompts_count >= MAX_SILENCE_PROMPTS:
                self.current_call_stats["ended_by_silence"] = True
            logger.info("--- Post-Call Summary ---")
            for key, value in self.current_call_stats.items():
                logger.info(f"  {key.replace('_', ' ').title()}: {value}")
            logger.info("-------------------------")

    def setup_pipeline_hook(self, pipeline_params: PipelineParams, room_url: str, token: str):
        self._reset_call_stats()
        self.pipeline = Pipeline([self.stt_service, self.llm_service, self.tts_service])
        self.silence_processor = SilenceAndCallLogicProcessor(
            tts_service=self.tts_service,
            pipeline=self.pipeline,
            app=self
        )
        self.pipeline.processors.append(self.silence_processor)
        pipeline_params.pipeline = self.pipeline
        pipeline_params.params = PipelineParams(allow_interruptions=True, enable_metrics=True)
        return pipeline_params

    def validate_voice_id(self, voice_id: str) -> bool:
        try:
            client = ElevenLabs(api_key=os.environ.get("elevenlabs"))
            client.voices.get(voice_id=voice_id)
            return True
        except Exception as e:
            logger.error(f"Failed to validate ElevenLabs voice ID {voice_id}: {e}")
            return False

    async def run(self):
        # Validate environment variables (Hugging Face Secrets)
        required_keys = [
            "deepgram", "elevenlabs", "dailyco", "azure_openai"
        ]
        missing_keys = [key for key in required_keys if not os.environ.get(key)]
        if missing_keys:
            logger.error(f"Missing Hugging Face Secrets: {', '.join(missing_keys)}")
            sys.exit(1)

        # Validate ElevenLabs voice ID
        voice_id = os.environ.get("ELEVENLABS_VOICE_ID", "cgSgspJ2msm6clMCkdW9")
        if not self.validate_voice_id(voice_id):
            logger.error(f"Invalid ElevenLabs voice ID: {voice_id}")
            sys.exit(1)

        self.stt_service = DeepgramSTTService(
            api_key=os.environ.get("deepgram"),
            input_audio_format="linear16"
        )
        self.tts_service = ElevenLabsTTSService(
            api_key=os.environ.get("elevenlabs"),
            voice_id=voice_id
        )
        self.llm_service = AzureOpenAILLMService(
            preprompt="You are a friendly and helpful phone assistant."
        )
        self.daily_transport = DailyTransport(
            os.environ.get("DAILY_DOMAIN", "your-username.daily.co"),
            os.environ.get("dailyco"),
            None,
            None,
            "Pipecat Phone Demo",
            vad_analyzer=SileroVADAnalyzer(),
            daily_params=DailyParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                transcription_enabled=False
            )
        )
        self.daily_transport.pipeline_params_hook = self.setup_pipeline_hook

        runner = PipelineRunner()
        try:
            await runner.run(self.daily_transport)
        except KeyboardInterrupt:
            logger.info("Ctrl+C pressed, shutting down")
        except Exception as e:
            logger.error(f"An error occurred: {e}", exc_info=True)
        finally:
            await self._log_call_summary()
            if self.pipeline:
                await self.pipeline.stop_when_done()
            if self.silence_processor:
                await self.silence_processor.stop()

async def main():
    app = PhoneChatbotApp()
    await app.run()

if __name__ == "__main__":
    asyncio.run(main())