from __future__ import annotations import logging import os import threading import time import webbrowser from pathlib import Path from typing import Any import uvicorn from backend import server_runtime logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", ) logger = logging.getLogger("aiforecast-launcher") DEFAULT_HOST: str = "127.0.0.1" SERVER_READY_TIMEOUT_SECONDS: float = 15.0 SERVER_READY_POLL_INTERVAL_SECONDS: float = 0.25 PROJECT_ROOT: Path = Path(__file__).resolve().parent.parent ENV_FILE: Path = PROJECT_ROOT / ".env" CONSOLE_MODE_ENV: str = "AIFORECAST_CONSOLE_MODE" def load_runtime_env() -> None: """Load local runtime settings before resolving the server port.""" server_runtime.load_runtime_env(ENV_FILE, override=False) load_runtime_env() BANNER = r""" _ _____ _____ _ / \ |_ _\ \ / / ___|__ _ __ ___ ___ __ _| |_ / _ \ | | \ V /| |_ / _ \| '__/ _ \/ __/ _` | __| / ___ \ | | | | | _| (_) | | | __/ (_| (_| | |_ /_/ \_\___| |_| |_| \___/|_| \___|\___\__,_|\__| """ def print_banner() -> None: """Render a clean startup banner in the local console.""" os.system("cls" if os.name == "nt" else "clear") print("\033[96m" + BANNER + "\033[0m") print(" [*] Dang khoi dong SuperAI Forecast...") print(" [*] Dang khoi tao he thong du bao Kronos / TimesFM / Chronos...") print(" [*] Trinh duyet se tu dong mo khi server san sang.") print() def apply_console_window_mode() -> None: """Minimize or hide this console window on Windows without affecting logs.""" if os.name != "nt": return mode = os.getenv(CONSOLE_MODE_ENV, "minimize").strip().lower() if mode in {"", "show", "visible", "keep"}: return try: import ctypes kernel32 = ctypes.windll.kernel32 user32 = ctypes.windll.user32 console_window = kernel32.GetConsoleWindow() if not console_window: return show_code = 0 if mode == "hide" else 6 user32.ShowWindow(console_window, show_code) except Exception as exc: logger.debug("Khong the doi che do cua so console: %s", exc) def load_app() -> Any: """Import the FastAPI app after runtime settings are in place.""" return server_runtime.load_fastapi_app(PROJECT_ROOT) def resolve_server_port() -> int: """Use PORT when available and free, otherwise choose a free local port.""" return server_runtime.resolve_server_port(DEFAULT_HOST, logger=logger) def wait_for_server_start(server: uvicorn.Server, startup_finished: threading.Event) -> bool: """Wait until this Uvicorn instance reports a successful startup.""" deadline = time.monotonic() + SERVER_READY_TIMEOUT_SECONDS while time.monotonic() < deadline: if server.started: return True if startup_finished.is_set(): return False time.sleep(SERVER_READY_POLL_INTERVAL_SECONDS) return False def open_browser(url: str, server: uvicorn.Server, startup_finished: threading.Event) -> None: """Open the dashboard only after this backend instance is ready.""" if not wait_for_server_start(server, startup_finished): logger.warning("Bo qua mo trinh duyet vi backend khong san sang tai %s", url) return logger.info("Dang mo bang dieu khien tai %s", url) webbrowser.open(url) if __name__ == "__main__": print_banner() apply_console_window_mode() host = DEFAULT_HOST port = resolve_server_port() url = f"http://{host}:{port}" app = load_app() config = uvicorn.Config(app, host=host, port=port, log_level="warning") server = uvicorn.Server(config) startup_finished = threading.Event() logger.info("Khoi dong SuperAI Forecast Backend tren %s", url) threading.Thread( target=open_browser, args=(url, server, startup_finished), daemon=True, ).start() try: server.run() except SystemExit as exc: if exc.code not in (0, None): logger.error("Loi khoi dong he thong tren %s (exit=%s)", url, exc.code) raise except Exception as exc: logger.exception("Loi khoi dong he thong: %s", exc) raise finally: startup_finished.set()