Skip to content

Instantly share code, notes, and snippets.

@danyaljj
Created May 5, 2021 16:24
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save danyaljj/e0792f6a71153db4122bcb151c87b342 to your computer and use it in GitHub Desktop.
Save danyaljj/e0792f6a71153db4122bcb151c87b342 to your computer and use it in GitHub Desktop.
"""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