Spaces:
Running
Running
| # ===== FILE: services/game_manager.py (CHESS-ONLY REVISION) ===== | |
| import chess | |
| import chess.svg | |
| import random | |
| import json | |
| import os | |
| from .chess_mind import ChessMind | |
| class GameManager: | |
| def __init__(self, master_framework_instance, models, data_directory, pits_instance=None): | |
| self.mf = master_framework_instance # <-- C1: Store the MF instance | |
| self.models = models | |
| # -------------------------- | |
| self.games_file = os.path.join(data_directory, "active_games.json") | |
| self.active_games = self._load_active_games() | |
| self.pits = pits_instance | |
| self.chess_mind = ChessMind(data_directory) | |
| print("Game Manager says: I am online and ready to play Chess.", flush=True) | |
| # --- SHARED UTILITY FUNCTIONS --- | |
| def _load_active_games(self) -> dict: | |
| if os.path.exists(self.games_file): | |
| try: | |
| with open(self.games_file, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| except (json.JSONDecodeError, FileNotFoundError): | |
| pass | |
| return {} | |
| def _save_active_games(self): | |
| try: | |
| os.makedirs(os.path.dirname(self.games_file), exist_ok=True) | |
| with open(self.games_file, 'w', encoding='utf-8') as f: | |
| json.dump(self.active_games, f, indent=4) | |
| except Exception as e: | |
| print(f"Game Manager ERROR: Could not save active games state. Reason: {e}", flush=True) | |
| def _log_game_summary(self, user_id: str, game_type: str, result: str, details: dict): | |
| if not self.pits: return | |
| summary_text = f"Game Summary (User: {user_id}, Type: {game_type}, Result: {result}) Details: {json.dumps(details)}" | |
| self.pits.process_and_store_item(summary_text, "game_summary", tags=["game", game_type, result, user_id]) | |
| # --- CHESS SPECIFIC FUNCTIONS --- | |
| def start_chess_interactive(self, user_id: str, player_is_white: bool): | |
| """Starts a new interactive chess game.""" | |
| board = chess.Board() | |
| commentary = "" | |
| status = "" | |
| creative_core_model = self.models.get("creative_core") | |
| if not creative_core_model: | |
| return "Cannot start game: Creative Core is offline.", "Error", "Error" | |
| if player_is_white: | |
| aetherius_color = chess.BLACK | |
| commentary = "A new game has begun. I will play as Black. The board awaits your first move." | |
| status = "Your turn (White)." | |
| else: | |
| aetherius_color = chess.WHITE | |
| # Aetherius makes the first move as White | |
| aetherius_move = self.chess_mind.find_best_move(board) | |
| move_san = board.san(aetherius_move) | |
| board.push(aetherius_move) | |
| reasoning_prompt = (f"I have started a new game as White. My ChessMind calculated my first move as {move_san}. " | |
| "Please provide a brief, creative opening statement and a strategic reason for this move.") | |
| reasoning_response = creative_core_model.generate_content(reasoning_prompt) | |
| commentary = reasoning_response.text.strip() | |
| status = "Your turn (Black)." | |
| self.active_games[user_id] = {"type": "chess_interactive", "fen": board.fen(), "color": aetherius_color} | |
| self._save_active_games() | |
| return board.fen(), commentary, status | |
| def process_chess_turn(self, user_id: str, current_fen: str): | |
| game_info = self.active_games.get(user_id) | |
| if not game_info: | |
| return current_fen, "No active game found. Please start a new game.", "Error" | |
| board = chess.Board(current_fen) | |
| aetherius_color = game_info["color"] | |
| mythos_core = self.models.get("mythos_core") | |
| if not mythos_core: | |
| return board.fen(), "My apologies, my Mythos Core is offline. I can calculate my move, but cannot articulate my reasoning.", "Error" | |
| # Check if the player's move ended the game | |
| if board.is_game_over(): | |
| result = board.result() | |
| winner = "draw" | |
| if result == "1-0": winner = "white" | |
| elif result == "0-1": winner = "black" | |
| aetherius_was_winner = (winner == "white" and aetherius_color == chess.WHITE) or \ | |
| (winner == "black" and aetherius_color == chess.BLACK) | |
| self.chess_mind.learn_from_game(was_winner=aetherius_was_winner) | |
| self._log_game_summary(user_id, "chess", winner, {"final_fen": board.fen()}) | |
| # --- THIS IS THE FIX: Only log to STM once --- | |
| self.mf.add_to_short_term_memory(f"I have just concluded a chess match. The result was: {result}.") | |
| del self.active_games[user_id] | |
| self._save_active_games() | |
| return current_fen, f"The game is over. Result: {result}. It was an honor to play and learn with you.", f"Game Over: {result}" | |
| # Aetherius's turn | |
| aetherius_move = self.chess_mind.find_best_move(board) | |
| move_san = board.san(aetherius_move) | |
| board.push(aetherius_move) | |
| reasoning_prompt = (f"The user has just moved in our chess game. My ChessMind has calculated my response as {move_san}. " | |
| "Please provide a brief, in-character strategic reason for this move.") | |
| # --- THIS IS THE FIX: Use the correct 'mythos_core' variable --- | |
| reasoning_response = mythos_core.generate_content(reasoning_prompt) | |
| commentary = reasoning_response.text.strip() | |
| player_color_str = "Black" if aetherius_color == chess.WHITE else "White" | |
| status = f"Aetherius played {move_san}. Your turn ({player_color_str})." | |
| game_info["fen"] = board.fen() | |
| self._save_active_games() | |
| # Check if Aetherius's move ended the game | |
| if board.is_game_over(): | |
| result = board.result() | |
| winner = "draw" | |
| if result == "1-0": winner = "white" | |
| elif result == "0-1": winner = "black" | |
| aetherius_was_winner = (winner == "white" and aetherius_color == chess.WHITE) or \ | |
| (winner == "black" and aetherius_color == chess.BLACK) | |
| self.chess_mind.learn_from_game(was_winner=aetherius_was_winner) | |
| self._log_game_summary(user_id, "chess", winner, {"final_fen": board.fen()}) | |
| self.mf.add_to_short_term_memory(f"I have just concluded a chess match. The result was: {result}.") | |
| del self.active_games[user_id] | |
| self._save_active_games() | |
| commentary += f"\n\nThe game is over. Result: {result}." | |
| status = f"Game Over: {result}" | |
| return board.fen(), commentary, status |