Private
Public Access
1
0

Compare commits

..

7 Commits

Author SHA1 Message Date
6484133199 new: dev: initialize the GTK client #11 2026-04-28 08:28:51 +02:00
199bb7e525 chg: pkg: fix all eslint issues - & add example of the testing env #10 2026-04-28 08:10:18 +02:00
5daaf71ae7 chg: pkg: new version release !skipChangelog 2026-04-23 21:42:21 +02:00
0aeec47996 new: pkg: add tracking code for the app #10
All checks were successful
Deploy to Production / deploy (push) Successful in 30s
2026-04-23 21:41:47 +02:00
3d67b8f2d9 chg: pkg: new version release !skipChangelog 2026-04-22 12:15:29 +02:00
dd9a190fd9 fix: usr: the error message cannot be seen during avatar changing #10
All checks were successful
Deploy to Production / deploy (push) Successful in 3m7s
2026-04-22 12:15:06 +02:00
f5e5019ea8 chg: pkg: new version release !skipChangelog 2026-04-21 22:47:04 +02:00
35 changed files with 2885 additions and 26 deletions

View File

@@ -1,6 +1,27 @@
# Changelog # Changelog
## v2026.2.9-0 (2026-04-23)
### New
* Add tracking code for the app #10. [Lang]
## v2026.2.8-3 (2026-04-22)
### Fix
* The error message cannot be seen during avatar changing #10. [Lang]
## v2026.2.8-2 (2026-04-21)
### Changes
* Increase the 2 MB avatar maximum file size to 10 MB #10. [Lang]
## v2026.2.8-1 (2026-04-21) ## v2026.2.8-1 (2026-04-21)
### Changes ### Changes

View File

@@ -28,6 +28,21 @@
box-shadow: 0 8px 48px rgba(0, 0, 0, 0.4); box-shadow: 0 8px 48px rgba(0, 0, 0, 0.4);
} }
#profile-avatar-root {
display: flex;
flex-direction: column;
align-items: center;
gap: 8px;
}
.profile-avatar__error {
font-size: 11px;
color: #e57373;
text-align: center;
max-width: 120px;
line-height: 1.3;
}
.profile-avatar { .profile-avatar {
position: relative; position: relative;
width: 80px; width: 80px;

View File

@@ -7,7 +7,7 @@
* file that was distributed with this source code. * file that was distributed with this source code.
*/ */
import React, { useMemo, useRef } from 'react'; import React, { Fragment, useMemo, useRef } from 'react';
import { string } from 'prop-types'; import { string } from 'prop-types';
import { useProfileDataProvider } from '@mine-hooks/useGameDataProvider'; import { useProfileDataProvider } from '@mine-hooks/useGameDataProvider';
@@ -16,6 +16,8 @@ export const AvatarUpload = ({ uploadUrl, initialThumbUrl, initials }) => {
const [thumbUrl, setThumbUrl] = React.useState(initialThumbUrl || null); const [thumbUrl, setThumbUrl] = React.useState(initialThumbUrl || null);
const { uploadAvatarMutation: { isPending, error, mutate } } = useProfileDataProvider(); const { uploadAvatarMutation: { isPending, error, mutate } } = useProfileDataProvider();
const errorMessage = useMemo(() => error?.message ?? null, [error]);
const handleChange = e => { const handleChange = e => {
const file = e.target.files?.[0]; const file = e.target.files?.[0];
if (!file) return; if (!file) return;
@@ -40,32 +42,32 @@ export const AvatarUpload = ({ uploadUrl, initialThumbUrl, initials }) => {
}); });
}; };
const errorMessage = useMemo(() => error?.message ?? null, [error]);
return ( return (
<div <Fragment>
className={`profile-avatar${isPending ? ' profile-avatar--loading' : ''}`} <div
title="Click to change profile picture" className={`profile-avatar${isPending ? ' profile-avatar--loading' : ''}`}
onClick={() => inputRef.current?.click()} title="Click to change profile picture"
> onClick={() => inputRef.current?.click()}
{thumbUrl >
? <img src={thumbUrl} alt={initials} className="profile-avatar__img" /> {thumbUrl
: <span className="profile-avatar__initials">{initials}</span> ? <img src={thumbUrl} alt={initials} className="profile-avatar__img" />
} : <span className="profile-avatar__initials">{initials}</span>
<div className="profile-avatar__overlay"> }
<i className="fa fa-camera" /> <div className="profile-avatar__overlay">
<i className="fa fa-camera" />
</div>
<input
ref={inputRef}
type="file"
accept="image/jpeg,image/png,image/gif,image/webp"
style={{ display: 'none' }}
onChange={handleChange}
/>
</div> </div>
<input
ref={inputRef}
type="file"
accept="image/jpeg,image/png,image/gif,image/webp"
style={{ display: 'none' }}
onChange={handleChange}
/>
{errorMessage && <div className="profile-avatar__error">{errorMessage}</div>} {errorMessage && <div className="profile-avatar__error">{errorMessage}</div>}
</div> </Fragment>
); );
} };
AvatarUpload.propTypes = { AvatarUpload.propTypes = {
uploadUrl: string.isRequired, uploadUrl: string.isRequired,

View File

@@ -212,7 +212,7 @@ const useServerCommunication = (gameAssoc, gameInherited, opponentName, isEnvDev
*/ */
if (!endRef.current && (!isTrueRestoredRef.current || 0 !== opponentLastSeenRef.current)) { if (!endRef.current && (!isTrueRestoredRef.current || 0 !== opponentLastSeenRef.current)) {
hideOverlay(); hideOverlay();
sounds.current.starting.play(); sounds.current.starting.play();
} }
}; };

View File

@@ -2,6 +2,25 @@
MineSeeker-specific testing setup and workflows. For general PHPUnit/Symfony testing, see [Symfony Testing Docs](https://symfony.com/doc/current/testing.html). MineSeeker-specific testing setup and workflows. For general PHPUnit/Symfony testing, see [Symfony Testing Docs](https://symfony.com/doc/current/testing.html).
## Example of the current tests
```shell
$ bin/phpunit (master->origin/master|✚1…2⚑1)
PHPUnit 13.1.7 by Sebastian Bergmann and contributors.
Runtime: PHP 8.5.5
Configuration: /var/www/splendid/Mine/phpunit.dist.xml
................................................................. 65 / 71 ( 91%)
...... 71 / 71 (100%)
Time: 00:07.319, Memory: 86.50 MB
OK (71 tests, 227 assertions)
Faker seed used: 918823
```
## Quick Start ## Quick Start
```bash ```bash
@@ -143,9 +162,9 @@ class MyControllerTest extends WebTestCase
{ {
$user = UserFactory::createOne(); $user = UserFactory::createOne();
$client = static::createClient(); $client = static::createClient();
$client->request('GET', '/profile'); $client->request('GET', '/profile');
self::assertResponseRedirects('/login'); self::assertResponseRedirects('/login');
} }
} }

8
gtk-client/.env.example Normal file
View File

@@ -0,0 +1,8 @@
# MineSeeker GTK4 Desktop Client — configuration
# Copy this file to .env and fill in your values.
# Base URL of the MineSeeker server (no trailing slash)
MINESEEKER_BASE_URL=https://mineseeker.example.com
# Public Mercure hub URL (SSE endpoint)
MINESEEKER_MERCURE_URL=https://mineseeker.example.com/.well-known/mercure

11
gtk-client/.gitignore vendored Normal file
View File

@@ -0,0 +1,11 @@
.env
__pycache__/
*.pyc
*.pyo
*.pyd
.Python
*.egg-info/
dist/
build/
.venv/
venv/

47
gtk-client/main.py Normal file
View File

@@ -0,0 +1,47 @@
"""
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.
"""
import sys
import logging
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
gi.require_version("Gst", "1.0")
from gi.repository import Gtk, Adw, Gst, GLib
# Validate config early (raises EnvironmentError if .env is missing)
from mineseeker import config # noqa: F401
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
class MineSeekerApp(Adw.Application):
def __init__(self) -> None:
super().__init__(application_id="org.splendidbear.mineseeker")
self.connect("activate", self._on_activate)
def _on_activate(self, app: Adw.Application) -> None:
# Import here so GTK/Adw is already initialised before building widgets
from mineseeker.ui.app_window import AppWindow
window = AppWindow(application=app)
window.present()
def main() -> int:
Gst.init(None)
app = MineSeekerApp()
return app.run(sys.argv)
if __name__ == "__main__":
sys.exit(main())

View File

View File

View File

