AI TCP Server

game/handlers/aiSimulation.py

@@ -0,0 +1,125 @@
+"""Run one isolated game simulation for the AI HTTP handler."""
+
+from copy import deepcopy
+from threading import Condition, Thread
+from time import sleep
+
+from game.game import Game
+from game.gameState import GameState
+from game.map import Map
+
+
+class AiSimulation:
+    """Own a single Game instance and the callbacks used by that instance."""
+
+    def __init__(self, game_map: Map):
+        self.game = Game(deepcopy(game_map), GameState())
+        min_moves = self.game.map.get_min_moves()
+        self.max_moves = min_moves * 2 if min_moves is not None else None
+        self._condition = Condition()
+        self._pending_move = None
+        self._waiting_for_move = False
+        self._submitted_move = False
+        self._turn_finished = False
+        self._finished = False
+        self._error = None
+        self._thread = None
+        self.moves = []
+        self.game.set_waiting_for_input_callback(self._on_input_request)
+        self.game.set_position_update_callback(self._on_position_update)
+        self.game.set_game_stop_callback(self._should_stop_game)
+
+    def start(self) -> None:
+        # print("Simulation thread starting")
+        self._thread = Thread(target=self._run_game, daemon=True)
+        self._thread.start()
+        with self._condition:
+            self._condition.wait_for(lambda: self._waiting_for_move or self._finished)
+            self._raise_error_if_needed()
+
+    def step(self, move: str | None) -> dict:
+        if move is None:
+            self.stop_for_generation()
+            return self.result()
+        if move not in GameState.PLAYER_MOVEMENTS:
+            raise ValueError(
+                f"Invalid move '{move}'. Must be one of {GameState.PLAYER_MOVEMENTS}."
+            )
+
+        with self._condition:
+            if self._finished:
+                return self.result()
+            self.moves.append(move)
+            self._pending_move = move
+            self._turn_finished = False
+            self._condition.notify_all()
+            self._condition.wait_for(lambda: self._turn_finished or self._finished)
+            self._raise_error_if_needed()
+            return self.result()
+
+    def result(self) -> dict:
+        state = self.game.gameState
+        return {
+            "won": state.gameWon,
+            "coins": state.coins,
+            "moves": state.movesCount,
+            "position": list(state.get_player_position()),
+            "finished": state.stopGame,
+        }
+
+    def close(self) -> None:
+        self.stop_for_generation()
+
+    def stop_for_generation(self) -> None:
+        with self._condition:
+            self._finished = True
+            self.game.stop_game()
+            self._turn_finished = True
+            self._condition.notify_all()
+
+    def _run_game(self) -> None:
+        try:
+            self.game.start_game()
+        except Exception as error:
+            with self._condition:
+                self._error = error
+                self._finished = True
+                self._condition.notify_all()
+            # print(f"Simulation failed: {error}")
+
+    def _on_input_request(self) -> str:
+        with self._condition:
+            if self._submitted_move:
+                self._submitted_move = False
+                self._turn_finished = True
+                self._condition.notify_all()
+            self._waiting_for_move = True
+            self._condition.notify_all()
+            self._condition.wait_for(lambda: self._pending_move is not None or self._finished)
+            if self._finished:
+                return -1
+            move = self._pending_move
+            self._pending_move = None
+            self._waiting_for_move = False
+            self._submitted_move = True
+            return move
+
+    def _on_position_update(self) -> None:
+        with self._condition:
+            if self.game.gameState.stopGame:
+                self._finished = True
+                self._turn_finished = True
+                self._condition.notify_all()
+
+    def _should_stop_game(self, game_state: GameState) -> bool:
+        should_stop = self.max_moves is not None and game_state.movesCount >= self.max_moves
+        if should_stop:
+            with self._condition:
+                self._finished = True
+                self._turn_finished = True
+                self._condition.notify_all()
+        return should_stop
+
+    def _raise_error_if_needed(self) -> None:
+        if self._error:
+            raise self._error