src/submission/submit.py

import gradio as gr

from src.about import Tasks
from src.display.formatting import styled_error, styled_message
from src.display.i18n import t
from src.leaderboard.store import save_submission
from src.leaderboard.sync import sync_results_from_hub
from src.populate import get_all_leaderboard_outputs, get_empty_leaderboard_notice_update


def _submission_outputs():
    sync_results_from_hub()
    outputs = get_all_leaderboard_outputs()
    return (*outputs, get_empty_leaderboard_notice_update(outputs[0]))


def _keep_form():
    return gr.skip(), gr.skip(), gr.skip()


def _reset_form():
    return "", "", None


def add_zip_submission(submission_zip, model: str, team: str):
    if submission_zip is None:
        return (
            styled_error(t("submit_error_no_zip")),
            *_submission_outputs(),
            *_keep_form(),
        )

    zip_path = submission_zip if isinstance(submission_zip, str) else submission_zip.name
    if not zip_path:
        return (
            styled_error(t("submit_error_read_zip")),
            *_submission_outputs(),
            *_keep_form(),
        )

    try:
        from src.submission.process_zip import process_submission_zip

        payload, _parsed_results, missing_files, missing_benchmarks = process_submission_zip(
            zip_path=zip_path,
            model_name=model,
            team=team,
        )
    except ValueError as exc:
        return (styled_error(str(exc)), *_submission_outputs(), *_keep_form())
    except Exception:
        return (
            styled_error(t("submit_error_process")),
            *_submission_outputs(),
            *_keep_form(),
        )

    resolved_model = payload["model"]
    _saved_payload, upload_error = save_submission(
        resolved_model,
        team,
        payload["results"],
        zip_path=zip_path,
    )

    warning = ""
    if missing_files:
        warning = (
            "<br><br><span style='color: orange;'>"
            + t("warn_missing_files").format(items=", ".join(missing_files))
            + "</span>"
        )
    if missing_benchmarks:
        warning += (
            "<br><br><span style='color: orange;'>"
            + t("warn_missing_benchmarks").format(items=", ".join(missing_benchmarks))
            + "</span>"
        )
    if upload_error:
        warning += f"<br><br><span style='color: orange;'>Warning: {upload_error}</span>"

    benchmarks = ", ".join(task.value.col_name for task in Tasks)
    message = styled_message(
        f"{t('submit_success')}"
        f"<br>{t('submit_model')}: <b>{resolved_model}</b>"
        f"<br>{t('submit_benchmarks')}: {benchmarks}"
        f"{warning}"
    )

    return (message, *_submission_outputs(), *_reset_form())