Spaces:
Running
Running
File size: 3,519 Bytes
c2ea5ed 96a9109 c2ea5ed |
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 |
"""
FastAPI Application Server Entry Point
This module provides a server entry point for the FastAPI application defined in app.py.
"""
import logging
import os
import sys
import argparse
import webbrowser
import socket
import uvicorn
# Import from backend modules
from backend.server_config import PORT, HOST, MAX_PORT_ATTEMPTS, LOG_LEVEL, ensure_directories
from backend.app import app # Import the FastAPI app from app.py
# Setup logging
logging.basicConfig(
level=getattr(logging, LOG_LEVEL),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("agent_monitoring_server")
def find_available_port(start_port: int, max_attempts: int) -> int:
"""Find an available port starting from the given port"""
for port_offset in range(max_attempts):
port = start_port + port_offset
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind((HOST, port))
logger.info(f"Found available port: {port}")
return port
except OSError:
logger.debug(f"Port {port} is not available")
continue
# If we reach here, try some common alternative ports
alternative_ports = [8080, 8000, 3000, 9000, 5000]
for port in alternative_ports:
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind((HOST, port))
logger.info(f"Found available alternative port: {port}")
return port
except OSError:
continue
# If still no port found, raise an error
raise RuntimeError(
f"Could not find an available port after trying {max_attempts} "
f"sequential ports and {len(alternative_ports)} alternatives"
)
def parse_arguments():
"""Parse command line arguments"""
parser = argparse.ArgumentParser(description="Agent Monitoring System Server")
parser.add_argument("--port", type=int, default=PORT, help=f"Server port (default: {PORT})")
parser.add_argument("--host", default=HOST, help=f"Server host (default: {HOST})")
parser.add_argument("--log-level", default=LOG_LEVEL, help=f"Log level (default: {LOG_LEVEL})")
parser.add_argument("--no-open-browser", action="store_true", help="Do not open browser automatically")
return parser.parse_args()
def main():
"""Main entry point for running the server"""
# Parse command line arguments
args = parse_arguments()
# Ensure required directories exist
ensure_directories()
# Find an available port
try:
port = find_available_port(args.port, MAX_PORT_ATTEMPTS)
except RuntimeError as e:
logger.error(str(e))
sys.exit(1)
# Log startup message
logger.info(f"Starting Agent Monitoring System server on {args.host}:{port}")
# Open browser if not disabled
if not args.no_open_browser:
webbrowser.open(f"http://{args.host}:{port}")
# Start the server
try:
uvicorn.run(
"backend.app:app", # Update to use app.py instead of main.py
host=args.host,
port=port,
log_level=args.log_level.lower(),
reload=False
)
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Error running server: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
main() |