Spaces:
Sleeping
Sleeping
File size: 5,034 Bytes
77da9e2 |
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 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
"""
Unified Entry Point - API Architecture
This file now uses a unified API-based architecture for all deployments.
Both local development and Hugging Face Spaces use the same API layer.
Architecture:
1. Starts API server in background (subprocess)
2. Starts Gradio UI that connects to the API
3. Everything goes through HTTP/REST
Benefits:
- Single code path to maintain
- Consistent behavior everywhere
- Easy to test and debug
- Proper separation of concerns
Usage:
python app.py
The script will automatically:
- Start the API server on http://localhost:8000
- Start the Gradio UI on http://localhost:7860
"""
import os
os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1'
import subprocess
import time
import sys
import signal
import requests
from functools import partial
# Use shared UI components
from ui.shared_interface import create_interface
from ui.detection_wrapper import detect_with_api
# Configuration
API_HOST = os.getenv("API_HOST", "0.0.0.0")
API_PORT = int(os.getenv("API_PORT", "8000"))
API_URL = f"http://localhost:{API_PORT}"
UI_HOST = os.getenv("GRADIO_SERVER_NAME", "0.0.0.0")
UI_PORT = int(os.getenv("GRADIO_SERVER_PORT", "7860"))
def start_api_server():
"""Start the API server in a subprocess"""
print("🚀 Starting API server...")
# Start API server as subprocess
api_process = subprocess.Popen(
[sys.executable, "app_api.py"],
env={**os.environ, "UVICORN_HOST": API_HOST, "UVICORN_PORT": str(API_PORT)},
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1
)
# Wait for API to be ready
max_wait = 60 # seconds
wait_interval = 0.5
elapsed = 0
print(f"⏳ Waiting for API server at {API_URL}...")
while elapsed < max_wait:
try:
response = requests.get(f"{API_URL}/health", timeout=2)
if response.status_code == 200:
print(f"✅ API server ready at {API_URL}")
return api_process
except requests.exceptions.RequestException:
pass
time.sleep(wait_interval)
elapsed += wait_interval
# Check if process died
if api_process.poll() is not None:
print("❌ API server failed to start!")
print("\nAPI server output:")
if api_process.stdout:
print(api_process.stdout.read())
sys.exit(1)
print(f"❌ API server did not start within {max_wait} seconds")
api_process.terminate()
sys.exit(1)
def main():
"""Main entry point - Unified API architecture"""
print("=" * 70)
print("🎯 CU-1 UI Element Detector - Unified API Mode")
print("=" * 70)
print("\n📡 Architecture: All traffic goes through API layer")
print(f" - API Server: {API_URL}")
print(f" - Gradio UI: http://localhost:{UI_PORT}")
print("\n🏗️ Benefits:")
print(" - Single code path (easier to maintain)")
print(" - Consistent behavior everywhere")
print(" - Proper microservices architecture")
print("=" * 70 + "\n")
# Start API server in background
api_process = start_api_server()
# Setup cleanup on exit
def cleanup(signum=None, frame=None):
print("\n\n🛑 Shutting down...")
if api_process and api_process.poll() is None:
print(" Stopping API server...")
api_process.terminate()
try:
api_process.wait(timeout=5)
except subprocess.TimeoutExpired:
api_process.kill()
print(" Goodbye! 👋")
sys.exit(0)
signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)
try:
# Create Gradio interface with API detection function
detection_fn = partial(detect_with_api, api_url=API_URL)
demo = create_interface(
detection_fn=detection_fn,
title_suffix="Unified API Mode",
show_api_info=True,
api_url=API_URL
)
print(f"\n🎨 Starting Gradio UI on http://localhost:{UI_PORT}...\n")
# Launch Gradio with automatic port fallback
try:
demo.queue().launch(
server_name=UI_HOST,
server_port=UI_PORT,
share=False
)
except OSError as e:
if "Cannot find empty port" in str(e):
print(f"⚠️ Port {UI_PORT} is busy, trying to find a free port...")
demo.queue().launch(
server_name=UI_HOST,
server_port=None, # Auto-select free port
share=False
)
else:
raise
except KeyboardInterrupt:
cleanup()
except Exception as e:
print(f"\n❌ Error: {e}")
cleanup()
finally:
cleanup()
if __name__ == "__main__":
main()
|