215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
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()
|