Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
File size: 5,841 Bytes
e7abd9e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
import os
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List, Tuple
from huggingface_hub import HfApi
from dotenv import load_dotenv
# Get the backend directory path
BACKEND_DIR = Path(__file__).parent.parent
ROOT_DIR = BACKEND_DIR.parent
# Load environment variables from .env file in root directory
load_dotenv(ROOT_DIR / ".env")
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(message)s'
)
logger = logging.getLogger(__name__)
# Initialize Hugging Face API
HF_TOKEN = os.getenv("HF_TOKEN")
if not HF_TOKEN:
raise ValueError("HF_TOKEN not found in environment variables")
api = HfApi(token=HF_TOKEN)
# Default organization
HF_ORGANIZATION = os.getenv('HF_ORGANIZATION', 'open-llm-leaderboard')
def get_last_votes(limit: int = 5) -> List[Dict]:
"""Get the last votes from the votes dataset"""
try:
logger.info("\nFetching last votes...")
# Download and read votes file
logger.info("Downloading votes file...")
votes_file = api.hf_hub_download(
repo_id=f"{HF_ORGANIZATION}/votes",
filename="votes_data.jsonl",
repo_type="dataset"
)
logger.info("Reading votes file...")
votes = []
with open(votes_file, 'r') as f:
for line in f:
try:
vote = json.loads(line)
votes.append(vote)
except json.JSONDecodeError:
continue
# Sort by timestamp and get last n votes
logger.info("Sorting votes...")
votes.sort(key=lambda x: x.get('timestamp', ''), reverse=True)
last_votes = votes[:limit]
logger.info(f"✓ Found {len(last_votes)} recent votes")
return last_votes
except Exception as e:
logger.error(f"Error reading votes: {str(e)}")
return []
def get_last_models(limit: int = 5) -> List[Dict]:
"""Get the last models from the requests dataset using commit history"""
try:
logger.info("\nFetching last model submissions...")
# Get commit history
logger.info("Getting commit history...")
commits = list(api.list_repo_commits(
repo_id=f"{HF_ORGANIZATION}/requests",
repo_type="dataset"
))
logger.info(f"Found {len(commits)} commits")
# Track processed files to avoid duplicates
processed_files = set()
models = []
# Process commits until we have enough models
for i, commit in enumerate(commits):
logger.info(f"Processing commit {i+1}/{len(commits)} ({commit.created_at})")
# Look at added/modified files in this commit
files_to_process = [f for f in (commit.added + commit.modified) if f.endswith('.json')]
if files_to_process:
logger.info(f"Found {len(files_to_process)} JSON files in commit")
for file in files_to_process:
if file in processed_files:
continue
processed_files.add(file)
logger.info(f"Downloading {file}...")
try:
# Download and read the file
content = api.hf_hub_download(
repo_id=f"{HF_ORGANIZATION}/requests",
filename=file,
repo_type="dataset"
)
with open(content, 'r') as f:
model_data = json.load(f)
models.append(model_data)
logger.info(f"✓ Added model {model_data.get('model', 'Unknown')}")
if len(models) >= limit:
logger.info("Reached desired number of models")
break
except Exception as e:
logger.error(f"Error reading file {file}: {str(e)}")
continue
if len(models) >= limit:
break
logger.info(f"✓ Found {len(models)} recent model submissions")
return models
except Exception as e:
logger.error(f"Error reading models: {str(e)}")
return []
def main():
"""Display last activities from the leaderboard"""
try:
# Get last votes
logger.info("\n=== Last Votes ===")
last_votes = get_last_votes()
if last_votes:
for vote in last_votes:
logger.info(f"\nModel: {vote.get('model')}")
logger.info(f"User: {vote.get('username')}")
logger.info(f"Timestamp: {vote.get('timestamp')}")
else:
logger.info("No votes found")
# Get last model submissions
logger.info("\n=== Last Model Submissions ===")
last_models = get_last_models()
if last_models:
for model in last_models:
logger.info(f"\nModel: {model.get('model')}")
logger.info(f"Submitter: {model.get('sender', 'Unknown')}")
logger.info(f"Status: {model.get('status', 'Unknown')}")
logger.info(f"Submission Time: {model.get('submitted_time', 'Unknown')}")
logger.info(f"Precision: {model.get('precision', 'Unknown')}")
logger.info(f"Weight Type: {model.get('weight_type', 'Unknown')}")
else:
logger.info("No models found")
except Exception as e:
logger.error(f"Global error: {str(e)}")
if __name__ == "__main__":
main() |