Created
May 5, 2021 16:24
-
-
Save danyaljj/e0792f6a71153db4122bcb151c87b342 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Interaction with and data models for the leaderboard API.""" | |
from __future__ import annotations | |
import dataclasses | |
from typing import ClassVar, Iterator, List, Optional, Dict | |
from urllib.parse import urljoin | |
import requests | |
try: | |
import settings | |
except ImportError: | |
try: | |
from annotator import settings | |
except ImportError: | |
print("Could not import settings . . . ") | |
# data model | |
from requests import HTTPError | |
@dataclasses.dataclass(frozen=True) | |
class EvaluationStatus: | |
"""The status of the submission's *automatic* evaluations.""" | |
VALUES: ClassVar = { | |
"pending", | |
"succeeded", | |
"failed", | |
} | |
"""All possible values for the evaluation status.""" | |
value: str | |
def __post_init__(self): | |
if self.value not in self.VALUES: | |
raise ValueError(f"Unrecognized value: {self.value}") | |
@dataclasses.dataclass(frozen=True) | |
class PublicationStatus: | |
"""The submission's publication status.""" | |
VALUES: ClassVar = {"unpublished", "published"} | |
"""All possible values for the publication status.""" | |
value: str | |
def __post_init__(self): | |
if self.value not in self.VALUES: | |
raise ValueError(f"Unrecognized value: {self.value}.") | |
@dataclasses.dataclass(frozen=True) | |
class Score: | |
"""A score for a given metric.""" | |
metric_id: str | |
score: Optional[float] | |
@dataclasses.dataclass(frozen=True) | |
class LeaderboardSubmission: | |
"""A data class representing a submission.""" | |
id: str | |
evaluation_status: EvaluationStatus | |
publication_status: PublicationStatus | |
is_archived: bool | |
scores: Dict[str, Score] | |
name: str | |
contributor: str | |
description: str | |
email: str | |
created: str | |
@property | |
def is_annotation_eligible(self): | |
return ( | |
self.evaluation_status.value == 'succeeded' | |
and self.publication_status.value == 'unpublished' | |
and not self.is_archived | |
) | |
@property | |
def as_list(self): | |
return [ | |
self.id, | |
self.name, | |
self.contributor, | |
self.email, | |
self.created, | |
self.description, | |
self.evaluation_status.value, | |
self.publication_status.value, | |
self.is_archived, | |
] | |
@classmethod | |
def deserialize(cls, data): | |
return cls( | |
id=data["id"], | |
name=data['name'], | |
contributor=data['contributors'], | |
description=data.get('description', None), | |
created=data['created'], | |
email=data.get('email'), | |
evaluation_status=EvaluationStatus(value=data['blind_status']), | |
publication_status=PublicationStatus(value=data['publish_status']), | |
is_archived=data['is_archived'], | |
scores={ | |
k: Score( | |
metric_id=v['metric_key'], | |
score=v.get('blind_score') | |
) | |
for k, v in data['metric_scores'].items() | |
}, | |
) | |
@classmethod | |
def all(cls) -> Iterator[LeaderboardSubmission]: | |
response = requests.get( | |
url=urljoin( | |
settings.LEADERBOARD_SITE_URL, | |
f"/api/leaderboards/{settings.LEADERBOARD_ID}/submissions/all" | |
), | |
cookies={"User-Token": settings.BEAKER_USER_TOKEN}, | |
) | |
response.raise_for_status() | |
for data in response.json(): | |
yield cls.deserialize(data) | |
@classmethod | |
def get_predictions(self, submission_id: str) -> Dict[str, str]: | |
url = urljoin( | |
settings.LEADERBOARD_SITE_URL, | |
f"/api/submissions/{submission_id}/predictions" | |
) | |
response = requests.get( | |
url=url, | |
cookies={'User-Token': settings.BEAKER_USER_TOKEN}, | |
) | |
response.raise_for_status() | |
return response.json() | |
@classmethod | |
def upload_metrics(self, submission_id: str, metrics: Dict[str, float]) -> None: | |
for name, value in metrics.items(): | |
url = urljoin( | |
settings.LEADERBOARD_SITE_URL, | |
f"/api/submissions/{submission_id}/score/{name}" | |
) | |
try: | |
response = requests.post( | |
url=url, | |
json={'blind_score': value}, | |
cookies={"User-Token": settings.BEAKER_USER_TOKEN} | |
) | |
response.raise_for_status() | |
except HTTPError as e: | |
print(e) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment