|
|
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)") |
|
|
|
|
|
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) |
|
|
|
