|
|
""" |
|
|
Submission Handler |
|
|
================== |
|
|
|
|
|
Handles model submission workflow for the leaderboard. |
|
|
""" |
|
|
|
|
|
import json |
|
|
import os |
|
|
from datetime import datetime |
|
|
from pathlib import Path |
|
|
from typing import Dict, Optional |
|
|
from dataclasses import dataclass, asdict |
|
|
import uuid |
|
|
|
|
|
|
|
|
@dataclass |
|
|
class Submission: |
|
|
"""Model submission data.""" |
|
|
id: str |
|
|
model_name: str |
|
|
model_type: str |
|
|
model_path: str |
|
|
precision: str |
|
|
base_model: Optional[str] |
|
|
license: str |
|
|
submitted_at: str |
|
|
status: str |
|
|
results: Optional[Dict] = None |
|
|
error_message: Optional[str] = None |
|
|
|
|
|
|
|
|
class SubmissionHandler: |
|
|
"""Handles model submissions and evaluation queue.""" |
|
|
|
|
|
def __init__(self, submissions_dir: str = "data/submissions"): |
|
|
self.submissions_dir = Path(submissions_dir) |
|
|
self.submissions_dir.mkdir(parents=True, exist_ok=True) |
|
|
|
|
|
def create_submission( |
|
|
self, |
|
|
model_name: str, |
|
|
model_type: str, |
|
|
model_path: str, |
|
|
precision: str = "float16", |
|
|
base_model: Optional[str] = None, |
|
|
license: str = "Apache-2.0" |
|
|
) -> Submission: |
|
|
"""Create a new submission.""" |
|
|
submission = Submission( |
|
|
id=str(uuid.uuid4())[:8], |
|
|
model_name=model_name, |
|
|
model_type=model_type, |
|
|
model_path=model_path, |
|
|
precision=precision, |
|
|
base_model=base_model, |
|
|
license=license, |
|
|
submitted_at=datetime.now().isoformat(), |
|
|
status="pending" |
|
|
) |
|
|
|
|
|
|
|
|
self._save_submission(submission) |
|
|
return submission |
|
|
|
|
|
def _save_submission(self, submission: Submission): |
|
|
"""Save submission to file.""" |
|
|
filepath = self.submissions_dir / f"{submission.id}.json" |
|
|
with open(filepath, 'w', encoding='utf-8') as f: |
|
|
json.dump(asdict(submission), f, ensure_ascii=False, indent=2) |
|
|
|
|
|
def get_submission(self, submission_id: str) -> Optional[Submission]: |
|
|
"""Load a submission by ID.""" |
|
|
filepath = self.submissions_dir / f"{submission_id}.json" |
|
|
if not filepath.exists(): |
|
|
return None |
|
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f: |
|
|
data = json.load(f) |
|
|
|
|
|
return Submission(**data) |
|
|
|
|
|
def update_status( |
|
|
self, |
|
|
submission_id: str, |
|
|
status: str, |
|
|
results: Optional[Dict] = None, |
|
|
error_message: Optional[str] = None |
|
|
): |
|
|
"""Update submission status.""" |
|
|
submission = self.get_submission(submission_id) |
|
|
if submission: |
|
|
submission.status = status |
|
|
if results: |
|
|
submission.results = results |
|
|
if error_message: |
|
|
submission.error_message = error_message |
|
|
self._save_submission(submission) |
|
|
|
|
|
def get_pending_submissions(self) -> list: |
|
|
"""Get all pending submissions.""" |
|
|
submissions = [] |
|
|
for filepath in self.submissions_dir.glob("*.json"): |
|
|
with open(filepath, 'r', encoding='utf-8') as f: |
|
|
data = json.load(f) |
|
|
if data.get('status') == 'pending': |
|
|
submissions.append(Submission(**data)) |
|
|
return sorted(submissions, key=lambda x: x.submitted_at) |
|
|
|
|
|
def get_all_submissions(self) -> list: |
|
|
"""Get all submissions.""" |
|
|
submissions = [] |
|
|
for filepath in self.submissions_dir.glob("*.json"): |
|
|
with open(filepath, 'r', encoding='utf-8') as f: |
|
|
data = json.load(f) |
|
|
submissions.append(Submission(**data)) |
|
|
return sorted(submissions, key=lambda x: x.submitted_at, reverse=True) |
|
|
|
