poemsforaphrodite's picture
Upload folder using huggingface_hub
b72ab63 verified
raw
history blame
1.88 kB
"""
Defines helper methods useful for setting up ports, launching servers, and
creating tunnels.
"""
from __future__ import annotations
import os
import time
import warnings
import httpx
from gradio.routes import App # HACK: to avoid circular import # noqa: F401
from gradio.tunneling import Tunnel
GRADIO_API_SERVER = "https://api.gradio.app/v2/tunnel-request"
GRADIO_SHARE_SERVER_ADDRESS = os.getenv("GRADIO_SHARE_SERVER_ADDRESS")
def setup_tunnel(
local_host: str, local_port: int, share_token: str, share_server_address: str | None
) -> str:
share_server_address = (
GRADIO_SHARE_SERVER_ADDRESS
if share_server_address is None
else share_server_address
)
if share_server_address is None:
try:
response = httpx.get(GRADIO_API_SERVER, timeout=30)
payload = response.json()[0]
remote_host, remote_port = payload["host"], int(payload["port"])
except Exception as e:
raise RuntimeError(
"Could not get share link from Gradio API Server."
) from e
else:
remote_host, remote_port = share_server_address.split(":")
remote_port = int(remote_port)
try:
tunnel = Tunnel(remote_host, remote_port, local_host, local_port, share_token)
address = tunnel.start_tunnel()
return address
except Exception as e:
raise RuntimeError(str(e)) from e
def url_ok(url: str) -> bool:
try:
for _ in range(5):
with warnings.catch_warnings():
warnings.filterwarnings("ignore")
r = httpx.head(url, timeout=3, verify=False)
if r.status_code in (200, 401, 302): # 401 or 302 if auth is set
return True
time.sleep(0.500)
except (ConnectionError, httpx.ConnectError):
return False
return False