File size: 5,625 Bytes
193db9d |
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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 |
import json
import os
import traceback
from datetime import datetime, timedelta, timezone
from typing import Optional
import gradio as gr
import yaml
from src.display.formatting import styled_error, styled_message
from src.envs import API, DAILY_SUBMISSION_LIMIT_PER_USER, EVAL_REQUESTS_PATH, QUEUE_REPO
from src.submission.structs import CompetitionType, Submission, SubmissionStatus
from workflows.structs import Workflow
def get_user_submissions_today(username: str, competition_type: str) -> list[Submission]:
today = datetime.now(timezone.utc).strftime("%Y%m%d")
if username is None:
raise gr.Error("Authentication required. Please log in to view your submissions.")
out_dir = f"{EVAL_REQUESTS_PATH}/{username}"
submissions = []
if not os.path.exists(out_dir):
return submissions
for file in os.listdir(out_dir):
if not file.startswith(f"{competition_type}_"):
continue
with open(os.path.join(out_dir, file), "r") as f:
submission = Submission.from_dict(json.load(f))
if submission.created_at.startswith(today):
submissions.append(submission)
return submissions
def get_time_until_next_submission(tz: timezone = timezone.utc) -> str:
next_day_00 = datetime.now(tz) + timedelta(days=1)
next_day_00 = next_day_00.replace(hour=0, minute=0, second=0, microsecond=0)
remaining_time = next_day_00 - datetime.now(tz)
hours = remaining_time.seconds // 3600
minutes = (remaining_time.seconds % 3600) // 60
remaining_time_str = f"{hours} hours {minutes} mins"
return remaining_time_str
def create_submission(
username: str,
model_name: str,
description: str,
workflow: Workflow,
competition_type: CompetitionType,
) -> Submission:
"""
Create a submission for a tossup model.
Args:
name: Display name of the submission
description: Detailed description of what the submission does
user_email: Email of the user who created the submission
workflow: The workflow configuration for the tossup model
Returns:
Submission object if successful, None if validation fails
"""
# Create the submission
dt = datetime.now(timezone.utc)
submission = Submission(
id=f"{competition_type}_{dt.strftime('%Y%m%d_%H%M%S')}_{model_name.lower().replace(' ', '_')}",
model_name=model_name,
username=username,
description=description,
competition_type=competition_type,
submission_type="simple_workflow",
workflow=workflow,
status="submitted",
created_at=dt.isoformat(),
updated_at=dt.isoformat(),
)
return submission
def submit_model(
model_name: str,
description: str,
workflow: Workflow,
competition_type: CompetitionType,
profile: gr.OAuthProfile | None,
) -> str:
"""
Submit a tossup model for evaluation.
Args:
name: Display name of the submission
description: Detailed description of what the submission does
user_email: Email of the user who created the submission
workflow: The workflow configuration for the tossup model
Returns:
Status message
"""
if profile is None:
return styled_error("Authentication required. Please log in first to submit your model.")
username = profile.username
if len(get_user_submissions_today(username)) >= DAILY_SUBMISSION_LIMIT_PER_USER:
time_str = get_time_until_next_submission()
return styled_error(
f"Daily submission limit of {DAILY_SUBMISSION_LIMIT_PER_USER} reached. Please try again in \n {time_str}."
)
try:
submission = create_submission(
username=username,
model_name=model_name,
description=description,
workflow=workflow,
competition_type=competition_type,
)
# Convert to dictionary format
submission_dict = submission.to_dict()
# Create output directory path
out_dir = f"{EVAL_REQUESTS_PATH}/{username}"
out_path = f"{out_dir}/{submission.id}.json"
# Upload to HuggingFace dataset
API.upload_file(
path_or_fileobj=json.dumps(submission_dict, indent=2).encode(),
path_in_repo=out_path.split("eval-queue/")[1],
repo_id=QUEUE_REPO,
repo_type="dataset",
commit_message=f"Add tossup submission {submission.id}",
)
return styled_message(
f"Successfully submitted tossup model!\n"
f"Submission ID: {submission.id}\n"
f"Name: {username}/{model_name}\n"
f"Please wait for up to an hour for the model to show in the PENDING list."
)
except Exception as e:
traceback.print_exc()
return styled_error(f"Error submitting model: {str(e)}")
if __name__ == "__main__":
# Example usage
from workflows.factory import create_quizbowl_simple_step_initial_setup
# Create workflow
model_step = create_quizbowl_simple_step_initial_setup()
model_step.model = "gpt-4"
model_step.provider = "openai"
model_step.temperature = 0.7
workflow = Workflow(
inputs=["question_text"],
outputs={"answer": "A.answer", "confidence": "A.confidence"},
steps={"A": model_step},
)
# Submit model
result = submit_model(
model_name="GPT-4 Tossup",
description="A simple GPT-4 model for tossup questions",
workflow=workflow,
competition_type="tossup",
)
print(result)
|