"""Network protocol for client-server communication.""" import json from enum import Enum from typing import Any, Dict class MessageType(Enum): """Types of messages exchanged between client and server.""" # Client -> Server JOIN = "JOIN" MOVE = "MOVE" START_GAME = "START_GAME" LEAVE = "LEAVE" # Server -> Client WELCOME = "WELCOME" STATE_UPDATE = "STATE_UPDATE" PLAYER_JOINED = "PLAYER_JOINED" PLAYER_LEFT = "PLAYER_LEFT" GAME_STARTED = "GAME_STARTED" GAME_OVER = "GAME_OVER" ERROR = "ERROR" PLAYER_INPUT = "PLAYER_INPUT" # Broadcast player input for prediction class Message: """Represents a protocol message.""" def __init__(self, msg_type: MessageType, data: Dict[str, Any] = None): """Initialize a message. Args: msg_type: The type of message data: Optional message data """ self.type = msg_type self.data = data or {} def to_json(self) -> str: """Serialize message to JSON string.""" return json.dumps({ "type": self.type.value, "data": self.data }) @classmethod def from_json(cls, json_str: str) -> "Message": """Deserialize message from JSON string.""" obj = json.loads(json_str) msg_type = MessageType(obj["type"]) data = obj.get("data", {}) return cls(msg_type, data) def __repr__(self) -> str: """String representation of message.""" return f"Message({self.type.value}, {self.data})" # Helper functions for creating specific messages def create_join_message(player_name: str) -> Message: """Create a JOIN message.""" return Message(MessageType.JOIN, {"player_name": player_name}) def create_move_message(direction: tuple) -> Message: """Create a MOVE message.""" return Message(MessageType.MOVE, {"direction": direction}) def create_start_game_message() -> Message: """Create a START_GAME message.""" return Message(MessageType.START_GAME) def create_leave_message() -> Message: """Create a LEAVE message.""" return Message(MessageType.LEAVE) def create_welcome_message(player_id: str) -> Message: """Create a WELCOME message.""" return Message(MessageType.WELCOME, {"player_id": player_id}) def create_state_update_message(game_state: dict) -> Message: """Create a STATE_UPDATE message.""" return Message(MessageType.STATE_UPDATE, {"game_state": game_state}) def create_player_joined_message(player_id: str, player_name: str) -> Message: """Create a PLAYER_JOINED message.""" return Message(MessageType.PLAYER_JOINED, { "player_id": player_id, "player_name": player_name }) def create_player_left_message(player_id: str) -> Message: """Create a PLAYER_LEFT message.""" return Message(MessageType.PLAYER_LEFT, {"player_id": player_id}) def create_game_started_message() -> Message: """Create a GAME_STARTED message.""" return Message(MessageType.GAME_STARTED) def create_game_over_message(winner_id: str = None) -> Message: """Create a GAME_OVER message.""" return Message(MessageType.GAME_OVER, {"winner_id": winner_id}) def create_error_message(error: str) -> Message: """Create an ERROR message.""" return Message(MessageType.ERROR, {"error": error}) def create_player_input_message(player_id: str, direction: tuple, input_buffer: list) -> Message: """Create a PLAYER_INPUT message for client-side prediction. Args: player_id: ID of the player who sent the input direction: Current direction tuple input_buffer: List of buffered direction tuples (max 3) """ return Message(MessageType.PLAYER_INPUT, { "player_id": player_id, "direction": direction, "input_buffer": input_buffer })