File size: 3,124 Bytes
78cb487 |
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
import time
from pathlib import Path
from typing import List, Optional, Tuple, Callable
import numpy as np
from .audio_io import save_audio_file, play_audio
from .audio_queue import AudioGenerationQueue
def generate_and_play_sentences(
sentences: List[str],
generator,
speed: float = 1.0,
play_function: Callable = play_audio,
check_interrupt: Optional[Callable] = None,
output_dir: Optional[Path] = None,
sample_rate: Optional[int] = None,
) -> Tuple[bool, Optional[np.ndarray], List[Path]]:
"""
Generates and plays audio for each sentence with optional interruption checking.
Args:
sentences (List[str]): A list of sentences to generate audio for.
generator: The audio generator object.
speed (float, optional): The speed of audio generation. Defaults to 1.0.
play_function (Callable, optional): The function to use for playing audio. Defaults to play_audio.
check_interrupt (Callable, optional): An optional function to check for interruptions. Defaults to None.
output_dir (Path, optional): The directory to save generated audio files. Defaults to None.
sample_rate (int, optional): The sample rate of the audio. Defaults to None.
Returns:
Tuple[bool, Optional[np.ndarray], List[Path]]: A tuple containing:
- A boolean indicating if the process was interrupted.
- Optional audio data if the process was interrupted.
- A list of paths to the generated audio files.
"""
audio_queue = AudioGenerationQueue(generator, speed, output_dir)
audio_queue.start()
audio_queue.add_sentences(sentences)
audio_files = []
was_interrupted = False
interrupt_audio = None
try:
while True:
if check_interrupt:
interrupted, audio_data = check_interrupt()
if interrupted:
was_interrupted = True
interrupt_audio = audio_data
break
audio_data, output_path = audio_queue.get_next_audio()
if audio_data is not None:
if output_path:
audio_files.append(output_path)
if play_function:
try:
was_interrupted, interrupt_data = (
play_function(audio_data, sample_rate)
if sample_rate
else play_function(audio_data)
)
if was_interrupted:
interrupt_audio = interrupt_data
break
except Exception as e:
print(f"Error playing audio: {str(e)}")
continue
if audio_queue.sentence_queue.empty() and audio_queue.audio_queue.empty():
break
time.sleep(0.01)
except Exception as e:
print(f"Error in generate_and_play_sentences: {str(e)}")
finally:
audio_queue.stop()
return was_interrupted, interrupt_audio, audio_files
|