Spaces:
Runtime error
Runtime error
File size: 3,447 Bytes
1354c32 | 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 85 86 87 88 89 90 91 92 93 94 95 96 | import numpy as np
import librosa
import io
import os
import warnings
from pydub import AudioSegment
from dotenv import load_dotenv
from fastrtc import get_cloudflare_turn_credentials_async, get_cloudflare_turn_credentials
try:
import torch
except ModuleNotFoundError:
torch = None # type: ignore
warnings.filterwarnings("ignore")
# load_dotenv(override = True)
# --- Device Configuration ---
def get_device():
"""Gets the best available device for PyTorch."""
if torch is None:
return "cpu"
if torch.cuda.is_available():
return "cuda"
elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
return "mps"
else:
return "cpu"
device = get_device()
print(f"Using device: {device}")
# --- Cloud Credentials ---
async def get_async_credentials():
"""Asynchronously fetches Cloudflare TURN credentials."""
return await get_cloudflare_turn_credentials_async(hf_token=os.getenv('HF_TOKEN'))
def get_sync_credentials(ttl=360_000):
"""Synchronously fetches Cloudflare TURN credentials."""
return get_cloudflare_turn_credentials(ttl=ttl)
def setup_gcp_credentials():
"""Sets up Google Cloud credentials from an environment variable."""
gcp_service_account_json_str = os.getenv("GCP_SERVICE_ACCOUNT_JSON")
if gcp_service_account_json_str:
print("GCP service account JSON loaded from environment variable.")
else:
print("Warning: GCP_SERVICE_ACCOUNT_JSON is not set; Google Cloud clients may fail.")
return gcp_service_account_json_str
# --- Audio Processing ---
def audiosegment_to_numpy(audio, target_sample_rate=16000):
samples = np.array(audio.get_array_of_samples(), dtype=np.float32)
if audio.channels > 1:
samples = samples.reshape((-1, audio.channels)).mean(axis=1)
samples /= np.iinfo(audio.array_type).max
if audio.frame_rate != target_sample_rate:
samples = librosa.resample(samples, orig_sr=audio.frame_rate, target_sr=target_sample_rate)
return samples
def preprocess_audio(audio, target_channels=1, target_frame_rate=16000):
"""
Preprocess the audio using pydub AudioSegment by setting the number of channels and frame rate.
Args:
audio (tuple): A tuple (sample_rate, audio_array) where audio_array is a NumPy array.
target_channels (int): Desired number of channels (default is 1 for mono).
target_frame_rate (int): Desired frame rate (default is 16000 Hz).
Returns:
np.ndarray: The processed audio as a NumPy array.
"""
sample_rate, audio_array = audio
target_frame_rate = sample_rate
audio_array_int16 = audio_array.astype(np.int16)
audio_bytes = audio_array_int16.tobytes()
audio_io = io.BytesIO(audio_bytes)
segment = AudioSegment.from_raw(audio_io, sample_width=2, frame_rate=sample_rate, channels=1)
segment = segment.set_channels(target_channels)
segment = segment.set_frame_rate(target_frame_rate)
return audiosegment_to_numpy(segment)
# --- Conversation Utilities ---
def is_valid_turn(turn):
"""Return True if turn is a valid dict with non-empty 'role' and 'content' strings."""
return (
isinstance(turn, dict)
and "role" in turn
and "content" in turn
and isinstance(turn["role"], str)
and isinstance(turn["content"], str)
and turn["role"].strip() != ""
and turn["content"].strip() != ""
)
|