From ef72806c0616c7f1eb477187bf77fa5726c72248 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Florian=20Schr=C3=B6der?=
<fschroeder@techfak.uni-bielefeld.de>
Date: Thu, 29 Feb 2024 15:38:01 +0100
Subject: [PATCH] Add Movement class for handling player and counter
interactions
This commit introduces a new Movement class which handles player movements, collisions, and interactions with counters in the 'cooperative_cuisine' environment. The commit also includes necessary changes to related configurations and adjustments in the 'environment.py' and 'player.py' files to accommodate the introduction of this new class. This encapsulation enhances readability, promotes code organization, and adheres to the principle of Single Responsibility.
---
cooperative_cuisine/__init__.py | 3 +-
.../configs/agents/random_agent.py | 2 +
.../configs/environment_config.yaml | 2 +-
.../configs/study/level1/level1_config.yaml | 2 +-
.../configs/study/level2/level2_config.yaml | 2 +-
.../configs/study/study_config.yaml | 4 +-
.../configs/tutorial_env_config.yaml | 2 +-
cooperative_cuisine/counter_factory.py | 128 +++++-
cooperative_cuisine/environment.py | 374 ++----------------
cooperative_cuisine/movement.py | 181 +++++++++
cooperative_cuisine/player.py | 2 +-
.../environment_config_rl.yaml | 2 +-
.../reinforcement_learning/gym_env.py | 6 +-
tests/test_start.py | 14 +-
14 files changed, 359 insertions(+), 365 deletions(-)
create mode 100644 cooperative_cuisine/movement.py
diff --git a/cooperative_cuisine/__init__.py b/cooperative_cuisine/__init__.py
index c32875fb..b733be00 100644
--- a/cooperative_cuisine/__init__.py
+++ b/cooperative_cuisine/__init__.py
@@ -309,7 +309,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 8
+ speed_units_per_seconds: 8
interaction_range: 1.6
```
@@ -367,6 +367,7 @@ On the left you can find the navigation panel that brings you to the implementat
websockets,
- the **hooks**, you can adapt and extend recording, scoring, msgs to players, etc. via hooks,
- the **info msgs** to players, based on hook events, short text msgs can be generated,
+- the calculation of player **movement**,
- the **orders**, how to sample incoming orders and their attributes,
- the **player**/agent, that interacts in the environment,
- the **pygame 2d visualization**, GUI, drawing, and video generation,
diff --git a/cooperative_cuisine/configs/agents/random_agent.py b/cooperative_cuisine/configs/agents/random_agent.py
index 1c36a303..db05c7a9 100644
--- a/cooperative_cuisine/configs/agents/random_agent.py
+++ b/cooperative_cuisine/configs/agents/random_agent.py
@@ -61,6 +61,8 @@ async def agent():
json.dumps({"type": "get_state", "player_hash": args.player_hash})
)
state = json.loads(await websocket.recv())
+ if not state["all_players_ready"]:
+ continue
if counters is None:
counters = defaultdict(list)
diff --git a/cooperative_cuisine/configs/environment_config.yaml b/cooperative_cuisine/configs/environment_config.yaml
index d589789d..6de17894 100644
--- a/cooperative_cuisine/configs/environment_config.yaml
+++ b/cooperative_cuisine/configs/environment_config.yaml
@@ -82,7 +82,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 6
+ speed_units_per_seconds: 6
interaction_range: 1.6
restricted_view: False
view_angle: 70
diff --git a/cooperative_cuisine/configs/study/level1/level1_config.yaml b/cooperative_cuisine/configs/study/level1/level1_config.yaml
index 7dad55ac..6732fa29 100644
--- a/cooperative_cuisine/configs/study/level1/level1_config.yaml
+++ b/cooperative_cuisine/configs/study/level1/level1_config.yaml
@@ -81,7 +81,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 6
+ speed_units_per_seconds: 6
interaction_range: 1.6
restricted_view: False
view_angle: 70
diff --git a/cooperative_cuisine/configs/study/level2/level2_config.yaml b/cooperative_cuisine/configs/study/level2/level2_config.yaml
index 918a6533..5bc7dd67 100644
--- a/cooperative_cuisine/configs/study/level2/level2_config.yaml
+++ b/cooperative_cuisine/configs/study/level2/level2_config.yaml
@@ -81,7 +81,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 6
+ speed_units_per_seconds: 6
interaction_range: 1.6
restricted_view: True
view_angle: 100
diff --git a/cooperative_cuisine/configs/study/study_config.yaml b/cooperative_cuisine/configs/study/study_config.yaml
index e20ee61f..4feba6bf 100644
--- a/cooperative_cuisine/configs/study/study_config.yaml
+++ b/cooperative_cuisine/configs/study/study_config.yaml
@@ -19,5 +19,5 @@ levels:
name: "Level 1-4: Bottleneck"
-num_players: 1
-num_bots: 0
+num_players: 2
+num_bots: 12
diff --git a/cooperative_cuisine/configs/tutorial_env_config.yaml b/cooperative_cuisine/configs/tutorial_env_config.yaml
index 1a41ba31..9b9011ae 100644
--- a/cooperative_cuisine/configs/tutorial_env_config.yaml
+++ b/cooperative_cuisine/configs/tutorial_env_config.yaml
@@ -79,7 +79,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 6
+ speed_units_per_seconds: 6
interaction_range: 1.6
restricted_view: False
view_angle: 70
diff --git a/cooperative_cuisine/counter_factory.py b/cooperative_cuisine/counter_factory.py
index 2dc90d33..5480a4c9 100644
--- a/cooperative_cuisine/counter_factory.py
+++ b/cooperative_cuisine/counter_factory.py
@@ -34,7 +34,7 @@ layout_chars:
import inspect
import sys
from random import Random
-from typing import Any, Type, TypeVar
+from typing import Any, Type, TypeVar, Tuple
import numpy as np
import numpy.typing as npt
@@ -384,3 +384,129 @@ class CounterFactory:
def get_counter_of_type(counter_type: Type[T], counters: list[Counter]) -> list[T]:
"""Filter all counters in the environment for a counter type."""
return list(filter(lambda counter: isinstance(counter, counter_type), counters))
+
+ def parse_layout_file(
+ self,
+ layout_config: str,
+ ) -> Tuple[list[Counter], list[npt.NDArray], list[npt.NDArray], int, int]:
+ """Creates layout of kitchen counters in the environment based on layout file.
+ Counters are arranged in a fixed size grid starting at [0,0]. The center of the first counter is at
+ [counter_size/2, counter_size/2], counters are directly next to each other (of no empty space is specified
+ in layout).
+
+ Args:
+ layout_config: the layout config string. Each character corresponds to a counter and each line to a row.
+ """
+
+ starting_at: float = 0.0
+ current_y: float = starting_at
+ counters: list[Counter] = []
+ designated_player_positions: list[npt.NDArray] = []
+ free_positions: list[npt.NDArray] = []
+
+ lines = layout_config.split("\n")
+
+ grid = []
+
+ max_width = 0
+
+ lines = list(filter(lambda l: l != "", lines))
+ for line in lines:
+ line = line.replace(" ", "")
+ if not line or line.startswith(";"):
+ break
+ current_x: float = starting_at
+ grid_line = []
+
+ for character in line:
+ # character = character.capitalize()
+ pos = np.array([current_x, current_y])
+
+ assert self.can_map(
+ character
+ ), f"{character=} in layout file can not be mapped"
+ if self.is_counter(character):
+ counters.append(self.get_counter_object(character, pos))
+ grid_line.append(1)
+ else:
+ grid_line.append(0)
+ match self.map_not_counter(character):
+ case "Agent":
+ designated_player_positions.append(pos)
+ case "Free":
+ free_positions.append(np.array([current_x, current_y]))
+
+ current_x += 1
+
+ if len(line) >= max_width:
+ max_width = len(line)
+
+ grid.append(grid_line)
+ current_y += 1
+
+ grid = [line + ([0] * (max_width - len(line))) for line in grid]
+
+ kitchen_width = int(max_width + starting_at)
+ kitchen_height = int(current_y)
+
+ determine_counter_orientations(
+ counters, grid, np.array([kitchen_width / 2, kitchen_height / 2])
+ )
+
+ self.post_counter_setup(counters)
+
+ return (
+ counters,
+ designated_player_positions,
+ free_positions,
+ kitchen_width,
+ kitchen_height,
+ )
+
+
+def determine_counter_orientations(counters, grid, kitchen_center):
+ grid = np.array(grid).T
+
+ grid_width = grid.shape[0]
+ grid_height = grid.shape[1]
+
+ last_counter = None
+ fst_counter_in_row = None
+ for c in counters:
+ grid_idx = np.floor(c.pos).astype(int)
+ neighbour_offsets = np.array([[0, 1], [0, -1], [1, 0], [-1, 0]], dtype=int)
+
+ neighbours_free = []
+ for offset in neighbour_offsets:
+ neighbour_pos = grid_idx + offset
+ if (
+ neighbour_pos[0] > (grid_width - 1)
+ or neighbour_pos[0] < 0
+ or neighbour_pos[1] > (grid_height - 1)
+ or neighbour_pos[1] < 0
+ ):
+ pass
+ else:
+ if grid[neighbour_pos[0]][neighbour_pos[1]] == 0:
+ neighbours_free.append(offset)
+ if len(neighbours_free) > 0:
+ vector_to_center = c.pos - kitchen_center
+ vector_to_center /= np.linalg.norm(vector_to_center)
+ n_idx = np.argmin(
+ np.linalg.norm(vector_to_center - n) for n in neighbours_free
+ )
+ nearest_vec = neighbours_free[n_idx]
+ # print(nearest_vec, type(nearest_vec))
+ c.set_orientation(nearest_vec)
+
+ elif grid_idx[0] == 0:
+ if grid_idx[1] == 0 or fst_counter_in_row is None:
+ # counter top left
+ c.set_orientation(np.array([1, 0]))
+ else:
+ c.set_orientation(fst_counter_in_row.orientation)
+ fst_counter_in_row = c
+ else:
+ c.set_orientation(last_counter.orientation)
+
+ last_counter = c
diff --git a/cooperative_cuisine/environment.py b/cooperative_cuisine/environment.py
index 8b289e15..16c568ea 100644
--- a/cooperative_cuisine/environment.py
+++ b/cooperative_cuisine/environment.py
@@ -12,19 +12,19 @@ from datetime import timedelta, datetime
from enum import Enum
from pathlib import Path
from random import Random
-from typing import Literal, TypedDict, Callable, Tuple
+from typing import Literal, TypedDict, Callable
import networkx
import numpy as np
import numpy.typing as npt
import yaml
from networkx import DiGraph
-from scipy.spatial import distance_matrix
from cooperative_cuisine import ROOT_DIR
-from cooperative_cuisine.counter_factory import CounterFactory
+from cooperative_cuisine.counter_factory import (
+ CounterFactory,
+)
from cooperative_cuisine.counters import (
- Counter,
PlateConfig,
)
from cooperative_cuisine.effect_manager import EffectManager
@@ -52,13 +52,17 @@ from cooperative_cuisine.hooks import (
ITEM_INFO_CONFIG,
POST_STEP,
)
+from cooperative_cuisine.movement import Movement
from cooperative_cuisine.orders import (
OrderManager,
OrderConfig,
)
from cooperative_cuisine.player import Player, PlayerConfig
from cooperative_cuisine.state_representation import InfoMsg
-from cooperative_cuisine.utils import create_init_env_time, get_closest
+from cooperative_cuisine.utils import (
+ create_init_env_time,
+ get_closest,
+)
log = logging.getLogger(__name__)
@@ -165,6 +169,8 @@ class Environment:
if self.as_files:
with open(env_config, "r") as file:
env_config = file.read()
+ with open(layout_config, "r") as layout_file:
+ layout_config = layout_file.read()
self.environment_config: EnvironmentConfig = yaml.load(
env_config, Loader=yaml.Loader
@@ -218,11 +224,6 @@ class Environment:
)
"""The manager for the orders and score update."""
- self.kitchen_height: int = 0
- """The height of the kitchen, is set by the `Environment.parse_layout_file` method"""
- self.kitchen_width: int = 0
- """The width of the kitchen, is set by the `Environment.parse_layout_file` method"""
-
self.counter_factory = CounterFactory(
layout_chars_config=self.environment_config["layout_chars"],
item_info=self.item_info,
@@ -253,22 +254,20 @@ class Environment:
self.counters,
self.designated_player_positions,
self.free_positions,
- ) = self.parse_layout_file()
+ self.kitchen_width,
+ self.kitchen_height,
+ ) = self.counter_factory.parse_layout_file(self.layout_config)
self.hook(LAYOUT_FILE_PARSED)
- self.world_borders = np.array(
- [[-0.5, self.kitchen_width - 0.5], [-0.5, self.kitchen_height - 0.5]],
- dtype=float,
+ self.movement = Movement(
+ counter_positions=np.array([c.pos for c in self.counters]),
+ player_config=self.environment_config["player_config"],
+ world_borders=np.array(
+ [[-0.5, self.kitchen_width - 0.5], [-0.5, self.kitchen_height - 0.5]],
+ dtype=float,
+ ),
)
- self.player_movement_speed = self.environment_config["player_config"][
- "player_speed_units_per_seconds"
- ]
- self.player_radius = self.environment_config["player_config"]["radius"]
- self.player_interaction_range = self.environment_config["player_config"][
- "interaction_range"
- ]
-
progress_counter_classes = list(
filter(
lambda cl: hasattr(cl, "progress"),
@@ -287,8 +286,6 @@ class Environment:
)
"""Counters that needs to be called in the step function via the `progress` method."""
- self.counter_positions = np.array([c.pos for c in self.counters])
-
self.order_manager.create_init_orders(self.env_time)
self.start_time = self.env_time
"""The relative env time when it started."""
@@ -316,7 +313,7 @@ class Environment:
def overwrite_counters(self, counters):
self.counters = counters
- self.counter_positions = np.array([c.pos for c in self.counters])
+ self.movement.counter_positions = np.array([c.pos for c in self.counters])
progress_counter_classes = list(
filter(
@@ -340,15 +337,6 @@ class Environment:
"""Whether the game is over or not based on the calculated `Environment.env_time_end`"""
return self.env_time >= self.env_time_end
- def set_collision_arrays(self):
- number_players = len(self.players)
- self.world_borders_lower = self.world_borders[np.newaxis, :, 0].repeat(
- number_players, axis=0
- )
- self.world_borders_upper = self.world_borders[np.newaxis, :, 1].repeat(
- number_players, axis=0
- )
-
def get_env_time(self):
"""the internal time of the environment. An environment starts always with the time from `create_init_env_time`.
@@ -442,134 +430,6 @@ class Environment:
"""TODO"""
raise NotImplementedError
- def parse_layout_file(
- self,
- ) -> Tuple[list[Counter], list[npt.NDArray], list[npt.NDArray]]:
- """Creates layout of kitchen counters in the environment based on layout file.
- Counters are arranged in a fixed size grid starting at [0,0]. The center of the first counter is at
- [counter_size/2, counter_size/2], counters are directly next to each other (of no empty space is specified
- in layout).
- """
-
- starting_at: float = 0.0
- current_y: float = starting_at
- counters: list[Counter] = []
- designated_player_positions: list[npt.NDArray] = []
- free_positions: list[npt.NDArray] = []
-
- if self.as_files:
- with open(self.layout_config, "r") as layout_file:
- self.layout_config = layout_file.read()
- lines = self.layout_config.split("\n")
-
- grid = []
-
- max_width = 0
-
- lines = list(filter(lambda l: l != "", lines))
- for line in lines:
- line = line.replace(" ", "")
- if not line or line.startswith(";"):
- break
- current_x: float = starting_at
- grid_line = []
-
- for character in line:
- # character = character.capitalize()
- pos = np.array([current_x, current_y])
-
- assert self.counter_factory.can_map(
- character
- ), f"{character=} in layout file can not be mapped"
- if self.counter_factory.is_counter(character):
- counters.append(
- self.counter_factory.get_counter_object(character, pos)
- )
- grid_line.append(1)
- else:
- grid_line.append(0)
- match self.counter_factory.map_not_counter(character):
- case "Agent":
- designated_player_positions.append(pos)
- case "Free":
- free_positions.append(np.array([current_x, current_y]))
-
- current_x += 1
-
- if len(line) >= max_width:
- max_width = len(line)
-
- grid.append(grid_line)
- current_y += 1
-
- grid = [line + ([0] * (max_width - len(line))) for line in grid]
-
- self.kitchen_width: float = max_width + starting_at
- self.kitchen_height = current_y
-
- self.determine_counter_orientations(
- counters, grid, np.array([self.kitchen_width / 2, self.kitchen_height / 2])
- )
-
- self.counter_factory.post_counter_setup(counters)
-
- return counters, designated_player_positions, free_positions
-
- def determine_counter_orientations(self, counters, grid, kitchen_center):
- grid = np.array(grid).T
-
- grid_width = grid.shape[0]
- grid_height = grid.shape[1]
-
- last_counter = None
- fst_counter_in_row = None
- for c in counters:
- grid_idx = np.floor(c.pos).astype(int)
- neighbour_offsets = np.array([[0, 1], [0, -1], [1, 0], [-1, 0]], dtype=int)
-
- neighbours_free = []
- for offset in neighbour_offsets:
- neighbour_pos = grid_idx + offset
- if (
- neighbour_pos[0] > (grid_width - 1)
- or neighbour_pos[0] < 0
- or neighbour_pos[1] > (grid_height - 1)
- or neighbour_pos[1] < 0
- ):
- pass
- else:
- if grid[neighbour_pos[0]][neighbour_pos[1]] == 0:
- neighbours_free.append(offset)
- if len(neighbours_free) > 0:
- vector_to_center = c.pos - kitchen_center
- vector_to_center /= np.linalg.norm(vector_to_center)
- n_idx = np.argmin(
- np.linalg.norm(vector_to_center - n) for n in neighbours_free
- )
- nearest_vec = neighbours_free[n_idx]
- # print(nearest_vec, type(nearest_vec))
- c.set_orientation(nearest_vec)
-
- elif grid_idx[0] == 0:
- if grid_idx[1] == 0 or fst_counter_in_row is None:
- # counter top left
- c.set_orientation(np.array([1, 0]))
- else:
- c.set_orientation(fst_counter_in_row.orientation)
- fst_counter_in_row = c
- else:
- c.set_orientation(last_counter.orientation)
-
- last_counter = c
-
- # for c in counters:
- # near_counters = [
- # other
- # for other in counters
- # if np.isclose(np.linalg.norm(c.pos - other.pos), 1)
- # ]
- # # print(c.pos, len(near_counters))
-
def perform_action(self, action: Action):
"""Performs an action of a player in the environment. Maps different types of action inputs to the
correct execution of the players.
@@ -588,7 +448,7 @@ class Environment:
self.env_time + timedelta(seconds=action.duration),
)
else:
- counter = self.get_facing_counter(player)
+ counter = get_closest(player.facing_point, self.counters)
if player.can_reach(counter):
if action.action_type == ActionType.PUT:
player.put_action(counter)
@@ -606,169 +466,6 @@ class Environment:
self.hook(POST_PERFORM_ACTION, action=action)
- def get_facing_counter(self, player: Player):
- """Determines the counter which the player is looking at.
- Adds a multiple of the player facing direction onto the player position and finds the closest
- counter for that point.
-
- Args:
- player: The player for which to find the facing counter.
-
- Returns:
-
- """
- facing_counter = get_closest(player.facing_point, self.counters)
- return facing_counter
-
- def get_counter_collisions(self, player_positions):
- counter_diff_vecs = (
- player_positions[:, np.newaxis, :]
- - self.counter_positions[np.newaxis, :, :]
- )
- counter_distances = np.max((np.abs(counter_diff_vecs)), axis=2)
- closest_counter_positions = self.counter_positions[
- np.argmin(counter_distances, axis=1)
- ]
- nearest_counter_to_player = player_positions - closest_counter_positions
- relevant_axes = np.abs(nearest_counter_to_player).argmax(axis=1)
-
- distances = np.linalg.norm(
- np.max(
- [
- np.abs(counter_diff_vecs) - 0.5,
- np.zeros(counter_diff_vecs.shape),
- ],
- axis=0,
- ),
- axis=2,
- )
-
- collided = np.any(distances < self.player_radius, axis=1)
-
- return collided, relevant_axes, nearest_counter_to_player
-
- def get_player_push(self, player_positions):
- distances_players_after_scipy = distance_matrix(
- player_positions, player_positions
- )
-
- player_diff_vecs = -(
- player_positions[:, np.newaxis, :] - player_positions[np.newaxis, :, :]
- )
- collisions = distances_players_after_scipy < (2 * self.player_radius)
- eye_idxs = np.eye(len(player_positions), len(player_positions), dtype=bool)
- collisions[eye_idxs] = False
- player_diff_vecs[collisions == False] = 0
- push_vectors = np.sum(player_diff_vecs, axis=0)
- collisions = np.any(collisions, axis=1)
- return collisions, push_vectors
-
- def perform_movement(self, duration: timedelta):
- """Moves a player in the direction specified in the action.action. If the player collides with a
- counter or other player through this movement, then they are not moved.
- (The extended code with the two ifs is for sliding movement at the counters, which feels a bit smoother.
- This happens, when the player moves diagonally against the counters or world boundary.
- This just checks if the single axis party of the movement could move the player and does so at a lower rate.)
-
- The movement action is a unit 2d vector.
-
- Detects collisions with other players and pushes them out of the way.
-
- Args:
- duration: The duration for how long the movement to perform.
- """
- d_time = duration.total_seconds()
-
- player_positions = np.array([p.pos for p in self.players.values()], dtype=float)
- player_movement_vectors = np.array(
- [
- p.current_movement if self.env_time <= p.movement_until else [0, 0]
- for p in self.players.values()
- ],
- dtype=float,
- )
-
- targeted_positions = player_positions + (
- player_movement_vectors * (self.player_movement_speed * d_time)
- )
-
- # Collisions player between player
- force_factor = 1.2
- _, push_vectors = self.get_player_push(targeted_positions)
- updated_movement = (force_factor * push_vectors) + player_movement_vectors
- new_targeted_positions = player_positions + (
- updated_movement * (self.player_movement_speed * d_time)
- )
- # same again to prevent squeezing into other players
- _, push_vectors2 = self.get_player_push(new_targeted_positions)
- updated_movement = (force_factor * push_vectors2) + updated_movement
- new_targeted_positions = player_positions + (
- updated_movement * (self.player_movement_speed * d_time)
- )
-
- # Check collisions with counters
- (
- collided,
- relevant_axes,
- nearest_counter_to_player,
- ) = self.get_counter_collisions(new_targeted_positions)
-
- # Check if sliding against counters is possible
- for idx, player in enumerate(player_positions):
- axis = relevant_axes[idx]
- if collided[idx]:
- # collide with counter left or top
- if nearest_counter_to_player[idx][axis] > 0:
- updated_movement[idx, axis] = np.max(
- [updated_movement[idx, axis], 0]
- )
- # collide with counter right or bottom
- if nearest_counter_to_player[idx][axis] < 0:
- updated_movement[idx, axis] = np.min(
- [updated_movement[idx, axis], 0]
- )
- new_positions = player_positions + (
- updated_movement * (self.player_movement_speed * d_time)
- )
-
- # Check collisions with counters again, now absolute with no sliding possible
- (
- collided,
- relevant_axes,
- nearest_counter_to_player,
- ) = self.get_counter_collisions(new_positions)
- new_positions[collided] = player_positions[collided]
-
- # Check player collisions a final time
- # collided, _ = self.get_player_push(new_positions)
- # if np.any(collided):
- # print(".", end="")
-
- # Collisions player world borders
- new_positions = np.clip(
- new_positions,
- self.world_borders_lower + self.player_radius,
- self.world_borders_upper - self.player_radius,
- )
-
- for idx, p in enumerate(self.players.values()):
- if not (new_positions[idx] == player_positions[idx]).all():
- p.pos = new_positions[idx]
-
- p.turn(player_movement_vectors[idx])
-
- facing_distances = np.linalg.norm(
- p.facing_point - self.counter_positions, axis=1
- )
- closest_counter = self.counters[facing_distances.argmin()]
- p.current_nearest_counter = (
- closest_counter
- if facing_distances.min() <= self.player_interaction_range
- else None
- )
- if p.last_interacted_counter != p.current_nearest_counter:
- p.perform_interact_stop()
-
def add_player(self, player_name: str, pos: npt.NDArray = None):
"""Add a player to the environment.
@@ -805,27 +502,9 @@ class Environment:
log.debug("No free positions left in kitchens")
player.update_facing_point()
- self.set_collision_arrays()
+ self.movement.set_collision_arrays(len(self.players))
self.hook(PLAYER_ADDED, player_name=player_name, pos=pos)
- def detect_collision_world_bounds(self, player: Player):
- """Checks for detections of the player and the world bounds.
-
- Args:
- player: The player which to not let escape the world.
-
- Returns: True if the player touches the world bounds, False if not.
- """
- collisions_lower = any(
- (player.pos - (player.radius))
- < [self.world_borders_x[0], self.world_borders_y[0]]
- )
- collisions_upper = any(
- (player.pos + (player.radius))
- > [self.world_borders_x[1], self.world_borders_y[1]]
- )
- return collisions_lower or collisions_upper
-
def step(self, passed_time: timedelta):
"""Performs a step of the environment. Affects time based events such as cooking or cutting things, orders
and time limits.
@@ -839,7 +518,9 @@ class Environment:
for player in self.players.values():
player.progress(passed_time, self.env_time)
- self.perform_movement(passed_time)
+ self.movement.perform_movement(
+ passed_time, self.env_time, self.players, self.counters
+ )
for counter in self.progressing_counters:
counter.progress(passed_time=passed_time, now=self.env_time)
@@ -872,7 +553,6 @@ class Environment:
Args:
player_id: The player for which to get the state.
- play_beep: Signal the GUI to play a beep when all connected players are ready to play the game.
Returns: The state of the game formatted as a json-string
diff --git a/cooperative_cuisine/movement.py b/cooperative_cuisine/movement.py
new file mode 100644
index 00000000..3a082ed5
--- /dev/null
+++ b/cooperative_cuisine/movement.py
@@ -0,0 +1,181 @@
+from datetime import timedelta, datetime
+
+import numpy as np
+from scipy.spatial import distance_matrix
+
+from cooperative_cuisine.counters import Counter
+from cooperative_cuisine.player import Player
+
+
+class Movement:
+ world_borders_lower = None
+ world_borders_upper = None
+
+ def __init__(self, counter_positions, player_config, world_borders):
+ self.counter_positions = counter_positions
+ self.player_radius = player_config["radius"]
+ self.player_interaction_range = player_config["interaction_range"]
+ self.player_movement_speed = player_config["speed_units_per_seconds"]
+ self.world_borders = world_borders
+ self.set_collision_arrays(1)
+
+ def set_collision_arrays(self, number_players):
+ self.world_borders_lower = self.world_borders[np.newaxis, :, 0].repeat(
+ number_players, axis=0
+ )
+ self.world_borders_upper = self.world_borders[np.newaxis, :, 1].repeat(
+ number_players, axis=0
+ )
+
+ def get_counter_collisions(self, player_positions):
+ counter_diff_vecs = (
+ player_positions[:, np.newaxis, :]
+ - self.counter_positions[np.newaxis, :, :]
+ )
+ counter_distances = np.max((np.abs(counter_diff_vecs)), axis=2)
+ closest_counter_positions = self.counter_positions[
+ np.argmin(counter_distances, axis=1)
+ ]
+ nearest_counter_to_player = player_positions - closest_counter_positions
+ relevant_axes = np.abs(nearest_counter_to_player).argmax(axis=1)
+
+ distances = np.linalg.norm(
+ np.max(
+ [
+ np.abs(counter_diff_vecs) - 0.5,
+ np.zeros(counter_diff_vecs.shape),
+ ],
+ axis=0,
+ ),
+ axis=2,
+ )
+
+ collided = np.any(distances < self.player_radius, axis=1)
+
+ return collided, relevant_axes, nearest_counter_to_player
+
+ def get_player_push(self, player_positions):
+ distances_players_after_scipy = distance_matrix(
+ player_positions, player_positions
+ )
+
+ player_diff_vecs = -(
+ player_positions[:, np.newaxis, :] - player_positions[np.newaxis, :, :]
+ )
+ collisions = distances_players_after_scipy < (2 * self.player_radius)
+ eye_idxs = np.eye(len(player_positions), len(player_positions), dtype=bool)
+ collisions[eye_idxs] = False
+ player_diff_vecs[collisions == False] = 0
+ push_vectors = np.sum(player_diff_vecs, axis=0)
+ collisions = np.any(collisions, axis=1)
+ return collisions, push_vectors
+
+ def perform_movement(
+ self,
+ duration: timedelta,
+ env_time: datetime,
+ players: dict[str, Player],
+ counters: list[Counter],
+ ):
+ """Moves a player in the direction specified in the action.action. If the player collides with a
+ counter or other player through this movement, then they are not moved.
+ (The extended code with the two ifs is for sliding movement at the counters, which feels a bit smoother.
+ This happens, when the player moves diagonally against the counters or world boundary.
+ This just checks if the single axis party of the movement could move the player and does so at a lower rate.)
+
+ The movement action is a unit 2d vector.
+
+ Detects collisions with other players and pushes them out of the way.
+
+ Args:
+ duration: The duration for how long the movement to perform.
+ env_time: The current time of the environment.
+ players: The players in the environment.
+ counters: The counters in the environment.
+ """
+ d_time = duration.total_seconds()
+
+ player_positions = np.array([p.pos for p in players.values()], dtype=float)
+ player_movement_vectors = np.array(
+ [
+ p.current_movement if env_time <= p.movement_until else [0, 0]
+ for p in players.values()
+ ],
+ dtype=float,
+ )
+
+ targeted_positions = player_positions + (
+ player_movement_vectors * (self.player_movement_speed * d_time)
+ )
+
+ # Collisions player between player
+ force_factor = 1.2
+ _, push_vectors = self.get_player_push(targeted_positions)
+ updated_movement = (force_factor * push_vectors) + player_movement_vectors
+ new_targeted_positions = player_positions + (
+ updated_movement * (self.player_movement_speed * d_time)
+ )
+ # same again to prevent squeezing into other players
+ _, push_vectors2 = self.get_player_push(new_targeted_positions)
+ updated_movement = (force_factor * push_vectors2) + updated_movement
+ new_targeted_positions = player_positions + (
+ updated_movement * (self.player_movement_speed * d_time)
+ )
+
+ # Check collisions with counters
+ (
+ collided,
+ relevant_axes,
+ nearest_counter_to_player,
+ ) = self.get_counter_collisions(new_targeted_positions)
+
+ # Check if sliding against counters is possible
+ for idx, player in enumerate(player_positions):
+ axis = relevant_axes[idx]
+ if collided[idx]:
+ # collide with counter left or top
+ if nearest_counter_to_player[idx][axis] > 0:
+ updated_movement[idx, axis] = np.max(
+ [updated_movement[idx, axis], 0]
+ )
+ # collide with counter right or bottom
+ if nearest_counter_to_player[idx][axis] < 0:
+ updated_movement[idx, axis] = np.min(
+ [updated_movement[idx, axis], 0]
+ )
+ new_positions = player_positions + (
+ updated_movement * (self.player_movement_speed * d_time)
+ )
+
+ # Check collisions with counters again, now absolute with no sliding possible
+ (
+ collided,
+ relevant_axes,
+ nearest_counter_to_player,
+ ) = self.get_counter_collisions(new_positions)
+ new_positions[collided] = player_positions[collided]
+
+ # Collisions player world borders
+ new_positions = np.clip(
+ new_positions,
+ self.world_borders_lower + self.player_radius,
+ self.world_borders_upper - self.player_radius,
+ )
+
+ for idx, p in enumerate(players.values()):
+ if not (new_positions[idx] == player_positions[idx]).all():
+ p.pos = new_positions[idx]
+
+ p.turn(player_movement_vectors[idx])
+
+ facing_distances = np.linalg.norm(
+ p.facing_point - self.counter_positions, axis=1
+ )
+ closest_counter = counters[facing_distances.argmin()]
+ p.current_nearest_counter = (
+ closest_counter
+ if facing_distances.min() <= self.player_interaction_range
+ else None
+ )
+ if p.last_interacted_counter != p.current_nearest_counter:
+ p.perform_interact_stop()
diff --git a/cooperative_cuisine/player.py b/cooperative_cuisine/player.py
index b85c7666..82ed1ede 100644
--- a/cooperative_cuisine/player.py
+++ b/cooperative_cuisine/player.py
@@ -29,7 +29,7 @@ class PlayerConfig:
radius: float = 0.4
"""The size of the player. The size of a counter is 1"""
- player_speed_units_per_seconds: float | int = 8
+ speed_units_per_seconds: float | int = 8
"""The move distance/speed of the player per action call."""
interaction_range: float = 1.6
"""How far player can interact with counters."""
diff --git a/cooperative_cuisine/reinforcement_learning/environment_config_rl.yaml b/cooperative_cuisine/reinforcement_learning/environment_config_rl.yaml
index 5e75d9bd..6b02ef60 100644
--- a/cooperative_cuisine/reinforcement_learning/environment_config_rl.yaml
+++ b/cooperative_cuisine/reinforcement_learning/environment_config_rl.yaml
@@ -82,7 +82,7 @@ orders:
player_config:
radius: 0.4
- player_speed_units_per_seconds: 1
+ speed_units_per_seconds: 1
interaction_range: 1.6
restricted_view: False
view_angle: 95
diff --git a/cooperative_cuisine/reinforcement_learning/gym_env.py b/cooperative_cuisine/reinforcement_learning/gym_env.py
index 7bacc68c..d5f79214 100644
--- a/cooperative_cuisine/reinforcement_learning/gym_env.py
+++ b/cooperative_cuisine/reinforcement_learning/gym_env.py
@@ -105,14 +105,18 @@ with open(ROOT_DIR / "pygame_2d_vis" / "visualization.yaml", "r") as file:
def shuffle_counters(env):
sample_counter = []
+ other_counters = []
for counter in env.counters:
if counter.__class__ != Counter:
sample_counter.append(counter)
+ else:
+ other_counters.append()
new_counter_pos = [c.pos for c in sample_counter]
random.shuffle(new_counter_pos)
for counter, new_pos in zip(sample_counter, new_counter_pos):
counter.pos = new_pos
- env.counter_positions = np.array([c.pos for c in env.counters])
+ sample_counter.extend(other_counters)
+ env.overwrite_counters(sample_counter)
class EnvGymWrapper(Env):
diff --git a/tests/test_start.py b/tests/test_start.py
index e69b1ac9..a17bb737 100644
--- a/tests/test_start.py
+++ b/tests/test_start.py
@@ -91,7 +91,7 @@ def test_movement(env_config, layout_empty_config, item_info):
player_name = "1"
start_pos = np.array([3, 4])
env.add_player(player_name, start_pos)
- env.player_movement_speed = 1
+ env.movement.player_movement_speed = 1
move_direction = np.array([1, 0])
move_action = Action(player_name, ActionType.MOVEMENT, move_direction, duration=0.1)
do_moves_number = 3
@@ -100,7 +100,7 @@ def test_movement(env_config, layout_empty_config, item_info):
env.step(timedelta(seconds=0.1))
expected = start_pos + do_moves_number * (
- move_direction * env.player_movement_speed * move_action.duration
+ move_direction * env.movement.player_movement_speed * move_action.duration
)
assert np.isclose(
np.linalg.norm(expected - env.players[player_name].pos), 0
@@ -112,7 +112,7 @@ def test_player_movement_speed(env_config, layout_empty_config, item_info):
player_name = "1"
start_pos = np.array([3, 4])
env.add_player(player_name, start_pos)
- env.player_movement_speed = 2
+ env.movement.player_movement_speed = 2
move_direction = np.array([1, 0])
move_action = Action(player_name, ActionType.MOVEMENT, move_direction, duration=0.1)
do_moves_number = 3
@@ -121,7 +121,7 @@ def test_player_movement_speed(env_config, layout_empty_config, item_info):
env.step(timedelta(seconds=0.1))
expected = start_pos + do_moves_number * (
- move_direction * env.player_movement_speed * move_action.duration
+ move_direction * env.movement.player_movement_speed * move_action.duration
)
assert np.isclose(
@@ -140,7 +140,7 @@ def test_player_reach(env_config, layout_empty_config, item_info):
counter = Counter(pos=counter_pos, hook=Hooks(env))
env.overwrite_counters([counter])
env.add_player("1", np.array([2, 4]))
- env.player_movement_speed = 1
+ env.movement.player_movement_speed = 1
player = env.players["1"]
assert not player.can_reach(counter), "Player is too far away."
@@ -162,7 +162,7 @@ def test_pickup(env_config, layout_config, item_info):
env.add_player("1", np.array([2, 3]))
player = env.players["1"]
- env.player_movement_speed = 1
+ env.movement.player_movement_speed = 1
move_down = Action("1", ActionType.MOVEMENT, np.array([0, -1]), duration=1)
move_up = Action("1", ActionType.MOVEMENT, np.array([0, 1]), duration=1)
@@ -224,7 +224,7 @@ def test_processing(env_config, layout_config, item_info):
tomato = Item(name="Tomato", item_info=None)
env.add_player("1", np.array([2, 3]))
player = env.players["1"]
- env.player_movement_speed = 1
+ env.movement.player_movement_speed = 1
player.holding = tomato
move = Action("1", ActionType.MOVEMENT, np.array([0, -1]), duration=1)
--
GitLab