File size: 3,713 Bytes
566d03e |
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 |
"""
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 # "huggingface", "api", "local"
model_path: str
precision: str
base_model: Optional[str]
license: str
submitted_at: str
status: str # "pending", "running", "completed", "failed"
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"
)
# Save submission
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)
|