"""
Defines helper methods useful for setting up ports, launching servers, and
creating tunnels.
"""
from __future__ import annotations
import os
import socket
import threading
import time
import warnings
from typing import TYPE_CHECKING, Tuple
import requests
import uvicorn
from gradio.routes import App
from gradio.tunneling import Tunnel
if TYPE_CHECKING: # Only import for type checking (to avoid circular imports).
from gradio.blocks import Blocks
# By default, the local server will try to open on localhost, port 7860.
# If that is not available, then it will try 7861, 7862, ... 7959.
INITIAL_PORT_VALUE = int(os.getenv("GRADIO_SERVER_PORT", "7860"))
TRY_NUM_PORTS = int(os.getenv("GRADIO_NUM_PORTS", "100"))
LOCALHOST_NAME = os.getenv("GRADIO_SERVER_NAME", "127.0.0.1")
GRADIO_API_SERVER = "https://api.gradio.app/v2/tunnel-request"
class Server(uvicorn.Server):
def install_signal_handlers(self):
pass
def run_in_thread(self):
self.thread = threading.Thread(target=self.run, daemon=True)
self.thread.start()
while not self.started:
time.sleep(1e-3)
def close(self):
self.should_exit = True
self.thread.join()
def get_first_available_port(initial: int, final: int) -> int:
"""
Gets the first open port in a specified range of port numbers
Parameters:
initial: the initial value in the range of port numbers
final: final (exclusive) value in the range of port numbers, should be greater than `initial`
Returns:
port: the first open port in the range
"""
for port in range(initial, final):
try:
s = socket.socket() # create a socket object
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((LOCALHOST_NAME, port)) # Bind to the port
s.close()
return port
except OSError:
pass
raise OSError(
"All ports from {} to {} are in use. Please close a port.".format(
initial, final - 1
)
)
def configure_app(app: App, blocks: Blocks) -> App:
auth = blocks.auth
if auth is not None:
if not callable(auth):
app.auth = {account[0]: account[1] for account in auth}
else:
app.auth = auth
else:
app.auth = None
app.blocks = blocks
app.cwd = os.getcwd()
app.favicon_path = blocks.favicon_path
app.tokens = {}
return app
def start_server(
blocks: Blocks,
server_name: str | None = None,
server_port: int | None = None,
ssl_keyfile: str | None = None,
ssl_certfile: str | None = None,
ssl_keyfile_password: str | None = None,
) -> Tuple[str, int, str, App, Server]:
"""Launches a local server running the provided Interface
Parameters:
blocks: The Blocks object to run on the server
server_name: to make app accessible on local network, set this to "0.0.0.0". Can be set by environment variable GRADIO_SERVER_NAME.
server_port: will start gradio app on this port (if available). Can be set by environment variable GRADIO_SERVER_PORT.
auth: If provided, username and password (or list of username-password tuples) required to access the Blocks. Can also provide function that takes username and password and returns True if valid login.
ssl_keyfile: If a path to a file is provided, will use this as the private key file to create a local server running on https.
ssl_certfile: If a path to a file is provided, will use this as the signed certificate for https. Needs to be provided if ssl_keyfile is provided.
ssl_keyfile_password: If a password is provided, will use this with the ssl certificate for https.
Returns:
port: the port number the server is running on
path_to_local_server: the complete address that the local server can be accessed at
app: the FastAPI app object
server: the server object that is a subclass of uvicorn.Server (used to close the server)
"""
server_name = server_name or LOCALHOST_NAME
# if port is not specified, search for first available port
if server_port is None:
port = get_first_available_port(
INITIAL_PORT_VALUE, INITIAL_PORT_VALUE + TRY_NUM_PORTS
)
else:
try:
s = socket.socket()
s.bind((LOCALHOST_NAME, server_port))
s.close()
except OSError:
raise OSError(
"Port {} is in use. If a gradio.Blocks is running on the port, you can close() it or gradio.close_all().".format(
server_port
)
)
port = server_port
url_host_name = "localhost" if server_name == "0.0.0.0" else server_name
if ssl_keyfile is not None:
if ssl_certfile is None:
raise ValueError(
"ssl_certfile must be provided if ssl_keyfile is provided."
)
path_to_local_server = "https://{}:{}/".format(url_host_name, port)
else:
path_to_local_server = "http://{}:{}/".format(url_host_name, port)
app = App.create_app(blocks)
if blocks.save_to is not None: # Used for selenium tests
blocks.save_to["port"] = port
config = uvicorn.Config(
app=app,
port=port,
host=server_name,
log_level="warning",
ssl_keyfile=ssl_keyfile,
ssl_certfile=ssl_certfile,
ssl_keyfile_password=ssl_keyfile_password,
ws_max_size=1024 * 1024 * 1024, # Setting max websocket size to be 1 GB
)
server = Server(config=config)
server.run_in_thread()
return server_name, port, path_to_local_server, app, server
def setup_tunnel(local_host: str, local_port: int) -> str:
response = requests.get(GRADIO_API_SERVER)
if response and response.status_code == 200:
try:
payload = response.json()[0]
remote_host, remote_port = payload["host"], int(payload["port"])
tunnel = Tunnel(remote_host, remote_port, local_host, local_port)
address = tunnel.start_tunnel()
return address
except Exception as e:
raise RuntimeError(str(e))
else:
raise RuntimeError("Could not get share link from Gradio API Server.")
def url_ok(url: str) -> bool:
try:
for _ in range(5):
with warnings.catch_warnings():
warnings.filterwarnings("ignore")
r = requests.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, requests.exceptions.ConnectionError):
return False
return False