You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
307 lines
9.8 KiB
307 lines
9.8 KiB
#!/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!")
|
|
|