From 3138dfc99fedffce2ff499c6c8ba29b6995e2a9c Mon Sep 17 00:00:00 2001 From: Hans Goor Date: Fri, 13 Feb 2026 20:40:40 +0100 Subject: [PATCH] The vibe is real --- .gitignore | 216 +++++++++++++++++++++++++++++ quarto/__init__.py | 0 quarto/client/__init__.py | 3 + quarto/client/ai.py | 208 ++++++++++++++++++++++++++++ quarto/client/core.py | 215 +++++++++++++++++++++++++++++ quarto/client/main.py | 4 + quarto/client/net.py | 47 +++++++ quarto/client/protocol.py | 82 +++++++++++ quarto/server/__init__.py | 3 + quarto/server/core.py | 159 ++++++++++++++++++++++ quarto/server/main.py | 4 + quarto/server/match.py | 35 +++++ quarto/server/net.py | 70 ++++++++++ quarto/server/protocol.py | 73 ++++++++++ quarto/server/session.py | 238 ++++++++++++++++++++++++++++++++ quarto/shared/__init__.py | 3 + quarto/shared/game.py | 277 ++++++++++++++++++++++++++++++++++++++ 17 files changed, 1637 insertions(+) create mode 100644 .gitignore create mode 100644 quarto/__init__.py create mode 100644 quarto/client/__init__.py create mode 100644 quarto/client/ai.py create mode 100644 quarto/client/core.py create mode 100644 quarto/client/main.py create mode 100644 quarto/client/net.py create mode 100644 quarto/client/protocol.py create mode 100644 quarto/server/__init__.py create mode 100644 quarto/server/core.py create mode 100644 quarto/server/main.py create mode 100644 quarto/server/match.py create mode 100644 quarto/server/net.py create mode 100644 quarto/server/protocol.py create mode 100644 quarto/server/session.py create mode 100644 quarto/shared/__init__.py create mode 100644 quarto/shared/game.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e15106e --- /dev/null +++ b/.gitignore @@ -0,0 +1,216 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[codz] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py.cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +# Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +# poetry.lock +# poetry.toml + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. +# https://pdm-project.org/en/latest/usage/project/#working-with-version-control +# pdm.lock +# pdm.toml +.pdm-python +.pdm-build/ + +# pixi +# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. +# pixi.lock +# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one +# in the .venv directory. It is recommended not to include this directory in version control. +.pixi + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# Redis +*.rdb +*.aof +*.pid + +# RabbitMQ +mnesia/ +rabbitmq/ +rabbitmq-data/ + +# ActiveMQ +activemq-data/ + +# SageMath parsed files +*.sage.py + +# Environments +.env +.envrc +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +# .idea/ + +# Abstra +# Abstra is an AI-powered process automation framework. +# Ignore directories containing user credentials, local state, and settings. +# Learn more at https://abstra.io/docs +.abstra/ + +# Visual Studio Code +# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore +# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore +# and can be added to the global gitignore or merged into this file. However, if you prefer, +# you could uncomment the following to ignore the entire vscode folder +# .vscode/ + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file +.pypirc + +# Marimo +marimo/_static/ +marimo/_lsp/ +__marimo__/ + +# Streamlit +.streamlit/secrets.toml diff --git a/quarto/__init__.py b/quarto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/quarto/client/__init__.py b/quarto/client/__init__.py new file mode 100644 index 0000000..3a7338f --- /dev/null +++ b/quarto/client/__init__.py @@ -0,0 +1,3 @@ +""" +Client package for Quarto AI bot. +""" diff --git a/quarto/client/ai.py b/quarto/client/ai.py new file mode 100644 index 0000000..ba325f1 --- /dev/null +++ b/quarto/client/ai.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import math +import time +from typing import Optional, Tuple, List + +from quarto.shared.game import ( + GameState, + InitialMove, + NormalMove, + Player, + generate_initial_moves, + generate_normal_moves, + detect_quarto, +) + + +class QuartoAI: + """ + Time-bounded AI using iterative deepening alpha-beta search. + Designed to avoid obvious blunders and always return a legal move. + """ + + def __init__(self) -> None: + # Evaluation parameters + self.win_value = 10_000 + self.lose_value = -10_000 + + def choose_initial_piece(self, state: GameState, time_limit: float) -> InitialMove: + moves = generate_initial_moves(state) + if not moves: + raise RuntimeError("No initial moves available") + # For initial piece selection, use a simple heuristic: avoid obviously strong pieces + # like "balanced" ones (7, 8) – but this is minor; just pick first for now. + return moves[0] + + def choose_normal_move(self, state: GameState, time_limit: float) -> NormalMove: + legal_moves = generate_normal_moves(state) + if not legal_moves: + raise RuntimeError("No legal moves available") + + start = time.time() + deadline = start + max(0.1, time_limit - 0.2) + + best_move = legal_moves[0] + best_val = -math.inf + + depth = 1 + while True: + if time.time() >= deadline: + break + val, move = self._search_root(state, depth, deadline) + if move is not None: + best_move = move + best_val = val + depth += 1 + # Early stopping if decisive win found + if best_val >= self.win_value - 1: + break + + return best_move + + def _search_root( + self, state: GameState, depth: int, deadline: float + ) -> Tuple[float, Optional[NormalMove]]: + legal_moves = generate_normal_moves(state) + if not legal_moves: + return self._evaluate(state), None + + best_val = -math.inf + best_move: Optional[NormalMove] = None + alpha = -math.inf + beta = math.inf + + for move in legal_moves: + if time.time() >= deadline: + break + # Clone state for simulation + sim_state = self._clone_state(state) + self._apply_simulated_move(sim_state, move) + val = -self._alphabeta(sim_state, depth - 1, -beta, -alpha, deadline) + if val > best_val: + best_val = val + best_move = move + if val > alpha: + alpha = val + if alpha >= beta: + break + + return best_val, best_move + + def _alphabeta( + self, + state: GameState, + depth: int, + alpha: float, + beta: float, + deadline: float, + ) -> float: + if time.time() >= deadline: + # Return heuristic evaluation at cutoff + return self._evaluate(state) + + if state.game_over or depth == 0: + return self._evaluate(state) + + legal_moves = generate_normal_moves(state) + if not legal_moves: + return self._evaluate(state) + + value = -math.inf + for move in legal_moves: + if time.time() >= deadline: + break + sim_state = self._clone_state(state) + self._apply_simulated_move(sim_state, move) + score = -self._alphabeta(sim_state, depth - 1, -beta, -alpha, deadline) + if score > value: + value = score + if score > alpha: + alpha = score + if alpha >= beta: + break + return value + + def _evaluate(self, state: GameState) -> float: + """ + Static evaluation from perspective of current_player. + """ + if state.game_over: + if state.winner is None: + return 0.0 + return self.win_value if state.winner == state.current_player else self.lose_value + + # Heuristic: count potential lines for both players + board = state.board + my_score = 0 + opp_score = 0 + + from quarto.shared.game import LINES + + for line in LINES: + pieces = [board.squares[i] for i in line] + if all(p is not None for p in pieces): + # Completed line; check if it is a Quarto + if detect_quarto(board): + # If Quarto exists and nobody claimed, it's neutral until claimed. + # Slight bonus to current player for potential claim. + my_score += 20 + continue + # For partially filled lines, count how many shared attributes are still possible. + occupied = [p for p in pieces if p is not None] + if not occupied: + continue + common_bits = 0b1111 + p0 = occupied[0] + for p in occupied[1:]: + common_bits &= ~(p ^ p0) + if common_bits: + # More shared bits and more empty slots => more potential + empty_slots = sum(1 for p in pieces if p is None) + my_score += (bin(common_bits).count("1") * empty_slots) + + # Very rough balancing: opponent score approximated similarly by symmetry + # but we don't track explicit opponent perspective, so we just scale. + return float(my_score - opp_score) + + def _clone_state(self, state: GameState) -> GameState: + new_state = GameState() + new_state.board = state.board.clone() + new_state.current_player = state.current_player + new_state.assigned_piece = state.assigned_piece + new_state.remaining_pieces = set(state.remaining_pieces) + new_state.move_count = state.move_count + new_state.game_over = state.game_over + new_state.winner = state.winner + new_state.last_move_was_claim = state.last_move_was_claim + new_state.last_move_was_draw_marker = state.last_move_was_draw_marker + return new_state + + def _apply_simulated_move(self, state: GameState, move: NormalMove) -> None: + """ + Apply a move in simulation only; we don't need full validation here + because moves are generated from generate_normal_moves. + """ + from quarto.shared.game import _apply_normal_move_no_rules, detect_quarto + + # Apply the move structure + _apply_normal_move_no_rules(state, move) + + # Post-apply rules for claims and draw markers in simulation + if move.claim_quarto: + # Evaluate claim after placement + has_quarto = detect_quarto(state.board) + if has_quarto: + state.game_over = True + state.winner = state.current_player + else: + state.game_over = True + state.winner = state.current_player.other() + elif move.final_pass: + state.game_over = True + state.winner = None + else: + # If board full and no pieces left, it's draw + if state.board.is_full() and not state.remaining_pieces: + state.game_over = True + state.winner = None diff --git a/quarto/client/core.py b/quarto/client/core.py new file mode 100644 index 0000000..4a3dac2 --- /dev/null +++ b/quarto/client/core.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import argparse +import sys +import time +from typing import Optional + +from quarto.client.net import ClientConnection +from quarto.client.protocol import ( + parse_server_line, + format_hello, + format_queue, + format_move_initial, + format_move_normal, + HelloServer, + NewGame, + MoveBroadcast, + GameOver, + UnknownCommand, +) +from quarto.client.ai import QuartoAI +from quarto.shared.game import ( + GameState, + new_game_state, + InitialMove, + NormalMove, + Player, + apply_initial_move, + validate_and_apply_normal_move, +) + + +class ClientApp: + def __init__(self, host: str, port: int, name: str) -> None: + self.host = host + self.port = port + self.name = name + self.conn = ClientConnection(host, port) + self.ai = QuartoAI() + self.state: Optional[GameState] = None + self.our_player: Optional[Player] = None + + def run(self) -> None: + self.conn.connect() + try: + self._handshake() + while True: + self._queue_and_play_one_game() + finally: + self.conn.close() + + def _handshake(self) -> None: + self.conn.send_line(format_hello(self.name)) + line = self.conn.read_line() + if line is None: + print("Connection closed during HELLO", file=sys.stderr) + sys.exit(1) + msg = parse_server_line(line) + if not isinstance(msg, HelloServer): + print(f"Expected HELLO from server, got: {line}", file=sys.stderr) + sys.exit(1) + + def _queue_and_play_one_game(self) -> None: + self.conn.send_line(format_queue()) + # Wait for NEWGAME + while True: + line = self.conn.read_line() + if line is None: + print("Disconnected while waiting for NEWGAME", file=sys.stderr) + sys.exit(1) + msg = parse_server_line(line) + if isinstance(msg, NewGame): + self._start_game(msg) + break + # ignore others silently + + self._game_loop() + + def _start_game(self, newgame: NewGame) -> None: + self.state = new_game_state() + if newgame.player1 == self.name: + self.our_player = Player.PLAYER1 + elif newgame.player2 == self.name: + self.our_player = Player.PLAYER2 + else: + # Fallback: assume we are player1 if name mismatch + self.our_player = Player.PLAYER1 + + # If we are player1, we must immediately choose initial piece + if self.our_player == Player.PLAYER1: + assert self.state is not None + init_move = self.ai.choose_initial_piece(self.state, 5.0) + # Do NOT apply locally; wait for server broadcast MOVE~M + self.conn.send_line(format_move_initial(init_move.chosen_piece)) + + def _game_loop(self) -> None: + assert self.state is not None + while True: + line = self.conn.read_line() + if line is None: + print("Disconnected during game", file=sys.stderr) + sys.exit(1) + msg = parse_server_line(line) + + if isinstance(msg, MoveBroadcast): + self._handle_move_broadcast(msg) + elif isinstance(msg, GameOver): + # End of game; just break and re-queue + break + else: + # Ignore HELLO/NEWGAME/Unknown during game + continue + + def _handle_move_broadcast(self, msg: MoveBroadcast) -> None: + assert self.state is not None + state = self.state + + if msg.square is None: + # Initial move: a piece has been chosen for the next player. + # Both players receive this broadcast and must update state. + chosen_piece = msg.value + init_move = InitialMove(chosen_piece=chosen_piece) + apply_initial_move(state, init_move) + else: + # Normal move + square = msg.square + m_value = msg.value + if m_value == 16: + move = NormalMove(square=square, next_piece=None, + claim_quarto=True, final_pass=False) + elif m_value == 17: + move = NormalMove(square=square, next_piece=None, + claim_quarto=False, final_pass=True) + else: + move = NormalMove(square=square, next_piece=m_value, + claim_quarto=False, final_pass=False) + + # Keep our state consistent using shared validation logic + valid, _winner = validate_and_apply_normal_move(state, move) + if not valid: + # According to spec, server ensures opponent moves are valid; + # if this happens, log for debugging. + print(f"Received invalid MOVE from server: {msg}", file=sys.stderr) + + # After applying the broadcast move, if it's now our turn and game not over, act. + if not state.game_over and state.current_player == self.our_player: + self._do_our_turn() + + def _do_our_turn(self) -> None: + assert self.state is not None + state = self.state + if state.assigned_piece is None and state.move_count == 0: + # Very beginning and we are player1; handled in _start_game + return + + # Use AI to choose move within 5 seconds + move = self.ai.choose_normal_move(state, 5.0) + + # Validate locally via simulation before sending + tmp_state = self._clone_state(state) + valid, _winner = validate_and_apply_normal_move(tmp_state, move) + if not valid: + print("AI produced invalid move; falling back to first legal move", file=sys.stderr) + from quarto.shared.game import generate_normal_moves + + legal_moves = generate_normal_moves(state) + if not legal_moves: + print("No legal moves available", file=sys.stderr) + return + move = legal_moves[0] + + # Do NOT apply the move to self.state here. + # We wait for the server's MOVE broadcast, which we will handle + # in _handle_move_broadcast. + + # Send to server + if move.claim_quarto: + m_value = 16 + elif move.final_pass: + m_value = 17 + else: + assert move.next_piece is not None + m_value = move.next_piece + self.conn.send_line(format_move_normal(move.square, m_value)) + + def _clone_state(self, state: GameState) -> GameState: + from quarto.shared.game import GameState as GS, Board + + ns = GS() + ns.board = Board(state.board.squares.copy()) + ns.current_player = state.current_player + ns.assigned_piece = state.assigned_piece + ns.remaining_pieces = set(state.remaining_pieces) + ns.move_count = state.move_count + ns.game_over = state.game_over + ns.winner = state.winner + ns.last_move_was_claim = state.last_move_was_claim + ns.last_move_was_draw_marker = state.last_move_was_draw_marker + return ns + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--host", default="localhost") + parser.add_argument("--port", type=int, default=12345) + parser.add_argument("--name", default="QuartoBot", + help="Client name/description to send in HELLO") + args = parser.parse_args() + + client = ClientApp(host=args.host, port=args.port, name=args.name) + client.run() + + +if __name__ == "__main__": + main() diff --git a/quarto/client/main.py b/quarto/client/main.py new file mode 100644 index 0000000..7cca4f1 --- /dev/null +++ b/quarto/client/main.py @@ -0,0 +1,4 @@ +from quarto.client.core import main + +if __name__ == "__main__": + main() diff --git a/quarto/client/net.py b/quarto/client/net.py new file mode 100644 index 0000000..6fee17d --- /dev/null +++ b/quarto/client/net.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import socket +from typing import Optional + + +class ClientConnection: + def __init__(self, host: str, port: int, timeout: float = 30.0) -> None: + self.host = host + self.port = port + self.timeout = timeout + self.sock: Optional[socket.socket] = None + self._buffer = b"" + + def connect(self) -> None: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(self.timeout) + s.connect((self.host, self.port)) + self.sock = s + + def send_line(self, line: str) -> None: + if self.sock is None: + raise RuntimeError("Not connected") + data = (line + "\n").encode("utf-8") + self.sock.sendall(data) + + def read_line(self) -> Optional[str]: + if self.sock is None: + raise RuntimeError("Not connected") + while True: + if b"\n" in self._buffer: + line, self._buffer = self._buffer.split(b"\n", 1) + return line.decode("utf-8", errors="replace").rstrip("\r") + try: + chunk = self.sock.recv(4096) + except socket.timeout: + return None + if not chunk: + return None + self._buffer += chunk + + def close(self) -> None: + if self.sock is not None: + try: + self.sock.close() + finally: + self.sock = None diff --git a/quarto/client/protocol.py b/quarto/client/protocol.py new file mode 100644 index 0000000..d1a5255 --- /dev/null +++ b/quarto/client/protocol.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional, Union + + +@dataclass +class HelloServer: + description: str + + +@dataclass +class NewGame: + player1: str + player2: str + + +@dataclass +class MoveBroadcast: + square: Optional[int] + value: int # piece index or 16/17 + + +@dataclass +class GameOver: + reason: str # "VICTORY" | "DRAW" | "DISCONNECT" + winner: Optional[str] + + +@dataclass +class UnknownCommand: + raw: str + + +ServerMessage = Union[HelloServer, NewGame, MoveBroadcast, GameOver, UnknownCommand] + + +def parse_server_line(line: str) -> ServerMessage: + parts = line.strip().split("~") + if not parts: + return UnknownCommand(raw=line) + cmd = parts[0] + if cmd == "HELLO" and len(parts) == 2: + return HelloServer(description=parts[1]) + if cmd == "NEWGAME" and len(parts) == 3: + return NewGame(player1=parts[1], player2=parts[2]) + if cmd == "MOVE": + if len(parts) == 2: + try: + value = int(parts[1]) + except ValueError: + return UnknownCommand(raw=line) + return MoveBroadcast(square=None, value=value) + if len(parts) == 3: + try: + square = int(parts[1]) + value = int(parts[2]) + except ValueError: + return UnknownCommand(raw=line) + return MoveBroadcast(square=square, value=value) + if cmd == "GAMEOVER": + if len(parts) == 2: + return GameOver(reason=parts[1], winner=None) + if len(parts) == 3: + return GameOver(reason=parts[1], winner=parts[2]) + return UnknownCommand(raw=line) + + +def format_hello(description: str) -> str: + return f"HELLO~{description}" + + +def format_queue() -> str: + return "QUEUE" + + +def format_move_initial(piece: int) -> str: + return f"MOVE~{piece}" + + +def format_move_normal(square: int, m_value: int) -> str: + return f"MOVE~{square}~{m_value}" diff --git a/quarto/server/__init__.py b/quarto/server/__init__.py new file mode 100644 index 0000000..ad10b54 --- /dev/null +++ b/quarto/server/__init__.py @@ -0,0 +1,3 @@ +""" +Server package for Quarto tournament server. +""" diff --git a/quarto/server/core.py b/quarto/server/core.py new file mode 100644 index 0000000..513334f --- /dev/null +++ b/quarto/server/core.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import argparse +import threading +import sys +from typing import Optional + +from .net import ServerListener, ServerClient +from .protocol import ( + parse_client_line, + ClientHello, + ClientQueue, + ClientMove, + UnknownCommand, + format_hello, +) +from .match import Matchmaker +from .session import SessionManager + + +class ServerApp: + def __init__(self, port: int) -> None: + self.port = port + self.clients: list[ServerClient] = [] + self.listener = ServerListener(port) + self.matchmaker = Matchmaker() + self.sessions = SessionManager() + + def run(self) -> None: + def on_new_client(client: ServerClient) -> None: + self.clients.append(client) + + t = threading.Thread(target=self._handle_client, args=(client,), daemon=True) + t.start() + + self.listener.accept_loop(on_new_client) + + def _handle_client(self, client: ServerClient) -> None: + # Expect HELLO + line = client.read_line() + if line is None: + client.close() + return + cmd = parse_client_line(line) + if isinstance(cmd, ClientHello): + desired_desc = cmd.description or "" + + # Enforce unique descriptions: if name is already taken, add suffix + existing_names = {c.description for c in self.clients if c.description is not None} + final_desc = desired_desc + if final_desc in existing_names: + i = 2 + while f"{desired_desc}#{i}" in existing_names: + i += 1 + final_desc = f"{desired_desc}#{i}" + + client.description = final_desc + client.send_line(format_hello("QuartoServer")) + + print( + f"[SERVER] Client #{client.id} connected from {client.addr}, " + f"description='{client.description}' (requested '{desired_desc}')", + file=sys.stderr, + ) + else: + client.close() + return + try: + while True: + line = client.read_line() + if line is None: + # disconnect + self._handle_disconnect(client) + break + cmd = parse_client_line(line) + if isinstance(cmd, ClientQueue): + self._handle_queue(client) + elif isinstance(cmd, ClientMove): + self._handle_move(client, cmd) + elif isinstance(cmd, UnknownCommand): + # Silently ignore unknown commands + continue + else: + # Unknown type; ignore + continue + finally: + client.close() + + def _handle_queue(self, client: ServerClient) -> None: + # If already in a game, ignore + if client.in_game: + return + self.matchmaker.enqueue(client) + pair = self.matchmaker.dequeue_pair() + if pair is None: + return + c1, c2 = pair + c1.in_game = True + c2.in_game = True + session = self.sessions.create_session(c1, c2) + session.send_newgame() + + # LOG: new game created + print( + f"[SERVER] New game #{session.id} created: " + f"P1=#{c1.id} '{session.player1_name}' vs " + f"P2=#{c2.id} '{session.player2_name}'", + file=sys.stderr, + ) + + def _handle_move(self, client: ServerClient, move: ClientMove) -> None: + session = self.sessions.session_for_client(client) + if session is None: + # Not in a game; ignore + return + + # LOG: raw MOVE from client + print( + f"[SERVER] Game #{session.id} - Client #{client.id} " + f"('{session.player1_name if client is session.player1 else session.player2_name}') " + f"sends MOVE: square={move.square}, m_value={move.m_value}", + file=sys.stderr, + ) + + session.handle_client_move(client, move) + # If session ended, clean up + if session.state.game_over: + self.sessions.end_session(session) + session.player1.in_game = False + session.player2.in_game = False + + def _handle_disconnect(self, client: ServerClient) -> None: + # If in a game, inform session + session = self.sessions.session_for_client(client) + if session is not None: + session.handle_disconnect(client) + self.sessions.end_session(session) + session.player1.in_game = False + session.player2.in_game = False + # If in lobby, remove from queue + self.matchmaker.remove(client) + if client in self.clients: + self.clients.remove(client) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Quarto tournament server") + parser.add_argument("--port", type=int, default=5000, help="Listen port") + args = parser.parse_args() + + app = ServerApp(args.port) + try: + app.run() + except KeyboardInterrupt: + print("Server shutting down", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/quarto/server/main.py b/quarto/server/main.py new file mode 100644 index 0000000..25c09da --- /dev/null +++ b/quarto/server/main.py @@ -0,0 +1,4 @@ +from quarto.server.core import main + +if __name__ == "__main__": + main() diff --git a/quarto/server/match.py b/quarto/server/match.py new file mode 100644 index 0000000..563edb1 --- /dev/null +++ b/quarto/server/match.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import threading +from collections import deque +from typing import Deque, Optional, Tuple + +from .net import ServerClient + + +class Matchmaker: + def __init__(self) -> None: + self._queue: Deque[ServerClient] = deque() + self._lock = threading.Lock() + + def enqueue(self, client: ServerClient) -> None: + with self._lock: + # Avoid duplicates + if client in self._queue: + return + self._queue.append(client) + + def dequeue_pair(self) -> Optional[Tuple[ServerClient, ServerClient]]: + with self._lock: + if len(self._queue) >= 2: + c1 = self._queue.popleft() + c2 = self._queue.popleft() + return c1, c2 + return None + + def remove(self, client: ServerClient) -> None: + with self._lock: + try: + self._queue.remove(client) + except ValueError: + pass diff --git a/quarto/server/net.py b/quarto/server/net.py new file mode 100644 index 0000000..2a197de --- /dev/null +++ b/quarto/server/net.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import socket +import threading +from dataclasses import dataclass, field +from typing import Callable, Optional + + +@dataclass +class ServerClient: + sock: socket.socket + addr: tuple + id: int + description: Optional[str] = None + lock: threading.Lock = field(default_factory=threading.Lock) + buffer: bytes = b"" + in_game: bool = False + + def send_line(self, line: str) -> None: + data = (line + "\n").encode("utf-8") + with self.lock: + try: + self.sock.sendall(data) + except OSError: + pass + + def read_line(self) -> Optional[str]: + while True: + if b"\n" in self.buffer: + line, self.buffer = self.buffer.split(b"\n", 1) + return line.decode("utf-8", errors="replace").rstrip("\r") + try: + chunk = self.sock.recv(4096) + except OSError: + return None + if not chunk: + return None + self.buffer += chunk + + def close(self) -> None: + try: + self.sock.close() + except OSError: + pass + + +class ServerListener: + def __init__(self, port: int) -> None: + self.port = port + self._next_id = 1 + self._lock = threading.Lock() + + def _alloc_id(self) -> int: + with self._lock: + cid = self._next_id + self._next_id += 1 + return cid + + def accept_loop(self, on_new_client: Callable[[ServerClient], None]) -> None: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(("", self.port)) + srv.listen() + try: + while True: + conn, addr = srv.accept() + client = ServerClient(sock=conn, addr=addr, id=self._alloc_id()) + on_new_client(client) + finally: + srv.close() diff --git a/quarto/server/protocol.py b/quarto/server/protocol.py new file mode 100644 index 0000000..25322c5 --- /dev/null +++ b/quarto/server/protocol.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional, Union + + +@dataclass +class ClientHello: + description: str + + +class ClientQueue: + pass + + +@dataclass +class ClientMove: + square: Optional[int] # None for initial move MOVE~M + m_value: int # 0-17 + + +@dataclass +class UnknownCommand: + raw: str + + +ClientCommand = Union[ClientHello, ClientQueue, ClientMove, UnknownCommand] + + +def parse_client_line(line: str) -> ClientCommand: + parts = line.strip().split("~") + if not parts: + return UnknownCommand(raw=line) + cmd = parts[0] + if cmd == "HELLO" and len(parts) == 2: + return ClientHello(description=parts[1]) + if cmd == "QUEUE" and len(parts) == 1: + return ClientQueue() + if cmd == "MOVE": + if len(parts) == 2: + try: + m_value = int(parts[1]) + except ValueError: + return UnknownCommand(raw=line) + return ClientMove(square=None, m_value=m_value) + if len(parts) == 3: + try: + square = int(parts[1]) + m_value = int(parts[2]) + except ValueError: + return UnknownCommand(raw=line) + return ClientMove(square=square, m_value=m_value) + return UnknownCommand(raw=line) + + +def format_hello(description: str) -> str: + return f"HELLO~{description}" + + +def format_newgame(player1: str, player2: str) -> str: + return f"NEWGAME~{player1}~{player2}" + + +def format_move_broadcast(square: Optional[int], m_value: int) -> str: + if square is None: + return f"MOVE~{m_value}" + return f"MOVE~{square}~{m_value}" + + +def format_gameover(reason: str, winner: Optional[str]) -> str: + if winner is None: + return f"GAMEOVER~{reason}" + return f"GAMEOVER~{reason}~{winner}" diff --git a/quarto/server/session.py b/quarto/server/session.py new file mode 100644 index 0000000..51ce15b --- /dev/null +++ b/quarto/server/session.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import threading +import sys +from dataclasses import dataclass +from typing import Optional, Dict + +from .net import ServerClient +from .protocol import ClientMove, format_newgame, format_move_broadcast, format_gameover +from quarto.shared.game import ( + GameState, + new_game_state, + Player, + InitialMove, + NormalMove, + apply_initial_move, + validate_and_apply_normal_move, +) + + +@dataclass +class GameSession: + id: int + player1: ServerClient + player2: ServerClient + state: GameState + current_player: Player + lock: threading.Lock + + @property + def player1_name(self) -> str: + return self.player1.description or f"Player{self.player1.id}" + + @property + def player2_name(self) -> str: + return self.player2.description or f"Player{self.player2.id}" + + def send_newgame(self) -> None: + msg = format_newgame(self.player1_name, self.player2_name) + self.player1.send_line(msg) + self.player2.send_line(msg) + + def client_for_player(self, player: Player) -> ServerClient: + return self.player1 if player is Player.PLAYER1 else self.player2 + + def player_for_client(self, client: ServerClient) -> Optional[Player]: + if client is self.player1: + return Player.PLAYER1 + if client is self.player2: + return Player.PLAYER2 + return None + + def handle_client_move(self, client: ServerClient, move_cmd: ClientMove) -> None: + with self.lock: + # Ignore moves once game over + if self.state.game_over: + return + + player = self.player_for_client(client) + if player is None: + return + if player is not self.current_player: + # Not this player's turn; ignore + return + + # Initial move if move_count == 0 and square is None + if self.state.move_count == 0 and move_cmd.square is None: + m_value = move_cmd.m_value + if not (0 <= m_value <= 15): + return + if m_value not in self.state.remaining_pieces: + return + init_move = InitialMove(chosen_piece=m_value) + apply_initial_move(self.state, init_move) + + # LOG: initial move + print( + f"[SERVER] Game #{self.id} - {self.player1_name if player is Player.PLAYER1 else self.player2_name} " + f"(P{'1' if player is Player.PLAYER1 else '2'}) chooses initial piece {m_value}", + file=sys.stderr, + ) + + # Broadcast MOVE~M + msg = format_move_broadcast(square=None, m_value=m_value) + self.player1.send_line(msg) + self.player2.send_line(msg) + # After initial move, it's player2's turn + self.current_player = Player.PLAYER2 + return + + # From now on, we expect a normal move + if move_cmd.square is None: + return + + square = move_cmd.square + m_value = move_cmd.m_value + + if m_value == 16: + nm = NormalMove(square=square, next_piece=None, + claim_quarto=True, final_pass=False) + move_desc = f"place at {square} and CLAIM QUARTO (M=16)" + elif m_value == 17: + nm = NormalMove(square=square, next_piece=None, + claim_quarto=False, final_pass=True) + move_desc = f"place at {square} and FINAL PASS / DRAW MARKER (M=17)" + else: + nm = NormalMove(square=square, next_piece=m_value, + claim_quarto=False, final_pass=False) + move_desc = f"place at {square}, give piece {m_value}" + + valid, winner = validate_and_apply_normal_move(self.state, nm) + if not valid: + # Per spec, ignore invalid moves silently + print( + f"[SERVER] Game #{self.id} - INVALID MOVE from " + f"{self.player1_name if player is Player.PLAYER1 else self.player2_name}: " + f"square={square}, m_value={m_value}", + file=sys.stderr, + ) + return + + # LOG: valid move applied + print( + f"[SERVER] Game #{self.id} - " + f"{self.player1_name if player is Player.PLAYER1 else self.player2_name} " + f"(P{'1' if player is Player.PLAYER1 else '2'}) {move_desc}", + file=sys.stderr, + ) + + # Broadcast MOVE as received + self.player1.send_line(format_move_broadcast(square=square, m_value=m_value)) + self.player2.send_line(format_move_broadcast(square=square, m_value=m_value)) + + # After applying normal move, the GameState has already handled + # board full -> draw, claims, etc. + if self.state.game_over: + if self.state.winner is None: + # DRAW + # If this was not an explicit 17, we may need to broadcast 17 then draw + if not nm.final_pass and m_value != 17: + # Convert to final-pass broadcast per spec 8: + # board full & draw => broadcast M=17 immediately, then GAMEOVER + self.player1.send_line(format_move_broadcast(square=square, m_value=17)) + self.player2.send_line(format_move_broadcast(square=square, m_value=17)) + go = format_gameover("DRAW", None) + self.player1.send_line(go) + self.player2.send_line(go) + + # LOG: draw + print( + f"[SERVER] Game #{self.id} ended in DRAW", + file=sys.stderr, + ) + else: + winner_name = ( + self.player1_name if self.state.winner is Player.PLAYER1 else self.player2_name + ) + go = format_gameover("VICTORY", winner_name) + self.player1.send_line(go) + self.player2.send_line(go) + + # LOG: victory + print( + f"[SERVER] Game #{self.id} ended in VICTORY for {winner_name}", + file=sys.stderr, + ) + return + + # Game continues: switch current_player already done in GameState + self.current_player = self.state.current_player + + def handle_disconnect(self, client: ServerClient) -> None: + with self.lock: + if self.state.game_over: + return + # Determine remaining player + if client is self.player1 and self.player2 is not None: + winner_name = self.player2_name + elif client is self.player2 and self.player1 is not None: + winner_name = self.player1_name + else: + return + self.state.game_over = True + self.state.winner = None # gameover reason is DISCONNECT, winner is remaining + msg = format_gameover("DISCONNECT", winner_name) + if client is not self.player1: + self.player1.send_line(msg) + if client is not self.player2: + self.player2.send_line(msg) + + # LOG: disconnect + print( + f"[SERVER] Game #{self.id} ended due to DISCONNECT; remaining player '{winner_name}'", + file=sys.stderr, + ) + + +class SessionManager: + """ + Keeps track of active sessions and mapping from clients to sessions. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._next_id = 1 + self._sessions: Dict[int, GameSession] = {} + self._by_client: Dict[int, int] = {} # client.id -> session.id + + def create_session(self, p1: ServerClient, p2: ServerClient) -> GameSession: + with self._lock: + sid = self._next_id + self._next_id += 1 + state = new_game_state() + session = GameSession( + id=sid, + player1=p1, + player2=p2, + state=state, + current_player=Player.PLAYER1, + lock=threading.Lock(), + ) + self._sessions[sid] = session + self._by_client[p1.id] = sid + self._by_client[p2.id] = sid + return session + + def session_for_client(self, client: ServerClient) -> Optional[GameSession]: + with self._lock: + sid = self._by_client.get(client.id) + if sid is None: + return None + return self._sessions.get(sid) + + def end_session(self, session: GameSession) -> None: + with self._lock: + self._sessions.pop(session.id, None) + self._by_client.pop(session.player1.id, None) + self._by_client.pop(session.player2.id, None) diff --git a/quarto/shared/__init__.py b/quarto/shared/__init__.py new file mode 100644 index 0000000..548f586 --- /dev/null +++ b/quarto/shared/__init__.py @@ -0,0 +1,3 @@ +""" +Shared package for Quarto game logic used by both client and server. +""" diff --git a/quarto/shared/game.py b/quarto/shared/game.py new file mode 100644 index 0000000..08f0887 --- /dev/null +++ b/quarto/shared/game.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import List, Optional, Set, Tuple + + +class Player(Enum): + PLAYER1 = auto() + PLAYER2 = auto() + + def other(self) -> "Player": + return Player.PLAYER1 if self is Player.PLAYER2 else Player.PLAYER2 + + +# Precomputed Quarto lines (rows, columns, diagonals) +LINES: List[Tuple[int, int, int, int]] = [ + (0, 1, 2, 3), + (4, 5, 6, 7), + (8, 9, 10, 11), + (12, 13, 14, 15), + (0, 4, 8, 12), + (1, 5, 9, 13), + (2, 6, 10, 14), + (3, 7, 11, 15), + (0, 5, 10, 15), + (3, 6, 9, 12), +] + + +@dataclass +class Board: + squares: List[Optional[int]] = field(default_factory=lambda: [None] * 16) + + def clone(self) -> "Board": + return Board(self.squares.copy()) + + def is_full(self) -> bool: + return all(s is not None for s in self.squares) + + def place_piece(self, index: int, piece: int) -> None: + if self.squares[index] is not None: + raise ValueError("Square already occupied") + self.squares[index] = piece + + +@dataclass +class GameState: + board: Board = field(default_factory=Board) + current_player: Player = Player.PLAYER1 + assigned_piece: Optional[int] = None # piece given to current player + remaining_pieces: Set[int] = field(default_factory=lambda: set(range(16))) + move_count: int = 0 + game_over: bool = False + winner: Optional[Player] = None + last_move_was_claim: bool = False + last_move_was_draw_marker: bool = False + + +@dataclass +class InitialMove: + chosen_piece: int # 0-15 + + +@dataclass +class NormalMove: + square: int # 0-15 + next_piece: Optional[int] # 0-15 or None if claim/draw marker + claim_quarto: bool # M=16 + final_pass: bool # M=17 + + +def new_game_state() -> GameState: + return GameState() + + +def detect_quarto(board: Board) -> bool: + """ + Detect if there is any Quarto line on the board. + A line is a Quarto if all 4 squares occupied and pieces share at least one attribute bit. + """ + b = board.squares + for a, c, d, e in LINES: + p0 = b[a] + if p0 is None: + continue + p1, p2, p3 = b[c], b[d], b[e] + if p1 is None or p2 is None or p3 is None: + continue + # bits same between p and p0: ~(p ^ p0); intersect across all + common_bits = 0b1111 + for p in (p1, p2, p3): + common_bits &= ~(p ^ p0) + if common_bits & 0b1111: + return True + return False + + +def generate_initial_moves(state: GameState) -> List[InitialMove]: + if state.move_count != 0 or state.assigned_piece is not None: + return [] + return [InitialMove(piece) for piece in sorted(state.remaining_pieces)] + + +def apply_initial_move(state: GameState, move: InitialMove) -> None: + if state.move_count != 0: + raise ValueError("Initial move only allowed at start") + if move.chosen_piece not in state.remaining_pieces: + raise ValueError("Chosen piece not available") + # Give piece to opponent + state.remaining_pieces.remove(move.chosen_piece) + state.assigned_piece = move.chosen_piece + state.current_player = state.current_player.other() + state.move_count += 1 + state.last_move_was_claim = False + state.last_move_was_draw_marker = False + + +def _apply_normal_move_no_rules(state: GameState, move: NormalMove) -> None: + """ + Apply a normal move assuming it is already validated. + This is used by both rules engine and AI search. + """ + # Place assigned piece + piece_to_place = state.assigned_piece + if piece_to_place is None: + raise ValueError("No assigned piece to place") + state.board.place_piece(move.square, piece_to_place) + + # Remove placed piece from remaining if it hasn't been removed yet + state.remaining_pieces.discard(piece_to_place) + + state.move_count += 1 + + state.last_move_was_claim = move.claim_quarto + state.last_move_was_draw_marker = move.final_pass + + if move.claim_quarto: + # Claim: decide winner after this placement from the active game rules code + state.game_over = True + # winner will be set by validator using detect_quarto + elif move.final_pass: + # Final pass: board must be full; game is a draw by rules + state.game_over = True + state.winner = None + else: + # Normal continuing move: set next assigned piece and switch current player + if move.next_piece is None: + raise ValueError("next_piece must not be None when not claim/draw") + state.assigned_piece = move.next_piece + state.remaining_pieces.remove(move.next_piece) + state.current_player = state.current_player.other() + + +def generate_normal_moves(state: GameState) -> List[NormalMove]: + """ + Generate all logically legal moves (from the client's perspective) + for the current player, given assigned_piece. + This does not validate Quarto claims (AI can also skip generating + losing claims if desired). + """ + if state.assigned_piece is None or state.game_over: + return [] + + moves: List[NormalMove] = [] + board = state.board + assigned = state.assigned_piece + + empties = [i for i, p in enumerate(board.squares) if p is None] + remaining_without_assigned = sorted(state.remaining_pieces - {assigned}) + + for sq in empties: + # Option 1: normal move with giving a next piece + for np in remaining_without_assigned: + moves.append(NormalMove(square=sq, next_piece=np, + claim_quarto=False, final_pass=False)) + + # Option 2: claim Quarto after placing here, if it would be a real Quarto + # We simulate placement + temp_board = board.clone() + temp_board.place_piece(sq, assigned) + if detect_quarto(temp_board): + moves.append(NormalMove(square=sq, next_piece=None, + claim_quarto=True, final_pass=False)) + + # Option 3: if this placement fills the board and there are no remaining pieces, + # use final_pass (M=17) draw marker + would_be_full = all( + (temp_board.squares[i] is not None) for i in range(16) + ) + if would_be_full and not remaining_without_assigned: + moves.append(NormalMove(square=sq, next_piece=None, + claim_quarto=False, final_pass=True)) + + return moves + + +def validate_and_apply_normal_move(state: GameState, move: NormalMove) -> Tuple[bool, Optional[Player]]: + """ + Server-side: validate a MOVE~N~M relative to GameState and apply it + if valid. Returns (valid, winner_after_move). winner_after_move is: + - None if no winner yet or draw + - Player enum if there is a winner + Draw is represented by state.game_over and winner == None. + """ + if state.game_over: + return False, None + if state.assigned_piece is None: + return False, None + if not (0 <= move.square < 16): + return False, None + if state.board.squares[move.square] is not None: + return False, None + + assigned = state.assigned_piece + + # Validate markers and next_piece consistency + if move.claim_quarto and move.final_pass: + return False, None + + remaining_without_assigned = state.remaining_pieces - {assigned} + + if move.final_pass: + # M=17: allowed only if placing assigned piece fills board and no remaining pieces + temp_board = state.board.clone() + temp_board.place_piece(move.square, assigned) + if not temp_board.is_full(): + return False, None + if remaining_without_assigned: + return False, None + # Valid final-pass; apply + _apply_normal_move_no_rules(state, move) + # Per rules: even if a Quarto exists, it's still a draw + state.game_over = True + state.winner = None + return True, None + + if move.claim_quarto: + # M=16: claim Quarto; no next_piece allowed + if move.next_piece is not None: + return False, None + # Apply placement temporarily and check Quarto + temp_board = state.board.clone() + temp_board.place_piece(move.square, assigned) + has_quarto = detect_quarto(temp_board) + if not has_quarto: + # Mis-claim: loss for current player + _apply_normal_move_no_rules(state, move) + state.game_over = True + state.winner = state.current_player.other() + return True, state.winner + else: + # Correct claim: current player wins + _apply_normal_move_no_rules(state, move) + state.game_over = True + state.winner = state.current_player + return True, state.winner + + # Normal move with next_piece + if move.next_piece is None: + return False, None + if move.next_piece == assigned: + return False, None + if move.next_piece not in remaining_without_assigned: + return False, None + + _apply_normal_move_no_rules(state, move) + + # After a normal move, check if board full and no remaining pieces, but note: + # per protocol, last move should have been M=17; if client sent a normal piece, + # the server will instead post-process and convert to M=17 + DRAW (spec 8). + if state.board.is_full() and not state.remaining_pieces and not state.game_over: + # No one claimed; automatic draw per rule 8. + state.game_over = True + state.winner = None + + return True, state.winner