mirror of
https://github.com/Nighthawk42/Qwen3-TTS-streaming.git
synced 2026-08-30 09:42:26 +00:00
141 lines
3.6 KiB
Python
141 lines
3.6 KiB
Python
"""
|
|
Utility functions for audio processing and API operations.
|
|
"""
|
|
|
|
import base64
|
|
import io
|
|
import os
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Tuple, Optional
|
|
|
|
import numpy as np
|
|
import soundfile as sf
|
|
|
|
|
|
def generate_request_id() -> str:
|
|
"""Generate a unique request ID."""
|
|
return f"req_{uuid.uuid4().hex[:12]}"
|
|
|
|
|
|
def get_unix_timestamp() -> int:
|
|
"""Get current Unix timestamp."""
|
|
return int(datetime.utcnow().timestamp())
|
|
|
|
|
|
def audio_to_base64(audio: np.ndarray, sample_rate: int) -> str:
|
|
"""
|
|
Convert numpy audio array to base64-encoded WAV string.
|
|
|
|
Args:
|
|
audio: Audio waveform as numpy array
|
|
sample_rate: Sample rate in Hz
|
|
|
|
Returns:
|
|
Base64-encoded audio data
|
|
"""
|
|
with io.BytesIO() as wav_buffer:
|
|
sf.write(wav_buffer, audio, sample_rate, format='WAV')
|
|
wav_bytes = wav_buffer.getvalue()
|
|
|
|
return base64.b64encode(wav_bytes).decode('utf-8')
|
|
|
|
|
|
def base64_to_audio(audio_base64: str) -> Tuple[np.ndarray, int]:
|
|
"""
|
|
Convert base64-encoded audio to numpy array.
|
|
|
|
Args:
|
|
audio_base64: Base64-encoded audio data
|
|
|
|
Returns:
|
|
Tuple of (audio waveform, sample_rate)
|
|
"""
|
|
wav_bytes = base64.b64decode(audio_base64)
|
|
audio, sr = sf.read(io.BytesIO(wav_bytes), dtype='float32')
|
|
return audio, int(sr)
|
|
|
|
|
|
def get_audio_duration(audio: np.ndarray, sample_rate: int) -> float:
|
|
"""
|
|
Calculate audio duration in seconds.
|
|
|
|
Args:
|
|
audio: Audio waveform as numpy array
|
|
sample_rate: Sample rate in Hz
|
|
|
|
Returns:
|
|
Duration in seconds
|
|
"""
|
|
return float(len(audio) / sample_rate)
|
|
|
|
|
|
def load_voice_clone_files(voice_clone_dir: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""
|
|
Load ref_audio.wav and ref_text.txt from voice cloning directory.
|
|
|
|
Args:
|
|
voice_clone_dir: Path to voice cloning directory
|
|
|
|
Returns:
|
|
Tuple of (ref_audio_path, ref_text) or (None, None) if files not found
|
|
"""
|
|
voice_clone_path = Path(voice_clone_dir)
|
|
|
|
ref_audio_path = voice_clone_path / "ref_audio.wav"
|
|
ref_text_path = voice_clone_path / "ref_text.txt"
|
|
|
|
if not ref_audio_path.exists() or not ref_text_path.exists():
|
|
return None, None
|
|
|
|
with open(ref_text_path, 'r', encoding='utf-8') as f:
|
|
ref_text = f.read().strip()
|
|
|
|
return str(ref_audio_path), ref_text
|
|
|
|
|
|
def validate_audio_file(file_path: str) -> Tuple[bool, str]:
|
|
"""
|
|
Validate that the audio file exists and is readable.
|
|
|
|
Args:
|
|
file_path: Path to audio file
|
|
|
|
Returns:
|
|
Tuple of (is_valid, message)
|
|
"""
|
|
path = Path(file_path)
|
|
|
|
if not path.exists():
|
|
return False, f"File not found: {file_path}"
|
|
|
|
if not path.is_file():
|
|
return False, f"Path is not a file: {file_path}"
|
|
|
|
if path.suffix.lower() not in ['.wav', '.mp3', '.flac', '.ogg']:
|
|
return False, f"Unsupported audio format: {path.suffix}"
|
|
|
|
try:
|
|
sf.read(file_path, frames=1)
|
|
return True, "Valid audio file"
|
|
except Exception as e:
|
|
return False, f"Error reading audio file: {str(e)}"
|
|
|
|
|
|
def concatenate_audio_chunks(chunks: list, sample_rate: int) -> Tuple[np.ndarray, int]:
|
|
"""
|
|
Concatenate multiple audio chunks into a single array.
|
|
|
|
Args:
|
|
chunks: List of audio chunks as numpy arrays
|
|
sample_rate: Sample rate in Hz
|
|
|
|
Returns:
|
|
Tuple of (concatenated audio, sample_rate)
|
|
"""
|
|
if not chunks:
|
|
return np.array([]), sample_rate
|
|
|
|
return np.concatenate(chunks), sample_rate
|