308 lines
9.8 KiB
Python
308 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Score API Client for Mice Game
|
|
Client module to integrate with the FastAPI score server
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
from typing import Optional, List, Dict, Any
|
|
import time
|
|
|
|
|
|
class ScoreAPIClient:
|
|
"""Client for communicating with the Mice Game Score API"""
|
|
|
|
def __init__(self, api_base_url: str = "http://localhost:8000", timeout: int = 5):
|
|
"""
|
|
Initialize the API client
|
|
|
|
Args:
|
|
api_base_url: Base URL of the API server
|
|
timeout: Request timeout in seconds
|
|
"""
|
|
self.api_base_url = api_base_url.rstrip('/')
|
|
self.timeout = timeout
|
|
|
|
def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Make HTTP request to API
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, etc.)
|
|
endpoint: API endpoint
|
|
data: Request data for POST requests
|
|
|
|
Returns:
|
|
Response JSON or None if error
|
|
"""
|
|
url = f"{self.api_base_url}{endpoint}"
|
|
|
|
try:
|
|
if method.upper() == "GET":
|
|
response = requests.get(url, timeout=self.timeout)
|
|
elif method.upper() == "POST":
|
|
response = requests.post(url, json=data, timeout=self.timeout)
|
|
else:
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
elif response.status_code in [400, 404, 409]:
|
|
# Client errors - return the error details
|
|
return {"error": True, "status": response.status_code, "detail": response.json()}
|
|
else:
|
|
return {"error": True, "status": response.status_code, "detail": "Server error"}
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
return {"error": True, "detail": "Could not connect to score server"}
|
|
except requests.exceptions.Timeout:
|
|
return {"error": True, "detail": "Request timeout"}
|
|
except Exception as e:
|
|
return {"error": True, "detail": str(e)}
|
|
|
|
def signup_user(self, device_id: str, user_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Register a new user
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
user_id: User identifier
|
|
|
|
Returns:
|
|
Response dictionary with success/error status
|
|
"""
|
|
endpoint = f"/signup/{device_id}/{user_id}"
|
|
response = self._make_request("POST", endpoint)
|
|
|
|
if response is None:
|
|
return {"success": False, "message": "Failed to connect to server"}
|
|
|
|
if response.get("error"):
|
|
return {"success": False, "message": response.get("detail", "Unknown error")}
|
|
|
|
return response
|
|
|
|
def submit_score(self, device_id: str, user_id: str, score: int, game_completed: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
Submit a score for a user
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
user_id: User identifier
|
|
score: Game score
|
|
game_completed: Whether the game was completed
|
|
|
|
Returns:
|
|
Response dictionary with success/error status
|
|
"""
|
|
endpoint = f"/score/{device_id}/{user_id}"
|
|
data = {
|
|
"user_id": user_id,
|
|
"device_id": device_id,
|
|
"score": score,
|
|
"game_completed": game_completed
|
|
}
|
|
|
|
response = self._make_request("POST", endpoint, data)
|
|
|
|
if response is None:
|
|
return {"success": False, "message": "Failed to connect to server"}
|
|
|
|
if response.get("error"):
|
|
return {"success": False, "message": response.get("detail", "Unknown error")}
|
|
|
|
return response
|
|
|
|
def get_device_users(self, device_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all users registered for a device
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
|
|
Returns:
|
|
List of user dictionaries
|
|
"""
|
|
endpoint = f"/users/{device_id}"
|
|
response = self._make_request("GET", endpoint)
|
|
|
|
if response is None:
|
|
return []
|
|
|
|
# Check if it's an error response (dict with error field) or success (list)
|
|
if isinstance(response, dict) and response.get("error"):
|
|
return []
|
|
|
|
# If it's a list (successful response), return it
|
|
if isinstance(response, list):
|
|
return response
|
|
|
|
return []
|
|
|
|
def get_user_scores(self, device_id: str, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get recent scores for a user
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
user_id: User identifier
|
|
limit: Maximum number of scores to return
|
|
|
|
Returns:
|
|
List of score dictionaries
|
|
"""
|
|
endpoint = f"/scores/{device_id}/{user_id}?limit={limit}"
|
|
response = self._make_request("GET", endpoint)
|
|
|
|
if response is None:
|
|
return []
|
|
|
|
# Check if it's an error response (dict with error field) or success (list)
|
|
if isinstance(response, dict) and response.get("error"):
|
|
return []
|
|
|
|
# If it's a list (successful response), return it
|
|
if isinstance(response, list):
|
|
return response
|
|
|
|
return []
|
|
|
|
def get_leaderboard(self, device_id: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get leaderboard for a device
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
limit: Maximum number of entries to return
|
|
|
|
Returns:
|
|
List of leaderboard entries
|
|
"""
|
|
endpoint = f"/leaderboard/{device_id}?limit={limit}"
|
|
response = self._make_request("GET", endpoint)
|
|
|
|
if response is None:
|
|
return []
|
|
|
|
# Check if it's an error response (dict with error field) or success (list)
|
|
if isinstance(response, dict) and response.get("error"):
|
|
return []
|
|
|
|
# If it's a list (successful response), return it
|
|
if isinstance(response, list):
|
|
return response
|
|
|
|
return []
|
|
|
|
def get_global_leaderboard(self, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get global leaderboard across all devices
|
|
|
|
Args:
|
|
limit: Maximum number of entries to return
|
|
|
|
Returns:
|
|
List of global leaderboard entries
|
|
"""
|
|
endpoint = f"/leaderboard/global/top?limit={limit}"
|
|
response = self._make_request("GET", endpoint)
|
|
|
|
if response is None:
|
|
return []
|
|
|
|
# Check if it's an error response (dict with error field) or success (list)
|
|
if isinstance(response, dict) and response.get("error"):
|
|
return []
|
|
|
|
# If it's a list (successful response), return it
|
|
if isinstance(response, list):
|
|
return response
|
|
|
|
return []
|
|
|
|
def is_server_available(self) -> bool:
|
|
"""
|
|
Check if the API server is available
|
|
|
|
Returns:
|
|
True if server is reachable, False otherwise
|
|
"""
|
|
response = self._make_request("GET", "/")
|
|
return response is not None and not response.get("error")
|
|
|
|
def user_exists(self, device_id: str, user_id: str) -> bool:
|
|
"""
|
|
Check if a user is registered for a device
|
|
|
|
Args:
|
|
device_id: Device identifier
|
|
user_id: User identifier
|
|
|
|
Returns:
|
|
True if user exists, False otherwise
|
|
"""
|
|
users = self.get_device_users(device_id)
|
|
return any(user["user_id"] == user_id for user in users)
|
|
|
|
|
|
# Convenience functions for easy integration
|
|
def create_api_client(api_url: str = "http://localhost:8000") -> ScoreAPIClient:
|
|
"""Create and return an API client instance"""
|
|
return ScoreAPIClient(api_url)
|
|
|
|
def test_connection(api_url: str = "http://localhost:8000") -> bool:
|
|
"""Test if the API server is available"""
|
|
client = ScoreAPIClient(api_url)
|
|
return client.is_server_available()
|
|
|
|
|
|
# Example usage and testing
|
|
if __name__ == "__main__":
|
|
# Example usage
|
|
print("Testing Score API Client...")
|
|
|
|
# Create client
|
|
client = ScoreAPIClient()
|
|
|
|
# Test server connection
|
|
if not client.is_server_available():
|
|
print("ERROR: API server is not available. Start it with: python score_api.py")
|
|
exit(1)
|
|
|
|
print("API server is available!")
|
|
|
|
# Example device and user
|
|
device_id = "DEV-CLIENT01"
|
|
user_id = "ClientTestUser"
|
|
|
|
# Test user signup
|
|
print(f"\nTesting user signup: {user_id}")
|
|
result = client.signup_user(device_id, user_id)
|
|
print(f"Signup result: {result}")
|
|
|
|
# Test score submission
|
|
print(f"\nTesting score submission...")
|
|
result = client.submit_score(device_id, user_id, 1750, True)
|
|
print(f"Score submission result: {result}")
|
|
|
|
# Test getting users
|
|
print(f"\nGetting users for device {device_id}:")
|
|
users = client.get_device_users(device_id)
|
|
for user in users:
|
|
print(f" User: {user['user_id']}, Best Score: {user['best_score']}")
|
|
|
|
# Test getting user scores
|
|
print(f"\nGetting scores for {user_id}:")
|
|
scores = client.get_user_scores(device_id, user_id)
|
|
for score in scores:
|
|
print(f" Score: {score['score']}, Time: {score['timestamp']}")
|
|
|
|
# Test leaderboard
|
|
print(f"\nLeaderboard for device {device_id}:")
|
|
leaderboard = client.get_leaderboard(device_id)
|
|
for entry in leaderboard:
|
|
print(f" Rank {entry['rank']}: {entry['user_id']} - {entry['best_score']} pts")
|
|
|
|
print("\nClient testing completed!")
|