""" This file is part of the Tetris project. Written 2023 by Maarten Tromp Website ------- https://www.geekabit.nl/projects/tetris-window/ License ------- CC0 (no copyright, public domain) The person who associated a work with this deed has dedicated the work to the public domain by waiving all of his or her rights to the work worldwide under copyright law, including all related and neighbouring rights, to the extent allowed by law. You can copy, modify, distribute and perform the work, even for commercial purposes, all without asking permission. https://creativecommons.org/share-your-work/public-domain/cc0/ """ # imports import re import json import logging from collections import deque # project imports import config class Player: """ Player class. It holds player metadata and getters/setters. """ # constants buttons = ("up", "down", "left", "right", "b", "a") def __init__(self, websocket, outbox): """ An instance of this class is created when a new player connects to the controller websocket. Parameters ---------- websocket: websocket object outbox: send queue """ # public variables self.id = id(websocket) self.name = "" self.game = "" self.state = "connected" # connected, ready, queued, playing, finished, disconnected self.replay = [] # private variables self._websocket = websocket self._outbox = outbox self._button_changes = deque() self._button_state = {"up" : 0, "down" : 0, "left" : 0, "right" : 0, "b" : 0, "a" : 0} self._last_button_state = self._button_state.copy() self._rle_counter = 0 logging.info("player %s connected", self.id) def send(self, message): """ Queue message for sending. Parameters ---------- message: string Returns ------- none """ self._outbox.appendleft((self._websocket, message)) def handle_message(self, message_raw): """ A message has been received. Decode it and send to appropriate handler. Parameters ---------- message_raw: string JSON serialized message coming from controller. Returns ------- bool True on valid message, False otherwise. """ try: # decode json message_obj = json.loads(message_raw) except json.decoder.JSONDecodeError as exception: logging.warning("received invalid json message from player %s (%s): %s %s", self.name, self.id, message_raw, repr(exception)) self.send(json.dumps({"error": "invalid json"})) return False # send message to appropriate handler message_ok = False if "button" in message_obj: message_ok = self._handle_message_button(message_obj["button"]) elif "name" in message_obj: message_ok = self._handle_message_name(message_obj["name"]) elif "game" in message_obj: message_ok = self._handle_message_game(message_obj["game"]) if not message_ok: logging.warning("received invalid message from player %s (%s): %s", self.name, self.id, message_obj) self.send(json.dumps({"error": "invalid message"})) return False return True def _handle_message_button(self, message): """ Handle message containing button changes. Validate and queue. Parameters ---------- message: object Contains a single button change message. Returns ------- bool True on valid message, False otherwise. """ # validate message for button, value in message.items(): # check for valid button if not button in self.buttons: logging.warning("player %s (%s) invalid button in message: %s", self.name, self.id, button) return False # check for valid value if value not in (0, 1): logging.warning("player %s (%s) invalid value in message: %s", self.name, self.id, message) return False # queue change self._button_changes.append(message) return True def _handle_message_name(self, name): """ Handle messages containing name. Validate contents and update internal data. Parameters ---------- name: string Returns ------- bool True on valid name, False otherwise. """ # validate name = name.upper() if not re.match(r"^[A-Z]{3}$", name): logging.warning("player %s invalid name: %s", self.id, name) self.send(json.dumps({"error": "invalid name"})) return False self.name = name logging.info("player %s (%s) entered name", self.name, self.id) self._is_ready() return True def _handle_message_game(self, game): """ Handle messages containing game selection. Validate contents and update internal data. Parameters ---------- game: string Returns ------- bool True on valid game, False otherwise. """ # validate game game = game.lower() if not game in config.GAMES: logging.warning("player %s (%s) invalid game: %s", self.name, self.id, game) self.send(json.dumps({"error": "invalid game"})) return False self.game = game logging.info("player %s (%s) selected game: %s", self.name, self.id, self.game) self._is_ready() return True def _is_ready(self): """ Check if player is ready to be put in queue. Parameters ---------- none Returns ------- none """ if self.name and self.game: self.state = "ready" def get_button_state(self): """ Getter for _button_state. Parameters ---------- none Returns ------- dict Copy of internal dict with all buttons and number of frames the button has been pressed. """ # update state for held buttons for button, value in self._button_state.items(): if value == 1: self._button_state[button] = 2 # update state from queue if self._button_changes: change = self._button_changes.popleft() for button, value in change.items(): self._button_state[button] = value # old style #self.replay.append(self._button_state.copy()) # new style if self._button_state != self._last_button_state: self.replay.append(self._rle_counter) self._rle_counter = 0 self.replay.append(self._button_state.copy()) self._last_button_state = self._button_state.copy() else: self._rle_counter += 1 return self._button_state.copy()