temp commit, woring on playht

examples/foundational/07d-interruptible-cartesia.py

@@ -19,6 +19,8 @@ from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
+from pipecat.processors.logger import FrameLogger
+
 
 from runner import configure
 
@@ -71,7 +73,9 @@ async def main(room_url: str, token):
             tma_in,              # User responses
             llm,                 # LLM
             tts,                 # TTS
+            FrameLogger("tts out"),
             transport.output(),  # Transport bot output
+            FrameLogger("transport out"),
             tma_out              # Assistant spoken responses
         ])
 

examples/foundational/07e-interruptible-playht.py

@@ -0,0 +1,98 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import aiohttp
+import os
+import sys
+
+from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
+from pipecat.services.playht import PlayHTTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+from pipecat.processors.logger import FrameLogger
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main(room_url: str, token):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                audio_out_sample_rate=16000,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer()
+            )
+        )
+
+        tts = PlayHTTTSService(
+            user_id=os.getenv("PLAYHT_USER_ID"),
+            api_key=os.getenv("PLAYHT_API_KEY"),
+            voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json",
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        tma_in = LLMUserResponseAggregator(messages)
+        tma_out = LLMAssistantResponseAggregator(messages)
+
+        pipeline = Pipeline([
+            transport.input(),   # Transport user input
+            tma_in,              # User responses
+            llm,                 # LLM
+            tts,                 # TTS
+            FrameLogger("tts out"),
+            transport.output(),  # Transport bot output
+            FrameLogger("transport out"),
+            tma_out              # Assistant spoken responses
+        ])
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append(
+                {"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url, token))

src/pipecat/services/playht.py

@@ -6,6 +6,7 @@
 
 import io
 import struct
+import time
 
 from typing import AsyncGenerator
 
@@ -25,7 +26,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-class PlayHTAIService(TTSService):
+class PlayHTTTSService(TTSService):
 
     def __init__(self, *, api_key: str, user_id: str, voice_url: str, **kwargs):
         super().__init__(**kwargs)
@@ -47,28 +48,40 @@ class PlayHTAIService(TTSService):
         self._client.close()
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
-        b = bytearray()
+        start_time = time.time()
-        in_header = True
+        ttfb = None
-        for chunk in self._client.tts(text, self._options):
+        logger.debug(f"Generating TTS: [{text}]")
-            # skip the RIFF header.
+
-            if in_header:
+        try:
-                b.extend(chunk)
+            b = bytearray()
-                if len(b) <= 36:
+            in_header = True
-                    continue
+            sync_gen = self._client.tts(
-                else:
+                text,
-                    fh = io.BytesIO(b)
+                voice_engine="PlayHT2.0-turbo",
-                    fh.seek(36)
+                options=self._options)
-                    (data, size) = struct.unpack('<4sI', fh.read(8))
+
-                    logger.debug(
+            # need to ask Aleix about this. frames are getting pushed.
-                        f"first attempt: data: {data}, size: {hex(size)}, position: {fh.tell()}")
+            # but playback is blocked
-                    while data != b'data':
+            for chunk in sync_gen:
-                        fh.read(size)
+                # skip the RIFF header.
+                if in_header:
+                    b.extend(chunk)
+                    if len(b) <= 36:
+                        continue
+                    else:
+                        fh = io.BytesIO(b)
+                        fh.seek(36)
                         (data, size) = struct.unpack('<4sI', fh.read(8))
-                        logger.debug(
+                        while data != b'data':
-                            f"subsequent data: {data}, size: {hex(size)}, position: {fh.tell()}, data != data: {data != b'data'}")
+                            fh.read(size)
-                    logger.debug("position: ", fh.tell())
+                            (data, size) = struct.unpack('<4sI', fh.read(8))
-                    in_header = False
+                        in_header = False
-            else:
+                else:
-                if len(chunk):
+                    if len(chunk):
-                    frame = AudioRawFrame(chunk, 16000, 1)
+                        if ttfb is None:
-                    yield frame
+                            ttfb = time.time() - start_time
+                            logger.debug(f"TTS ttfb: {ttfb}")
+                        frame = AudioRawFrame(chunk, 16000, 1)
+                        yield frame
+        except Exception as e:
+            logger.error(f"Error generating TTS: {e}")