@@ -0,0 +1,93 @@
"""
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
from mineseeker.api import client
class AuthError(Exception):
"""Raised on authentication failure."""
class TotpRequired(Exception):
"""Raised when the server requires a TOTP code after password login."""
def login(username: str, password: str) -> None:
"""
Authenticate with username + password via the dedicated JSON endpoint
POST /api/auth/login, which bypasses the reCAPTCHA gate.
Raises:
TotpRequired server confirmed credentials but TOTP is required next.
AuthError credentials wrong, account inactive, or server error.
"""
session = client.get_session()
resp = session.post(
client.url("/api/auth/login"),
json={"username": username, "password": password},
# The endpoint sets a session cookie; follow any redirects
allow_redirects=True,
)
# Non-2xx means a hard server error (500 etc.) — let it propagate
if resp.status_code >= 500:
resp.raise_for_status()
data = resp.json()
if not data.get("success"):
raise AuthError(data.get("error", "Login failed."))
if data.get("requiresTwoFactor"):
raise TotpRequired()
def submit_totp(code: str) -> None:
"""
Submit the 6-digit TOTP code after login() raises TotpRequired.
The scheb/2fa bundle processes POST /2fa_check directly as a firewall
listener — no CSRF token required, no JSON body. The code goes as a
form-encoded field named _auth_code, same as the browser form.
The LoginCaptchaListener already skips /2fa_check paths.
Raises:
AuthError if the code is wrong or the session is no longer in
IS_AUTHENTICATED_2FA_IN_PROGRESS state.
"""
session = client.get_session()
resp = session.post(
client.url("/2fa_check"),
data={"_auth_code": code},
headers={"Content-Type": "application/x-www-form-urlencoded"},
allow_redirects=True,
)
resp.raise_for_status()
# If we land back on /2fa the code was wrong
if "/2fa" in resp.url:
raise AuthError("Invalid authentication code.")
def login_as_guest() -> None:
"""
Start an anonymous session.
A GET to the homepage is enough for Symfony to create a session and
assign the anon_<session_id> identity used by ResolveUserNamesService.
"""
session = client.get_session()
resp = session.get(client.url("/"), headers={"Accept": "text/html"})
resp.raise_for_status()
def logout() -> None:
"""Discard the local session (client-side only)."""
client.reset_session()

View File

@@ -0,0 +1,52 @@
"""
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 requests
from mineseeker import config
# Module-level singleton session shared by all API modules.
# Holds cookies (Symfony session cookie after login) across all requests.
_session: requests.Session | None = None
def get_session() -> requests.Session:
global _session
if _session is None:
_session = requests.Session()
_session.headers.update({
"Accept": "application/json",
"Content-Type": "application/json",
})
return _session
def reset_session() -> None:
"""Discard the current session (logout / new guest session)."""
global _session
_session = None
def url(path: str) -> str:
"""Build an absolute URL from a server-relative path."""
return f"{config.BASE_URL}/{path.lstrip('/')}"
def get(path: str, **kwargs) -> requests.Response:
resp = get_session().get(url(path), **kwargs)
resp.raise_for_status()
return resp
def post(path: str, json: dict | None = None, **kwargs) -> requests.Response:
resp = get_session().post(url(path), json=json, **kwargs)
resp.raise_for_status()
return resp

View File

@@ -0,0 +1,117 @@
"""
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 base64
import json
from mineseeker.api import client
def fetch_token() -> dict:
"""
GET /api/game/token
Returns { "mercureJwt": str, "gameAssoc": str }
"""
resp = client.get("/api/game/token")
return resp.json()
def connect(game_assoc: str) -> dict:
"""
GET /api/game/connect/{gameAssoc}
Returns the decoded connect-information dict.
"""
resp = client.get(f"/api/game/connect/{game_assoc}")
raw = resp.text.strip()
decoded = base64.b64decode(raw).decode("utf-8")
return json.loads(decoded)
def start(game_assoc: str) -> bool:
"""POST /api/game/start — initialise the grid for a new game."""
resp = client.post("/api/game/start", json={"gameAssoc": game_assoc})
return bool(resp.json().get("success", False))
def join(game_assoc: str) -> bool:
"""POST /api/game/join/{gameAssoc} — announce this player's presence."""
resp = client.post(f"/api/game/join/{game_assoc}", json={})
return bool(resp.json().get("success", False))
def step(
game_assoc: str,
coords: list[int],
player: str,
bomb: bool,
resign: str | None,
step_elapsed: float,
) -> dict:
"""
POST /api/game/step/{gameAssoc}
Returns the full step result dict published by TopicManager::publish().
"""
resp = client.post(
f"/api/game/step/{game_assoc}",
json={
"coords": coords,
"player": player,
"bomb": bomb,
"resign": resign,
"stepElapsed": step_elapsed,
},
)
return resp.json()
def leave(game_assoc: str) -> None:
"""POST /api/game/leave/{gameAssoc} — fire-and-forget on window close."""
try:
client.post(f"/api/game/leave/{game_assoc}", json={})
except Exception:
pass # best-effort
def heartbeat(game_assoc: str, color: str) -> None:
"""POST /api/game/heartbeat/{gameAssoc} — keep-alive ping."""
try:
client.post(f"/api/game/heartbeat/{game_assoc}", json={"color": color})
except Exception:
pass # best-effort
def waiting() -> list[dict]:
"""GET /api/game/waiting — list of waiting players in the lobby."""
resp = client.get("/api/game/waiting")
return resp.json()
def challenge(target_game_assoc: str, challenger_game_assoc: str) -> None:
"""POST /api/game/challenge/{targetGameAssoc}"""
client.post(
f"/api/game/challenge/{target_game_assoc}",
json={"challengerGameAssoc": challenger_game_assoc},
)
def challenge_respond(
challenger_game_assoc: str,
accepted: bool,
target_game_assoc: str,
) -> None:
"""POST /api/game/challenge/respond/{challengerGameAssoc}"""
client.post(
f"/api/game/challenge/respond/{challenger_game_assoc}",
json={
"accepted": accepted,
"targetGameAssoc": target_game_assoc,
},
)

View File

@@ -0,0 +1,163 @@
"""
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 json
import logging
import threading
import time
from collections.abc import Callable
from typing import Any
import requests
from gi.repository import GLib
from mineseeker import config
from mineseeker.api import client
from mineseeker.constants import SSE_RECONNECT_INITIAL, SSE_RECONNECT_MAX
log = logging.getLogger(__name__)
class SseListener:
"""
Opens a Mercure SSE connection in a daemon thread and dispatches
parsed JSON messages back to the GTK main thread via GLib.idle_add().
Message routing mirrors useServerCommunication.jsx handleMercureMessage():
payload.type == "challenge" → on_challenge(payload)
payload.type == "challenge-response" → on_challenge_response(payload)
payload.type == "heartbeat" → on_heartbeat(payload)
"data" key present → on_topic(payload)
"msg" key present → on_unsubscribe(payload)
(none of the above) → on_subscribe(payload)
"""
def __init__(
self,
game_assoc: str,
mercure_jwt: str,
*,
on_subscribe: Callable[[dict], Any] | None = None,
on_unsubscribe: Callable[[dict], Any] | None = None,
on_topic: Callable[[dict], Any] | None = None,
on_challenge: Callable[[dict], Any] | None = None,
on_challenge_response: Callable[[dict], Any] | None = None,
on_heartbeat: Callable[[dict], Any] | None = None,
) -> None:
self._game_assoc = game_assoc
self._mercure_jwt = mercure_jwt
self._handlers = {
"subscribe": on_subscribe,
"unsubscribe": on_unsubscribe,
"topic": on_topic,
"challenge": on_challenge,
"challenge-response": on_challenge_response,
"heartbeat": on_heartbeat,
}
self._stop_event = threading.Event()
self._thread: threading.Thread | None = None
# ------------------------------------------------------------------
# Public control
# ------------------------------------------------------------------
def start(self) -> None:
"""Start the background SSE listener thread."""
self._stop_event.clear()
self._thread = threading.Thread(
target=self._run, daemon=True, name="sse-listener"
)
self._thread.start()
def stop(self) -> None:
"""Signal the background thread to stop."""
self._stop_event.set()
# ------------------------------------------------------------------
# Background thread
# ------------------------------------------------------------------
def _build_url(self) -> str:
topic = f"mineseeker/channel/{self._game_assoc}"
return f"{config.MERCURE_URL}?topic={topic}"
def _run(self) -> None:
backoff = SSE_RECONNECT_INITIAL
while not self._stop_event.is_set():
try:
self._stream()
backoff = SSE_RECONNECT_INITIAL # reset on clean disconnect
except Exception as exc:
if self._stop_event.is_set():
break
log.warning("SSE connection lost (%s), reconnecting in %.1fs", exc, backoff)
time.sleep(backoff)
backoff = min(backoff * 2, SSE_RECONNECT_MAX)
def _stream(self) -> None:
"""Open the SSE stream and process events until stopped or error."""
url = self._build_url()
headers = {
"Authorization": f"Bearer {self._mercure_jwt}",
"Accept": "text/event-stream",
"Cache-Control": "no-cache",
}
# Use the requests session from client.py so cookies are included
resp = client.get_session().get(
url, headers=headers, stream=True, timeout=(10, None)
)
resp.raise_for_status()
# Parse the raw SSE stream manually (sseclient-py would also work
# but avoids an extra dependency on GLib-aware loops)
data_lines: list[str] = []
for raw_line in resp.iter_lines(decode_unicode=True):
if self._stop_event.is_set():
break
if raw_line.startswith("data:"):
data_lines.append(raw_line[5:].lstrip(" "))
elif raw_line == "" and data_lines:
# Empty line signals end of event — dispatch it
payload_str = "\n".join(data_lines)
data_lines = []
try:
payload = json.loads(payload_str)
GLib.idle_add(self._dispatch, payload)
except json.JSONDecodeError:
log.debug("Non-JSON SSE data ignored: %s", payload_str)
def _dispatch(self, payload: dict) -> bool:
"""Called on the GTK main thread via GLib.idle_add."""
msg_type = payload.get("type")
if msg_type == "challenge":
self._call("challenge", payload)
elif msg_type == "challenge-response":
self._call("challenge-response", payload)
elif msg_type == "heartbeat":
self._call("heartbeat", payload)
elif "data" in payload:
self._call("topic", payload)
elif "msg" in payload:
self._call("unsubscribe", payload)
else:
self._call("subscribe", payload)
return GLib.SOURCE_REMOVE # run once only
def _call(self, key: str, payload: dict) -> None:
handler = self._handlers.get(key)
if handler is not None:
try:
handler(payload)
except Exception:
log.exception("Error in SSE handler '%s'", key)

View File

@@ -0,0 +1,108 @@
"""
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 io
import logging
import gi
gi.require_version("GdkPixbuf", "2.0")
gi.require_version("Gst", "1.0")
from gi.repository import GdkPixbuf, Gst
from mineseeker import config
from mineseeker.api import client
from mineseeker.constants import IMAGE_NAMES, SOUND_NAMES
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Image cache { filename: GdkPixbuf.Pixbuf }
# ---------------------------------------------------------------------------
_images: dict[str, GdkPixbuf.Pixbuf] = {}
def load_images(cell_size: int = 40) -> None:
"""
Fetch all game images from the server and cache them as GdkPixbuf.Pixbuf.
Call once at startup (blocking; run in a thread if you want a splash screen).
"""
for name in IMAGE_NAMES:
url = f"{config.BASE_URL}/images/{name}"
try:
resp = client.get_session().get(url, timeout=10)
resp.raise_for_status()
loader = GdkPixbuf.PixbufLoader()
loader.write(resp.content)
loader.close()
pixbuf = loader.get_pixbuf()
# Scale to the cell size used by the grid widget
pixbuf = pixbuf.scale_simple(cell_size, cell_size, GdkPixbuf.InterpType.BILINEAR)
_images[name] = pixbuf
except Exception as exc:
log.warning("Could not load image %s: %s", name, exc)
def get_image(name: str) -> GdkPixbuf.Pixbuf | None:
"""Return a cached Pixbuf by filename, or None if not loaded."""
return _images.get(name)
def get_image_or_fallback(name: str, fallback: str) -> GdkPixbuf.Pixbuf | None:
return _images.get(name) or _images.get(fallback)
# ---------------------------------------------------------------------------
# Sound — via GStreamer playbin
# ---------------------------------------------------------------------------
_sounds: dict[str, str] = {} # { key: URI }
def load_sounds() -> None:
"""
Build the URI map for the six game sound effects.
GStreamer will stream them on-demand from the server.
"""
Gst.init(None)
for filename in SOUND_NAMES:
key = filename.split(".")[0] # "click", "bomb", etc.
_sounds[key] = f"{config.BASE_URL}/sound/{filename}"
def play_sound(key: str) -> None:
"""
Play a sound by key ("click", "mine", "warning", "bomb", "won", "starting").
Each call spawns a fresh GStreamer playbin — fire-and-forget.
"""
uri = _sounds.get(key)
if not uri:
return
try:
player = Gst.ElementFactory.make("playbin", None)
if player is None:
return
player.set_property("uri", uri)
player.set_state(Gst.State.PLAYING)
# Connect to bus to clean up after playback
bus = player.get_bus()
bus.add_signal_watch()
def _on_message(bus, msg, player=player):
if msg.type in (Gst.MessageType.EOS, Gst.MessageType.ERROR):
player.set_state(Gst.State.NULL)
bus.remove_signal_watch()
return True
bus.connect("message", _on_message)
except Exception as exc:
log.debug("Sound play failed (%s): %s", key, exc)

