File size: 4,024 Bytes
4412065 | 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 125 126 127 128 129 130 | """Gradio app runner — launches Gradio apps as subprocess servers.
Manages the lifecycle of Gradio app processes: start, status check, and stop.
"""
from __future__ import annotations
import logging
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# ─── Registry for running Gradio subprocesses ───────────────────────────
_running_gradio_procs: dict[str, subprocess.Popen] = {}
def run_gradio_app(code: str, port: int = 7861) -> dict[str, Any]:
"""Launch a Gradio app as a subprocess and return its URL.
The Gradio app is run on the specified port. We modify the code
to ensure it launches on the correct port and is accessible.
"""
# Kill any previously running Gradio app
_stop_all_procs()
# Patch the code: ensure launch uses correct server_name and server_port
patched_code = code
# Replace .launch() with correct params
patched_code = re.sub(
r"(\w+)\.launch\([^)]*\)",
f'\\1.launch(server_name="0.0.0.0", server_port={port}, share=False)',
patched_code,
)
# If no .launch() found, add one
if ".launch(" not in patched_code:
patched_code += (
f'\n\nif __name__ == "__main__":\n'
f' iface.launch(server_name="0.0.0.0", server_port={port}, share=False)\n'
)
with tempfile.TemporaryDirectory(prefix="gradio_app_") as tmp:
app_path = Path(tmp) / "gradio_app.py"
app_path.write_text(patched_code, encoding="utf-8")
env = {
**os.environ,
"PYTHONUNBUFFERED": "1",
"GRADIO_SERVER_NAME": "0.0.0.0",
"GRADIO_SERVER_PORT": str(port),
}
try:
proc = subprocess.Popen(
[sys.executable, str(app_path)],
cwd=tmp,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
proc_id = f"gradio_{port}"
_running_gradio_procs[proc_id] = proc
# Wait a bit for the server to start
import time as _time
_time.sleep(3)
# Check if process is still running
poll = proc.poll()
if poll is not None:
stdout = proc.stdout.read() if proc.stdout else ""
stderr = proc.stderr.read() if proc.stderr else ""
return {
"success": False,
"url": "",
"message": f"Gradio app exited with code {poll}",
"stdout": stdout[-2000:] if stdout else "",
"stderr": stderr[-2000:] if stderr else "",
}
gradio_url = f"http://localhost:{port}"
return {
"success": True,
"url": gradio_url,
"message": f"Gradio app running at {gradio_url}",
"port": port,
}
except Exception as exc:
logger.exception("Failed to launch Gradio app")
return {
"success": False,
"url": "",
"message": f"Failed to launch: {exc}",
}
def stop_gradio_app() -> dict[str, Any]:
"""Stop any running Gradio app subprocess."""
stopped = _stop_all_procs()
return {"success": True, "message": f"Stopped {stopped} Gradio app(s)"}
def _stop_all_procs() -> int:
"""Stop all running Gradio processes. Returns count of stopped procs."""
stopped = 0
for pid, proc in list(_running_gradio_procs.items()):
try:
proc.terminate()
proc.wait(timeout=3)
stopped += 1
except Exception:
try:
proc.kill()
stopped += 1
except Exception:
pass
_running_gradio_procs.clear()
return stopped
|