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

Play all maps each generation

Browse Source 
Also disabled some too hard maps
This commit is contained in:
Matthias Guillitte
2026-10-26 15:53:27 +01:00
parent f56b8c1e9c
commit a7a3d973a3
5 changed files with 105 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
game/handlers/aiHttpGameHandler.py
+100 -45
View File
@@ -47,10 +47,13 @@ class AiHttpGameHandler(GameHandler):
VALIDATION_DIFFICULTY = 90
REPLAY_FRAME_DELAY = 0.1
# Scoring weights TOTAL MUST BE 100
COINS_WEIGHT = 50
COINS_WEIGHT = 35
STEPS_WEIGHT = 10
DISTANCE_WEIGHT = 10
WIN_WEIGHT = 30
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.
CONSISTENCY_PENALTY_WEIGHT = 0.4 # Penalty for inconsistent map scores
def __init__(self, host: str | None = None, port: int | None = None,
max_workers: int | None = None):
@@ -80,6 +83,9 @@ class AiHttpGameHandler(GameHandler):
"replay_condition": Condition(),
"client_connected": True,
"stagnant_generations": 0,
"map_round": 0,
"cumulative_scores": {},
"last_results": {}
}
replay_thread = Thread(target=handler._replay_best_runs, args=(session,))
replay_thread.start()
@@ -132,15 +138,34 @@ class AiHttpGameHandler(GameHandler):
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,
# )
# Initialize generation round tracking and select 3 maps upfront if starting fresh
if session.get("map_round", 0) == 0:
session["generation"] = session.get("generation", 0) + 1
session["map_scores"] = {model.get("id"): [] for model in models}
# Pre-select the 3 maps for this entire generation based on current performance
if validation:
session["generation_maps"] = [
random.choice(self.validation_maps)
for _ in range(self.MAPS_PER_GENERATION)
]
else:
perf = session.get("performance", 0.0)
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
]
# Fetch the pre-selected map for the current round
map_round = session.get("map_round", 0)
selected_map = session["generation_maps"][map_round]
session["last_result"] = None
simulations = {}
for model in models:
@@ -150,9 +175,7 @@ class AiHttpGameHandler(GameHandler):
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
@@ -169,7 +192,6 @@ class AiHttpGameHandler(GameHandler):
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")
@@ -204,42 +226,75 @@ class AiHttpGameHandler(GameHandler):
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["map_round"] += 1
# Accumulate scores for this map round
for model_id, res in results.items():
session["map_scores"][model_id].append(res["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
# 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:
# 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:
session["stagnant_generations"] = 0 # Reset on breakthrough
# All 3 maps completed -> finalize generation results (Average scores)
response["type"] = "simulation_result"
final_results = {}
# Calculate final scores for each model based on their score on every map
for model_id, scores in session["map_scores"].items():
# Calculate Mean Squared Error from perfection (100)
squared_errors = [(100.0 - score) ** 2 for score in scores]
mse = sum(squared_errors) / len(squared_errors)
rmse = math.sqrt(mse)
# Final score penalizes variance/drops sharply based on RMSE
final_score = round(max(0.0, min(100.0, 100.0 - rmse)), 2)
# Attach final structural properties from the last results set
final_results[model_id] = deepcopy(results[model_id])
final_results[model_id]["score"] = final_score
# 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)
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
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()
# --- 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)
# Reset round counter for the next generation
session["map_round"] = 0
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:
game/mapSet.py
+3
View File
@@ -21,6 +21,9 @@ class MapSet:
raise IndexError("Map index out of range.")
return self.maps[index]
def get_maps(self):
return sorted(self.maps, key=lambda m: m.difficulty)
def get_maps_by_difficulty(self, desired_difficulty: int, num_maps: int):
"""
Returns a list of maps with difficulty levels as close as possible to the desired difficulty level.
maps/big_peanut_pirate.toml
+1 -1
View File
@@ -1,4 +1,4 @@
difficulty = 100
difficulty = 85
min_moves = 40
layout = """
#####
maps/home.toml → maps/home.toml.disabled
View File
maps/lava_bucket.toml → maps/lava_bucket.toml.disabled
+1
View File
@@ -1,6 +1,7 @@
difficulty = 70
min_moves = 6
layout = """
# #
# S #
# ##### #
# #