View File

@@ -0,0 +1,30 @@
"""
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.
"""
import os
from pathlib import Path
from dotenv import load_dotenv
# Load .env from the gtk-client/ directory (parent of this package)
_env_path = Path(__file__).parent.parent / ".env"
load_dotenv(dotenv_path=_env_path)
BASE_URL: str = os.environ.get("MINESEEKER_BASE_URL", "").rstrip("/")
MERCURE_URL: str = os.environ.get("MINESEEKER_MERCURE_URL", "").rstrip("/")
if not BASE_URL:
raise EnvironmentError(
"MINESEEKER_BASE_URL is not set. "
"Copy gtk-client/.env.example to gtk-client/.env and fill in the values."
)
if not MERCURE_URL:
# Fall back to <BASE_URL>/.well-known/mercure if not explicitly set
MERCURE_URL = f"{BASE_URL}/.well-known/mercure"

View File

@@ -0,0 +1,90 @@
"""
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.
"""
# Grid dimensions
GRID_ROWS: int = 16
GRID_COLS: int = 16
GRID_SIZE: int = GRID_ROWS * GRID_COLS # 256
# Game rules
TOTAL_MINES: int = 51
WIN_THRESHOLD: int = 26 # first player to reach this wins
# Cell pixel size in the grid widget (each cell rendered as a square)
CELL_SIZE: int = 40
# Player colours (match backend "red" / "blue" strings)
PLAYER_RED: str = "red"
PLAYER_BLUE: str = "blue"
# Bomb reveal diamond half-width (matches PHP getBombRadius / JS bombRadius)
BOMB_RADIUS: int = 2
# Heartbeat interval in milliseconds (mirrors JS 1500 ms)
HEARTBEAT_INTERVAL_MS: int = 1500
# SSE reconnect back-off (seconds)
SSE_RECONNECT_INITIAL: float = 1.0
SSE_RECONNECT_MAX: float = 30.0
# Bonus stat display labels (mirrors JS BONUS_LABELS)
BONUS_LABELS: dict[str, str] = {
"blindHits": "Blind Hits",
"chainBest": "Best Chain",
"chainCurrent": "Current Chain",
"lastMineHits": "Endgame Mines",
"edgeMines": "Edge Mines",
"biggestReveal": "Biggest Reveal",
}
# Image URL path fragments served from BASE_URL/images/
IMAGE_NAMES: list[str] = [
"bg-target-outbg.png",
"bg-bomb-outbg.png",
"bg-bomb-disabled-outbg.png",
"bg-bomb-exploded-outbg.png",
"bg-bomb-empty-outbg.png",
"bg-left-mine-outbg.png",
"bg-cursor-red-outbg.png",
"bg-cursor-blue-outbg.png",
"bg-figure-red-outbg.png",
"bg-figure-blue-outbg.png",
"bg-flag-red-outbg.png",
"bg-flag-blue-outbg.png",
"bg-last-red-outbg.png",
"bg-last-blue-outbg.png",
"bg-wave-1-outbg.png",
"bg-wave-2-outbg.png",
"bg-corner-outbg.png",
"bg-bomb-top-left-outbg.png",
"bg-bomb-top-center-outbg.png",
"bg-bomb-top-right-outbg.png",
"bg-bomb-middle-left-outbg.png",
"bg-bomb-middle-center-outbg.png",
"bg-bomb-middle-right-outbg.png",
"bg-bomb-bottom-left-outbg.png",
"bg-bomb-bottom-center-outbg.png",
"bg-bomb-bottom-right-outbg.png",
]
# Sound file names served from BASE_URL/sound/
SOUND_NAMES: list[str] = [
"click.mp3",
"bomb.mp3",
"mine.mp3",
"warning.mp3",
"won.mp3",
"starting.mp3",
]
# Bomb position image name helper
# horizontal: "top" | "middle" | "bottom"
# vertical: "left" | "center" | "right"
def bomb_pos_image(horizontal: str, vertical: str) -> str:
return f"bg-bomb-{horizontal}-{vertical}-outbg.png"

View File

View File

