|
""" |
|
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: |
|
from gradio.blocks import Blocks |
|
|
|
|
|
|
|
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() |
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
|
s.bind((LOCALHOST_NAME, 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 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: |
|
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, |
|
) |
|
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, share_token: str) -> 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, share_token |
|
) |
|
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): |
|
return True |
|
time.sleep(0.500) |
|
except (ConnectionError, requests.exceptions.ConnectionError): |
|
return False |
|
return False |
|
|
