import tkinter as tk from tkinter import messagebox import torch import chess import hashlib import numpy as np import subprocess import threading import time # ------------------------------------------------------------ # 1. Feature extraction (same as before) # ------------------------------------------------------------ def board_to_tensor(board: chess.Board) -> torch.Tensor: piece_map = board.piece_map() tensor = torch.zeros(64, 12, dtype=torch.float32) piece_to_idx = { chess.PAWN: 0, chess.KNIGHT: 1, chess.BISHOP: 2, chess.ROOK: 3, chess.QUEEN: 4, chess.KING: 5 } for sq, piece in piece_map.items(): idx = piece_to_idx[piece.piece_type] if piece.color == chess.BLACK: idx += 6 tensor[sq, idx] = 1.0 return tensor.flatten() # ------------------------------------------------------------ # 2. Telepathic memory generator (same) # ------------------------------------------------------------ class TelepathicChessMemory: def __init__(self, board_features: torch.Tensor, hidden_dim=32, gamma=0.2, epsilon=1e-4, max_iter=100): self.board = board_features self.hidden_dim = hidden_dim self.gamma = gamma self.epsilon = epsilon self.max_iter = max_iter self.W_board = torch.randn(len(board_features), hidden_dim) * 0.1 self.W_mem = torch.randn(hidden_dim, hidden_dim) * 0.1 self.xi = self._compute_xi() self.memory = torch.zeros(hidden_dim) def _compute_xi(self) -> torch.Tensor: board_bytes = self.board.numpy().tobytes() h = hashlib.md5(board_bytes).hexdigest() n = self.hidden_dim bytes_needed = n * 4 hex_bytes = h[:bytes_needed*2] if len(hex_bytes) < bytes_needed*2: hex_bytes = hex_bytes.ljust(bytes_needed*2, '0') arr = np.frombuffer(bytes.fromhex(hex_bytes), dtype=np.float32) if len(arr) < n: arr = np.pad(arr, (0, n - len(arr))) else: arr = arr[:n] xi = torch.tensor(arr) return xi / (xi.norm() + 1e-8) def telepath(self) -> torch.Tensor: board_proj = self.board @ self.W_board mem_proj = self.memory @ self.W_mem target = torch.sigmoid(board_proj + mem_proj) grad = target - torch.sigmoid(self.memory) return grad def addp(self, grad: torch.Tensor): self.memory += self.gamma * grad self.memory = torch.clamp(self.memory, -1.0, 1.0) def collapse_if_divergent(self) -> bool: if torch.norm(self.memory - self.xi) > 0.8: self.memory = self.xi.clone() return True return False def run(self) -> torch.Tensor: for _ in range(self.max_iter): prev = self.memory.clone() grad = self.telepath() self.addp(grad) self.collapse_if_divergent() if torch.norm(self.memory - prev) < self.epsilon: break return self.memory # ------------------------------------------------------------ # 3. GNUChess engine wrapper (XBoard protocol) # ------------------------------------------------------------ class GNUChessEngine: def __init__(self, master, status_callback): self.master = master self.status_callback = status_callback self.process = None self.engine_ready = False self.start_engine() def start_engine(self): try: self.process = subprocess.Popen( ["gnuchess", "-x"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1 ) # Send initial commands self._send_command("new") self._send_command("post") # show thinking self._send_command("level 0 0 0") # no time limit self._flush_output() self.engine_ready = True self.status_callback("Engine ready") except FileNotFoundError: self.engine_ready = False self.status_callback("GNUChess not found. Install 'gnuchess'.") messagebox.showerror("Engine Error", "GNUChess is not installed or not in PATH.\nPlease install it:\nsudo apt install gnuchess (Linux)\nbrew install gnuchess (Mac)\nDownload from https://www.gnu.org/software/chess/") def _send_command(self, cmd): if self.process: self.process.stdin.write(cmd + "\n") self.process.stdin.flush() def _flush_output(self): """Clear any pending output (non‑blocking).""" if self.process: import select while select.select([self.process.stdout], [], [], 0)[0]: self.process.stdout.readline() def make_move(self, uci_move): """Send a move to the engine (called after human moves).""" if not self.engine_ready: return None # Convert UCI (e2e4) to XBoard format (e2e4 same) # Force engine to stop thinking, send move, then think self._send_command("force") self._send_command(f"usermove {uci_move}") # Now let engine think self._send_command("go") # Wait for engine's move return self._read_engine_move() def _read_engine_move(self): """Read from stdout until we see 'My move is: ...'.""" if not self.process: return None while True: line = self.process.stdout.readline() if not line: return None if "My move is" in line: parts = line.split() # format: "My move is: e2e4" move_str = parts[-1].strip() return move_str # Also handle game end messages if "checkmate" in line or "stalemate" in line: self.status_callback(line.strip()) return None def set_position(self, fen): """Set engine position from FEN.""" if not self.engine_ready: return self._send_command("force") self._send_command(f"setboard {fen}") def quit(self): if self.process: self._send_command("quit") self.process.terminate() # ------------------------------------------------------------ # 4. GUI with engine # ------------------------------------------------------------ class TelepathicChessGUI: def __init__(self, master): self.master = master master.title("♜ Telepathic Chess Memory vs GNUChess ♞") master.geometry("950x650") self.board = chess.Board() self.selected_square = None self.prev_memory = None self.engine_thinking = False # Canvas for chess board self.canvas = tk.Canvas(master, width=500, height=500, bg="white") self.canvas.pack(side=tk.LEFT, padx=10, pady=10) self.canvas.bind("", self.on_click) # Right panel self.info_frame = tk.Frame(master) self.info_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=10, pady=10) self.metric_label = tk.Label(self.info_frame, text="Memory Metric", font=("Arial", 14, "bold")) self.metric_label.pack(pady=5) self.change_label = tk.Label(self.info_frame, text="Change: --", font=("Arial", 12)) self.change_label.pack(anchor="w") self.xi_label = tk.Label(self.info_frame, text="Distance to Ξ: --", font=("Arial", 12)) self.xi_label.pack(anchor="w") self.memory_values_label = tk.Label(self.info_frame, text="Memory (first 5): --", font=("Arial", 10), wraplength=300) self.memory_values_label.pack(anchor="w", pady=10) self.status_label = tk.Label(self.info_frame, text="Status: Initializing engine...", font=("Arial", 10), fg="blue") self.status_label.pack(pady=10) self.reset_button = tk.Button(self.info_frame, text="Reset Board", command=self.reset_board) self.reset_button.pack(pady=5) self.quit_button = tk.Button(self.info_frame, text="Quit", command=self.on_quit) self.quit_button.pack(pady=5) # Start engine self.engine = GNUChessEngine(self, self.update_status) self.update_status("Waiting for your move (White)") # Initial memory self.update_memory_for_current_board() self.draw_board() def update_status(self, msg): self.status_label.config(text=f"Status: {msg}") self.master.update_idletasks() def get_square_from_click(self, x, y): size = 500 // 8 col = x // size row = 7 - (y // size) if 0 <= col < 8 and 0 <= row < 8: return row * 8 + col return None def on_click(self, event): if self.engine_thinking: self.update_status("Engine is thinking, please wait...") return if self.board.turn != chess.WHITE: self.update_status("It's Black's turn (engine).") return sq = self.get_square_from_click(event.x, event.y) if sq is None: return if self.selected_square is None: piece = self.board.piece_at(sq) if piece and piece.color == chess.WHITE: self.selected_square = sq self.draw_board(highlight=sq) else: pass else: move = chess.Move(self.selected_square, sq) # Promotion handling if self.board.piece_at(self.selected_square) and self.board.piece_at(self.selected_square).piece_type == chess.PAWN: if chess.square_rank(sq) in (0, 7): # Ask user for promotion piece (simplified: always queen) move = chess.Move(self.selected_square, sq, promotion=chess.QUEEN) if move in self.board.legal_moves: self.board.push(move) self.update_memory_for_current_board() self.draw_board() self.selected_square = None # Check game over after human move if self.board.is_game_over(): self.game_over() return # Now let engine move self.engine_move() else: self.selected_square = None self.draw_board() def engine_move(self): if self.board.is_game_over(): return if self.board.turn != chess.BLACK: return self.engine_thinking = True self.update_status("Engine is thinking...") # Run engine move in a separate thread to avoid GUI freeze thread = threading.Thread(target=self._engine_move_thread, daemon=True) thread.start() def _engine_move_thread(self): # Set engine position self.engine.set_position(self.board.fen()) # Get best move uci_move = self.engine.make_move(None) # None because we already forced position # Actually we need to call make_move with last human move? Simpler: send "go" after setting position. # Let's reimplement: call engine.make_move with the move we just made? Better: engine has its own state. # Because we used "force" and "setboard", we must now send "go". if self.engine.process: self.engine._send_command("go") uci_move = self.engine._read_engine_move() self.master.after(0, lambda: self._apply_engine_move(uci_move)) def _apply_engine_move(self, uci_move): self.engine_thinking = False if uci_move is None: self.update_status("Engine failed to produce a move.") return try: move = chess.Move.from_uci(uci_move) if move in self.board.legal_moves: self.board.push(move) self.update_memory_for_current_board() self.draw_board() if self.board.is_game_over(): self.game_over() else: self.update_status("Your turn (White)") else: self.update_status(f"Engine proposed illegal move: {uci_move}") except Exception as e: self.update_status(f"Engine error: {e}") def game_over(self): result = self.board.result() msg = f"Game over: {result}" self.update_status(msg) messagebox.showinfo("Game Over", msg) self.reset_board() def update_memory_for_current_board(self): features = board_to_tensor(self.board) tmem = TelepathicChessMemory(features, hidden_dim=32, gamma=0.2) new_memory = tmem.run() if self.prev_memory is None: change = 0.0 else: change = torch.norm(new_memory - self.prev_memory).item() divergence_xi = torch.norm(new_memory - tmem.xi).item() mem_vals = new_memory[:5].tolist() mem_str = ", ".join(f"{v:.3f}" for v in mem_vals) self.change_label.config(text=f"Change from previous: {change:.4f}") self.xi_label.config(text=f"Distance to Ξ: {divergence_xi:.4f}") self.memory_values_label.config(text=f"Memory (first 5): [{mem_str}]") self.prev_memory = new_memory def draw_board(self, highlight=None): self.canvas.delete("all") size = 500 // 8 colors = ["#F0D9B5", "#B58863"] for row in range(8): for col in range(8): x1 = col * size y1 = (7 - row) * size x2 = x1 + size y2 = y1 + size color = colors[(row + col) % 2] self.canvas.create_rectangle(x1, y1, x2, y2, fill=color, outline="black") sq = row * 8 + col piece = self.board.piece_at(sq) if piece: symbols = { "K": "♔", "Q": "♕", "R": "♖", "B": "♗", "N": "♘", "P": "♙", "k": "♚", "q": "♛", "r": "♜", "b": "♝", "n": "♞", "p": "♟" } symbol = symbols[piece.symbol()] self.canvas.create_text(x1 + size/2, y1 + size/2, text=symbol, font=("Arial", size-10), fill="black") if highlight is not None: row = highlight // 8 col = highlight % 8 x1 = col * size y1 = (7 - row) * size self.canvas.create_rectangle(x1, y1, x1+size, y1+size, outline="yellow", width=4) def reset_board(self): self.board = chess.Board() self.selected_square = None self.prev_memory = None self.engine_thinking = False self.update_memory_for_current_board() self.draw_board() self.update_status("Board reset. Your turn (White).") if self.engine and self.engine.engine_ready: self.engine.set_position(self.board.fen()) # Make sure engine is not thinking self.engine._send_command("force") self.engine._send_command("new") self.engine._send_command("post") self.engine._send_command("level 0 0 0") def on_quit(self): if self.engine: self.engine.quit() self.master.quit() # ------------------------------------------------------------ # 5. Run # ------------------------------------------------------------ if __name__ == "__main__": root = tk.Tk() app = TelepathicChessGUI(root) root.mainloop()