@@ -0,0 +1,267 @@
"""
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 random
from dataclasses import dataclass, field
from typing import Any
from mineseeker.constants import GRID_ROWS, GRID_COLS, WIN_THRESHOLD
# ---------------------------------------------------------------------------
# Cell representation
# ---------------------------------------------------------------------------
@dataclass
class Cell:
"""Mirrors a single entry in the JS cells array."""
row: int
col: int
# "hidden" | "safe" | "mine"
state: str = "hidden"
# Numeric adjacent-mine count (0-8) or "m" for mine, None when unknown
value: Any = None
# "red", "blue", or None — which player claimed this mine
owner: str | None = None
# Whether this was the last cell clicked (for highlight)
is_last: bool = False
# Wave background variant (1 or 2) for unrevealed cells
wave: int = 1
# Bomb position overlay strings ("top"/"middle"/"bottom", "left"/"center"/"right") or None
bomb_h: str | None = None
bomb_v: str | None = None
def _init_cells() -> list[list[Cell]]:
"""Create the initial 16×16 grid of hidden cells (random wave image)."""
return [
[Cell(row=r, col=c, wave=random.choice([1, 1, 2]))
for c in range(GRID_COLS)]
for r in range(GRID_ROWS)
]
# ---------------------------------------------------------------------------
# Bonus stats
# ---------------------------------------------------------------------------
@dataclass
class BonusStats:
blind_hits: int = 0
chain_best: int = 0
chain_current: int = 0
last_mine_hits: int = 0
edge_mines: int = 0
biggest_reveal: int = 0
@classmethod
def from_dict(cls, d: dict) -> "BonusStats":
return cls(
blind_hits=d.get("blindHits", 0),
chain_best=d.get("chainBest", 0),
chain_current=d.get("chainCurrent", 0),
last_mine_hits=d.get("lastMineHits", 0),
edge_mines=d.get("edgeMines", 0),
biggest_reveal=d.get("biggestReveal", 0),
)
# ---------------------------------------------------------------------------
# Player snapshot
# ---------------------------------------------------------------------------
@dataclass
class PlayerState:
name: str = ""
anon_name: str = ""
avatar_url: str | None = None
mines: int = 0
bonus_points: float = 0.0
bonus_stats: BonusStats = field(default_factory=BonusStats)
have_bomb: bool = True
bomb_used: bool = False
@property
def display_name(self) -> str:
return self.name or self.anon_name or "Guest"
@property
def bomb_enabled(self) -> bool:
"""Mirrors JS: bomb only enabled when player is NOT ahead."""
return self.have_bomb and not self.bomb_used
# ---------------------------------------------------------------------------
# Game state
# ---------------------------------------------------------------------------
@dataclass
class GameState:
cells: list[list[Cell]] = field(default_factory=_init_cells)
red: PlayerState = field(default_factory=PlayerState)
blue: PlayerState = field(default_factory=PlayerState)
# Whose turn it is ("red" or "blue") — blue always starts
turn: str = "blue"
# Has the game ended?
finished: bool = False
# Winner ("red", "blue", "draw", or None)
winner: str | None = None
# Resignation ("red" or "blue" resigned, or None)
resigned: str | None = None
# Shareable UUID assigned by the server after the first step
uuid: str | None = None
# The last step coordinates per player { "red": (r,c) | None, ... }
last_step: dict[str, tuple[int, int] | None] = field(
default_factory=lambda: {"red": None, "blue": None}
)
# ---------------------------------------------------------------------------
# Apply a step result from the server
# ---------------------------------------------------------------------------
def apply_step(self, data: dict) -> None:
"""
Update state from a step payload (TopicManager::publish() result).
Mirrors JS applyStep() in useServerCommunication.jsx.
"""
if data.get("resign"):
self.resigned = data["resign"]
self.finished = True
self.winner = "blue" if data["resign"] == "red" else "red"
if data.get("uuid"):
self.uuid = data["uuid"]
return
player: str = data.get("player", "")
coords = data.get("coords")
if coords:
self.last_step[player] = (coords[0], coords[1])
# Clear previous last-step highlights for this player
for row in self.cells:
for cell in row:
if cell.is_last and cell.owner == player:
cell.is_last = False
self.cells[coords[0]][coords[1]].is_last = True
# Reveal cells
for rc in data.get("revealedCells", []):
r, c, v = rc["row"], rc["col"], rc["value"]
cell = self.cells[r][c]
if v == "m":
cell.state = "mine"
cell.value = "m"
cell.owner = player
else:
cell.state = "safe"
cell.value = v
# Reveal leftover mines at game end
for rc in data.get("leftMines", []):
r, c = rc["row"], rc["col"]
cell = self.cells[r][c]
cell.state = "mine"
cell.value = "m"
# Scores
self.red.mines = data.get("redPoints", self.red.mines)
self.blue.mines = data.get("bluePoints", self.blue.mines)
self.red.bonus_points = data.get("redBonusPoints", self.red.bonus_points)
self.blue.bonus_points = data.get("blueBonusPoints", self.blue.bonus_points)
if "redBonusStats" in data:
self.red.bonus_stats = BonusStats.from_dict(data["redBonusStats"])
if "blueBonusStats" in data:
self.blue.bonus_stats = BonusStats.from_dict(data["blueBonusStats"])
if data.get("bomb"):
if player == "red":
self.red.bomb_used = True
else:
self.blue.bomb_used = True
if data.get("uuid") and not self.finished:
self.uuid = data["uuid"]
# Win check
if self.red.mines >= WIN_THRESHOLD:
self.finished = True
self.winner = "red"
elif self.blue.mines >= WIN_THRESHOLD:
self.finished = True
self.winner = "blue"
# Advance turn (switches after every move)
if not self.finished:
self.turn = "blue" if player == "red" else "red"
# ---------------------------------------------------------------------------
# Restore from server connect information
# ---------------------------------------------------------------------------
def apply_connect(self, data: dict) -> None:
"""
Restore an existing game from the /api/game/connect payload.
Mirrors JS wInit() in useServerCommunication.jsx.
"""
if not data.get("users"):
return # fresh game, nothing to restore
users = data["users"]
self.red.name = users.get("red", "")
self.red.anon_name = users.get("redAnon", "")
self.blue.name = users.get("blue", "")
self.blue.anon_name = users.get("blueAnon", "")
self.red.mines = data.get("redPoints", 0)
self.blue.mines = data.get("bluePoints", 0)
self.red.bonus_points = data.get("redBonusPoints", 0.0)
self.blue.bonus_points = data.get("blueBonusPoints", 0.0)
if data.get("redBonusStats"):
self.red.bonus_stats = BonusStats.from_dict(data["redBonusStats"])
if data.get("blueBonusStats"):
self.blue.bonus_stats = BonusStats.from_dict(data["blueBonusStats"])
# Restore revealed cells (enriched with player colour)
for rc in data.get("revealedCells") or []:
r, c, v = rc["row"], rc["col"], rc["value"]
p = rc.get("player")
cell = self.cells[r][c]
if v == "m":
cell.state = "mine"
cell.value = "m"
cell.owner = p
else:
cell.state = "safe"
cell.value = v
# Restore last-step highlights
last = data.get("lastStep", {})
for color in ("red", "blue"):
ls = last.get(color)
if ls:
r, c = ls["row"], ls["col"]
self.last_step[color] = (r, c)
self.cells[r][c].is_last = True
# Determine whose turn it is from mostRecentStep
mrs = data.get("mostRecentStep")
if mrs:
self.turn = "blue" if mrs["player"] == "red" else "red"
self.finished = bool(data.get("gameFinished", False))

View File

@@ -0,0 +1,45 @@
"""
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
from dataclasses import dataclass, field
@dataclass
class Session:
"""Holds the current player's identity and game association."""
# Username (real user) or "anon_<session_id>" for guests
username: str = ""
# Whether this is an authenticated (non-guest) user
is_authenticated: bool = False
# Current game association UUID
game_assoc: str = ""
# "red" or "blue" — assigned when both players are subscribed
color: str = ""
# Mercure subscriber JWT for the current game
mercure_jwt: str = ""
# Module-level singleton; reset on logout
_session: Session = Session()
def get() -> Session:
return _session
def reset() -> None:
global _session
_session = Session()

View File

View File

