Files
video_upscaler/start.py
T

157 lines
5.4 KiB
Python
Executable File

#!/usr/bin/env python3
"""
start.py — Start the AI Video Upscaler web application.
Usage:
python start.py # Start on default port 8000 (binds to 0.0.0.0 for LAN access)
python start.py --port 9000 # Start on a custom port
python start.py --host 127.0.0.1 # Bind to loopback only (no LAN access)
"""
import argparse
import os
import signal
import socket
import subprocess
import sys
import time
import webbrowser
# ── Paths ────────────────────────────────────────────────────────────────────
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PID_FILE = os.path.join(BASE_DIR, ".uvicorn.pid")
LOG_FILE = os.path.join(BASE_DIR, "server.log")
# Detect the right Python interpreter:
# 1. If running inside an activated venv, use that interpreter
# 2. Otherwise look for venv/ then .venv/ in the project
if sys.prefix != sys.base_prefix:
# Already inside an activated venv — use it
VENV_PYTHON = sys.executable
else:
VENV_PYTHON = os.path.join(BASE_DIR, "venv", "bin", "python")
if not os.path.isfile(VENV_PYTHON):
VENV_PYTHON = os.path.join(BASE_DIR, ".venv", "bin", "python")
if not os.path.isfile(VENV_PYTHON):
VENV_PYTHON = sys.executable # last resort: system python
def _is_running(pid: int) -> bool:
"""Check whether a process with the given PID is still alive and is our server."""
try:
os.kill(pid, 0)
except OSError:
return False
# Check /proc/{pid}/cmdline to verify it's our server process
try:
cmdline_path = f"/proc/{pid}/cmdline"
if os.path.exists(cmdline_path):
with open(cmdline_path, "r") as f:
cmdline = f.read()
# Check for indicators of our server
return "uvicorn" in cmdline or "app.main" in cmdline
except Exception:
pass
# If we can't read cmdline, default to True (safer fallback)
return True
def _read_pid() -> int | None:
"""Read and validate the PID from the pidfile."""
if not os.path.isfile(PID_FILE):
return None
try:
pid = int(open(PID_FILE).read().strip())
return pid if _is_running(pid) else None
except (ValueError, OSError):
return None
def _get_local_ip() -> str:
"""Get the primary local network IP address."""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
# Dummy connection to google DNS to get local route IP
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
except Exception:
try:
ip = socket.gethostbyname(socket.gethostname())
except Exception:
ip = "127.0.0.1"
finally:
s.close()
return ip
def start(host: str, port: int, open_browser: bool = True) -> None:
existing = _read_pid()
if existing:
print(f"⚠ Server is already running (PID {existing}).")
if host == "0.0.0.0":
local_ip = _get_local_ip()
print(f" → Local: http://localhost:{port}")
if local_ip and local_ip != "127.0.0.1":
print(f" → Network: http://{local_ip}:{port}")
else:
print(f" → http://{host}:{port}")
return
print(f"🚀 Starting AI Video Upscaler …")
print(f" Host : {host}")
print(f" Port : {port}")
print(f" Log : {LOG_FILE}")
log_fh = open(LOG_FILE, "a")
proc = subprocess.Popen(
[
VENV_PYTHON, "-m", "uvicorn",
"app.main:app",
"--host", host,
"--port", str(port),
],
cwd=BASE_DIR,
stdout=log_fh,
stderr=log_fh,
start_new_session=True, # detach from terminal
)
# Write PID
with open(PID_FILE, "w") as f:
f.write(str(proc.pid))
# Wait briefly to make sure it didn't crash immediately
time.sleep(1.5)
if proc.poll() is not None:
print("❌ Server failed to start. Check server.log for details.")
os.remove(PID_FILE)
sys.exit(1)
url = f"http://{'localhost' if host == '0.0.0.0' else host}:{port}"
print(f"✅ Server started (PID {proc.pid})")
if host == "0.0.0.0":
local_ip = _get_local_ip()
print(f" → Local: http://localhost:{port}")
if local_ip and local_ip != "127.0.0.1":
print(f" → Network: http://{local_ip}:{port}")
else:
print(f" → {url}")
if open_browser:
webbrowser.open(url)
# ── CLI ──────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Start the AI Video Upscaler")
parser.add_argument("--host", default="0.0.0.0",
help="Bind address (default: 0.0.0.0 to bind all interfaces, use 127.0.0.1 for local loopback only)")
parser.add_argument("--port", type=int, default=8000,
help="Port to listen on (default: 8000)")
parser.add_argument("--no-browser", action="store_true",
help="Don't open the browser automatically")
args = parser.parse_args()
start(args.host, args.port, open_browser=not args.no_browser)