Implement main menu, profile list, and profile stats screens; add screen manager and user profile integration
This commit is contained in:
@@ -0,0 +1,307 @@
|
||||
#!/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!")
|
||||
Reference in New Issue
Block a user