@@ -0,0 +1,109 @@
"""
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 gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Gtk, Adw
from mineseeker.ui.login_page import LoginPage
from mineseeker.ui.totp_page import TotpPage
from mineseeker.ui.lobby_page import LobbyPage
from mineseeker.ui.game_page import GamePage
class AppWindow(Adw.ApplicationWindow):
"""
Main application window containing a Gtk.Stack that navigates between:
- "login" : LoginPage
- "totp" : TotpPage
- "lobby" : LobbyPage
- "game" : GamePage (replaced on each new game)
"""
def __init__(self, application: Adw.Application) -> None:
super().__init__(application=application)
self.set_title("MineSeeker")
self.set_default_size(980, 680)
# Stack — child names serve as page IDs
self._stack = Gtk.Stack()
self._stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE)
self._stack.set_transition_duration(200)
# Build pages
self._login_page = LoginPage(on_success=self._on_login_success, on_guest=self._on_guest)
self._totp_page = TotpPage(on_success=self._on_totp_success, on_back=self._show_login)
self._lobby_page = LobbyPage(on_game_start=self._on_game_start)
self._game_page: GamePage | None = None
self._stack.add_named(self._login_page, "login")
self._stack.add_named(self._totp_page, "totp")
self._stack.add_named(self._lobby_page, "lobby")
# Wrap in a NavigationView-style container using Adw.ToolbarView
toolbar_view = Adw.ToolbarView()
header = Adw.HeaderBar()
toolbar_view.add_top_bar(header)
toolbar_view.set_content(self._stack)
self.set_content(toolbar_view)
# Start on the login page
self._stack.set_visible_child_name("login")
# ------------------------------------------------------------------
# Navigation helpers
# ------------------------------------------------------------------
def _show_login(self) -> None:
self._stack.set_visible_child_name("login")
def _show_totp(self) -> None:
self._stack.set_visible_child_name("totp")
def _show_lobby(self) -> None:
self._lobby_page.refresh()
self._stack.set_visible_child_name("lobby")
# ------------------------------------------------------------------
# Callbacks from child pages
# ------------------------------------------------------------------
def _on_login_success(self, needs_totp: bool) -> None:
if needs_totp:
self._show_totp()
else:
self._show_lobby()
def _on_guest(self) -> None:
self._show_lobby()
def _on_totp_success(self) -> None:
self._show_lobby()
def _on_game_start(self, game_assoc: str, mercure_jwt: str, color: str) -> None:
"""Replace or create the GamePage and switch to it."""
# Remove previous game page if present
if self._game_page is not None:
self._stack.remove(self._game_page)
self._game_page = GamePage(
game_assoc=game_assoc,
mercure_jwt=mercure_jwt,
color=color,
on_leave=self._on_game_leave,
)
self._stack.add_named(self._game_page, "game")
self._stack.set_visible_child_name("game")
def _on_game_leave(self) -> None:
self._show_lobby()

View File

@@ -0,0 +1,96 @@
"""
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 gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Gtk, Adw
from mineseeker.state.game_state import BonusStats
from mineseeker.constants import BONUS_LABELS
class BonusDialog(Adw.Dialog):
"""Modal dialog displaying bonus stats for both players."""
def __init__(
self,
parent: Gtk.Widget,
red_name: str,
blue_name: str,
red_points: float,
blue_points: float,
red_stats: BonusStats,
blue_stats: BonusStats,
) -> None:
super().__init__()
self.set_title("Bonus Statistics")
self.set_content_width(480)
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
header = Adw.HeaderBar()
box.append(header)
content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
content.set_margin_top(16)
content.set_margin_bottom(16)
content.set_margin_start(16)
content.set_margin_end(16)
# Totals row
totals = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
totals.add_css_class("card")
red_total = Gtk.Label(label=f"{red_name}: {red_points:.1f} pts")
red_total.add_css_class("red-player")
red_total.set_hexpand(True)
red_total.set_xalign(0)
red_total.set_margin_start(12)
red_total.set_margin_top(8)
red_total.set_margin_bottom(8)
totals.append(red_total)
blue_total = Gtk.Label(label=f"{blue_name}: {blue_points:.1f} pts")
blue_total.add_css_class("blue-player")
blue_total.set_hexpand(True)
blue_total.set_xalign(1)
blue_total.set_margin_end(12)
totals.append(blue_total)
content.append(totals)
# Per-stat rows
group = Adw.PreferencesGroup(title="Breakdown")
stat_fields = [
("blind_hits", red_stats.blind_hits, blue_stats.blind_hits),
("chain_best", red_stats.chain_best, blue_stats.chain_best),
("last_mine_hits",red_stats.last_mine_hits,blue_stats.last_mine_hits),
("edge_mines", red_stats.edge_mines, blue_stats.edge_mines),
("biggest_reveal",red_stats.biggest_reveal,blue_stats.biggest_reveal),
]
key_map = {
"blind_hits": "blindHits",
"chain_best": "chainBest",
"last_mine_hits": "lastMineHits",
"edge_mines": "edgeMines",
"biggest_reveal": "biggestReveal",
}
for field_name, rv, bv in stat_fields:
label = BONUS_LABELS.get(key_map[field_name], field_name)
row = Adw.ActionRow(title=label)
row.set_subtitle(f"Red: {rv} Blue: {bv}")
group.add(row)
content.append(group)
box.append(content)
self.set_child(box)
self.present(parent)

View File

