From 5dc58220bda85f812d5682f4c78ab882315eb238 Mon Sep 17 00:00:00 2001 From: LuckyLuke255 <105583948+LuckyLuke255@users.noreply.github.com> Date: Mon, 19 Jan 2026 13:42:41 +0100 Subject: [PATCH 1/3] Improve agent movement logic (anti-loop) --- src/agent.py | 254 ++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 190 insertions(+), 64 deletions(-) diff --git a/src/agent.py b/src/agent.py index 80c6526..af4db67 100644 --- a/src/agent.py +++ b/src/agent.py @@ -1,15 +1,20 @@ """Agent module containing the Agent class and related utilities for simulation.""" from __future__ import annotations -from food import FoodSource + +import logging import math import random -import pygame import typing -import logging -if typing.TYPE_CHECKING: - from environment import Environment import uuid +from collections import deque + +import pygame + from area import Area +from food import FoodSource + +if typing.TYPE_CHECKING: + from environment import Environment def _clamp(x: float, lo: float, hi: float) -> float: @@ -30,75 +35,101 @@ def _sqrt_scale(points: float, k: float) -> float: """Scales points using square root with coefficient k.""" return k * math.sqrt(max(0.0, points)) -def _sign(x: float): + +def _sign(x: float) -> float: """Returns sign of value x.""" return 1.0 if x >= 0 else -1.0 + def dist2(ax: float, ay: float, bx: float, by: float) -> float: """Returns squared distance between (ax, ay) and (bx, by) in torus space.""" dx, dy = torus_diff(ax, ay, bx, by) return dx * dx + dy * dy + def torus_diff(ax: float, ay: float, bx: float, by: float) -> typing.Tuple[float, float]: - """Returns dirrection (dx, dy) from (ax, ay) to (bx, by) in torus space.""" + """Returns direction (dx, dy) from (ax, ay) to (bx, by) in torus space.""" max_x = float(Agent.bound_x) max_y = float(Agent.bound_y) tx = float(bx - ax) ty = float(by - ay) - dx = tx if abs(tx) < max_x / 2 else tx - _sign(tx) * max_x + dx = tx if abs(tx) < max_x / 2 else tx - _sign(tx) * max_x dy = ty if abs(ty) < max_y / 2 else ty - _sign(ty) * max_y return (dx, dy) - + class Agent: """Represents an agent in the simulation with neural network-based decision making.""" # Sense vector (normalized inputs for neural network): - # 0 hp_n -> current HP / max HP (0..1) - # 1 en_n -> current energy / max energy (0..1) - # 2 age_n -> current age / max age (0..1) - # 3 x_n -> x position normalized to world width (0..1) - # 4 y_n -> y position normalized to world height (0..1) - # 5 head_x -> facing direction x (cos(angle)) - # 6 head_y -> facing direction y (-sin(angle)) - # 7 food_d_n -> dist2ance to nearest food normalized by sight (0..1) - # 8 food_dx_n -> x direction to nearest food (normalized) - # 9 food_dy_n -> y direction to nearest food (normalized) - # 10 food_in_sight -> whether food is within sight range (0 or 1) - # 11 friend_count_n -> nearby friends count (normalized) - # 11 enemy_count_n -> nearby enemies count (normalized) - # 12 enemy_d_n -> dist2ance to nearest enemy normalized by sight (0..1) - # 13 enemy_dx_n -> x direction to nearest enemy (normalized) - # 14 enemy_dy_n -> y direction to nearest enemy (normalized) - # 15 ground -> type of ground where agent actually is - # 16 bias -> constant bias input (always 1.0) + # 0 hp_n + # 1 en_n + # 2 age_n + # 3 x_n + # 4 y_n + # 5 head_x + # 6 head_y + # 7 food_d_n + # 8 food_dx_n + # 9 food_dy_n + # 10 food_in_sight (0/1) <-- kluczowa flaga od kolegi + # 11 friend_count_n + # 12 friend_d_n + # 13 friend_dx_n + # 14 friend_dy_n + # 15 friend_in_sight (0/1) <-- dodane analogicznie + # 16 enemy_count_n + # 17 enemy_d_n + # 18 enemy_dx_n + # 19 enemy_dy_n + # 20 enemy_in_sight (0/1) <-- dodane analogicznie + # 21 ground + # 22 loop_n (0..1) <-- detekcja kręcenia kółek (pomocniczy sygnał) + # 23 bias (1.0) + bound_x = 0 bound_y = 0 cell_size = 1 - - actions = {"ACTION_MOVE" : 0, - "ACTION_IDLE" : 1, - "ACTION_FLEE" : 2, - "ACTION_MATE" : 3, - "ACTION_ATTACK" : 4} + actions = { + "ACTION_MOVE": 0, + "ACTION_IDLE": 1, + "ACTION_FLEE": 2, + "ACTION_MATE": 3, + "ACTION_ATTACK": 4, + } body_points_total = 100 logging.basicConfig(level=logging.INFO) logger = logging.getLogger() - def __init__(self, position: tuple, environment : Environment,/, decision_matrix : typing.List[typing.List[int]] = None, genome = None, species = None ): + # --- anti-loop / drift tuning --- + _HISTORY_LEN = 40 # ile ostatnich pozycji trzymamy + _MIN_PROGRESS_PX = 12.0 # poniżej tej "drogi netto" uznajemy, że jest pętla/stagnacja + _LOOP_CONFIRM_STEPS = 25 # ile klatek musi to trwać, żeby uznać pętlę + _DRIFT_MAX_DEG = 20.0 # maksymalny dryf w stopniach (nie robi chaosu) + _DRIFT_STEP_DEG = 2.0 # jak szybko dryf się zmienia + _DRIFT_DECAY = 0.85 # jak szybko dryf zanika, gdy nie ma problemu + + def __init__( + self, + position: tuple, + environment: "Environment", + /, + decision_matrix: typing.List[typing.List[float]] | None = None, + genome=None, + species=None, + ): from mating import Mating self.environment = environment - self.mate_module :Mating = Mating(self) + self.mate_module: Mating = Mating(self) self.x = float(position[0]) self.y = float(position[1]) self.uuid = uuid.uuid4() - - self.group_id = species if species else random.randint(0,1) # team/species id (same -> friend, different -> enemy) + self.group_id = species if species is not None else random.randint(0, 1) if genome: self.body_points = genome @@ -120,22 +151,51 @@ def __init__(self, position: tuple, environment : Environment,/, decision_matrix self.angle = random.random() * 360.0 self.last_action = self.actions["ACTION_MOVE"] + # --- memory for anti-loop --- + self._pos_history: deque[tuple[float, float]] = deque(maxlen=self._HISTORY_LEN) + self._loop_counter = 0 + self._drift_deg = 0.0 + + self._inputs_override = None + self._last_enemy = None + self._last_food = None + self._last_friend = None + + # infer input size from sense vector self.input_size = len(self.sense()) self.action_count = len(self.actions) self.output_size = self.action_count + 2 if decision_matrix: - self.weights = decision_matrix + self.weights = self._normalize_or_rebuild_weights(decision_matrix) else: self.weights = self._random_matrix(self.input_size, self.output_size, scale=0.6) - - self._inputs_override = None - self._last_enemy = None - self._last_food = None + def _normalize_or_rebuild_weights(self, decision_matrix: typing.List[typing.List[float]]): + """ + Backward compatibility: if someone passes a matrix with wrong shape, + we rebuild it to the correct input_size/output_size. + """ + try: + rows = len(decision_matrix) + cols = len(decision_matrix[0]) if rows else 0 + except Exception: + rows, cols = 0, 0 + + if rows == self.input_size and cols == self.output_size: + return decision_matrix + + # Rebuild to avoid runtime IndexError in think() + rebuilt = self._random_matrix(self.input_size, self.output_size, scale=0.6) + min_r = min(rows, self.input_size) + min_c = min(cols, self.output_size) + for i in range(min_r): + for j in range(min_c): + rebuilt[i][j] = float(decision_matrix[i][j]) + return rebuilt def _random_body_points(self, total: int): - """Randomly dist2ribute total points across body stat keys.""" + """Randomly distribute total points across body stat keys.""" keys = ["hp", "energy", "speed", "attack", "lifespan", "sight", "agility"] pts = {k: 0 for k in keys} for _ in range(total): @@ -155,6 +215,33 @@ def set_inputs(self, inputs): raise ValueError(f"expected input vector length {self.input_size}, got {len(inputs)}") self._inputs_override = [float(v) for v in inputs] + def _net_progress(self) -> float: + """Net displacement over history window.""" + if len(self._pos_history) < 2: + return 1e9 + x0, y0 = self._pos_history[0] + x1, y1 = self._pos_history[-1] + dx, dy = torus_diff(x0, y0, x1, y1) + return math.sqrt(dx * dx + dy * dy) + + def _update_loop_state(self, has_target_in_sight: bool) -> float: + """ + Returns loop_n in [0..1]. Increases when agent doesn't make progress and has no target. + """ + self._pos_history.append((self.x, self.y)) + + if has_target_in_sight: + self._loop_counter = max(0, self._loop_counter - 2) + return _clamp(self._loop_counter / float(self._LOOP_CONFIRM_STEPS), 0.0, 1.0) + + progress = self._net_progress() + if progress < self._MIN_PROGRESS_PX and len(self._pos_history) >= self._LOOP_CONFIRM_STEPS: + self._loop_counter = min(self._LOOP_CONFIRM_STEPS, self._loop_counter + 1) + else: + self._loop_counter = max(0, self._loop_counter - 1) + + return _clamp(self._loop_counter / float(self._LOOP_CONFIRM_STEPS), 0.0, 1.0) + def sense(self, foods=None, agents=None): """Generate normalized sensory input vector.""" bx = float(self.bound_x) if self.bound_x > 0 else 1.0 @@ -164,13 +251,14 @@ def sense(self, foods=None, agents=None): en_n = _clamp(self.energy / max(1e-9, self.max_energy), 0.0, 1.0) age_n = _clamp(self.age / max(1, self.max_age), 0.0, 1.0) - x_n = _clamp(self.x / max(1e-9, bx), 0.0, 1.0) # normalized position + x_n = _clamp(self.x / max(1e-9, bx), 0.0, 1.0) y_n = _clamp(self.y / max(1e-9, by), 0.0, 1.0) ang = math.radians(self.angle) - head_x = math.cos(ang) # facing direction (unit vector) + head_x = math.cos(ang) head_y = -math.sin(ang) + # --- food --- food_d_n, food_dx_n, food_dy_n, food_in_sight = 1.0, 0.0, 0.0, 0 if foods: best_d2 = 1e18 @@ -185,24 +273,27 @@ def sense(self, foods=None, agents=None): if best is not None: dx, dy = torus_diff(self.x, self.y, best[0], best[1]) d = math.sqrt(best_d2) + + # Normalizacja dystansu: nadal clamp 0..1, ale boolean mówi "czy w zasięgu" food_d_n = _clamp(d / max(1e-9, self.sight), 0.0, 1.0) food_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) food_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_food = best - if d <= self.sight: # Adding boolean to deal with semantic discontinuity ( 1 could mean food is far away and 0,95 mean is really close) + if d <= self.sight: food_in_sight = 1 + # --- friends / enemies --- friend_count_n = 0.0 - friend_d_n, friend_dx_n, friend_dy_n = 1.0, 0.0, 0.0 + friend_d_n, friend_dx_n, friend_dy_n, friend_in_sight = 1.0, 0.0, 0.0, 0 enemy_count_n = 0.0 - enemy_d_n, enemy_dx_n, enemy_dy_n = 1.0, 0.0, 0.0 + enemy_d_n, enemy_dx_n, enemy_dy_n, enemy_in_sight = 1.0, 0.0, 0.0, 0 if agents: r2 = self.sight * self.sight friends = 0 enemies = 0 best_friend_d2 = 1e18 - best_friend = None + best_friend = None best_enemy_d2 = 1e18 best_enemy = None @@ -237,26 +328,38 @@ def sense(self, foods=None, agents=None): enemy_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) enemy_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_enemy = best_enemy + if d <= self.sight: + enemy_in_sight = 1 else: self._last_enemy = None - - if best_friend is not None: # Adding this part to enable to agent getting information where are friends + if best_friend is not None: dx, dy = torus_diff(self.x, self.y, best_friend[0], best_friend[1]) d = math.sqrt(best_friend_d2) friend_d_n = _clamp(d / max(1e-9, self.sight), 0.0, 1.0) friend_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) friend_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_friend = best_friend + if d <= self.sight: + friend_in_sight = 1 else: self._last_friend = None cell = self.environment._get_agent_area(self) - if cell == Area.PLAINS: ground = 0 - elif cell == Area.FERTILE_VALLEY: ground = 0.9 - elif cell == Area.DESERT: ground = 0.1 - elif cell == Area.BERRY_CORNER: ground= 1 - else: raise ValueError(f"Wrong place! {cell}, {type(cell)}") + if cell == Area.PLAINS: + ground = 0.0 + elif cell == Area.FERTILE_VALLEY: + ground = 0.9 + elif cell == Area.DESERT: + ground = 0.1 + elif cell == Area.BERRY_CORNER: + ground = 1.0 + else: + raise ValueError(f"Wrong place! {cell}, {type(cell)}") + + # loop detector: tylko jeśli nie ma celu w zasięgu + has_target_in_sight = bool(food_in_sight or enemy_in_sight or friend_in_sight) + loop_n = self._update_loop_state(has_target_in_sight=has_target_in_sight) return [ hp_n, @@ -269,17 +372,20 @@ def sense(self, foods=None, agents=None): food_d_n, food_dx_n, food_dy_n, - food_in_sight, + float(food_in_sight), friend_count_n, friend_d_n, friend_dx_n, friend_dy_n, + float(friend_in_sight), enemy_count_n, enemy_d_n, enemy_dx_n, enemy_dy_n, + float(enemy_in_sight), ground, - 1.0 + loop_n, + 1.0, ] def think(self, inputs): @@ -321,9 +427,27 @@ def _apply_bounds(self, new_x: float, new_y: float, new_angle: float): self.y = (new_y + by) % max_y self.angle = new_angle % 360.0 - def _move(self, turn: float, intensity: float, speed_modifier: float = 1.0): + def _update_drift(self, loop_n: float): + """ + Lekki dryf: gdy loop_n rośnie, dryf rośnie (mała losowa korekta kierunku), + gdy loop_n maleje, dryf zanika. + """ + if loop_n > 0.5: + # random-walk drift + step = random.uniform(-self._DRIFT_STEP_DEG, self._DRIFT_STEP_DEG) + self._drift_deg = _clamp(self._drift_deg + step, -self._DRIFT_MAX_DEG, self._DRIFT_MAX_DEG) + else: + # decay to 0 + self._drift_deg *= self._DRIFT_DECAY + if abs(self._drift_deg) < 0.05: + self._drift_deg = 0.0 + + def _move(self, turn: float, intensity: float, speed_modifier: float = 1.0, loop_n: float = 0.0): + # turn from network + drift if looping + self._update_drift(loop_n) + turn_delta = turn * (self.agility * 0.5) - new_angle = self.angle + turn_delta + new_angle = self.angle + turn_delta + self._drift_deg * loop_n inten = _clamp(0.5 + 0.5 * intensity, 0.0, 1.0) sp = self.base_speed * (0.20 + 1.30 * inten) * speed_modifier @@ -355,7 +479,6 @@ def _attack(self): def _mate(self): self.mate_module.mate() - def _tick_body(self): self.age += 1 self.energy = max(0.0, self.energy - 0.01) @@ -380,14 +503,17 @@ def update(self, speed_modifier): action, turn, intensity = self.decide(outputs) self.last_action = action + # loop_n jest na przedostatniej pozycji (przed bias) + loop_n = float(inputs[-2]) + if action == self.actions["ACTION_MOVE"]: self.logger.debug(f"Agent - {self.uuid}; action - move") - self._move(turn, intensity) + self._move(turn, intensity, speed_modifier=speed_modifier, loop_n=loop_n) elif action == self.actions["ACTION_IDLE"]: self.logger.debug(f"Agent - {self.uuid}; action - idle") self._idle() elif action == self.actions["ACTION_FLEE"]: - self.logger.debug(f"Agent - {self.uuid}; action - free") + self.logger.debug(f"Agent - {self.uuid}; action - flee") self._flee(turn, intensity) elif action == self.actions["ACTION_MATE"]: self.logger.debug(f"Agent - {self.uuid}; action - mate") @@ -418,7 +544,7 @@ def render(self, window: pygame.window, cell_size: int, offset: tuple): rad = math.radians(self.angle - 135.0) points.append((env_x + math.cos(rad) * r, env_y - math.sin(rad) * r)) - color = (100, 200, 255) if self.group_id == 0 else (255, 255, 255) + color = (100, 200, 255) if self.group_id == 0 else (255, 255, 255) pygame.draw.polygon(window, color, points) bar_w = int(cell_size * 0.8) From 3c823587d1c620339251392e120278323bc1a7fb Mon Sep 17 00:00:00 2001 From: LuckyLuke255 <105583948+LuckyLuke255@users.noreply.github.com> Date: Tue, 20 Jan 2026 00:29:01 +0100 Subject: [PATCH 2/3] Update agent.py --- src/agent.py | 324 ++++++++++++++++++++++----------------------------- 1 file changed, 140 insertions(+), 184 deletions(-) diff --git a/src/agent.py b/src/agent.py index af4db67..4c0e416 100644 --- a/src/agent.py +++ b/src/agent.py @@ -1,20 +1,15 @@ """Agent module containing the Agent class and related utilities for simulation.""" from __future__ import annotations - -import logging import math import random -import typing -import uuid from collections import deque - import pygame - -from area import Area -from food import FoodSource - +import typing +import logging if typing.TYPE_CHECKING: from environment import Environment +import uuid +from area import Area def _clamp(x: float, lo: float, hi: float) -> float: @@ -35,20 +30,17 @@ def _sqrt_scale(points: float, k: float) -> float: """Scales points using square root with coefficient k.""" return k * math.sqrt(max(0.0, points)) - -def _sign(x: float) -> float: +def _sign(x: float): """Returns sign of value x.""" return 1.0 if x >= 0 else -1.0 - def dist2(ax: float, ay: float, bx: float, by: float) -> float: """Returns squared distance between (ax, ay) and (bx, by) in torus space.""" dx, dy = torus_diff(ax, ay, bx, by) return dx * dx + dy * dy - def torus_diff(ax: float, ay: float, bx: float, by: float) -> typing.Tuple[float, float]: - """Returns direction (dx, dy) from (ax, ay) to (bx, by) in torus space.""" + """Returns dirrection (dx, dy) from (ax, ay) to (bx, by) in torus space.""" max_x = float(Agent.bound_x) max_y = float(Agent.bound_y) tx = float(bx - ax) @@ -62,74 +54,58 @@ class Agent: """Represents an agent in the simulation with neural network-based decision making.""" # Sense vector (normalized inputs for neural network): - # 0 hp_n - # 1 en_n - # 2 age_n - # 3 x_n - # 4 y_n - # 5 head_x - # 6 head_y - # 7 food_d_n - # 8 food_dx_n - # 9 food_dy_n - # 10 food_in_sight (0/1) <-- kluczowa flaga od kolegi - # 11 friend_count_n - # 12 friend_d_n - # 13 friend_dx_n - # 14 friend_dy_n - # 15 friend_in_sight (0/1) <-- dodane analogicznie - # 16 enemy_count_n - # 17 enemy_d_n - # 18 enemy_dx_n - # 19 enemy_dy_n - # 20 enemy_in_sight (0/1) <-- dodane analogicznie - # 21 ground - # 22 loop_n (0..1) <-- detekcja kręcenia kółek (pomocniczy sygnał) - # 23 bias (1.0) - + # 0 hp_n -> current HP / max HP (0..1) + # 1 en_n -> current energy / max energy (0..1) + # 2 age_n -> current age / max age (0..1) + # 3 x_n -> x position normalized to world width (0..1) + # 4 y_n -> y position normalized to world height (0..1) + # 5 head_x -> facing direction x (cos(angle)) + # 6 head_y -> facing direction y (-sin(angle)) + # 7 food_d_n -> dist2ance to nearest food normalized by sight (0..1) + # 8 food_dx_n -> x direction to nearest food (normalized) + # 9 food_dy_n -> y direction to nearest food (normalized) + # 10 food_in_sight -> whether food is within sight range (0 or 1) + # 11 friend_count_n -> nearby friends count (normalized) + # 11 enemy_count_n -> nearby enemies count (normalized) + # 12 enemy_d_n -> dist2ance to nearest enemy normalized by sight (0..1) + # 13 enemy_dx_n -> x direction to nearest enemy (normalized) + # 14 enemy_dy_n -> y direction to nearest enemy (normalized) + # 15 ground -> type of ground where agent actually is + # 16 bias -> constant bias input (always 1.0) bound_x = 0 bound_y = 0 cell_size = 1 - actions = { - "ACTION_MOVE": 0, - "ACTION_IDLE": 1, - "ACTION_FLEE": 2, - "ACTION_MATE": 3, - "ACTION_ATTACK": 4, - } + + actions = {"ACTION_MOVE" : 0, + "ACTION_IDLE" : 1, + "ACTION_FLEE" : 2, + "ACTION_MATE" : 3, + "ACTION_ATTACK" : 4} body_points_total = 100 logging.basicConfig(level=logging.INFO) logger = logging.getLogger() - # --- anti-loop / drift tuning --- - _HISTORY_LEN = 40 # ile ostatnich pozycji trzymamy - _MIN_PROGRESS_PX = 12.0 # poniżej tej "drogi netto" uznajemy, że jest pętla/stagnacja - _LOOP_CONFIRM_STEPS = 25 # ile klatek musi to trwać, żeby uznać pętlę - _DRIFT_MAX_DEG = 20.0 # maksymalny dryf w stopniach (nie robi chaosu) - _DRIFT_STEP_DEG = 2.0 # jak szybko dryf się zmienia - _DRIFT_DECAY = 0.85 # jak szybko dryf zanika, gdy nie ma problemu - - def __init__( - self, - position: tuple, - environment: "Environment", - /, - decision_matrix: typing.List[typing.List[float]] | None = None, - genome=None, - species=None, - ): + + LOOP_WINDOW = 24 + LOOP_MAX_SPAN = 3.0 + LOOP_TURN_STD_EPS = 0.035 + LOOP_TURN_MEAN_MIN = 0.18 + ESCAPE_STEPS = 14 + + def __init__(self, position: tuple, environment : Environment,/, decision_matrix : typing.List[typing.List[int]] = None, genome = None, species = None ): from mating import Mating self.environment = environment - self.mate_module: Mating = Mating(self) + self.mate_module :Mating = Mating(self) self.x = float(position[0]) self.y = float(position[1]) self.uuid = uuid.uuid4() - self.group_id = species if species is not None else random.randint(0, 1) + + self.group_id = species if species else random.randint(0,1) # species id (same -> friend, different -> enemy) if genome: self.body_points = genome @@ -140,7 +116,7 @@ def __init__( self.max_energy = (10.0 + _sqrt_scale(self.body_points["energy"], 2.0)) * 2 self.base_speed = 0.5 + _sqrt_scale(self.body_points["speed"], 0.2) self.attack_power = 0.5 + _sqrt_scale(self.body_points["attack"], 0.06) - self.max_age = int(200 + _sqrt_scale(self.body_points["lifespan"], 14.0)) * 2 + self.max_age = int((300 + _sqrt_scale(self.body_points["lifespan"], 18.0)) * 2.5) self.sight = 70.0 + _sqrt_scale(self.body_points["sight"], 6.0) self.agility = 30.0 + _sqrt_scale(self.body_points["agility"], 2.0) @@ -151,51 +127,28 @@ def __init__( self.angle = random.random() * 360.0 self.last_action = self.actions["ACTION_MOVE"] - # --- memory for anti-loop --- - self._pos_history: deque[tuple[float, float]] = deque(maxlen=self._HISTORY_LEN) - self._loop_counter = 0 - self._drift_deg = 0.0 - - self._inputs_override = None - self._last_enemy = None - self._last_food = None - self._last_friend = None + #tracks recent motion; does not change the NN input/output interface. + self._loop_positions = deque(maxlen=self.LOOP_WINDOW) + self._loop_turns = deque(maxlen=self.LOOP_WINDOW) + self._escape_steps_left = 0 + self._loop_position_anchor = (self.x, self.y) - # infer input size from sense vector self.input_size = len(self.sense()) self.action_count = len(self.actions) self.output_size = self.action_count + 2 if decision_matrix: - self.weights = self._normalize_or_rebuild_weights(decision_matrix) + self.weights = decision_matrix else: self.weights = self._random_matrix(self.input_size, self.output_size, scale=0.6) - def _normalize_or_rebuild_weights(self, decision_matrix: typing.List[typing.List[float]]): - """ - Backward compatibility: if someone passes a matrix with wrong shape, - we rebuild it to the correct input_size/output_size. - """ - try: - rows = len(decision_matrix) - cols = len(decision_matrix[0]) if rows else 0 - except Exception: - rows, cols = 0, 0 - - if rows == self.input_size and cols == self.output_size: - return decision_matrix - - # Rebuild to avoid runtime IndexError in think() - rebuilt = self._random_matrix(self.input_size, self.output_size, scale=0.6) - min_r = min(rows, self.input_size) - min_c = min(cols, self.output_size) - for i in range(min_r): - for j in range(min_c): - rebuilt[i][j] = float(decision_matrix[i][j]) - return rebuilt + + self._inputs_override = None + self._last_enemy = None + self._last_food = None def _random_body_points(self, total: int): - """Randomly distribute total points across body stat keys.""" + """Randomly dist2ribute total points across body stat keys.""" keys = ["hp", "energy", "speed", "attack", "lifespan", "sight", "agility"] pts = {k: 0 for k in keys} for _ in range(total): @@ -215,32 +168,65 @@ def set_inputs(self, inputs): raise ValueError(f"expected input vector length {self.input_size}, got {len(inputs)}") self._inputs_override = [float(v) for v in inputs] - def _net_progress(self) -> float: - """Net displacement over history window.""" - if len(self._pos_history) < 2: - return 1e9 - x0, y0 = self._pos_history[0] - x1, y1 = self._pos_history[-1] - dx, dy = torus_diff(x0, y0, x1, y1) - return math.sqrt(dx * dx + dy * dy) - - def _update_loop_state(self, has_target_in_sight: bool) -> float: - """ - Returns loop_n in [0..1]. Increases when agent doesn't make progress and has no target. - """ - self._pos_history.append((self.x, self.y)) - - if has_target_in_sight: - self._loop_counter = max(0, self._loop_counter - 2) - return _clamp(self._loop_counter / float(self._LOOP_CONFIRM_STEPS), 0.0, 1.0) - - progress = self._net_progress() - if progress < self._MIN_PROGRESS_PX and len(self._pos_history) >= self._LOOP_CONFIRM_STEPS: - self._loop_counter = min(self._LOOP_CONFIRM_STEPS, self._loop_counter + 1) - else: - self._loop_counter = max(0, self._loop_counter - 1) + #keep random-weight agents from moving in perfect circles + def _record_motion_sample(self, applied_turn: float, did_move: bool) -> None: + if not did_move: + return + + if len(self._loop_positions) == 0: + self._loop_position_anchor = (self.x, self.y) + + self._loop_turns.append(float(_clamp(applied_turn, -1.0, 1.0))) + self._loop_positions.append((self.x, self.y)) + + def _unwrapped_positions(self) -> typing.List[typing.Tuple[float, float]]: + ax, ay = self._loop_position_anchor + out: typing.List[typing.Tuple[float, float]] = [] + for px, py in self._loop_positions: + dx, dy = torus_diff(ax, ay, px, py) + out.append((dx, dy)) + return out - return _clamp(self._loop_counter / float(self._LOOP_CONFIRM_STEPS), 0.0, 1.0) + @staticmethod + def _mean_std(values: typing.Sequence[float]) -> typing.Tuple[float, float]: + n = len(values) + if n == 0: + return 0.0, 0.0 + mean = sum(values) / n + var = sum((v - mean) * (v - mean) for v in values) / n + return mean, math.sqrt(var) + + def _looks_like_circle(self) -> bool: + if len(self._loop_turns) < self.LOOP_WINDOW: + return False + + mean_turn, std_turn = self._mean_std(self._loop_turns) + if abs(mean_turn) < self.LOOP_TURN_MEAN_MIN or std_turn > self.LOOP_TURN_STD_EPS: + return False + + pts = self._unwrapped_positions() + xs = [p[0] for p in pts] + ys = [p[1] for p in pts] + span = max(max(xs) - min(xs), max(ys) - min(ys)) + return span <= self.LOOP_MAX_SPAN + + def _apply_anti_loop(self, action: int, turn: float, intensity: float) -> typing.Tuple[int, float, float]: + if self._escape_steps_left > 0: + self._escape_steps_left -= 1 + escape_turn = _clamp(turn, -1.0, 1.0) + (random.random() * 2.0 - 1.0) * 0.55 + return self.actions["ACTION_MOVE"], _clamp(escape_turn, -1.0, 1.0), max(0.35, intensity) + + if action != self.actions["ACTION_MOVE"]: + return action, turn, intensity + + if self._looks_like_circle(): + self._escape_steps_left = self.ESCAPE_STEPS + mean_turn, _ = self._mean_std(self._loop_turns) + flipped = -_clamp(mean_turn, -1.0, 1.0) + escape_turn = flipped + (random.random() * 2.0 - 1.0) * 0.35 + return self.actions["ACTION_MOVE"], _clamp(escape_turn, -1.0, 1.0), max(0.45, intensity) + + return action, turn, intensity def sense(self, foods=None, agents=None): """Generate normalized sensory input vector.""" @@ -251,14 +237,13 @@ def sense(self, foods=None, agents=None): en_n = _clamp(self.energy / max(1e-9, self.max_energy), 0.0, 1.0) age_n = _clamp(self.age / max(1, self.max_age), 0.0, 1.0) - x_n = _clamp(self.x / max(1e-9, bx), 0.0, 1.0) + x_n = _clamp(self.x / max(1e-9, bx), 0.0, 1.0) # normalized position y_n = _clamp(self.y / max(1e-9, by), 0.0, 1.0) ang = math.radians(self.angle) - head_x = math.cos(ang) + head_x = math.cos(ang) # facing direction (unit vector) head_y = -math.sin(ang) - # --- food --- food_d_n, food_dx_n, food_dy_n, food_in_sight = 1.0, 0.0, 0.0, 0 if foods: best_d2 = 1e18 @@ -273,20 +258,17 @@ def sense(self, foods=None, agents=None): if best is not None: dx, dy = torus_diff(self.x, self.y, best[0], best[1]) d = math.sqrt(best_d2) - - # Normalizacja dystansu: nadal clamp 0..1, ale boolean mówi "czy w zasięgu" food_d_n = _clamp(d / max(1e-9, self.sight), 0.0, 1.0) food_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) food_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_food = best - if d <= self.sight: + if d <= self.sight: # Adding boolean to deal with semantic discontinuity ( 1 could mean food is far away and 0,95 mean is really close) food_in_sight = 1 - # --- friends / enemies --- friend_count_n = 0.0 - friend_d_n, friend_dx_n, friend_dy_n, friend_in_sight = 1.0, 0.0, 0.0, 0 + friend_d_n, friend_dx_n, friend_dy_n = 1.0, 0.0, 0.0 enemy_count_n = 0.0 - enemy_d_n, enemy_dx_n, enemy_dy_n, enemy_in_sight = 1.0, 0.0, 0.0, 0 + enemy_d_n, enemy_dx_n, enemy_dy_n = 1.0, 0.0, 0.0 if agents: r2 = self.sight * self.sight @@ -328,38 +310,26 @@ def sense(self, foods=None, agents=None): enemy_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) enemy_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_enemy = best_enemy - if d <= self.sight: - enemy_in_sight = 1 else: self._last_enemy = None - if best_friend is not None: + + if best_friend is not None: # Adding this part to enable to agent getting information where are friends dx, dy = torus_diff(self.x, self.y, best_friend[0], best_friend[1]) d = math.sqrt(best_friend_d2) friend_d_n = _clamp(d / max(1e-9, self.sight), 0.0, 1.0) friend_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) friend_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_friend = best_friend - if d <= self.sight: - friend_in_sight = 1 else: self._last_friend = None cell = self.environment._get_agent_area(self) - if cell == Area.PLAINS: - ground = 0.0 - elif cell == Area.FERTILE_VALLEY: - ground = 0.9 - elif cell == Area.DESERT: - ground = 0.1 - elif cell == Area.BERRY_CORNER: - ground = 1.0 - else: - raise ValueError(f"Wrong place! {cell}, {type(cell)}") - - # loop detector: tylko jeśli nie ma celu w zasięgu - has_target_in_sight = bool(food_in_sight or enemy_in_sight or friend_in_sight) - loop_n = self._update_loop_state(has_target_in_sight=has_target_in_sight) + if cell == Area.PLAINS: ground = 0 + elif cell == Area.FERTILE_VALLEY: ground = 0.9 + elif cell == Area.DESERT: ground = 0.1 + elif cell == Area.BERRY_CORNER: ground= 1 + else: raise ValueError(f"Wrong place! {cell}, {type(cell)}") return [ hp_n, @@ -372,20 +342,17 @@ def sense(self, foods=None, agents=None): food_d_n, food_dx_n, food_dy_n, - float(food_in_sight), + food_in_sight, friend_count_n, friend_d_n, friend_dx_n, friend_dy_n, - float(friend_in_sight), enemy_count_n, enemy_d_n, enemy_dx_n, enemy_dy_n, - float(enemy_in_sight), ground, - loop_n, - 1.0, + 1.0 ] def think(self, inputs): @@ -427,27 +394,9 @@ def _apply_bounds(self, new_x: float, new_y: float, new_angle: float): self.y = (new_y + by) % max_y self.angle = new_angle % 360.0 - def _update_drift(self, loop_n: float): - """ - Lekki dryf: gdy loop_n rośnie, dryf rośnie (mała losowa korekta kierunku), - gdy loop_n maleje, dryf zanika. - """ - if loop_n > 0.5: - # random-walk drift - step = random.uniform(-self._DRIFT_STEP_DEG, self._DRIFT_STEP_DEG) - self._drift_deg = _clamp(self._drift_deg + step, -self._DRIFT_MAX_DEG, self._DRIFT_MAX_DEG) - else: - # decay to 0 - self._drift_deg *= self._DRIFT_DECAY - if abs(self._drift_deg) < 0.05: - self._drift_deg = 0.0 - - def _move(self, turn: float, intensity: float, speed_modifier: float = 1.0, loop_n: float = 0.0): - # turn from network + drift if looping - self._update_drift(loop_n) - + def _move(self, turn: float, intensity: float, speed_modifier: float = 1.0): turn_delta = turn * (self.agility * 0.5) - new_angle = self.angle + turn_delta + self._drift_deg * loop_n + new_angle = self.angle + turn_delta inten = _clamp(0.5 + 0.5 * intensity, 0.0, 1.0) sp = self.base_speed * (0.20 + 1.30 * inten) * speed_modifier @@ -457,7 +406,7 @@ def _move(self, turn: float, intensity: float, speed_modifier: float = 1.0, loop new_y = self.y - math.sin(rad) * sp self._apply_bounds(new_x, new_y, new_angle) - cost = 0.02 + 0.06 * inten + cost = 0.01 + 0.05 * inten self.energy = max(0.0, self.energy - cost) def _idle(self): @@ -479,6 +428,7 @@ def _attack(self): def _mate(self): self.mate_module.mate() + def _tick_body(self): self.age += 1 self.energy = max(0.0, self.energy - 0.01) @@ -495,25 +445,28 @@ def update(self, speed_modifier): return if self._inputs_override is None: - inputs = self.sense(self.environment.food_sources, self.environment.get_agents()) + nearby_agents = self.environment.get_nearby_agents(self, self.sight * 2) + inputs = self.sense(self.environment.food_sources, nearby_agents) else: inputs = self._inputs_override outputs = self.think(inputs) action, turn, intensity = self.decide(outputs) + + # Anti-loop adjustment: breaks deterministic circles when turn stays nearly constant. + action, turn, intensity = self._apply_anti_loop(action, turn, intensity) self.last_action = action - # loop_n jest na przedostatniej pozycji (przed bias) - loop_n = float(inputs[-2]) + did_move = action == self.actions["ACTION_MOVE"] if action == self.actions["ACTION_MOVE"]: self.logger.debug(f"Agent - {self.uuid}; action - move") - self._move(turn, intensity, speed_modifier=speed_modifier, loop_n=loop_n) + self._move(turn, intensity) elif action == self.actions["ACTION_IDLE"]: self.logger.debug(f"Agent - {self.uuid}; action - idle") self._idle() elif action == self.actions["ACTION_FLEE"]: - self.logger.debug(f"Agent - {self.uuid}; action - flee") + self.logger.debug(f"Agent - {self.uuid}; action - free") self._flee(turn, intensity) elif action == self.actions["ACTION_MATE"]: self.logger.debug(f"Agent - {self.uuid}; action - mate") @@ -522,6 +475,9 @@ def update(self, speed_modifier): self.logger.debug(f"Agent - {self.uuid}; action - attack") self._attack() + # Anti-loop tracking: record motion samples after the action was applied. + self._record_motion_sample(applied_turn=turn, did_move=did_move) + self._tick_body() def render(self, window: pygame.window, cell_size: int, offset: tuple): From cd9c75ed69eaa92b465f1ce24e1cb850daa8856e Mon Sep 17 00:00:00 2001 From: LuckyLuke255 <105583948+LuckyLuke255@users.noreply.github.com> Date: Tue, 20 Jan 2026 00:31:28 +0100 Subject: [PATCH 3/3] Update agent.py --- src/agent.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/agent.py b/src/agent.py index 4c0e416..465339b 100644 --- a/src/agent.py +++ b/src/agent.py @@ -127,7 +127,7 @@ def __init__(self, position: tuple, environment : Environment,/, decision_matrix self.angle = random.random() * 360.0 self.last_action = self.actions["ACTION_MOVE"] - #tracks recent motion; does not change the NN input/output interface. + #tracks recent motion, does not change the NN input/output interface. self._loop_positions = deque(maxlen=self.LOOP_WINDOW) self._loop_turns = deque(maxlen=self.LOOP_WINDOW) self._escape_steps_left = 0 @@ -262,7 +262,7 @@ def sense(self, foods=None, agents=None): food_dx_n = _clamp(dx / max(1e-9, self.sight), -1.0, 1.0) food_dy_n = _clamp(dy / max(1e-9, self.sight), -1.0, 1.0) self._last_food = best - if d <= self.sight: # Adding boolean to deal with semantic discontinuity ( 1 could mean food is far away and 0,95 mean is really close) + if d <= self.sight: # Adding boolean to deal with discontinuity ( 1 could mean food is far away and 0,95 mean is really close) food_in_sight = 1 friend_count_n = 0.0 @@ -453,7 +453,7 @@ def update(self, speed_modifier): outputs = self.think(inputs) action, turn, intensity = self.decide(outputs) - # Anti-loop adjustment: breaks deterministic circles when turn stays nearly constant. + #breaks circles when turn stays nearly constant. action, turn, intensity = self._apply_anti_loop(action, turn, intensity) self.last_action = action @@ -475,7 +475,7 @@ def update(self, speed_modifier): self.logger.debug(f"Agent - {self.uuid}; action - attack") self._attack() - # Anti-loop tracking: record motion samples after the action was applied. + #to record motion samples after the action was applied. self._record_motion_sample(applied_turn=turn, did_move=did_move) self._tick_body()