mice/engine/score_api_client.py

#!/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!")