File size: 1,952 Bytes
5306da4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import wave
import io
from providers.chunk_provider import chunk_text
from kokoro import generate, generate_full
import numpy as np
import asyncio
from concurrent.futures import ThreadPoolExecutor
tts_pool = ThreadPoolExecutor(max_workers=10)
def save_audio_to_file(audio_data, file_number, sample_rate=24000):
filename = f"output-{file_number}.wav"
with wave.open(filename, 'wb') as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(sample_rate)
audio_int16 = (audio_data * 32767).astype(np.int16)
wav_file.writeframes(audio_int16.tobytes())
return filename
def get_audio_bytes(audio_data, sample_rate=24000):
wav_bytes = io.BytesIO()
with wave.open(wav_bytes, 'wb') as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(sample_rate)
audio_int16 = (audio_data * 32767).astype(np.int16)
wav_file.writeframes(audio_int16.tobytes())
wav_bytes.seek(0)
return wav_bytes.read()
def dummy_bytes():
buffer = io.BytesIO()
dummy_data = b"This is a test of dummy byte data."
buffer.write(dummy_data)
buffer.seek(0)
byte_value = buffer.getvalue()
return byte_value
def generate_audio_from_chunks(text, model, voicepack, voice_name):
chunks = chunk_text(text)
combined_audio = np.array([])
for chunk in chunks:
try:
audio, _ = generate(model, chunk, voicepack, lang=voice_name[0])
combined_audio = np.concatenate(
[combined_audio, audio]) if combined_audio.size > 0 else audio
except Exception:
pass
return combined_audio
async def generate_audio_stream(text, model, vp, name):
for chunk in chunk_text(text):
audio, _ = await asyncio.get_running_loop().run_in_executor(
tts_pool, generate_full, model, chunk, vp, name[0]
)
yield audio
|