File size: 3,167 Bytes
193db9d 02b7dec 193db9d 9756440 193db9d 02b7dec 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 |
from datetime import datetime
from typing import Dict, List, Literal, Optional
from pydantic import BaseModel, Field
from workflows.structs import TossupWorkflow, Workflow
CompetitionType = Literal["tossup", "bonus"]
SubmissionType = Literal["python_file", "simple_workflow", "complex_workflow"]
SubmissionStatus = Literal["submitted", "in_progress", "completed", "failed"]
class Submission(BaseModel):
"""
Represents a submission in the competition system, formatted for HuggingFace datasets.
This model is designed to be easily serializable to/from HuggingFace dataset format
while maintaining type safety and validation through Pydantic.
Attributes:
id: Unique identifier for the submission
name: Display name of the submission
description: Detailed description of what the submission does
user_email: Email of the user who created the submission
competition_type: Type of competition (Tossup or Bonus)
submission_type: Format of the submission (python file or workflow)
workflow: Optional workflow definition for workflow submissions, stored as JSON
code: Optional code content for python file submissions
status: Current status of the submission
created_at: ISO format timestamp of creation
updated_at: ISO format timestamp of last update
"""
id: str = Field(description="Unique identifier for the submission")
model_name: str = Field(description="Display name of the submission")
username: str = Field(description="HuggingFace username of the user who created the submission")
description: str = Field(description="Detailed description of what the submission does")
competition_type: CompetitionType = Field(description="Type of competition (tossup or bonus)")
submission_type: SubmissionType = Field(description="Format of the submission (python file or workflow)")
# TODO: Make workflow as json / yaml string instead of Workflow object
workflow: Optional[Workflow] = Field(default=None, description="Optional workflow definition stored as JSON dict")
code: Optional[str] = Field(default=None, description="Optional code content for python file submissions")
status: SubmissionStatus = Field(description="Current status of the submission")
created_at: str = Field(description="ISO format timestamp of creation")
updated_at: str = Field(description="ISO format timestamp of last update")
def to_dict(self) -> Dict:
"""Convert to dictionary format suitable for HF datasets"""
data = self.model_dump()
if self.workflow:
data["workflow"] = self.workflow.model_dump(exclude_defaults=True)
return data
@classmethod
def from_dict(cls, data: Dict) -> "Submission":
"""Create instance from dictionary format used in HF datasets"""
if data.get("workflow"):
if data["competition_type"] == "tossup":
data["workflow"] = TossupWorkflow.model_validate(data["workflow"])
else:
data["workflow"] = Workflow.model_validate(data["workflow"])
return cls.model_validate(data)
|