""" This file is part of the SplendidBear Websites' projects. Copyright (c) 2026 @ www.splendidbear.org For the full copyright and license information, please view the LICENSE file that was distributed with this source code. """ from __future__ import annotations import threading import time from collections.abc import Callable import gi gi.require_version("Gtk", "4.0") gi.require_version("Adw", "1") from gi.repository import Gtk, Adw, GLib from mineseeker.api import game as game_api from mineseeker.api.sse import SseListener from mineseeker import assets from mineseeker.constants import HEARTBEAT_INTERVAL_MS, WIN_THRESHOLD, PLAYER_RED, PLAYER_BLUE from mineseeker.state.game_state import GameState from mineseeker.state import session as session_mod from mineseeker.ui.grid_widget import GridWidget from mineseeker.ui.player_panel import PlayerPanel from mineseeker.ui.bonus_dialog import BonusDialog from mineseeker.ui.result_overlay import ResultOverlay class GamePage(Gtk.Overlay): """ Full game screen. Layout: [RedPanel] [GridWidget] [BluePanel] An Overlay places the ResultOverlay on top when the game ends. """ def __init__( self, game_assoc: str, mercure_jwt: str, color: str, on_leave: Callable[[], None], ) -> None: super().__init__() self._game_assoc = game_assoc self._mercure_jwt = mercure_jwt self._color = color # "red" | "blue" | "" (determined by subscribe) self._on_leave = on_leave self._state = GameState() self._bomb_mode = False self._step_start: float = time.monotonic() self._heartbeat_source: int | None = None # --- Layout --- main_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0) main_box.set_hexpand(True) main_box.set_vexpand(True) # Red player panel (left) self._red_panel = PlayerPanel( color=PLAYER_RED, is_local=(color == PLAYER_RED), on_bomb_toggle=self._on_bomb_toggle, on_resign=self._on_resign, ) main_box.append(self._red_panel) # Centre column: status bar + grid centre = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) centre.set_hexpand(True) centre.set_vexpand(True) # Status / turn label self._status_label = Gtk.Label(label="Connecting…") self._status_label.add_css_class("dim-label") self._status_label.set_margin_top(8) self._status_label.set_margin_bottom(8) centre.append(self._status_label) # Grid in a scrolled window so it never clips on small screens scrolled = Gtk.ScrolledWindow() scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) scrolled.set_hexpand(True) scrolled.set_vexpand(True) self._grid = GridWidget(on_cell_click=self._on_cell_click) scrolled.set_child(self._grid) centre.append(scrolled) # Bonus button row bonus_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) bonus_box.set_halign(Gtk.Align.CENTER) bonus_box.set_margin_top(6) bonus_box.set_margin_bottom(6) bonus_btn = Gtk.Button(label="Bonus Stats") bonus_btn.add_css_class("flat") bonus_btn.connect("clicked", self._show_bonus_dialog) bonus_box.append(bonus_btn) leave_btn = Gtk.Button(label="Leave") leave_btn.add_css_class("flat") leave_btn.connect("clicked", self._on_leave_clicked) bonus_box.append(leave_btn) centre.append(bonus_box) main_box.append(centre) # Blue player panel (right) self._blue_panel = PlayerPanel( color=PLAYER_BLUE, is_local=(color == PLAYER_BLUE), on_bomb_toggle=self._on_bomb_toggle, on_resign=self._on_resign, ) main_box.append(self._blue_panel) # Result overlay self._result_overlay = ResultOverlay( on_play_again=self._on_play_again, on_lobby=self._on_lobby, ) self.set_child(main_box) self.add_overlay(self._result_overlay) # Start async init threading.Thread(target=self._init_game, daemon=True).start() # ------------------------------------------------------------------ # Game initialisation # ------------------------------------------------------------------ def _init_game(self) -> None: """Connect, start SSE, join the channel, start/restore the game.""" try: # 1. Fetch existing game state connect_data = game_api.connect(self._game_assoc) GLib.idle_add(self._apply_connect_data, connect_data) # 2. Start the SSE listener self._sse = SseListener( game_assoc=self._game_assoc, mercure_jwt=self._mercure_jwt, on_subscribe=self._on_subscribe, on_unsubscribe=self._on_unsubscribe, on_topic=self._on_topic, on_challenge=self._on_challenge, on_challenge_response=self._on_challenge_response, on_heartbeat=self._on_heartbeat, ) self._sse.start() # 3. Join (announces presence via Mercure) game_api.join(self._game_assoc) # 4. If no existing game, create the grid if not connect_data.get("users"): game_api.start(self._game_assoc) # 5. Start heartbeat GLib.idle_add(self._start_heartbeat) except Exception as e: GLib.idle_add(self._set_status, f"Error: {e}") def _apply_connect_data(self, data: dict) -> bool: self._state.apply_connect(data) self._refresh_panels() self._grid.set_state(self._state) return GLib.SOURCE_REMOVE # ------------------------------------------------------------------ # SSE handlers (called on GTK main thread via GLib.idle_add) # ------------------------------------------------------------------ def _on_subscribe(self, payload: dict) -> None: """Two players connected → start game.""" users = payload.get("users", {}) user_cnt = payload.get("userCnt", 0) # Determine our colour if not yet assigned if not self._color: sess = session_mod.get() my_name = sess.username if my_name == users.get("blue") or my_name == users.get("blueAnon"): self._color = PLAYER_BLUE else: self._color = PLAYER_RED sess.color = self._color # Update player names from subscribe payload self._state.red.name = users.get("red", "") self._state.red.anon_name = users.get("redAnon", "") self._state.blue.name = users.get("blue", "") self._state.blue.anon_name = users.get("blueAnon", "") if user_cnt == 2: self._set_status("Game started!") assets.play_sound("starting") else: self._set_status("Waiting for opponent…") self._refresh_panels() def _on_unsubscribe(self, payload: dict) -> None: self._set_status("Opponent left the game.") def _on_topic(self, payload: dict) -> None: """A step was made — apply it and refresh.""" data = payload.get("data", {}) if not data: return player = data.get("player", "") is_mine = data.get("revealedCells") and any( rc.get("value") == "m" for rc in data["revealedCells"] ) # Play sounds if data.get("resign"): assets.play_sound("won") elif is_mine: my_state = self._state.red if player == PLAYER_RED else self._state.blue if my_state.mines > 20: assets.play_sound("warning") else: assets.play_sound("mine") else: assets.play_sound("click") self._state.apply_step(data) self._grid.refresh() self._refresh_panels() if self._state.finished: self._show_result() # uuid from server if data.get("uuid"): session_mod.get().game_assoc = data["uuid"] def _on_challenge(self, payload: dict) -> None: """Incoming challenge — show accept/decline dialog.""" challenger_name = payload.get("challengerName", "Someone") challenger_assoc = payload.get("challengerGameAssoc", "") GLib.idle_add(self._show_challenge_dialog, challenger_name, challenger_assoc) def _on_challenge_response(self, payload: dict) -> None: if payload.get("accepted"): # Switch to the new game assoc new_assoc = payload.get("targetGameAssoc", "") if new_assoc: GLib.idle_add(self._redirect_to_game, new_assoc) def _on_heartbeat(self, payload: dict) -> None: # Heartbeat from opponent received — game is live pass # ------------------------------------------------------------------ # Cell click / resign # ------------------------------------------------------------------ def _on_cell_click(self, row: int, col: int, bomb_mode: bool) -> None: if self._state.finished: return if self._state.turn != self._color: return # not our turn elapsed = time.monotonic() - self._step_start self._step_start = time.monotonic() threading.Thread( target=self._send_step, args=(row, col, bomb_mode, elapsed), daemon=True, ).start() def _send_step(self, row: int, col: int, bomb: bool, elapsed: float) -> None: try: result = game_api.step( game_assoc=self._game_assoc, coords=[row, col], player=self._color, bomb=bomb, resign=None, step_elapsed=elapsed, ) GLib.idle_add(self._apply_step_result, result) except Exception as e: GLib.idle_add(self._set_status, f"Step error: {e}") def _apply_step_result(self, data: dict) -> bool: self._state.apply_step(data) self._grid.refresh() self._refresh_panels() if self._bomb_mode: self._bomb_mode = False self._grid.set_bomb_mode(False) local_panel = self._red_panel if self._color == PLAYER_RED else self._blue_panel local_panel.reset_bomb_toggle() if self._state.finished: self._show_result() return GLib.SOURCE_REMOVE def _on_resign(self) -> None: threading.Thread(target=self._send_resign, daemon=True).start() def _send_resign(self) -> None: try: result = game_api.step( game_assoc=self._game_assoc, coords=[0, 0], player=self._color, bomb=False, resign=self._color, step_elapsed=0, ) GLib.idle_add(self._apply_step_result, result) except Exception: pass # ------------------------------------------------------------------ # Bomb toggle # ------------------------------------------------------------------ def _on_bomb_toggle(self, active: bool) -> None: self._bomb_mode = active self._grid.set_bomb_mode(active) # ------------------------------------------------------------------ # Heartbeat # ------------------------------------------------------------------ def _start_heartbeat(self) -> bool: interval_s = HEARTBEAT_INTERVAL_MS / 1000.0 self._heartbeat_source = GLib.timeout_add( HEARTBEAT_INTERVAL_MS, self._send_heartbeat, ) return GLib.SOURCE_REMOVE def _send_heartbeat(self) -> bool: if self._color: threading.Thread( target=game_api.heartbeat, args=(self._game_assoc, self._color), daemon=True, ).start() return GLib.SOURCE_CONTINUE # repeat # ------------------------------------------------------------------ # Result / game over # ------------------------------------------------------------------ def _show_result(self) -> None: assets.play_sound("won") self._result_overlay.show_result( winner=self._state.winner, resigned=self._state.resigned, local_color=self._color, red_mines=self._state.red.mines, blue_mines=self._state.blue.mines, red_name=self._state.red.display_name, blue_name=self._state.blue.display_name, ) self._stop_heartbeat() def _stop_heartbeat(self) -> None: if self._heartbeat_source is not None: GLib.source_remove(self._heartbeat_source) self._heartbeat_source = None # ------------------------------------------------------------------ # Navigation callbacks # ------------------------------------------------------------------ def _on_play_again(self) -> None: self._leave_game() self._on_leave() def _on_lobby(self) -> None: self._leave_game() self._on_leave() def _on_leave_clicked(self, *_) -> None: self._leave_game() self._on_leave() def _leave_game(self) -> None: self._stop_heartbeat() if hasattr(self, "_sse"): self._sse.stop() threading.Thread( target=game_api.leave, args=(self._game_assoc,), daemon=True ).start() def _redirect_to_game(self, new_assoc: str) -> bool: # Challenge accepted — leave current and open new game page self._leave_game() self._on_leave() return GLib.SOURCE_REMOVE # ------------------------------------------------------------------ # Challenge dialog # ------------------------------------------------------------------ def _show_challenge_dialog(self, challenger_name: str, challenger_assoc: str) -> bool: dialog = Adw.AlertDialog( heading=f"Challenge from {challenger_name}", body="Do you accept the challenge?", ) dialog.add_response("decline", "Decline") dialog.add_response("accept", "Accept") dialog.set_response_appearance("accept", Adw.ResponseAppearance.SUGGESTED) dialog.connect( "response", lambda d, resp: self._on_challenge_response_dialog(resp, challenger_assoc), ) dialog.present(self) return GLib.SOURCE_REMOVE def _on_challenge_response_dialog(self, response: str, challenger_assoc: str) -> None: accepted = response == "accept" threading.Thread( target=game_api.challenge_respond, args=(challenger_assoc, accepted, self._game_assoc), daemon=True, ).start() # ------------------------------------------------------------------ # Bonus dialog # ------------------------------------------------------------------ def _show_bonus_dialog(self, *_) -> None: BonusDialog( parent=self, red_name=self._state.red.display_name, blue_name=self._state.blue.display_name, red_points=self._state.red.bonus_points, blue_points=self._state.blue.bonus_points, red_stats=self._state.red.bonus_stats, blue_stats=self._state.blue.bonus_stats, ) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _refresh_panels(self) -> None: is_red_turn = self._state.turn == PLAYER_RED self._red_panel.update(self._state.red, is_red_turn) self._blue_panel.update(self._state.blue, not is_red_turn) self._grid.set_state(self._state) self._update_status_label() def _update_status_label(self) -> None: if self._state.finished: self._status_label.set_label("Game over") elif not self._color: self._status_label.set_label("Connecting…") elif self._state.turn == self._color: self._status_label.set_label("Your turn") else: opponent = self._state.blue if self._color == PLAYER_RED else self._state.red self._status_label.set_label(f"{opponent.display_name}'s turn") def _set_status(self, message: str) -> bool: self._status_label.set_label(message) return GLib.SOURCE_REMOVE