Skip to content
Snippets Groups Projects

Resolve "simple pathfinding"

Merged Resolve "simple pathfinding"
120-simple-pathfinding into dev
Merged Fabian Heinrich requested to merge 120-simple-pathfinding into dev
Compare
and
Show latest version
9 files
+ 424
− 123
Compare changes
  • Side-by-side
  • Inline
Files
9
cooperative_cuisine/configs/agents/random_agent.py
+ 62
− 37
@@ -22,6 +22,10 @@ from cooperative_cuisine.utils import custom_asdict_factory
TIME_TO_STOP_ACTION = 3.0
ADD_RANDOM_MOVEMENTS = False
DIAGONAL_MOVEMENTS = True
AVOID_OTHER_PLAYERS = True
def get_free_neighbours(
state: dict, counter_pos: list[float] | tuple[float, float] | npt.NDArray
@@ -30,7 +34,7 @@ def get_free_neighbours(
free_space = np.ones((width, height), dtype=bool)
for counter in state["counters"]:
grid_idx = np.array(counter["pos"]).astype(int)
free_space[*grid_idx] = False
free_space[grid_idx[0], grid_idx[1]] = False
i, j = np.array(counter_pos).astype(int)
free = []
@@ -49,7 +53,7 @@ async def agent():
args = parser.parse_args()
async with connect(args.uri) as websocket:
async with (connect(args.uri) as websocket):
await websocket.send(
json.dumps({"type": "ready", "player_hash": args.player_hash})
)
@@ -58,6 +62,7 @@ async def agent():
ended = False
counters = None
all_counters = None
movement_graph = None
@@ -87,12 +92,15 @@ async def agent():
continue
if movement_graph is None:
movement_graph = create_movement_graph(state, diagonal=True)
movement_graph = create_movement_graph(
state, diagonal=DIAGONAL_MOVEMENTS
)
if counters is None:
counters = defaultdict(list)
for counter in state["counters"]:
counters[counter["type"]].append(counter)
all_counters = state["counters"]
for player in state["players"]:
if player["id"] == args.player_id:
@@ -165,16 +173,17 @@ async def agent():
paths = []
for free in target_free_spaces:
try:
modified_graph = restrict_movement_graph(
graph=movement_graph,
player_positions=[
p["pos"]
for p in state["players"]
if p["id"] != args.player_id
],
)
path = networkx.astar_path(
modified_graph,
restrict_movement_graph(
graph=movement_graph,
player_positions=[
p["pos"]
for p in state["players"]
if p["id"] != args.player_id
],
)
if AVOID_OTHER_PLAYERS
else movement_graph,
source,
free,
heuristic=astar_heuristic,
@@ -195,29 +204,34 @@ async def agent():
movement = node_diff / node_dist
else:
movement = target_diff / target_dist
do_movement = True
else:
movement = np.array([0, 0])
task_type = None
task_args = None
# no paths available
print("NO PATHS")
# task_type = None
# task_args = None
do_movement = False
if target_dist > 1.2:
if target_dist > 1.2 and do_movement:
if target_dist != 0:
# random_small_rotation_angle = (
# np.random.random() * np.pi * 0.1
# )
# rotation_matrix = np.array(
# [
# [
# np.cos(random_small_rotation_angle),
# -np.sin(random_small_rotation_angle),
# ],
# [
# np.sin(random_small_rotation_angle),
# np.cos(random_small_rotation_angle),
# ],
# ]
# )
# movement = rotation_matrix @ movement
if ADD_RANDOM_MOVEMENTS:
random_small_rotation_angle = (
np.random.random() * np.pi * 0.1
)
rotation_matrix = np.array(
[
[
np.cos(random_small_rotation_angle),
-np.sin(random_small_rotation_angle),
],
[
np.sin(random_small_rotation_angle),
np.cos(random_small_rotation_angle),
],
]
)
movement = rotation_matrix @ movement
await websocket.send(
json.dumps(
@@ -238,6 +252,8 @@ async def agent():
)
await websocket.recv()
else:
# Target reached here.
print("TARGET REACHED")
task_type = None
task_args = None
case "INTERACT":
@@ -294,14 +310,23 @@ async def agent():
...
if not task_type:
# task_type = random.choice(["GOTO"])
task_type = random.choice(["GOTO", "PUT"])
# task_type = random.choice(["GOTO", "PUT", "INTERACT"])
task_type = random.choice(["GOTO"])
threshold = datetime.now() + timedelta(seconds=TIME_TO_STOP_ACTION)
if task_type == "GOTO":
counter_type = random.choice(list(counters.keys()))
task_args = random.choice(counters[counter_type])["pos"]
print(args.player_hash, args.player_id, task_type, counter_type)
# counter_type = random.choice(list(counters.keys()))
# task_args = random.choice(counters[counter_type])["pos"]
random_counter = random.choice(all_counters)
counter_type = random_counter["type"]
task_args = random_counter["pos"]
print(
args.player_hash,
args.player_id,
task_type,
counter_type,
task_args,
)
else:
print(args.player_hash, args.player_id, task_type)
task_args = None