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()