Spaces:
Running
Running
| import os | |
| import logging | |
| from typing import Optional | |
| import pandas as pd | |
| import requests | |
| class LeaderboardServer: | |
| def __init__(self): | |
| self._LOG = logging.getLogger('leaderboard_server') | |
| self._server_address = os.environ['LEADERBOARD_SERVER_ADDRESS'] | |
| def get_leaderboard(self, submission_type: str, dataset_version: str) -> pd.DataFrame: | |
| """ | |
| Gets the leaderboard of the given submission type | |
| Args: | |
| submission_type: the type of the submission to get the leaderboard of: | |
| 'SC' / 'MC-specific' / 'MC-agnostic' / 'MC-agnostic-all' | |
| dataset_version: the version of the dataset to get the leaderboard of ('Devset1' / 'Devset2' / ...) | |
| """ | |
| self._LOG.info(f'Getting leaderboard for submission type: {submission_type}') | |
| endpoint = f'{self._server_address}/leaderboard' | |
| submission_type = submission_type.lower().replace('-', '_') | |
| response = requests.get(endpoint, params={'submission_type': submission_type, | |
| 'dataset_version': dataset_version}) | |
| if response.status_code != 200: | |
| self._LOG.error(f'Error while fetching leaderboard, status code: {response.status_code}, ' | |
| f'response: {response.text}, endpoint: {endpoint}') | |
| return pd.DataFrame() | |
| return pd.DataFrame(response.json()) | |
| def get_submissions_by_hf_token(self, hf_token: str) -> pd.DataFrame: | |
| """ | |
| Gets the submissions of the given hf token | |
| Args: | |
| hf_token: the hf token to get the submissions of | |
| """ | |
| self._LOG.info(f'Fetching submissions') | |
| endpoint = f'{self._server_address}/submissions' | |
| response = requests.get(endpoint, params={'token': hf_token}) | |
| if response.status_code != 200: | |
| self._LOG.error(f'Error while fetching submissions, status code: {response.status_code}, ' | |
| f'response: {response.text}, endpoint: {endpoint}') | |
| return pd.DataFrame() | |
| return pd.DataFrame(response.json()) | |
| def is_hf_token_valid(self, hf_token: str) -> Optional[bool]: | |
| """ | |
| Validates the given hf token | |
| Args: | |
| hf_token: the hf token to validate | |
| """ | |
| self._LOG.info(f'Validating hf token') | |
| endpoint = f'{self._server_address}/validate_hf_token' | |
| response = requests.get(endpoint, params={'token': hf_token}) | |
| if response.status_code != 200: | |
| self._LOG.error(f'Error while validating hf token, status code: {response.status_code}, ' | |
| f'response: {response.text}, endpoint: {endpoint}') | |
| return None | |
| return response.json()['valid'] | |
| def get_submission_count_last_24_hours(self, hf_token: str) -> Optional[int]: | |
| """ | |
| Gets the number of submissions of the given hf token in the last 24 hours | |
| Args: | |
| hf_token: the hf token to get the submissions count of | |
| """ | |
| self._LOG.info(f'fetching submissions count for the last 24 hours') | |
| endpoint = f'{self._server_address}/submission_count_last_24_hours' | |
| response = requests.get(endpoint, params={'token': hf_token}) | |
| if response.status_code != 200: | |
| self._LOG.error(f'Error while fetching submissions count, status code: {response.status_code}, ' | |
| f'response: {response.text}, endpoint: {endpoint}') | |
| return None | |
| return int(response.json()['count']) | |
| def add_submission(self, token: str, file_path: str, metadata: dict) -> dict: | |
| """ | |
| Adds a submission to the leaderboard based on the given file and metadata | |
| Args: | |
| token: the token of the team | |
| file_path: the path of the file to submit | |
| metadata: the metadata of the submission | |
| """ | |
| self._LOG.info(f'Adding submission for team: {metadata["team_name"]}, ' | |
| f'submission type: {metadata["submission_type"]}') | |
| endpoint = f'{self._server_address}/add_submission' | |
| metadata['token'] = token | |
| metadata['submission_type'] = metadata['submission_type'].lower().replace('-', '_') | |
| with open(file_path, 'rb') as payload_file: | |
| files = {'zip_file': payload_file} | |
| response = requests.post(endpoint, files=files, params=metadata, timeout=600) | |
| if response.status_code != 200: | |
| self._LOG.error(f'Error while adding submission, status code: {int(response.status_code)}, ' | |
| f'response: {response.text}, endpoint: {endpoint}') | |
| return dict(error=response.json()['message']) | |
| return response.json() | |
| def main(): | |
| """ | |
| Usage of the LeaderboardServer class | |
| """ | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') | |
| server = LeaderboardServer() | |
| hf_token = str(os.environ['HF_TOKEN']) | |
| print('leaderboard:\n', server.get_leaderboard('notsofar_mc')) | |
| print('submissions by hf token:\n', server.get_submissions_by_hf_token(hf_token)) | |
| print('is hf token valid:\n', server.is_hf_token_valid(hf_token)) | |
| print('is hf token valid:\n', server.is_hf_token_valid(hf_token + '1')) | |
| print('add_submission:\n', server.add_submission( | |
| token=hf_token, | |
| file_path=fr"C:\Users\shaipeer\Downloads\submissions\notsofar_submission.zip", | |
| metadata={ | |
| 'challenge_name': 'NOTSOFAR1', | |
| 'team_name': 'NOTSOFAR Test Team', | |
| 'submission_type': 'notsofar_mc', | |
| 'description': 'Test NOTSOFAR submission', | |
| 'token': hf_token, | |
| 'file_name': 'notsofar_submission.zip', | |
| 'file_size_mb': 10, | |
| 'ip': '127.0.0.1' | |
| })) | |
| print('add_submission:\n', server.add_submission( | |
| token=hf_token, | |
| file_path=fr"C:\Users\shaipeer\Downloads\submissions\chime_submission.zip", | |
| metadata={ | |
| 'challenge_name': 'NOTSOFAR1', | |
| 'team_name': 'Chime Test Team', | |
| 'submission_type': 'dasr_unconstrained_lm', | |
| 'description': 'Test chime submission', | |
| 'token': hf_token, | |
| 'file_name': 'chime_submission.zip', | |
| 'file_size_mb': 10, | |
| 'ip': '127.0.0.1' | |
| })) | |
| if __name__ == '__main__': | |
| main() | |