Better performance score

This commit is contained in:
2026-10-31 14:50:25 +01:00
parent a7a3d973a3
commit 104c6fac63
2 changed files with 25 additions and 32 deletions
+25 -32
View File
@@ -51,8 +51,10 @@ class AiHttpGameHandler(GameHandler):
STEPS_WEIGHT = 10
DISTANCE_WEIGHT = 10
WIN_WEIGHT = 45
PLAY_ALL_MAPS_PER_GENERATION = True # Are all maps played during a generation or only MAPS_PER_GENERATION maps per generation are played
MAPS_PER_GENERATION = 9 # Evaluate across x maps per generation. Only used if PLAY_ALL_MAPS_PER_GENERATION is False. If True, all maps are played in a generation.
PLAY_ALL_MAPS_PER_GENERATION = False # Are all maps played during a generation or only MAPS_PER_GENERATION maps per generation are played
MAPS_PER_GENERATION = 3 # Evaluate across x maps per generation. Only used if PLAY_ALL_MAPS_PER_GENERATION is False. If True, all maps are played in a generation.
MAP_PASS_MIN_SCORE = 20 # Minimum score to consider a map passed. Only used if PLAY_ALL_MAPS_PER_GENERATION is False. If True, all maps are played in a generation.
MAP_DIFFICULTY_VARIANCE = 5 # Lower value will make the map selection more variable, higher value will make the map selection more consistent with the current performance. Only used if PLAY_ALL_MAPS_PER_GENERATION is False. A value too high may produce no evolving behavior
CONSISTENCY_PENALTY_WEIGHT = 0.4 # Penalty for inconsistent map scores
def __init__(self, host: str | None = None, port: int | None = None,
@@ -155,11 +157,7 @@ class AiHttpGameHandler(GameHandler):
if self.PLAY_ALL_MAPS_PER_GENERATION:
session["generation_maps"] = self.map_set.get_maps()
else:
available_maps = self.map_set.get_maps_by_difficulty(round(perf), self.MAPS_PER_GENERATION)
session["generation_maps"] = [
random.choice(available_maps)
for _ in range(self.MAPS_PER_GENERATION) # Randomly selects in case there is less maps available than MAPS_PER_GENERATION
]
session["generation_maps"] = sorted(self.map_set.get_maps_by_difficulty(round(perf), self.MAPS_PER_GENERATION), key=lambda m: m.difficulty)
# Fetch the pre-selected map for the current round
map_round = session.get("map_round", 0)
@@ -233,14 +231,14 @@ class AiHttpGameHandler(GameHandler):
session["map_scores"][model_id].append(res["score"])
# Check if we have completed all maps for this generation
if session["map_round"] < self.MAPS_PER_JSON_ROUND if hasattr(self, 'MAPS_PER_JSON_ROUND') else session["map_round"] < self.MAPS_PER_GENERATION:
if session["map_round"] < len(session["generation_maps"]):
# More maps remaining in this generation! Tell client to simulate next map.
response["type"] = "map_round_complete"
response["map_round"] = session["map_round"]
session["simulations"] = None
session["last_result"] = response
else:
# All 3 maps completed -> finalize generation results (Average scores)
# All maps completed -> finalize generation results (Average scores)
response["type"] = "simulation_result"
final_results = {}
@@ -261,26 +259,18 @@ class AiHttpGameHandler(GameHandler):
response["results"] = final_results
best_score = max(res["score"] for res in final_results.values())
best_model_id = max(final_results, key=lambda mid: final_results[mid]["score"])
map_difficulty = session["map"].difficulty
response["score"] = best_score
# --- Curriculum Regression & Performance Update ---
PASSING_THRESHOLD = 85.0
if best_score < PASSING_THRESHOLD:
session["stagnant_generations"] = session.get("stagnant_generations", 0) + 1
else:
session["stagnant_generations"] = 0
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
else:
difficulty_scaling = 10 * math.sqrt(map_difficulty)
normalized_score = best_score / 100.0
session["performance"] = difficulty_scaling * (0.5 + 0.5 * normalized_score)
# --- Performance Update ---
# Average score of the 90th percentile of the final results
sorted_scores = sorted((res["score"] for res in final_results.values()), reverse=True)
top_90_count = max(1, round(len(sorted_scores) * 0.9))
top_90_scores = sorted_scores[:top_90_count]
avg_top_90_score = sum(top_90_scores) / len(top_90_scores)
performance_delta = (avg_top_90_score - self.MAP_PASS_MIN_SCORE) / self.MAP_DIFFICULTY_VARIANCE
performance_delta = max(0.05, performance_delta)
session["performance"] = max(0.0, session.get("performance", 0.0) + performance_delta)
# print(f"Generation {session['generation']} completed on difficulty {map_difficulty}. Best score: {best_score:.2f}. Updated performance: {session['performance']:.2f}", flush=True)
# Reset round counter for the next generation
session["map_round"] = 0
@@ -290,6 +280,7 @@ class AiHttpGameHandler(GameHandler):
deepcopy(session["map"]),
list(simulations[best_model_id].moves),
session["generation"],
session["performance"],
)
replay_condition = session.get("replay_condition")
if replay_condition is not None:
@@ -311,7 +302,7 @@ class AiHttpGameHandler(GameHandler):
return
self._show_best_run(*best_run)
def _show_best_run(self, game_map, moves: list[str], generation: int) -> None:
def _show_best_run(self, game_map, moves: list[str], generation: int, performance: float) -> None:
game = Game(game_map, GameState())
remaining_moves = iter(moves)
@@ -327,12 +318,14 @@ class AiHttpGameHandler(GameHandler):
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)
elif index == 1:
line += "\tPerformance: " + str(round(performance, 2))
elif index == 2:
line += "\tCoins: " + str(game.gameState.coins)
elif index == 3:
line += "\tMoves: " + str(game.gameState.movesCount)
elif index == 4:
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)