app.py

"""Gradio deploy entrypoint for the customized FastChat UI."""

import os
from pathlib import Path
import traceback
from types import SimpleNamespace

import gradio as gr
from fastchat.serve.gradio_global_state import Context
from fastchat.serve.gradio_web_server import get_model_list
import fastchat.serve.gradio_web_server_multi as web_multi
from fastchat.serve.gradio_web_server_multi import (
    build_demo,
    set_global_vars,
    set_global_vars_anony,
    set_global_vars_named,
)


def _to_bool(value: str, default: bool) -> bool:
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _get_server_port():
    """Use explicit platform ports when provided; otherwise let Gradio pick locally."""
    env_port = os.getenv("GRADIO_SERVER_PORT") or os.getenv("PORT")
    if not env_port:
        return None
    return int(env_port)


def create_demo():
    repo_root = Path(__file__).resolve().parent
    register_file = os.getenv("FASTCHAT_REGISTER_API_FILE", "api_endpoint.json")
    register_path = repo_root / register_file
    register_api_file = str(register_path) if register_path.exists() else None
    server_port = _get_server_port()

    args = SimpleNamespace(
        host="0.0.0.0",
        port=server_port or 7860,
        share=False,
        controller_url=os.getenv("FASTCHAT_CONTROLLER_URL", ""),
        concurrency_count=int(os.getenv("FASTCHAT_CONCURRENCY", "10")),
        model_list_mode=os.getenv("FASTCHAT_MODEL_LIST_MODE", "once"),
        moderate=_to_bool(os.getenv("FASTCHAT_MODERATE"), False),
        show_terms_of_use=False,
        vision_arena=_to_bool(os.getenv("FASTCHAT_VISION_ARENA"), False),
        random_questions=None,
        register_api_endpoint_file=register_api_file,
        gradio_auth_path=None,
        elo_results_file=None,
        leaderboard_table_file=None,
        arena_hard_table=None,
        gradio_root_path=None,
        ga_id=None,
        use_remote_storage=False,
        password=None,
        show_visualizer=False,
        hide_arena_battle=_to_bool(os.getenv("FASTCHAT_HIDE_ARENA_BATTLE"), True),
    )

    print(
        "Space startup config:",
        {
            "controller_url": args.controller_url or None,
            "register_api_endpoint_file": args.register_api_endpoint_file,
            "model_list_mode": args.model_list_mode,
            "vision_arena": args.vision_arena,
            "hide_arena_battle": args.hide_arena_battle,
            "port": args.port,
        },
        flush=True,
    )

    # gradio_web_server_multi references a module-level `args`.
    web_multi.args = args

    set_global_vars(args.controller_url, args.moderate, args.use_remote_storage)
    set_global_vars_named(args.moderate)
    set_global_vars_anony(args.moderate)

    text_models, all_text_models = get_model_list(
        args.controller_url,
        args.register_api_endpoint_file,
        vision_arena=False,
    )
    vision_models, all_vision_models = get_model_list(
        args.controller_url,
        args.register_api_endpoint_file,
        vision_arena=True,
    )

    models = text_models + [
        model for model in vision_models if model not in text_models
    ]
    all_models = all_text_models + [
        model for model in all_vision_models if model not in all_text_models
    ]
    print(
        "Discovered models:",
        {
            "text_models": text_models,
            "vision_models": vision_models,
            "all_models": all_models,
        },
        flush=True,
    )
    context = Context(
        text_models,
        all_text_models,
        vision_models,
        all_vision_models,
        models,
        all_models,
    )

    demo = build_demo(
        context,
        args.elo_results_file,
        args.leaderboard_table_file,
        args.arena_hard_table,
    )
    return demo.queue(
        default_concurrency_limit=args.concurrency_count,
        status_update_rate=10,
        api_open=False,
    )


try:
    demo = create_demo()
except Exception:
    error_trace = traceback.format_exc()
    print("Space startup failed:\n", error_trace)
    with gr.Blocks(title="Oulution Demo Startup Error") as demo:
        gr.Markdown("## Oulution Demo failed to start")
        gr.Markdown(
            "The app encountered an initialization error. "
            "Check Space logs and fix configuration."
        )
        gr.Code(error_trace, language="python")


if __name__ == "__main__":
    launch_kwargs = {"server_name": "0.0.0.0"}
    server_port = _get_server_port()
    if server_port is not None:
        launch_kwargs["server_port"] = server_port
    demo.launch(**launch_kwargs)