@@ -0,0 +1,477 @@
"""
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

View File

@@ -0,0 +1,239 @@
"""
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
from collections.abc import Callable
from typing import Any
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0")
from gi.repository import Gtk, Gdk, GdkPixbuf
from mineseeker import assets
from mineseeker.constants import (
GRID_ROWS, GRID_COLS, CELL_SIZE, PLAYER_RED, PLAYER_BLUE, bomb_pos_image
)
from mineseeker.state.game_state import GameState, Cell
# ---------------------------------------------------------------------------
# Bomb diamond radius (mirrors JS bombRadius() / PHP getBombRadius())
# ---------------------------------------------------------------------------
def _bomb_cells(row: int, col: int) -> list[tuple[int, int]]:
"""Return all cells within the 5×5 diamond centred at (row, col)."""
result = []
for dr in range(-2, 3):
for dc in range(-2, 3):
if abs(dr) + abs(dc) <= 2:
r, c = row + dr, col + dc
if 0 <= r < GRID_ROWS and 0 <= c < GRID_COLS:
result.append((r, c))
return result
def _bomb_pos(dr: int, dc: int) -> tuple[str, str]:
"""Map (delta_row, delta_col) to (horizontal, vertical) overlay names."""
h = "top" if dr < 0 else ("bottom" if dr > 0 else "middle")
v = "left" if dc < 0 else ("right" if dc > 0 else "center")
return h, v
# ---------------------------------------------------------------------------
# GridWidget
# ---------------------------------------------------------------------------
class GridWidget(Gtk.DrawingArea):
"""
16×16 minesweeper grid rendered with Cairo + GdkPixbuf tile images.
Signals emitted (via callbacks, not GObject signals for simplicity):
on_cell_click(row, col, bomb_mode) — user clicked a cell
"""
def __init__(self, on_cell_click: Callable[[int, int, bool], None]) -> None:
super().__init__()
self._on_cell_click = on_cell_click
self._state: GameState | None = None
self._bomb_mode: bool = False
self._hover: tuple[int, int] | None = None # (row, col) under cursor
width = CELL_SIZE * GRID_COLS
height = CELL_SIZE * GRID_ROWS
self.set_content_width(width)
self.set_content_height(height)
self.set_draw_func(self._draw)
# Click gesture
click = Gtk.GestureClick()
click.connect("pressed", self._on_pressed)
self.add_controller(click)
# Motion controller for bomb hover preview
motion = Gtk.EventControllerMotion()
motion.connect("motion", self._on_motion)
motion.connect("leave", self._on_leave)
self.add_controller(motion)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def set_state(self, state: GameState) -> None:
self._state = state
self.queue_draw()
def set_bomb_mode(self, active: bool) -> None:
self._bomb_mode = active
self.queue_draw()
def refresh(self) -> None:
self.queue_draw()
# ------------------------------------------------------------------
# Drawing
# ------------------------------------------------------------------
def _draw(self, area, cr, width, height) -> None:
if self._state is None:
return
for r in range(GRID_ROWS):
for c in range(GRID_COLS):
cell = self._state.cells[r][c]
x = c * CELL_SIZE
y = r * CELL_SIZE
self._draw_cell(cr, x, y, cell, r, c)
# Bomb hover diamond overlay
if self._bomb_mode and self._hover:
hr, hc = self._hover
for (br, bc) in _bomb_cells(hr, hc):
dr, dc = br - hr, bc - hc
h_pos, v_pos = _bomb_pos(dr, dc)
img_name = bomb_pos_image(h_pos, v_pos)
pixbuf = assets.get_image(img_name)
if pixbuf:
self._paint_pixbuf(cr, bc * CELL_SIZE, br * CELL_SIZE, pixbuf)
def _draw_cell(self, cr, x: int, y: int, cell: Cell, row: int, col: int) -> None:
cs = CELL_SIZE
if cell.state == "hidden":
# Wave background
wave_name = f"bg-wave-{cell.wave}-outbg.png"
pixbuf = assets.get_image(wave_name) or assets.get_image("bg-wave-1-outbg.png")
if pixbuf:
self._paint_pixbuf(cr, x, y, pixbuf)
else:
# Fallback: solid dark tile
cr.set_source_rgb(0.15, 0.15, 0.25)
cr.rectangle(x, y, cs, cs)
cr.fill()
elif cell.state == "safe":
# Light tile with number
cr.set_source_rgb(0.85, 0.85, 0.85)
cr.rectangle(x, y, cs, cs)
cr.fill()
# Draw thin border
cr.set_source_rgb(0.6, 0.6, 0.6)
cr.set_line_width(0.5)
cr.rectangle(x + 0.5, y + 0.5, cs - 1, cs - 1)
cr.stroke()
if cell.value and cell.value != 0:
self._draw_number(cr, x, y, cs, int(cell.value))
elif cell.state == "mine":
# Mine flag — show the appropriate player flag
color = cell.owner or "red"
flag_name = f"bg-flag-{color}-outbg.png"
pixbuf = assets.get_image(flag_name)
if pixbuf:
self._paint_pixbuf(cr, x, y, pixbuf)
else:
# Fallback colour
if color == "red":
cr.set_source_rgb(0.8, 0.1, 0.1)
else:
cr.set_source_rgb(0.1, 0.3, 0.9)
cr.rectangle(x, y, cs, cs)
cr.fill()
# Last-step highlight overlay
if cell.is_last:
color = cell.owner or (self._state.turn if self._state else "red")
last_name = f"bg-last-{color}-outbg.png"
pixbuf = assets.get_image(last_name)
if pixbuf:
self._paint_pixbuf(cr, x, y, pixbuf)
# Target overlay on hover (non-bomb)
if not self._bomb_mode and self._hover == (row, col):
pixbuf = assets.get_image("bg-target-outbg.png")
if pixbuf:
self._paint_pixbuf(cr, x, y, pixbuf)
@staticmethod
def _paint_pixbuf(cr, x: int, y: int, pixbuf: GdkPixbuf.Pixbuf) -> None:
Gdk.cairo_set_source_pixbuf(cr, pixbuf, x, y)
cr.paint()
# Number colours matching standard minesweeper conventions
_NUM_COLOURS = {
1: (0.0, 0.0, 1.0),
2: (0.0, 0.5, 0.0),
3: (1.0, 0.0, 0.0),
4: (0.0, 0.0, 0.5),
5: (0.5, 0.0, 0.0),
6: (0.0, 0.5, 0.5),
7: (0.0, 0.0, 0.0),
8: (0.5, 0.5, 0.5),
}
def _draw_number(self, cr, x: int, y: int, cs: int, value: int) -> None:
r, g, b = self._NUM_COLOURS.get(value, (0, 0, 0))
cr.set_source_rgb(r, g, b)
cr.select_font_face("Sans", 0, 1) # normal, bold
cr.set_font_size(cs * 0.55)
text = str(value)
ext = cr.text_extents(text)
tx = x + (cs - ext.width) / 2 - ext.x_bearing
ty = y + (cs + ext.height) / 2 - ext.y_bearing - ext.height
cr.move_to(tx, ty)
cr.show_text(text)
# ------------------------------------------------------------------
# Input handlers
# ------------------------------------------------------------------
def _cell_at(self, px: float, py: float) -> tuple[int, int] | None:
col = int(px // CELL_SIZE)
row = int(py // CELL_SIZE)
if 0 <= row < GRID_ROWS and 0 <= col < GRID_COLS:
return row, col
return None
def _on_pressed(self, gesture, n_press, x, y) -> None:
pos = self._cell_at(x, y)
if pos and self._state and not self._state.finished:
self._on_cell_click(pos[0], pos[1], self._bomb_mode)
def _on_motion(self, controller, x, y) -> None:
pos = self._cell_at(x, y)
if pos != self._hover:
self._hover = pos
self.queue_draw()
def _on_leave(self, controller) -> None:
if self._hover is not None:
self._hover = None
self.queue_draw()

View File

@@ -0,0 +1,185 @@
"""
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
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 import assets
from mineseeker.state import session as session_mod
class LobbyPage(Gtk.Box):
"""
Lobby screen — shows waiting players and a "New Game" button.
Flow:
- "New Game" → fetch token → start game → on_game_start()
- Click a waiting player → challenge them → on_game_start() when accepted
"""
def __init__(self, on_game_start: Callable[[str, str, str], None]) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=0)
self._on_game_start = on_game_start
self._waiting: list[dict] = []
# Header bar action area
header_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
header_box.set_margin_top(12)
header_box.set_margin_bottom(12)
header_box.set_margin_start(16)
header_box.set_margin_end(16)
title = Gtk.Label(label="Lobby")
title.add_css_class("title-2")
title.set_hexpand(True)
title.set_xalign(0)
header_box.append(title)
self._refresh_btn = Gtk.Button(label="Refresh")
self._refresh_btn.connect("clicked", lambda *_: self.refresh())
header_box.append(self._refresh_btn)
new_game_btn = Gtk.Button(label="New Game")
new_game_btn.add_css_class("suggested-action")
new_game_btn.connect("clicked", self._on_new_game)
header_box.append(new_game_btn)
self.append(header_box)
self.append(Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL))
# Waiting players list
scrolled = Gtk.ScrolledWindow()
scrolled.set_vexpand(True)
scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
self._list_box = Gtk.ListBox()
self._list_box.set_selection_mode(Gtk.SelectionMode.NONE)
self._list_box.add_css_class("boxed-list")
self._list_box.set_margin_top(12)
self._list_box.set_margin_bottom(12)
self._list_box.set_margin_start(16)
self._list_box.set_margin_end(16)
scrolled.set_child(self._list_box)
self.append(scrolled)
self._status_label = Gtk.Label(label="No players waiting.")
self._status_label.add_css_class("dim-label")
self._status_label.set_margin_top(24)
self._status_label.set_visible(True)
self.append(self._status_label)
def refresh(self) -> None:
"""Fetch waiting players list from the server."""
self._refresh_btn.set_sensitive(False)
threading.Thread(target=self._do_refresh, daemon=True).start()
def _do_refresh(self) -> None:
try:
waiting = game_api.waiting()
GLib.idle_add(self._update_list, waiting)
except Exception:
GLib.idle_add(self._refresh_btn.set_sensitive, True)
def _update_list(self, waiting: list[dict]) -> bool:
self._waiting = waiting
# Clear existing rows
while True:
row = self._list_box.get_first_child()
if row is None:
break
self._list_box.remove(row)
my_assoc = session_mod.get().game_assoc
for player in waiting:
if player["gameAssoc"] == my_assoc:
continue # don't show ourselves
row = self._make_player_row(player)
self._list_box.append(row)
has_players = bool([p for p in waiting if p.get("gameAssoc") != my_assoc])
self._status_label.set_visible(not has_players)
self._refresh_btn.set_sensitive(True)
return GLib.SOURCE_REMOVE
def _make_player_row(self, player: dict) -> Adw.ActionRow:
row = Adw.ActionRow()
row.set_title(player.get("name", "Guest"))
row.set_subtitle(f"Waiting since {player.get('since', '')[:19].replace('T', ' ')}")
challenge_btn = Gtk.Button(label="Challenge")
challenge_btn.add_css_class("flat")
challenge_btn.set_valign(Gtk.Align.CENTER)
challenge_btn.connect(
"clicked",
lambda _btn, p=player: self._on_challenge(p),
)
row.add_suffix(challenge_btn)
return row
def _on_new_game(self, *_) -> None:
threading.Thread(target=self._do_new_game, daemon=True).start()
def _do_new_game(self) -> None:
try:
token_data = game_api.fetch_token()
game_assoc = token_data["gameAssoc"]
mercure_jwt = token_data["mercureJwt"]
sess = session_mod.get()
sess.game_assoc = game_assoc
sess.mercure_jwt = mercure_jwt
sess.color = "red" # first player always red
# Load images while we wait for an opponent
assets.load_images()
GLib.idle_add(self._on_game_start, game_assoc, mercure_jwt, "red")
except Exception as e:
GLib.idle_add(self._show_error_toast, str(e))
def _on_challenge(self, player: dict) -> None:
threading.Thread(
target=self._do_challenge, args=(player,), daemon=True
).start()
def _do_challenge(self, player: dict) -> None:
try:
token_data = game_api.fetch_token()
game_assoc = token_data["gameAssoc"]
mercure_jwt = token_data["mercureJwt"]
sess = session_mod.get()
sess.game_assoc = game_assoc
sess.mercure_jwt = mercure_jwt
game_api.challenge(
target_game_assoc=player["gameAssoc"],
challenger_game_assoc=game_assoc,
)
assets.load_images()
# GamePage will determine color from subscribe payload
GLib.idle_add(self._on_game_start, game_assoc, mercure_jwt, "")
except Exception as e:
GLib.idle_add(self._show_error_toast, str(e))
def _show_error_toast(self, message: str) -> bool:
# Find the nearest Adw.ToastOverlay ancestor if available, otherwise print
print(f"[LobbyPage] Error: {message}")
return GLib.SOURCE_REMOVE

View File

@@ -0,0 +1,146 @@
"""
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
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.auth import login, login_as_guest, TotpRequired, AuthError
from mineseeker import assets
class LoginPage(Gtk.Box):
"""Username + password login form with a 'Play as Guest' option."""
def __init__(
self,
on_success: Callable[[bool], None],
on_guest: Callable[[], None],
) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=0)
self._on_success = on_success
self._on_guest = on_guest
self.set_valign(Gtk.Align.CENTER)
self.set_halign(Gtk.Align.CENTER)
clamp = Adw.Clamp()
clamp.set_maximum_size(360)
inner = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
inner.set_margin_top(32)
inner.set_margin_bottom(32)
inner.set_margin_start(16)
inner.set_margin_end(16)
# Title
title = Gtk.Label(label="MineSeeker")
title.add_css_class("title-1")
inner.append(title)
subtitle = Gtk.Label(label="Sign in to play")
subtitle.add_css_class("dim-label")
inner.append(subtitle)
# Credentials group
group = Adw.PreferencesGroup()
self._username_row = Adw.EntryRow(title="Username")
group.add(self._username_row)
self._password_row = Adw.PasswordEntryRow(title="Password")
self._password_row.connect("entry-activated", self._on_login_clicked)
group.add(self._password_row)
inner.append(group)
# Error label
self._error_label = Gtk.Label(label="")
self._error_label.add_css_class("error")
self._error_label.set_visible(False)
inner.append(self._error_label)
# Login button
self._login_btn = Gtk.Button(label="Sign In")
self._login_btn.add_css_class("suggested-action")
self._login_btn.add_css_class("pill")
self._login_btn.connect("clicked", self._on_login_clicked)
inner.append(self._login_btn)
# Separator
sep = Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL)
inner.append(sep)
# Guest button
guest_btn = Gtk.Button(label="Play as Guest")
guest_btn.add_css_class("pill")
guest_btn.connect("clicked", self._on_guest_clicked)
inner.append(guest_btn)
clamp.set_child(inner)
self.append(clamp)
def _set_busy(self, busy: bool) -> None:
self._login_btn.set_sensitive(not busy)
self._username_row.set_sensitive(not busy)
self._password_row.set_sensitive(not busy)
if busy:
self._error_label.set_visible(False)
def _show_error(self, message: str) -> None:
self._error_label.set_label(message)
self._error_label.set_visible(True)
def _on_login_clicked(self, *_) -> None:
username = self._username_row.get_text().strip()
password = self._password_row.get_text()
if not username or not password:
self._show_error("Please enter username and password.")
return
self._set_busy(True)
threading.Thread(
target=self._do_login, args=(username, password), daemon=True
).start()
def _do_login(self, username: str, password: str) -> None:
try:
login(username, password)
# Load assets after successful authentication
assets.load_sounds()
GLib.idle_add(self._on_success, False)
except TotpRequired:
assets.load_sounds()
GLib.idle_add(self._on_success, True)
except AuthError as e:
GLib.idle_add(self._handle_auth_error, str(e))
except Exception as e:
GLib.idle_add(self._handle_auth_error, f"Connection error: {e}")
def _handle_auth_error(self, message: str) -> bool:
self._set_busy(False)
self._show_error(message)
return GLib.SOURCE_REMOVE
def _on_guest_clicked(self, *_) -> None:
self._set_busy(True)
threading.Thread(target=self._do_guest, daemon=True).start()
def _do_guest(self) -> None:
try:
login_as_guest()
assets.load_sounds()
GLib.idle_add(self._on_guest)
except Exception as e:
GLib.idle_add(self._handle_auth_error, f"Could not start guest session: {e}")

