Spaces:
Sleeping
Sleeping
File size: 5,474 Bytes
9595e79 |
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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
import os
from typing import Literal, Optional, Dict, Any, Callable, Awaitable
import requests
from fastrtc import (
get_hf_turn_credentials,
get_twilio_turn_credentials,
get_cloudflare_turn_credentials,
get_cloudflare_turn_credentials_async
)
def get_rtc_credentials(
provider: Literal["hf", "twilio", "cloudflare", "hf-cloudflare"] = "hf-cloudflare",
**kwargs
) -> Dict[str, Any]:
"""
Get RTC configuration for different TURN server providers.
Args:
provider: The TURN server provider to use ('hf', 'twilio', 'cloudflare', or 'hf-cloudflare')
**kwargs: Additional arguments passed to the specific provider's function
Returns:
Dictionary containing the RTC configuration
"""
try:
if provider == "hf":
# HF Community Server (Deprecated)
# 1. Create a Hugging Face account at huggingface.co
# 2. Visit: https://huggingface.co/settings/tokens to create a token
# 3. Set HF_TOKEN environment variable or pass token directly
token = kwargs.pop("token", os.environ.get("HF_TOKEN"))
if not token:
raise ValueError("HF_TOKEN environment variable not set")
return get_hf_turn_credentials(token=token)
elif provider == "twilio":
# Twilio TURN Server
# 1. Create a free Twilio account at: https://login.twilio.com/u/signup
# 2. Get your Account SID and Auth Token from the Twilio Console
# 3. Set environment variables: TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN
account_sid = kwargs.pop("account_sid", os.environ.get("TWILIO_ACCOUNT_SID"))
auth_token = kwargs.pop("auth_token", os.environ.get("TWILIO_AUTH_TOKEN"))
if not account_sid or not auth_token:
raise ValueError("Twilio credentials not found. Set TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN env vars")
return get_twilio_turn_credentials(account_sid=account_sid, auth_token=auth_token)
elif provider == "cloudflare":
# Cloudflare TURN Server
# 1. Create a free Cloudflare account
# 2. Go to Cloudflare dashboard -> Calls section
# 3. Create a TURN App and get the Turn Token ID and API Token
# 4. Set environment variables: TURN_KEY_ID and TURN_KEY_API_TOKEN
key_id = kwargs.pop("key_id", os.environ.get("TURN_KEY_ID"))
api_token = kwargs.pop("api_token", os.environ.get("TURN_KEY_API_TOKEN"))
ttl = kwargs.pop("ttl", 86400)
if not key_id or not api_token:
raise ValueError("Cloudflare credentials not found. Set TURN_KEY_ID and TURN_KEY_API_TOKEN env vars")
return get_cloudflare_turn_credentials(key_id=key_id, api_token=api_token, ttl=ttl)
elif provider == "hf-cloudflare":
# Cloudflare with Hugging Face Token (10GB free traffic per month)
# 1. Create a Hugging Face account at huggingface.co
# 2. Visit: https://huggingface.co/settings/tokens to create a token
# 3. Set HF_TOKEN environment variable or pass token directly
hf_token = kwargs.pop("hf_token", os.environ.get("HF_TOKEN"))
ttl = kwargs.pop("ttl", 86400)
if not hf_token:
raise ValueError("HF_TOKEN environment variable not set")
return get_cloudflare_turn_credentials(hf_token=hf_token, ttl=ttl)
else:
raise ValueError(f"Unknown provider: {provider}")
except Exception as e:
raise Exception(f"Failed to get RTC credentials ({provider}): {str(e)}")
async def get_rtc_credentials_async(
provider: Literal["hf-cloudflare"] = "hf-cloudflare",
**kwargs
) -> Dict[str, Any]:
"""
Get RTC configuration asynchronously for different TURN server providers.
Args:
provider: Currently only supports 'hf-cloudflare'
**kwargs: Additional arguments passed to the specific provider's function
Returns:
Dictionary containing the RTC configuration
"""
if provider != "hf-cloudflare":
raise NotImplementedError(f"Async credentials for {provider} not implemented")
try:
# Cloudflare with Hugging Face Token (10GB free traffic per month)
hf_token = kwargs.pop("hf_token", os.environ.get("HF_TOKEN"))
ttl = kwargs.pop("ttl", 600) # Default 10 minutes for client-side
if not hf_token:
raise ValueError("HF_TOKEN environment variable not set")
return await get_cloudflare_turn_credentials_async(hf_token=hf_token, ttl=ttl)
except Exception as e:
raise Exception(f"Failed to get async RTC credentials: {str(e)}")
def get_credential_function(provider: str, is_async: bool = False) -> Callable:
"""
Get the appropriate credential function based on provider and whether async is needed.
Args:
provider: The TURN server provider
is_async: Whether to return an async function
Returns:
Function that returns credentials (async or sync)
"""
if is_async and provider == "hf-cloudflare":
async def get_creds():
return await get_rtc_credentials_async(provider=provider)
return get_creds
else:
def get_creds():
return get_rtc_credentials(provider=provider)
return get_creds |