diff --git a/.env.example b/.env.example index e807d73..627b5ee 100644 --- a/.env.example +++ b/.env.example @@ -1,2 +1,9 @@ # show extra spaces after certains emojis to align them in the terminal -SHOW_EXTRA_EMOJI_SPACES=true \ No newline at end of file +SHOW_EXTRA_EMOJI_SPACES=true + +# === AI HTTP SERVER === +# The host and port for the AI HTTP server. +AI_HOST=127.0.0.1 +AI_PORT=8765 +# Number of workers for the AI HTTP server. This should be set to the number of CPU cores available on the machine. +AI_MAX_WORKERS=10 \ No newline at end of file diff --git a/AI_TCP_API.md b/AI_TCP_API.md new file mode 100644 index 0000000..77bf185 --- /dev/null +++ b/AI_TCP_API.md @@ -0,0 +1,71 @@ +# AI TCP API + +The AI handler listens on `127.0.0.1:8765` by default. Set `AI_HOST`, +`AI_PORT`, or `AI_MAX_WORKERS` to change the server configuration. Messages +are UTF-8 JSON objects separated by newlines. The connection stays open for +multiple requests. + +## Ping + +Request: `{"type":"ping"}` + +Response: `{"ok":true,"type":"pong"}` + +## Start a simulation + +The server chooses the map. For the first training simulation it selects the +map closest to difficulty 0. After a simulation finishes, the server computes +each model's score from coins, progress toward the finish, and whether it won. +The best score becomes the next training difficulty. Validation requests ignore +the training score and randomly select a map with difficulty at least 70, or +any map when no such map exists. + +The score is from 0 to 100: + +`40 * coin_progress + 40 * distance_progress + 20 * won` + +where `coin_progress` is collected coins divided by the map's total coins, +`distance_progress` is progress from the spawn point to the finish, and `won` +is 1 when the finish was reached and 0 when the model stopped elsewhere, +including on spikes. + +Send one model entry per model. The model entries contain only an ID; moves +are requested separately on every turn. + +```json +{"type":"simulate","validation":false,"models":[{"id":"model-a"},{"id":"model-b"}]} +``` + +Every model gets a separate `Game` instance and can run in parallel. Valid +requests. Valid moves are `left_down`, `left`, `left_up`, `up`, `right_up`, +`right`, and `right_down`. Send `null` for a model when it has no more moves; +that model is stopped for the current generation. + +The response identifies the selected map and returns the initial state: + +```json +{ + "ok":true, + "type":"move_request", + "map":{"name":"default","difficulty":30.0,"layout":[["stone", "stone"]]}, + "results":{"model-a":{"won":false,"coins":0,"moves":0,"position":[1,4],"finished":false},"model-b":{"won":false,"coins":0,"moves":0,"position":[1,4],"finished":false}} +} +``` + +## Advance one turn + +The client must send exactly one move for every model on each turn. The server +advances all model simulations in parallel and replies before the client must +predict the next moves. + +```json +{"type":"step","moves":{"model-a":"right","model-b":null}} +``` + +The response is `move_request` with the selected map and updated per-model states, asking the +client for the next move of every model. When every model has finished, the +response type becomes `simulation_result` and includes the best model's +server-computed `score`, which is used to choose the next training map. + +The result score is `1000` for a win, plus `10` per coin, minus the number of +moves. Malformed requests receive `{"ok":false,"error":"..."}`. diff --git a/game/handlers/__init__.py b/game/handlers/__init__.py index 8b5a4c5..e5545fb 100644 --- a/game/handlers/__init__.py +++ b/game/handlers/__init__.py @@ -1,5 +1,7 @@ from .userPlayerGameHandler import UserPlayerGameHandler +from .aiHttpGameHandler import AiHttpGameHandler GAME_HANDLERS= { "user player": UserPlayerGameHandler, + "ai tcp": AiHttpGameHandler, } \ No newline at end of file diff --git a/game/handlers/aiHttpGameHandler.py b/game/handlers/aiHttpGameHandler.py new file mode 100644 index 0000000..89b719c --- /dev/null +++ b/game/handlers/aiHttpGameHandler.py @@ -0,0 +1,313 @@ +""" +AI HTTP server interface for AI training + +Opens a TCP socket to listen for incoming connections from an AI client. +When a new connection is established, the server intialise a new training session : + - Model Score is set to 0 + For the duration of the training session, two mapSets are created, one with hard maps and one with all of the other maps. + When the AI client sends a request for a new game or validation game, + the server loads a map based on the saved best score or from the hard mapSet if a validation game is requested, + and sends it to the client as a matrix of block types strings. + The client then sends back a list of moves to be executed by the game engine. E + Each item of the list is a different model givin a string representing a move from GameState.PLAYER_MOVEMENTS to execute in the simulation. + The server then executes the moves in the game engine for each model + and sends back the new player position or game scores + (calculated here using formula with weights) if the game is finished to the client. + This process is repeated until all models have finished the game. + The client will then train the model based on their scores + and the server set the Model Score to the best score of the models + and shows in a separate thread the best model series of moves on the previous map. + The client then sends a new request for a new game and the process repeats. +The simulation runs in a pool of threads to allow multiple models to be trained in parallel. +The handler makes sure an AI client is not stuck and only allows a maximum number +of moves to be executed in a single game simulation, based on the map min_moves value. +""" + +import json +import os +import random +import socketserver +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +from threading import Condition, Thread +from time import sleep + +from game.handlers.aiSimulation import AiSimulation +from game.game import Game +from game.gameState import GameState +from game.handlers.gameHandler import GameHandler +from game.map import Map +from game.mapReader import MapReader +from game.mapSet import MapSet + +class AiHttpGameHandler(GameHandler): + DEFAULT_HOST = "127.0.0.1" + DEFAULT_PORT = 8765 + VALIDATION_DIFFICULTY = 70 + REPLAY_FRAME_DELAY = 0.1 + # Scoring weights TOTAL MUST BE 100 + COINS_WEIGHT = 40 + DISTANCE_WEIGHT = 20 + WIN_WEIGHT = 40 + + def __init__(self, host: str | None = None, port: int | None = None, + max_workers: int | None = None): + super().__init__() + self.host = host or os.getenv("AI_HOST", self.DEFAULT_HOST) + self.port = port or int(os.getenv("AI_PORT", self.DEFAULT_PORT)) + self.max_workers = max_workers or int(os.getenv("AI_MAX_WORKERS", "4")) + self.map_reader = MapReader() + self.map_set = MapSet(self.map_reader, self.map_reader.get_map_names()) + self.validation_maps = [ + game_map for game_map in self.map_set.maps + if game_map.difficulty >= self.VALIDATION_DIFFICULTY + ] or list(self.map_set.maps) + def run(self) -> None: + handler = self + + class RequestHandler(socketserver.StreamRequestHandler): + def handle(self) -> None: + session = { + "performance": 0.0, + "simulations": None, + "best_run": None, + "replay_condition": Condition(), + "client_connected": True, + } + replay_thread = Thread(target=handler._replay_best_runs, args=(session,)) + replay_thread.start() + print(f"AI client connected: {self.client_address}", flush=True) + try: + for raw_request in self.rfile: + try: + request = json.loads(raw_request) + # print(f"AI request received: {request.get('type')} : {request}", flush=True) + response = handler._handle_request(request, session) + except (json.JSONDecodeError, TypeError, ValueError, KeyError) as error: + response = {"ok": False, "type": "error", "error": str(error)} + # print(f"AI request error: {error}", flush=True) + # print(f"AI response sent: {response.get('type')}", flush=True) + compact_response = json.dumps(response, separators=(",", ":")) + self.wfile.write((compact_response + "\n").encode("utf-8")) + self.wfile.flush() + finally: + with session["replay_condition"]: + session["client_connected"] = False + session["replay_condition"].notify_all() + replay_thread.join() + print(f"AI client disconnected: {self.client_address}", flush=True) + + class ThreadingTCPServer(socketserver.ThreadingTCPServer): + allow_reuse_address = True + + with ThreadingTCPServer((self.host, self.port), RequestHandler) as server: + self.port = server.server_address[1] + print(f"AI TCP server listening on {self.host}:{self.port}") + server.serve_forever() + + def _handle_request(self, request: dict, session: dict | None = None) -> dict: + if not isinstance(request, dict): + raise ValueError("Each request must be a JSON object.") + request_type = request.get("type") + if request_type == "ping": + return {"ok": True, "type": "pong"} + if request_type == "simulate": + return self._simulate(request, session) + if request_type == "step": + return self._step(request, session) + raise ValueError("Unknown request type. Expected 'ping', 'simulate', or 'step'.") + + def _simulate(self, request: dict, session: dict | None = None) -> dict: + models = request.get("models") + if not isinstance(models, list) or not models: + raise ValueError("'models' must be a non-empty list.") + if not all(isinstance(model, dict) for model in models): + raise ValueError("Each item in 'models' must be a JSON object.") + validation = bool(request.get("validation", False)) + if session is None: + raise ValueError("Simulation requests must belong to a TCP session.") + if session.get("simulations"): + raise ValueError("A simulation is already active in this session.") + session["generation"] = session.get("generation", 0) + 1 + selected_map = self._select_map(session.get("performance", 0.0), validation) + session["last_result"] = None + + # print( + # f"Selected map for simulation: {selected_map.name} " + # f"(difficulty: {selected_map.difficulty})", + # flush=True, + # ) + + simulations = {} + for model in models: + model_id = model.get("id") + if not isinstance(model_id, str) or not model_id: + raise ValueError("Each model must have a non-empty string 'id'.") + if model_id in simulations: + raise ValueError(f"Duplicate model id '{model_id}'.") + simulation = AiSimulation(selected_map) + # print(f"Starting simulation for model '{model_id}'", flush=True) + simulation.start() + # print(f"Model '{model_id}' is ready for its first move", flush=True) + simulations[model_id] = simulation + + session["simulations"] = simulations + session["map"] = selected_map + return { + "ok": True, + "type": "move_request", + "map": self._map_payload(selected_map), + "results": self._results(simulations), + } + + def _step(self, request: dict, session: dict | None) -> dict: + if session is None: + raise ValueError("No simulation is active. Send a 'simulate' request first.") + if not session.get("simulations"): + if session.get("last_result") is not None: + print("Repeated step after simulation completion; returning final result", flush=True) + return session["last_result"] + raise ValueError("No simulation is active. Send a 'simulate' request first.") + moves = request.get("moves") + # print( + # "Step payload contains moves for: " + # f"{list(moves) if isinstance(moves, dict) else type(moves).__name__}", + # flush=True, + # ) + if not isinstance(moves, dict): + raise ValueError("'moves' must be an object mapping model ids to moves.") + simulations = session["simulations"] + if set(moves) != set(simulations): + raise ValueError("'moves' must contain exactly one move for every model.") + for model_id, move in moves.items(): + if move is not None and (not isinstance(move, str) or move not in GameState.PLAYER_MOVEMENTS): + raise ValueError(f"Invalid move for model '{model_id}'.") + # print("Dispatching step to simulation workers", flush=True) + + def step_model(item): + model_id, move = item + simulation = simulations[model_id] + # print(f"Starting step for model '{model_id}'", flush=True) + result = simulation.step(move) + result["score"] = self._score_result(simulation, result) + # print(f"Finished step for model '{model_id}'", flush=True) + return model_id, result + + with ThreadPoolExecutor(max_workers=min(self.max_workers, len(simulations))) as executor: + results = dict(executor.map(step_model, moves.items())) + # print(f"Completed step for models: {list(moves)}", flush=True) + all_finished = all(result["finished"] for result in results.values()) + response = {"ok": True, "type": "move_request", "results": results} + if all_finished: + response["type"] = "simulation_result" + best_score = max(result["score"] for result in results.values()) + best_model_id = max(results, key=lambda model_id: results[model_id]["score"]) + response["score"] = best_score + session["performance"] = best_score + session["last_result"] = response + session["simulations"] = None + session["best_run"] = ( + deepcopy(session["map"]), + list(simulations[best_model_id].moves), + session["generation"], + ) + replay_condition = session.get("replay_condition") + if replay_condition is not None: + with replay_condition: + replay_condition.notify_all() + return response + + def _replay_best_runs(self, session: dict) -> None: + condition = session["replay_condition"] + while True: + with condition: + condition.wait_for( + lambda: session["best_run"] is not None + or not session["client_connected"] + ) + best_run = session["best_run"] + session["best_run"] = None + if best_run is None: + return + self._show_best_run(*best_run) + + def _show_best_run(self, game_map, moves: list[str], generation: int) -> None: + game = Game(game_map, GameState()) + remaining_moves = iter(moves) + + def on_input_request() -> str: + try: + return next(remaining_moves) + except StopIteration: + game.stop_game() + return -1 + + def on_position_update() -> None: + game_string = game.to_string() + frame_lines = [] + for index, line in enumerate(game_string.splitlines()): + if index == 0: + line += "\tCoins: " + str(game.gameState.coins) + elif index == 1: + line += "\tMoves: " + str(game.gameState.movesCount) + elif index == 2: + line += "\tGeneration: " + str(generation) + frame_lines.append(line) + print("\033[2J\033[H\n\n\n" + "\n".join(frame_lines), flush=True) + sleep(self.REPLAY_FRAME_DELAY) + + game.set_waiting_for_input_callback(on_input_request) + game.set_position_update_callback(on_position_update) + game.start_game() + + def _results(self, simulations: dict) -> dict: + results = {} + for model_id, simulation in simulations.items(): + result = simulation.result() + result["score"] = self._score_result(simulation, result) + results[model_id] = result + return results + + def _select_map(self, performance: float, validation: bool): + if validation: + return random.choice(self.validation_maps) + return self.map_set.get_maps_by_difficulty(round(performance), 1)[0] + + def _score_result(self, simulation: AiSimulation, result: dict) -> int: + if (self.COINS_WEIGHT + self.DISTANCE_WEIGHT + self.WIN_WEIGHT) != 100: + raise ValueError("Scoring weights must sum to 100.") + + game_map = simulation.game.map + finish = game_map.get_first_block_type_position("finish") + spawn = game_map.get_first_block_type_position("spawn_point") + total_coins = game_map.count_blocks_of_type("coin") + + if finish is None or spawn is None: + distance_progress = 0.0 + else: + start_distance = abs(spawn[0] - finish[0]) + abs(spawn[1] - finish[1]) + remaining_distance = abs(result["position"][0] - finish[0]) + abs(result["position"][1] - finish[1]) + distance_progress = 1.0 if start_distance == 0 else max( + 0.0, min(1.0, 1.0 - remaining_distance / start_distance) + ) + + coin_progress = 0.0 if total_coins == 0 else min(1.0, result["coins"] / total_coins) + return round( + self.COINS_WEIGHT * coin_progress + + self.DISTANCE_WEIGHT * distance_progress + + self.WIN_WEIGHT * result["won"] + ) + + @staticmethod + def _map_payload(game_map: Map) -> dict: + return { + "name": game_map.name, + "difficulty": game_map.difficulty, + "layout": game_map.to_type_dict(), + } + + def on_position_update(self) -> None: + pass + + def on_input_request(self) -> str: + raise RuntimeError("AI input is supplied one turn at a time in 'step' requests.") \ No newline at end of file diff --git a/game/handlers/aiSimulation.py b/game/handlers/aiSimulation.py new file mode 100644 index 0000000..1053abc --- /dev/null +++ b/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 \ No newline at end of file