backend/debug_tts_stream.py

import asyncio
import time
import edge_tts
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("param_test")

async def test_streaming_library_behavior():
    text = "The quick brown fox jumps over the lazy dog. " * 10
    voice = "en-US-AriaNeural"
    rate = "+0%"
    pitch = "+0Hz"
    
    print(f"Testing direct library usage with text length: {len(text)}")
    print(f"Params: voice={voice}, rate={rate}, pitch={pitch}")
    communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch)
    
    start_time = time.time()
    first_byte_time = None
    chunks = 0
    total_bytes = 0
    
    print("Starting stream...")
    async for chunk in communicate.stream():
        if chunk["type"] == "audio":
            if first_byte_time is None:
                first_byte_time = time.time()
                print(f"FIRST BYTE received after: {first_byte_time - start_time:.4f}s")
            chunks += 1
            total_bytes += len(chunk["data"])
            # print(f"Chunk {chunks}: {len(chunk['data'])} bytes")
            
    total_time = time.time() - start_time
    print(f"Total time: {total_time:.4f}s")
    print(f"Total bytes: {total_bytes}")
    print(f"Chunks: {chunks}")

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