|
|
|
|
|
""" |
|
|
API Client module for MTEB Turkish Leaderboard |
|
|
""" |
|
|
|
|
|
from typing import Optional, Dict, Any |
|
|
import traceback |
|
|
import requests |
|
|
|
|
|
from config import API_BASE_URL, API_TIMEOUT, API_URL, USERNAME, PASSWORD |
|
|
|
|
|
|
|
|
def check_api_health() -> bool: |
|
|
"""Check if API is available""" |
|
|
try: |
|
|
response = requests.get(f"{API_BASE_URL}/api/v1/health", timeout=5) |
|
|
return response.status_code == 200 |
|
|
except: |
|
|
return False |
|
|
|
|
|
|
|
|
def send_evaluation_request_to_api(model_name: str, batch_size: int = 32, email: str = "user@example.com") -> Optional[Dict[str, Any]]: |
|
|
""" |
|
|
Send an evaluation request to the API for the specified model. |
|
|
Returns the API response as a dictionary if successful, otherwise None. |
|
|
""" |
|
|
try: |
|
|
payload = { |
|
|
"model_name": model_name, |
|
|
"model_repo": model_name.split("/")[0] if "/" in model_name else "unknown", |
|
|
"batch_size": batch_size, |
|
|
"email": email, |
|
|
"model_type": "sentence-transformer" |
|
|
} |
|
|
|
|
|
|
|
|
auth = (USERNAME, PASSWORD) |
|
|
|
|
|
response = requests.post( |
|
|
f"{API_URL}/api/mteb/request", |
|
|
json=payload, |
|
|
timeout=API_TIMEOUT, |
|
|
auth=auth |
|
|
) |
|
|
|
|
|
print(f"Response Status: {response.status_code}") |
|
|
|
|
|
if response.status_code == 200: |
|
|
result = response.json() |
|
|
return result |
|
|
else: |
|
|
print(f"API Error: {response.status_code}") |
|
|
try: |
|
|
error_detail = response.json() |
|
|
print(f" Error Detail: {error_detail}") |
|
|
except: |
|
|
print(f" Raw Response: {response.text}") |
|
|
return None |
|
|
|
|
|
except Exception as e: |
|
|
print(f"API Call Error: {e}") |
|
|
traceback.print_exc() |
|
|
return None |
|
|
|
|
|
|
|
|
def get_evaluation_status(request_id: str) -> Optional[Dict[str, Any]]: |
|
|
"""Get evaluation status from""" |
|
|
try: |
|
|
auth = (USERNAME, PASSWORD) |
|
|
|
|
|
response = requests.get( |
|
|
f"{API_URL}/api/mteb/status/{request_id}", |
|
|
timeout=API_TIMEOUT, |
|
|
auth=auth |
|
|
) |
|
|
|
|
|
if response.status_code == 200: |
|
|
return response.json() |
|
|
else: |
|
|
print(f"Status check error: {response.status_code}") |
|
|
return None |
|
|
|
|
|
except Exception as e: |
|
|
print(f"Status check error: {e}") |
|
|
return None |
|
|
|
|
|
|
|
|
def cancel_evaluation_request(request_id: str) -> bool: |
|
|
"""Cancel an evaluation request""" |
|
|
try: |
|
|
auth = (USERNAME, PASSWORD) |
|
|
|
|
|
response = requests.delete( |
|
|
f"{API_URL}/api/mteb/request/{request_id}", |
|
|
timeout=API_TIMEOUT, |
|
|
auth=auth |
|
|
) |
|
|
|
|
|
return response.status_code == 200 |
|
|
|
|
|
except Exception as e: |
|
|
print(f"Cancel request error: {e}") |
|
|
return False |
|
|
|
