Spaces:
Runtime error
Runtime error
File size: 4,293 Bytes
1354c32 6425857 1354c32 e9c465b 1354c32 e9c465b 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 97 98 99 100 101 102 103 104 105 106 107 108 109 | import numpy as np
import librosa
import io
import os
import warnings
import tempfile
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:
try:
# Create a temporary file to store the credentials
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix=".json") as temp_file:
temp_file.write(gcp_service_account_json_str)
gcp_credential_path = temp_file.name # Get the path to the temporary file
# Set the environment variable that Google Cloud libraries expect
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = gcp_credential_path
print(f"Google Cloud credentials set from secret to: {gcp_credential_path}")
except Exception as e:
print(f"Error setting up Google Cloud credentials: {e}")
else:
print("Warning: GCP_SERVICE_ACCOUNT_JSON secret not found. Google Cloud services may fail.")
# 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() != ""
)
|