legacy_to_delete/debug.py

import asyncio
import time
import traceback
from chat_pipeline import ChatPipeline
from clip_transform import CLIPTransform
from chat_service import ChatService
from dotenv import load_dotenv
from text_to_speech_service import TextToSpeechService
from concurrent.futures import ThreadPoolExecutor
from local_speaker_service import LocalSpeakerService
from chat_service import ChatService

def time_sentance_lenghts():
    load_dotenv()

    print ("Initializing Chat")
    # audio_processor = AudioStreamProcessor()
    user_speech_service0 = TextToSpeechService(voice_id="Adam")
    prompts = [
        "hello, i am a long sentance, how are you today? Tell me about your shadow self?",
        "a shorter sentance",
        "Jung believed that the process of self-discovery and personal growth involves confronting and integrating the shadow self into the conscious mind.",
        "By doing so, we become more self-aware and more fully actualized individuals.",
    ]

    print ("Timing prompts\n")
    for prompt in prompts:
        start_time = time.time()
        start_stream_time = time.time()
        stream = user_speech_service0.stream(prompt)
        audio = b""
        for chunk in stream:
            if chunk is not None:
                audio += chunk
        end_stream_time = time.time()
        from elevenlabs import play
        start_speech_time = time.time()
        play(audio)
        end_speech_time = time.time()
        end_time = time.time()
        total_time = (end_time - start_time)
        stream_time = (end_stream_time - start_stream_time)
        speech_time = (end_speech_time - start_speech_time)
        stream_multiple = speech_time / stream_time
        print(f"Stream time: {stream_time:.4f}, Acutual audio time: {speech_time:.4f}, a multiple of {stream_multiple:.2f}. for prompt: {prompt}")

    print ("\nChat success")

def test_sentance_lenghts():
    load_dotenv()

    print ("Initializing Chat")
    audio_processor = LocalSpeakerService()
    user_speech_service0 = TextToSpeechService(voice_id="Adam")
    user_speech_service1 = TextToSpeechService(voice_id="Adam")
    user_speech_service2 = TextToSpeechService(voice_id="Adam")
    user_speech_service3 = TextToSpeechService(voice_id="Adam")

    prompts = [
        "hello, i am a long sentance, how are you today? Tell me about your shadow self?",
        "a shorter sentance",
        "Jung believed that the process of self-discovery and personal growth involves confronting and integrating the shadow self into the conscious mind.",
        "By doing so, we become more self-aware and more fully actualized individuals.",
    ]
    first = True
    stream1 = user_speech_service1.stream(prompts[1])
    stream0 = user_speech_service0.stream(prompts[0])
    time.sleep(5)
    stream2 = user_speech_service2.stream(prompts[2])
    stream3 = user_speech_service3.stream(prompts[3])
    audio_processor.add_audio_stream(stream0)
    audio_processor.add_audio_stream(stream1)
    audio_processor.add_audio_stream(stream2)
    audio_processor.add_audio_stream(stream3)
    audio_processor.close()
    from elevenlabs import generate, play
    speech0 = generate(prompts[0], voice="Adam")
    speech1 = generate(prompts[1], voice="Adam")
    speech2 = generate(prompts[2], voice="Adam")
    speech3 = generate(prompts[3], voice="Adam")
    play(speech0)
    play(speech1)
    play(speech2)
    play(speech1)
    play(speech3)
    play(speech1)
    # for prompt in prompts:
    #     stream = user_speech_service.stream(prompt)
    #     if first:
    #         first = False
    #         time.sleep(5)
    #     audio_processor.add_audio_stream(stream)
    audio_processor.close()
    print ("Chat success")

def run_debug_code():
    load_dotenv()

    # print ("Initializing CLIP templates")
    # clip_transform = CLIPTransform()
    # print ("CLIP success")

    print ("Initializing Chat")
    # chat_service = ChatService()
    audio_processor = LocalSpeakerService()
    chat_service = ChatService(audio_processor, voice_id="2OviOUQc1JsQRQgNkVBj") # Chales003

    user_speech_service = TextToSpeechService(voice_id="Adam")

    # user_speech_service.print_voices() # if you want to see your custom voices

    prompts = [
        "hello, how are you today?",
        "tell me about your shadow self?",
        "hmm, interesting, tell me more about that.",
        "wait, that is so interesting, what else?",
    ]
    for prompt in prompts:
        print ("")
        print (f'prompt: "{prompt}"')
        stream = user_speech_service.stream(prompt)
        audio_processor.add_audio_stream(stream)

        print ("")
        print (f'response:')
        response = chat_service.respond_to(prompt)

    audio_processor.close()
    print ("Chat success")

async def run_pipeline():
    load_dotenv()

    try:
        chat_pipeline = ChatPipeline()
        await chat_pipeline.start()
        prompts = [
            "hello, how are you today?",
            "tell me about your shadow self?",
            "hmm, interesting, tell me more about that.",
            "wait, that is so interesting, what else?",
        ]
        for prompt in prompts:
            await chat_pipeline.enqueue(prompt)
        await chat_pipeline.wait_until_all_jobs_idle()
    except KeyboardInterrupt:
        print("Pipeline interrupted by user")
    except Exception as e:
        traceback.print_exc()
        print(f"An error occurred: {e}")

if __name__ == '__main__':
    # time_sentance_lenghts()
    # test_sentance_lenghts()
    # run_debug_code()
    asyncio.run(run_pipeline())