import json import socket from datetime import datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Optional from source import path_save, path_history from source.core.enums import BombState from source.core.error import InvalidBombPosition, PositionAlreadyShot from source.gui.scene import GameResult from source.gui.scene.abc import Scene from source.gui import widget, texture, scene from source.network.packet import PacketChat, PacketBombPlaced, PacketBoatPlaced, PacketBombState, PacketQuit, \ PacketAskSave, PacketResponseSave from source.type import Point2D from source.utils import StoppableThread if TYPE_CHECKING: from source.gui.window import Window class Game(Scene): def __init__(self, window: "Window", thread: StoppableThread, connection: socket.socket, boats_length: list, name_ally: str, name_enemy: str, my_turn: bool, grid_width: int = None, grid_height: int = None, board_ally_data: dict = None, board_enemy_data: dict = None, history: list[bool, Point2D] = None, **kwargs): super().__init__(window, **kwargs) self.thread = thread self.connection = connection self.boats_length = boats_length self.name_ally = name_ally self.name_enemy = name_enemy self.grid_width = grid_width self.grid_height = grid_height self.background = self.add_widget( widget.Image, x=0, y=0, width=1.0, height=1.0, image=texture.Background.game, ) self.grid_ally = self.add_widget( widget.GameGrid, x=75, y=0.25, width=0.35, height=0.5, boats_length=self.boats_length, grid_style=texture.Grid.Style1, boat_style=texture.Grid.Boat.Style1, rows=self.grid_height, columns=self.grid_width, board_data=board_ally_data ) def board_ally_ready(widget): self.boat_ready_ally = True PacketBoatPlaced().send_connection(connection) self.grid_ally.add_listener("on_all_boats_placed", board_ally_ready) self.grid_enemy = self.add_widget( widget.GameGrid, x=lambda widget: widget.scene.window.width - 75 - widget.width, y=0.25, width=0.35, height=0.5, grid_style=texture.Grid.Style1, boat_style=texture.Grid.Boat.Style1, rows=self.grid_height, columns=self.grid_width, board_data=board_enemy_data ) def board_enemy_bomb(widget, cell: Point2D): if not (self.boat_ready_ally and self.boat_ready_enemy): return if not self.my_turn: return PacketBombPlaced(position=cell).send_connection(connection) self.my_turn = False self.grid_enemy.add_listener("on_request_place_bomb", board_enemy_bomb) self.add_widget( widget.Text, x=0.27, y=0.995, text=self.name_ally, font_size=20, anchor_x="center", anchor_y="center" ) self.add_widget( widget.Text, x=0.73, y=0.995, text=self.name_enemy, font_size=20, anchor_x="center", anchor_y="center" ) self.score_ally = self.add_widget( widget.Text, x=0.44, y=0.995, text="0", font_size=25, anchor_x="center", anchor_y="center" ) self.score_enemy = self.add_widget( widget.Text, x=0.56, y=0.995, text="0", font_size=25, anchor_x="center", anchor_y="center" ) self.chat_log = self.add_widget( widget.Text, x=10, y=45, width=0.4, text="", anchor_x="left", anchor_y="bottom", multiline=True ) self.chat_input = self.add_widget( widget.Input, x=10, y=10, width=0.5, height=30, type_regex=".{0,60}", style=texture.Button.Style1 ) def send_chat(widget): text: str = widget.text widget.text = "" self.chat_new_message(self.name_ally, text) PacketChat(message=text).send_connection(connection) self.chat_input.add_listener("on_enter", send_chat) self.button_save = self.add_widget( widget.Button, x=0.7, y=0, width=0.15, height=0.1, label_text="Sauvegarder", style=texture.Button.Style1 ) def ask_save(widget, x, y, button, modifiers): if not (self._boat_ready_ally and self._boat_ready_enemy): self.chat_new_message("System", "Veuillez poser vos bateaux avant de sauvegarder.", system=True) return PacketAskSave().send_connection(self.connection) self.chat_new_message("System", "demande de sauvegarde envoyé.", system=True) self.button_save.add_listener("on_click_release", ask_save) self.button_quit = self.add_widget( widget.Button, x=0.85, y=0, width=0.15, height=0.1, label_text="Quitter", style=texture.Button.Style1 ) self.button_quit.add_listener("on_click_release", lambda *_: self.window.add_scene(scene.GameQuit, game_scene=self)) self.label_state = self.add_widget( widget.Text, x=0.5, y=0.15, anchor_x="center", font_size=20 ) self._my_turn = my_turn # is it the player turn ? self._boat_ready_ally: bool = False # does the player finished placing his boat ? self._boat_ready_enemy: bool = False # does the opponent finished placing his boat ? self.history: list[tuple[bool, Point2D]] = [] if history is None else history # liste des bombes posées if len(boats_length) == 0: # s'il n'y a pas de bateau à placé self._boat_ready_ally = True # défini l'état de notre planche comme prête PacketBoatPlaced().send_connection(connection) # indique à l'adversaire que notre planche est prête self.save_cooldown: Optional[datetime] = None self._refresh_turn_text() self._refresh_score_text() # refresh def _refresh_chat_box(self): # supprime des messages jusqu'à ce que la boite soit plus petite que un quart de la fenêtre while self.chat_log.label.content_height > (self.window.height / 4): chat_logs: list[str] = self.chat_log.text.split("\n") self.chat_log.text = "\n".join(chat_logs[1:]) # ajuste la boite de message pour être collé en bas self.chat_log.label.y = self.chat_log.y + self.chat_log.label.content_height def _refresh_turn_text(self): self.label_state.text = ( "Placer vos bateaux" if not self.boat_ready_ally else "L'adversaire place ses bateaux..." if not self.boat_ready_enemy else "Placer vos bombes" if self.my_turn else "L'adversaire place ses bombes..." ) def _refresh_score_text(self): self.score_ally.text = str(self.grid_enemy.board.get_score()) self.score_enemy.text = str(self.grid_ally.board.get_score()) # property @property def my_turn(self): return self._my_turn @my_turn.setter def my_turn(self, my_turn: bool): self._my_turn = my_turn self._refresh_turn_text() @property def boat_ready_ally(self): return self._boat_ready_ally @boat_ready_ally.setter def boat_ready_ally(self, boat_ready_ally: bool): self._boat_ready_ally = boat_ready_ally self._refresh_turn_text() @property def boat_ready_enemy(self): return self._boat_ready_enemy @boat_ready_enemy.setter def boat_ready_enemy(self, boat_ready_enemy: bool): self._boat_ready_enemy = boat_ready_enemy self._refresh_turn_text() # function def to_json(self) -> dict: return { "name_ally": self.name_ally, "name_enemy": self.name_enemy, "my_turn": self.my_turn, "grid_ally": self.grid_ally.board.to_json(), "grid_enemy": self.grid_enemy.board.to_json(), "history": self.history, } @classmethod def from_json(cls, data: dict, window: "Window", thread: StoppableThread, connection: socket.socket) -> "Game": return cls( window=window, thread=thread, connection=connection, boats_length=[], name_ally=data["name_ally"], name_enemy=data["name_enemy"], my_turn=data["my_turn"], board_ally_data=data["grid_ally"], board_enemy_data=data["grid_enemy"], history=data["history"] ) @property def save_path(self) -> Path: ip_address, port = self.connection.getpeername() # Le nom du fichier est l'IP de l'opposent, suivi d'un entier indiquant si c'est à notre tour ou non. # Cet entier permet aux localhost de toujours pouvoir sauvegarder et charger sans problème. return path_save / ( ip_address + (f"-{int(self.my_turn)}" if ip_address == "127.0.0.1" else "") + ".bn-save" ) @property def history_path(self): return path_history / ( datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".bn-history" ) def save_to_path(self, path: Path): with open(path, "w", encoding="utf-8") as file: json.dump(self.to_json(), file, ensure_ascii=False) def save(self, value: bool): self.chat_new_message( "System", "Sauvegarde de la partie..." if value else "Sauvegarde de la partie refusé.", system=True ) if not value: return self.save_to_path(self.save_path) def game_end(self, won: bool): # s'il existe une ancienne sauvegarde, efface la self.save_path.unlink(missing_ok=True) # sauvegarde cette partie dans l'historique self.save_to_path(self.history_path) self.window.add_scene(GameResult, game_scene=self, won=won) # affiche le résultat self.thread.stop() # coupe la connexion def chat_new_message(self, author: str, content: str, system: bool = False): deco_left, deco_right = "<>" if system else "[]" message: str = f"{deco_left}{author}{deco_right} - {content}" self.chat_log.text += "\n" + message self._refresh_chat_box() # network def network_on_chat(self, packet: PacketChat): self.chat_new_message(self.name_enemy, packet.message) def network_on_boat_placed(self, packet: PacketBoatPlaced): self.boat_ready_enemy = True def network_on_bomb_placed(self, packet: PacketBombPlaced): try: # essaye de poser la bombe sur la grille alliée bomb_state = self.grid_ally.place_bomb(packet.position) except (InvalidBombPosition, PositionAlreadyShot): # si une erreur se produit, signale l'erreur bomb_state = BombState.ERROR # l'opposant va rejouer, ce n'est donc pas notre tour self.my_turn = False else: # c'est à notre tour si l'opposant à loupé sa bombe self.my_turn = not bomb_state.success # envoie le résultat à l'autre joueur PacketBombState(position=packet.position, bomb_state=bomb_state).send_connection(self.connection) # sauvegarde la bombe dans l'historique self.history.append((False, packet.position)) self._refresh_score_text() # le score a changé, donc rafraichi son texte if bomb_state is BombState.WON: # si l'ennemie à gagner, alors l'on a perdu self.game_end(won=False) def network_on_bomb_state(self, packet: PacketBombState): if packet.bomb_state is BombState.ERROR: # si une erreur est survenue, on rejoue self.my_turn = True return # on rejoue uniquement si la bombe à toucher self.my_turn = packet.bomb_state.success self._refresh_score_text() # le score a changé, donc rafraichi son texte # place la bombe sur la grille ennemie visuelle self.grid_enemy.place_bomb(packet.position, force_touched=packet.bomb_state.success) # sauvegarde la bombe dans l'historique self.history.append((True, packet.position)) if packet.bomb_state is BombState.WON: # si cette bombe a touché le dernier bateau, alors l'on a gagné self.game_end(won=True) def network_on_quit(self, packet: PacketQuit): self.thread.stop() # coupe la connexion from source.gui.scene import GameError self.window.set_scene(GameError, text="L'adversaire a quitté la partie.") def network_on_ask_save(self, packet: PacketAskSave): now = datetime.now() if self.save_cooldown is not None: # si un cooldown est défini if self.save_cooldown + timedelta(seconds=40) >= now: # si l'action à déjà été faite dans les 40 dernières secondes, ignore la requête de sauvegarde return self.save_cooldown = now from source.gui.scene import GameSave self.window.add_scene(GameSave, game_scene=self) def network_on_response_save(self, packet: PacketResponseSave): self.save(value=packet.value) # event def on_resize_after(self, width: int, height: int): self._refresh_chat_box()