app.py

import json
import sys

import datasets
import gradio as gr
from apscheduler.schedulers.background import BackgroundScheduler
from loguru import logger

from app_configs import DEFAULT_SELECTIONS, THEME
from components.hf_pipelines import create_hf_pipeline_submission_interface
from components.quizbowl.bonus import BonusInterface
from components.quizbowl.tossup import TossupInterface
from components.typed_dicts import PipelineInterfaceDefaults, TossupInterfaceDefaults
from display.css_html_js import fonts_header, js_head, leaderboard_css
from display.custom_css import css_bonus, css_pipeline, css_tossup
from display.guide import BUILDING_MARKDOWN, QUICKSTART_MARKDOWN

# Constants
from envs import (
    API,
    COMPETITION_URL,
    CONTACT_EMAIL,
    DISCORD_URL,
    DOCS_REPO_URL,
    GITHUB_ISSUES_URL,
    LEADERBOARD_URL,
    LOCAL_REQUESTS_PATH,
    LOG_LEVEL,
    PLAYGROUND_DATASET,
    QUEUE_SYNC_INTERVAL,
    REGISTRATION_URL,
    REPO_ID,
    REQUESTS_REPO,
    SERVER_RESTART_INTERVAL,
)
from hf_datasets_utils import check_and_create_dataset_repo, download_dataset_snapshot
from shared.workflows import factory
from shared.workflows.configs import AVAILABLE_MODELS
from shared.workflows.llms import llm_cache


def restart_space():
    llm_cache.sync_to_hf()
    API.restart_space(repo_id=REPO_ID)


# Set the log level to INFO
logger.remove()
logger.add(sys.stdout, level=LOG_LEVEL, diagnose=False)


def filter_qids(qid: str, packet_ids: list[int]) -> bool:
    packet_id = int(qid.split("-")[-2])
    return packet_id in packet_ids


def load_dataset(config_name: str, max_questions: int = 10):
    ds = datasets.load_dataset(PLAYGROUND_DATASET, config_name, split="eval")

    return ds.filter(lambda x: filter_qids(x["qid"], [1])).select(range(max_questions))


def get_default_tab_id(request: gr.Request):
    logger.info(f"Request: {request}")
    tab_key_value = request.query_params.get("tab", "tossup")
    return gr.update(selected=tab_key_value)


def presave_pipeline_state(
    login_btn,
    browser_state: dict,
    tossup_pipeline_state: dict,
    tossup_output_state: dict,
    bonus_pipeline_state: dict,
    bonus_output_state: dict,
):
    browser_state.setdefault("tossup", {})
    browser_state["tossup"]["pipeline_state"] = tossup_pipeline_state
    browser_state["tossup"]["output_state"] = tossup_output_state
    browser_state.setdefault("bonus", {})
    browser_state["bonus"]["pipeline_state"] = bonus_pipeline_state
    browser_state["bonus"]["output_state"] = bonus_output_state
    logger.debug(
        f"Pipeline state before login. Login button: {login_btn}, browser state: {json.dumps(browser_state, indent=4)}"
    )
    return login_btn, browser_state


if __name__ == "__main__":
    check_and_create_dataset_repo(REQUESTS_REPO)
    download_dataset_snapshot(REQUESTS_REPO, LOCAL_REQUESTS_PATH)
    scheduler = BackgroundScheduler()
    scheduler.add_job(restart_space, "interval", seconds=SERVER_RESTART_INTERVAL)
    scheduler.start()

    css = css_pipeline + css_tossup + css_bonus + leaderboard_css
    head = fonts_header + js_head
    tossup_ds = load_dataset("tossup")
    logger.info(f"Tossup dataset size: {len(tossup_ds)}")
    bonus_ds = load_dataset("bonus")
    logger.info(f"Bonus dataset size: {len(bonus_ds)}")
    with gr.Blocks(
        css=css,
        head=head,
        theme=THEME,
        title="QANTA 2025 Quizbowl Competition",
    ) as demo:
        browser_state = gr.BrowserState(
            {
                "tossup": {"pipeline_state": None, "output_state": None},
                "bonus": {"pipeline_state": None, "output_state": None},
            }
        )
        with gr.Row():
            with gr.Column(scale=5):
                gr.Markdown(
                    f"## 🤖 Welcome to QANTA 2025 Quizbowl Arena!     👉 🏆 [Leaderboard]({LEADERBOARD_URL}) 👈"
                    "\n### 🎲 Create, play around, and submit your quizbowl agents."
                    f"<br>📋 [Register]({REGISTRATION_URL}) to participate in our [QANTA 2025 Human-AI Quizbowl Competition]({COMPETITION_URL}).",
                    elem_classes="welcome-text",
                )
            # leaderboard_btn = gr.Button("🏆 Leaderboard", elem_id="leaderboard-btn", link=LEADERBOARD_URL)
            login_btn = gr.LoginButton(scale=1)

        accent_color = "#B00000"
        style_text = f"color:{accent_color};font-weight:bold;"
        gr.Markdown(
            f"👀 **First time here?** Check out the [❓ Help](#help) tab for a quick introduction and "
            f"[QANTA25 Documentation]({DOCS_REPO_URL}) "
            "for detailed examples and tutorials on how to create and compete with your own QuizBowl agents. "
            f"<br>🤨 **Facing any issues?** Please contact us at [{CONTACT_EMAIL}](mailto:{CONTACT_EMAIL}), or raise an issue on our [GitHub repository]({GITHUB_ISSUES_URL})."
            f" &nbsp; 🧑🏻‍💻 **Engagements?** Join our [Discord server]({DISCORD_URL}) for discussions and getting help.",
            elem_classes="help-text",
        )
        with gr.Tabs() as gtab:
            with gr.Tab("🛎️ Tossup Agents", id="tossup"):
                defaults = TossupInterfaceDefaults(
                    **DEFAULT_SELECTIONS["tossup"], init_workflow=factory.create_simple_qb_tossup_workflow()
                )
                tossup_interface = TossupInterface(demo, browser_state, tossup_ds, AVAILABLE_MODELS, defaults)
            with gr.Tab("🙋🏻‍♂️ Bonus Round Agents", id="bonus"):
                defaults = PipelineInterfaceDefaults(
                    **DEFAULT_SELECTIONS["bonus"], init_workflow=factory.create_simple_qb_bonus_workflow()
                )
                bonus_interface = BonusInterface(demo, browser_state, bonus_ds, AVAILABLE_MODELS, defaults)
            with gr.Tab("🤗 HuggingFace Pipelines", elem_id="hf-pipeline-tab", id="hf-pipeline-tab"):
                hf_pipeline_interface = create_hf_pipeline_submission_interface(demo)
            with gr.Tab("❓ Help", id="help"):
                with gr.Row():
                    with gr.Column():
                        gr.Markdown(QUICKSTART_MARKDOWN)
                    with gr.Column():
                        gr.Markdown(BUILDING_MARKDOWN)

        # Event Listeners

        # This is used to retrieve the pipeline state user was working on before login.
        # makes things less annoying when progress is lost due to login.
        login_btn.click(
            fn=presave_pipeline_state,
            inputs=[
                login_btn,
                browser_state,
                tossup_interface.pipeline_state,
                tossup_interface.output_state,
                bonus_interface.pipeline_state,
                bonus_interface.output_state,
            ],
            outputs=[login_btn, browser_state],
        )

        demo.queue(default_concurrency_limit=40).launch()