import os import logging from typing import Optional import pandas as pd import requests LEADERBOARDS = { 'notsofar_sc': pd.DataFrame({'#': {0: 1, 1: 2, 2: 3, 3: 4}, 'Team': {0: 'ToTaTo', 1: 'ts', 2: '--', 3: 'NOTSOFAR baseline'}, 'TCP WER (%)': {0: 37.58, 1: 41.905, 2: 45.198, 3: 45.844}, 'tcORC WER (%)': {0: 25.833, 1: 36.107, 2: 37.536, 3: 38.604}, 'Entries': {0: 17, 1: 3, 2: 3, 3: 1}, 'Last': {0: '16d', 1: '29d', 2: '42d', 3: '74d'}}), 'notsofar_mc': pd.DataFrame({'#': {0: 1, 1: 2, 2: 3, 3: 4, 4: 5, 5: 6, 6: 7, 7: 8, 8: 9, 9: 10, 10: 11}, 'Team': {0: 'STS (DASR-LM)', 1: 'ts', 2: 'short', 3: 'long', 4: 'BUT', 5: 'IE', 6: 'NOTSOFAR baseline', 7: '--', 8: 'DASR Organizers (DASR-LM)', 9: 'DASR Organizers (NeMo) (DASR-LM)', 10: '---new'}, 'TCP WER (%)': {0: 22.447, 1: 23.857, 2: 26.887, 3: 30.738, 4: 31.35, 5: 31.392, 6: 31.551, 7: 35.276, 8: 48.136, 9: 61.342, 10: 63.598}, 'tcORC WER (%)': {0: 10000.0, 1: 19.864, 2: 15.721, 3: 20.311, 4: 26.417, 5: 26.376, 6: 26.597, 7: 26.426, 8: 36.303, 9: 47.87, 10: 26.438}, 'Entries': {0: 1, 1: 15, 2: 1, 3: 2, 4: 3, 5: 5, 6: 1, 7: 9, 8: 1, 9: 2, 10: 1}, 'Last': {0: '8d', 1: '22d', 2: '64d', 3: '69d', 4: '71d', 5: '30d', 6: '74d', 7: '8d', 8: '60d', 9: '59d', 10: '7d'}}), 'dasr_constrained_lm': pd.DataFrame({'#': {0: 1, 1: 2, 2: 3}, 'Team': {0: 'STS', 1: 'DASR Organizers (NeMo)', 2: 'DASR Organizers'}, 'TCP WER (%)': {0: 49.692, 1: 54.56, 2: 65.656}, 'chime6': {0: 82.367, 1: 56.532, 2: 88.708}, 'mixer6': {0: 28.169, 1: 24.888, 2: 29.235}, 'dipco': {0: 66.307, 1: 75.787, 2: 98.44}, 'notsofar1': {0: 21.925, 1: 61.031, 2: 46.24}, 'Entries': {0: 1, 1: 2, 2: 1}, 'Last': {0: '8d', 1: '59d', 2: '60d'}}), 'dasr_unconstrained_lm': pd.DataFrame() } 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: return LEADERBOARDS[submission_type] # 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()