Spaces:
Sleeping
Sleeping
File size: 5,474 Bytes
5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea 5ef2360 97f18ea |
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 |