Ninluc/plateformer-evolve
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Better scoring and map selection + Map delta

Browse Source
This commit is contained in:
Matthias Guillitte
2026-10-23 17:54:13 +02:00
parent ce720a92b2
commit f56b8c1e9c
16 changed files with 233 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
game/handlers/aiHttpGameHandler.py
+58 -15
View File
@@ -24,6 +24,7 @@ of moves to be executed in a single game simulation, based on the map min_moves
"""
import json
import math
import os
import random
import socketserver
@@ -43,12 +44,13 @@ from game.mapSet import MapSet
class AiHttpGameHandler(GameHandler):
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8765
VALIDATION_DIFFICULTY = 70
VALIDATION_DIFFICULTY = 90
REPLAY_FRAME_DELAY = 0.1
# Scoring weights TOTAL MUST BE 100
COINS_WEIGHT = 40
DISTANCE_WEIGHT = 20
WIN_WEIGHT = 40
COINS_WEIGHT = 50
STEPS_WEIGHT = 10
DISTANCE_WEIGHT = 10
WIN_WEIGHT = 30
def __init__(self, host: str | None = None, port: int | None = None,
max_workers: int | None = None):
@@ -62,6 +64,10 @@ class AiHttpGameHandler(GameHandler):
game_map for game_map in self.map_set.maps
if game_map.difficulty >= self.VALIDATION_DIFFICULTY
] or list(self.map_set.maps)
if (self.COINS_WEIGHT + self.DISTANCE_WEIGHT + self.STEPS_WEIGHT + self.WIN_WEIGHT) != 100:
raise ValueError("Scoring weights must sum to 100.")
def run(self) -> None:
handler = self
@@ -73,6 +79,7 @@ class AiHttpGameHandler(GameHandler):
"best_run": None,
"replay_condition": Condition(),
"client_connected": True,
"stagnant_generations": 0,
}
replay_thread = Thread(target=handler._replay_best_runs, args=(session,))
replay_thread.start()
@@ -81,12 +88,9 @@ class AiHttpGameHandler(GameHandler):
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()
@@ -189,7 +193,7 @@ class AiHttpGameHandler(GameHandler):
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)
result["score"] = self._score_simulation(simulation, result)
# print(f"Finished step for model '{model_id}'", flush=True)
return model_id, result
@@ -198,12 +202,33 @@ class AiHttpGameHandler(GameHandler):
# 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"])
map_difficulty = session["map"].difficulty # e.g., 20 or 60
response["score"] = best_score
session["performance"] = best_score
# --- Curriculum Regression (Fallback) & Performance Update ---
PASSING_THRESHOLD = 85.0 # Score percentage considered successful completion
if best_score < PASSING_THRESHOLD:
session["stagnant_generations"] = session.get("stagnant_generations", 0) + 1
else:
session["stagnant_generations"] = 0 # Reset on breakthrough
# If stuck on a difficulty wall for 15 generations, fallback to an easier map range
if session["stagnant_generations"] >= 15:
current_perf = session.get("performance", 10 * math.sqrt(map_difficulty))
regression_drop = 15.0
session["performance"] = max(0.0, current_perf - regression_drop)
print(f"Stuck on difficulty {map_difficulty} for {session['stagnant_generations']} generations. Dropping target performance to {session['performance']:.1f}", flush=True)
session["stagnant_generations"] = 0 # Reset stagnation counter after fallback
else:
difficulty_scaling = 10 * math.sqrt(map_difficulty)
normalized_score = best_score / 100.0 # Convert 0-100 percentage to 0.0-1.0 scale
session["performance"] = difficulty_scaling * (0.5 + 0.5 * normalized_score)
session["last_result"] = response
session["simulations"] = None
session["best_run"] = (
@@ -252,6 +277,8 @@ class AiHttpGameHandler(GameHandler):
line += "\tMoves: " + str(game.gameState.movesCount)
elif index == 2:
line += "\tGeneration: " + str(generation)
elif index == 3:
line += "\tMove: " + moves[game.gameState.movesCount - 1] if len(moves) >= game.gameState.movesCount else "\tMove: None"
frame_lines.append(line)
print("\033[2J\033[H\n\n\n" + "\n".join(frame_lines), flush=True)
sleep(self.REPLAY_FRAME_DELAY)
@@ -264,20 +291,34 @@ class AiHttpGameHandler(GameHandler):
results = {}
for model_id, simulation in simulations.items():
result = simulation.result()
result["score"] = self._score_result(simulation, result)
result["score"] = self._score_simulation(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]
maps = self.map_set.get_maps_by_difficulty(round(performance), 2)
return random.choice(maps)
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.")
def _score_simulation(self, simulation: AiSimulation, result: dict) -> int:
game_map = simulation.game.map
raw_score = self._score_result(game_map, result)
max_score = self._get_map_max_score(game_map)
if max_score == 0:
return 0
percentage = (raw_score / max_score) * 100.0
return round(max(0.0, min(100.0, percentage)))
def _get_map_max_score(self, game_map: Map) -> int:
result = {}
result["position"] = game_map.get_first_block_type_position("finish")
result["coins"] = game_map.count_blocks_of_type("coin")
result["won"] = True
result["moves"] = game_map.min_moves
return self._score_result(game_map, result)
def _score_result(self, game_map: Map, result: dict) -> int:
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")
@@ -290,11 +331,13 @@ class AiHttpGameHandler(GameHandler):
distance_progress = 1.0 if start_distance == 0 else max(
0.0, min(1.0, 1.0 - remaining_distance / start_distance)
)
steps_progress = 0.0 if result["moves"] == 0 else min(1.0, 1.0 - result["moves"] / game_map.min_moves)
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.STEPS_WEIGHT * steps_progress
+ self.WIN_WEIGHT * result["won"]
)