Spaces:
Running
Running
File size: 1,877 Bytes
b72ab63 |
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 |
"""
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
|