View File

@@ -0,0 +1,116 @@
"""
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
from collections.abc import Callable
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Gtk, Adw
from mineseeker.state.game_state import PlayerState
from mineseeker.constants import WIN_THRESHOLD
class PlayerPanel(Gtk.Box):
"""
Vertical sidebar panel showing one player's info:
- Name + colour indicator
- Mine count (e.g. "12 / 26")
- Bonus points
- Bomb toggle button
- Resign button (only for the local player)
"""
def __init__(
self,
color: str,
is_local: bool,
on_bomb_toggle: Callable[[bool], None],
on_resign: Callable[[], None],
) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8)
self._color = color
self._is_local = is_local
self._on_bomb_toggle = on_bomb_toggle
self._on_resign = on_resign
self._bomb_active = False
self.set_margin_top(12)
self.set_margin_bottom(12)
self.set_margin_start(12)
self.set_margin_end(12)
self.set_valign(Gtk.Align.START)
# Colour dot + name
name_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
dot = Gtk.Label(label="")
dot.add_css_class("red-player" if color == "red" else "blue-player")
name_box.append(dot)
self._name_label = Gtk.Label(label="Waiting…")
self._name_label.add_css_class("title-4")
self._name_label.set_ellipsize(3) # PANGO_ELLIPSIZE_END
name_box.append(self._name_label)
self.append(name_box)
# Mine count
self._mine_label = Gtk.Label(label=f"0 / {WIN_THRESHOLD}")
self._mine_label.add_css_class("title-2")
self.append(self._mine_label)
# Bonus points
self._bonus_label = Gtk.Label(label="Bonus: 0")
self._bonus_label.add_css_class("dim-label")
self.append(self._bonus_label)
# Bomb button — only meaningful for local player
if is_local:
self._bomb_btn = Gtk.ToggleButton(label="Bomb")
self._bomb_btn.set_sensitive(False)
self._bomb_btn.connect("toggled", self._on_bomb_toggled)
self.append(self._bomb_btn)
sep = Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL)
self.append(sep)
resign_btn = Gtk.Button(label="Resign")
resign_btn.add_css_class("destructive-action")
resign_btn.connect("clicked", lambda *_: self._on_resign())
self.append(resign_btn)
# ------------------------------------------------------------------
# Update from state
# ------------------------------------------------------------------
def update(self, player: PlayerState, is_turn: bool) -> None:
self._name_label.set_label(player.display_name)
self._mine_label.set_label(f"{player.mines} / {WIN_THRESHOLD}")
self._bonus_label.set_label(f"Bonus: {player.bonus_points:.1f}")
if self._is_local and hasattr(self, "_bomb_btn"):
can_use = player.bomb_enabled and not player.bomb_used and is_turn
self._bomb_btn.set_sensitive(can_use)
if player.bomb_used:
self._bomb_btn.set_label("Bomb Used")
def set_bomb_enabled(self, enabled: bool) -> None:
if self._is_local and hasattr(self, "_bomb_btn"):
self._bomb_btn.set_sensitive(enabled)
def reset_bomb_toggle(self) -> None:
"""Deactivate the bomb toggle (after a bomb move is sent)."""
if self._is_local and hasattr(self, "_bomb_btn"):
self._bomb_btn.set_active(False)
def _on_bomb_toggled(self, btn: Gtk.ToggleButton) -> None:
self._bomb_active = btn.get_active()
self._on_bomb_toggle(self._bomb_active)

View File

@@ -0,0 +1,102 @@
"""
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
from collections.abc import Callable
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Gtk, Adw
class ResultOverlay(Gtk.Box):
"""
Translucent overlay shown at game end.
Displays the winner, final scores, and action buttons.
"""
def __init__(
self,
on_play_again: Callable[[], None],
on_lobby: Callable[[], None],
) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=16)
self._on_play_again = on_play_again
self._on_lobby = on_lobby
self.set_visible(False)
self.set_valign(Gtk.Align.CENTER)
self.set_halign(Gtk.Align.CENTER)
self.set_margin_top(16)
self.set_margin_bottom(16)
self.set_margin_start(16)
self.set_margin_end(16)
self.add_css_class("card")
self._title_label = Gtk.Label(label="")
self._title_label.add_css_class("title-1")
self.append(self._title_label)
self._subtitle_label = Gtk.Label(label="")
self._subtitle_label.add_css_class("title-3")
self.append(self._subtitle_label)
self._score_label = Gtk.Label(label="")
self._score_label.add_css_class("dim-label")
self.append(self._score_label)
btn_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
btn_box.set_halign(Gtk.Align.CENTER)
play_again_btn = Gtk.Button(label="Play Again")
play_again_btn.add_css_class("suggested-action")
play_again_btn.add_css_class("pill")
play_again_btn.connect("clicked", lambda *_: self._on_play_again())
btn_box.append(play_again_btn)
lobby_btn = Gtk.Button(label="Back to Lobby")
lobby_btn.add_css_class("pill")
lobby_btn.connect("clicked", lambda *_: self._on_lobby())
btn_box.append(lobby_btn)
self.append(btn_box)
def show_result(
self,
winner: str | None,
resigned: str | None,
local_color: str,
red_mines: int,
blue_mines: int,
red_name: str,
blue_name: str,
) -> None:
if resigned:
loser_name = red_name if resigned == "red" else blue_name
self._title_label.set_label("Resignation")
self._subtitle_label.set_label(f"{loser_name} resigned.")
elif winner == "draw" or winner is None:
self._title_label.set_label("Draw!")
self._subtitle_label.set_label("Equal mines — it's a draw.")
elif winner == local_color:
self._title_label.set_label("You Win!")
self._subtitle_label.set_label("Congratulations!")
else:
self._title_label.set_label("You Lose")
self._subtitle_label.set_label("Better luck next time.")
self._score_label.set_label(
f"{red_name}: {red_mines} mines · {blue_name}: {blue_mines} mines"
)
self.set_visible(True)
def hide_result(self) -> None:
self.set_visible(False)

View File

