""" 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 os import logging import json import random from statemachine import StateMachine, State # project imports import config from replayer import Replayer from tetris import TetrisGame from snake import SnakeGame from game_start import GameStart from game_over import GameOver class GameManager(StateMachine): """ Game manager class, poor mans state machine. It controls game state and attaches and removes screen and players. """ # constants game_start = State() game_play = State() game_over = State() game_replay = State(initial=True) start_game_start = game_over.to(game_start) start_game_play = game_start.to(game_play) start_game_over = (game_play.to(game_over) | game_replay.to(game_over)) start_game_replay = (game_over.to(game_replay) | game_start.to(game_play)) def __init__(self, screen, waiting_players): """ Game manager. This function will select a game to run and attach players and displays. Parameters ---------- screen: screen object waiting_players: queue Deque of player objects. """ logging.debug("starting game manager") # public variables # private variables self._screen = screen self._waiting_players = waiting_players self._game_name = "" self._game_obj = None self._game_seed = 0 self._players = [] # init base class super().__init__() def on_enter_game_start(self): """ Enter game-start state. This shows the player name. """ logging.debug("entering game_start state") # grab waiting playes self._players = [self._waiting_players.popleft()] self._game_name = self._players[0].game # TODO; check if game is multiplayer game # then find if there is one more waiting player that selected the same game. # if so, grab that player and append to self._players[] # send game_start message to active players for player in self._players: player.send(json.dumps({"game_start": True})) # send queue update to all waiting players self.send_queue_status() # create game start animation instance self._game_obj = GameStart(self._screen, self._players) def on_exit_game_start(self): """ Exit game-start state. Clean up. """ logging.debug("exit game_start state") self._game_obj = None def on_enter_game_play(self): """ Enter game-play state. The actual game is started. """ logging.debug("entering game play state") logging.info("starting game: %s, player: %s", self._game_name, self._players[0].name) # update player state for player in self._players: player.state = "playing" # create game instance random.seed() self._game_seed = random.randint(0, 2**31 - 1) self._start_game() def on_exit_game_play(self): """ Exit game-play state. The play is finished, clean up. """ logging.debug("exit game_start state") # save game if self._game_obj.save: self._save_game() # send game-over message to active players if self._players: for player in self._players: player.send(json.dumps({"game_over": True})) # update player state #if self._players: # for player in self._players: # player.state = "finished" # remove active players if self._players: for player in self._players: self._players.remove(player) self._game_obj = None def on_enter_game_replay(self): """ Enter game-replay state. """ logging.debug("entering game replay state") # load game and create replayers self._load_game() logging.info("starting game: %s, seed: %s", self._game_name, self._game_seed) self._start_game() def on_exit_game_replay(self): """ Exit game-replay state. Clean up. """ logging.debug("exit game replay state") # remove replayers if self._players: for player in self._players: self._players.remove(player) self._game_obj = None def on_enter_game_over(self): """ Enter game-over state. Fade out the screen. """ logging.debug("entering game-over state") # create game over animation instance self._game_obj = GameOver(self._screen) def on_exit_game_over(self): """ Exit game-over state. """ logging.debug("exit game-over state") self._game_obj = None def queue_player(self, player): """ Put player in queue. Parameters ---------- player Returns ------- none """ # queue player self._waiting_players.append(player) player.state = "queued" queue_pos = self._waiting_players.index(player) logging.debug("put player %s (%s) in queue at position %s", player.name, player.id, queue_pos) # check if player can go directly to game if queue_pos == 0 and self.current_state in (self.game_replay, self.game_over): player.send(json.dumps({"game_start": True})) # update queue status for waiting players self.send_queue_status() def send_queue_status(self): """ Send queue update to all waiting players. Parameters ---------- none Returns ------- none """ if self._waiting_players: playing = "" if self._players: for player in self._players: playing += player.name waiting = "" for player in self._waiting_players: waiting += player.name message = json.dumps({"playing" : playing, "waiting" : waiting}) for player in self._waiting_players: player.send(message) def _start_game(self): """ Check is selected game is valid and create an instance of it. Parameters ---------- none Returns ------- none """ assert self._game_name in config.GAMES, f"invalid game: {self._game_name}" if self._game_name == "tetris": self._game_obj = TetrisGame(self._screen, self._players, self._game_seed) elif self._game_name == "snake": self._game_obj = SnakeGame(self._screen, self._players, self._game_seed) def update(self): """ Render another game frame, handle all events like new players, game over, etc. Parameters ---------- none Returns ------- none """ # generate next game frame finished = not self._game_obj.update() # find out what happens next if finished: # something has finished, find out what happens next if self.current_state == self.game_start: # game-start animation has finished if self._players: self.start_game_play() else: # this happens when player is disconnected during game-start animation self.start_game_replay() elif self.current_state == self.game_play: # game play has finished self.start_game_over() elif self.current_state == self.game_replay: # game replay play has finished self.start_game_over() elif self.current_state == self.game_over: # game-over animation has finished if len(self._waiting_players) > 0: # there's waiting players self.start_game_start() else: # no waiting players self.start_game_replay() else: if self._players: for player in self._players: if player.state == "disconnected": logging.warning("player %s (%s) is disconnected", player.name, player.id) if self.current_state == self.game_play: self.start_game_over() if self.current_state == self.game_replay and len(self._waiting_players) > 0: # a new player has been queued during game replay logging.debug("detected waiting player during game replay") self.start_game_over() def _save_game(self): """ Save replay data to file. Parameters ---------- none Returns ------- none """ logging.debug("saving game") # create data structure data = {} data["meta"] = {} data["meta"]["random_seed"] = self._game_seed data["meta"]["game"] = self._game_name data["meta"]["players"] = len(self._players) for i, player in enumerate(self._players): data[f"player{i}"] = player.replay # write to file filename = f"{config.REPLAY_PATH}/{self._game_name}/{self._game_seed}.json" logging.info("writing to file: %s", filename) with open(filename, "w", encoding="utf-8") as fp: json.dump(data, fp, separators = (',', ':')) def _load_game(self): """ Load replay data from a random stored game. Parameters ---------- none Returns ------- none """ logging.debug("loading game") # get a random replay file name random.seed() game = random.choice(config.GAMES) replay_dir = f"{config.REPLAY_PATH}/{game}" files = os.listdir(replay_dir) file = random.choice(files) filename = f"{replay_dir}/{file}" logging.debug("reading from file: %s", filename) # read from file with open(filename, "r", encoding="utf-8") as fp: data = json.load(fp) # break down data structure self._game_seed = data["meta"]["random_seed"] self._game_name = data["meta"]["game"] # create (re)player instances if "players" in data["meta"]: # new style num_players = data["meta"]["players"] for i in range(num_players): self._players.append(Replayer(data[f"player{i}"])) else: # old style self._players.append(Replayer(data["data"]))