@@ -0,0 +1,112 @@
"""
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
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.auth import submit_totp, AuthError
class TotpPage(Gtk.Box):
"""6-digit TOTP code entry shown after a successful password login."""
def __init__(
self,
on_success: Callable[[], None],
on_back: Callable[[], None],
) -> None:
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=0)
self._on_success = on_success
self._on_back = on_back
self.set_valign(Gtk.Align.CENTER)
self.set_halign(Gtk.Align.CENTER)
clamp = Adw.Clamp()
clamp.set_maximum_size(360)
inner = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
inner.set_margin_top(32)
inner.set_margin_bottom(32)
inner.set_margin_start(16)
inner.set_margin_end(16)
title = Gtk.Label(label="Two-Factor Authentication")
title.add_css_class("title-2")
inner.append(title)
subtitle = Gtk.Label(label="Enter the 6-digit code from your authenticator app.")
subtitle.set_wrap(True)
subtitle.add_css_class("dim-label")
inner.append(subtitle)
group = Adw.PreferencesGroup()
self._code_row = Adw.EntryRow(title="Authentication Code")
self._code_row.set_input_purpose(Gtk.InputPurpose.DIGITS)
self._code_row.connect("entry-activated", self._on_verify_clicked)
group.add(self._code_row)
inner.append(group)
self._error_label = Gtk.Label(label="")
self._error_label.add_css_class("error")
self._error_label.set_visible(False)
inner.append(self._error_label)
self._verify_btn = Gtk.Button(label="Verify")
self._verify_btn.add_css_class("suggested-action")
self._verify_btn.add_css_class("pill")
self._verify_btn.connect("clicked", self._on_verify_clicked)
inner.append(self._verify_btn)
back_btn = Gtk.Button(label="Back to Login")
back_btn.add_css_class("pill")
back_btn.connect("clicked", lambda *_: self._on_back())
inner.append(back_btn)
clamp.set_child(inner)
self.append(clamp)
def _set_busy(self, busy: bool) -> None:
self._verify_btn.set_sensitive(not busy)
self._code_row.set_sensitive(not busy)
if busy:
self._error_label.set_visible(False)
def _show_error(self, message: str) -> None:
self._error_label.set_label(message)
self._error_label.set_visible(True)
def _on_verify_clicked(self, *_) -> None:
code = self._code_row.get_text().strip()
if len(code) != 6 or not code.isdigit():
self._show_error("Code must be exactly 6 digits.")
return
self._set_busy(True)
threading.Thread(target=self._do_verify, args=(code,), daemon=True).start()
def _do_verify(self, code: str) -> None:
try:
submit_totp(code)
GLib.idle_add(self._on_success)
except AuthError as e:
GLib.idle_add(self._handle_error, str(e))
except Exception as e:
GLib.idle_add(self._handle_error, f"Connection error: {e}")
def _handle_error(self, message: str) -> bool:
self._set_busy(False)
self._show_error(message)
return GLib.SOURCE_REMOVE

View File

@@ -0,0 +1,17 @@
# PyGObject is NOT installable from PyPI into a plain venv.
# It must come from your system package manager, e.g.:
# Debian/Ubuntu: sudo apt install python3-gi python3-gi-cairo gir1.2-gtk-4.0 gir1.2-adw-1
# Fedora: sudo dnf install python3-gobject gtk4 libadwaita
# Arch: sudo pacman -S python-gobject gtk4 libadwaita
#
# Create the venv with --system-site-packages so the system gi module is visible:
# python3 -m venv --system-site-packages .venv
#
# Then install only the pure-Python deps below:
requests>=2.31.0
sseclient-py>=1.8.0
python-dotenv>=1.0.0
# GStreamer for sound is also a system package:
# Debian/Ubuntu: sudo apt install gstreamer1.0-plugins-good gstreamer1.0-libav python3-gst-1.0
# Fedora: sudo dnf install gstreamer1-plugins-good python3-gstreamer1

31
gtk-client/run.sh Executable file
View File

@@ -0,0 +1,31 @@
#!/usr/bin/env bash
# Convenience launcher for the MineSeeker GTK4 desktop client.
# Creates the venv on first run, then launches main.py.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
VENV="$SCRIPT_DIR/.venv"
PYTHON="$VENV/bin/python"
# ── Create venv if missing ──────────────────────────────────────────────────
if [ ! -f "$PYTHON" ]; then
echo "[run.sh] Creating venv with --system-site-packages…"
python3 -m venv --system-site-packages "$VENV"
fi
# ── Install / update pure-Python deps ──────────────────────────────────────
echo "[run.sh] Installing dependencies…"
"$VENV/bin/pip" install --quiet requests sseclient-py python-dotenv
# ── Check .env ──────────────────────────────────────────────────────────────
if [ ! -f "$SCRIPT_DIR/.env" ]; then
echo ""
echo "ERROR: $SCRIPT_DIR/.env not found."
echo "Copy .env.example to .env and set MINESEEKER_BASE_URL."
exit 1
fi
# ── Launch ──────────────────────────────────────────────────────────────────
echo "[run.sh] Starting MineSeeker…"
exec "$PYTHON" "$SCRIPT_DIR/main.py" "$@"

View File

@@ -0,0 +1,106 @@
<?php declare(strict_types=1);
/*
* 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.
*/
namespace App\Controller;
use App\Entity\User;
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Bundle\SecurityBundle\Security;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\HttpKernel\Attribute\AsController;
use Symfony\Component\PasswordHasher\Hasher\UserPasswordHasherInterface;
use Symfony\Component\Routing\Attribute\Route;
/**
* Class ApiAuthController
*
* Provides a JSON login endpoint for native desktop clients.
* This endpoint is intentionally exempt from the reCAPTCHA listener
* because desktop clients cannot execute reCAPTCHA v3.
*
* After a successful password login, if the user has TOTP enabled the response
* returns { requiresTwoFactor: true }. The client must then POST the 6-digit
* code to the standard /2fa_check endpoint (which is already exempt from
* the reCAPTCHA listener via LoginCaptchaListener).
*
* @package App\Controller
* @author Lang <https://www.splendidbear.org>
* @category Class
* @license https://www.gnu.org/licenses/lgpl-3.0.en.html GNU Lesser General Public License
* @link www.splendidbear.org
* @since 2026. 04. 26.
*/
#[AsController]
class ApiAuthController extends AbstractController
{
public function __construct(
private readonly EntityManagerInterface $em,
private readonly UserPasswordHasherInterface $passwordHasher,
private readonly Security $security,
) {
}
/**
* POST /api/auth/login
*
* Request body (JSON): { "username": "...", "password": "..." }
*
* Responses:
* 200 { "success": true, "requiresTwoFactor": false }
* 200 { "success": true, "requiresTwoFactor": true }
* 400 { "success": false, "error": "..." }
* 401 { "success": false, "error": "..." }
*/
#[Route('/api/auth/login', name: 'MineSeekerBundle_api_auth_login', methods: ['POST'])]
public function login(Request $request): JsonResponse
{
$data = $request->toArray();
$username = trim($data['username'] ?? '');
$password = $data['password'] ?? '';
if ($username === '' || $password === '') {
return $this->json(
['success' => false, 'error' => 'Username and password are required.'],
Response::HTTP_BAD_REQUEST
);
}
/** @var User|null $user */
$user = $this->em->getRepository(User::class)->findOneBy(['username' => $username]);
if ($user === null || !$this->passwordHasher->isPasswordValid($user, $password)) {
return $this->json(
['success' => false, 'error' => 'Invalid username or password.'],
Response::HTTP_UNAUTHORIZED
);
}
if (!$user->isVerified) {
return $this->json(
['success' => false, 'error' => 'Account not yet activated. Check your email.'],
Response::HTTP_UNAUTHORIZED
);
}
// Log the user in via the Symfony security system.
// If TOTP is enabled, scheb/2fa will place the session into
// IS_AUTHENTICATED_2FA_IN_PROGRESS state, and the client must
// complete 2FA by POSTing the code to /2fa_check.
$this->security->login($user, 'form_login');
return $this->json([
'success' => true,
'requiresTwoFactor' => $user->isTotpAuthenticationEnabled(),
]);
}
}

View File

@@ -12,6 +12,7 @@ namespace App\Controller;
use App\Entity\PlayedGame; use App\Entity\PlayedGame;
use App\Repository\PlayedGameRepository; use App\Repository\PlayedGameRepository;
use App\Service\MercureJwtService;
use App\Service\ResolveUserNamesService; use App\Service\ResolveUserNamesService;
use App\Util\RpcManager; use App\Util\RpcManager;
use App\Util\TopicManager; use App\Util\TopicManager;
@@ -23,6 +24,7 @@ use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response; use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\HttpKernel\Attribute\AsController; use Symfony\Component\HttpKernel\Attribute\AsController;
use Symfony\Component\Routing\Attribute\Route; use Symfony\Component\Routing\Attribute\Route;
use Symfony\Component\Uid\Uuid;
/** /**
* Class MercureController * Class MercureController
@@ -45,9 +47,28 @@ class MercureController extends AbstractController
private readonly TopicManager $topicManager, private readonly TopicManager $topicManager,
private readonly RpcManager $rpcManager, private readonly RpcManager $rpcManager,
private readonly ResolveUserNamesService $userNamesService, private readonly ResolveUserNamesService $userNamesService,
private readonly MercureJwtService $mercureJwtService,
) { ) {
} }
/**
* Returns a fresh Mercure subscriber JWT and a new gameAssoc UUID.
* Intended for native desktop clients that cannot parse the JWT from HTML.
*
* Response: { "mercureJwt": "<jwt>", "gameAssoc": "<uuid>" }
*/
#[Route('/api/game/token', name: 'MineSeekerBundle_api_game_token', methods: ['GET'])]
public function token(): JsonResponse
{
$gameAssoc = Uuid::v4()->toRfc4122();
$userName = $this->userNamesService->resolveUserName();
return $this->json([
'mercureJwt' => $this->mercureJwtService->mintSubscriberToken($gameAssoc, $userName),
'gameAssoc' => $gameAssoc,
]);
}
#[Route('/api/game/start', name: 'MineSeekerBundle_api_game_start', methods: ['POST'])] #[Route('/api/game/start', name: 'MineSeekerBundle_api_game_start', methods: ['POST'])]
public function start(Request $request): JsonResponse public function start(Request $request): JsonResponse
{ {

View File

@@ -28,6 +28,20 @@
<link rel="icon" href="{{ asset('/images/favicon/favicon.ico') }}" type="image/x-icon"> <link rel="icon" href="{{ asset('/images/favicon/favicon.ico') }}" type="image/x-icon">
{% block metas %}{% endblock %} {% block metas %}{% endblock %}
<title>MineSeeker{% block title %}{% endblock %}</title> <title>MineSeeker{% block title %}{% endblock %}</title>
<script
defer src="https://umami.splendidbear.org/script.js"
data-website-id="825e02a9-d675-4cbd-9e68-72b98de2e4e9"
>
</script>
<script
defer
src="https://umami.splendidbear.org/recorder.js"
data-website-id="825e02a9-d675-4cbd-9e68-72b98de2e4e9"
data-sample-rate="0.15"
data-mask-level="moderate"
data-max-duration="300000"
>
</script>
{% block stylesheets %}{% endblock %} {% block stylesheets %}{% endblock %}
</head> </head>