steffen/marl-factory-grid
1
0
Fork 0
mirror of https://github.com/illiumst/marl-factory-grid.git synced 2025-06-18 18:52:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

added simple route caching and fixed move 4 point graph in tspbaseagent

Browse Source
This commit is contained in:
Chanumask
2024-04-17 15:28:10 +02:00
parent fe5a97a413
commit 54d4e1ecb5
2 changed files with 57 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
marl_factory_grid/algorithms/static/TSP_base_agent.py
View File

@ -3,6 +3,8 @@ from random import choice
import numpy as np
from networkx.algorithms.approximation import traveling_salesman as tsp
import time
import copy
from marl_factory_grid.algorithms.static.utils import points_to_graph
from marl_factory_grid.modules.doors import constants as do
@ -31,8 +33,9 @@ class TSPBaseAgent(ABC):
self.local_optimization = True
self._env = state
self.state = self._env.state[c.AGENT][agent_i]
self._position_graph = points_to_graph(self._env.state.entities.floorlist)
self._position_graph = self.generate_pos_graph()
self._static_route = None
self.cached_route = None
@abstractmethod
def predict(self, *_, **__) -> int:
@ -72,21 +75,35 @@ class TSPBaseAgent(ABC):
:return: TSP route
:rtype: List[int]
"""
positions = [x for x in self._env.state[target_identifier].positions if x != c.VALUE_NO_POS]
if self.local_optimization:
nodes = \
[self.state.pos] + \
[x for x in positions if max(abs(np.subtract(x, self.state.pos))) < 3]
try:
while len(nodes) < 7:
nodes += [next(x for x in positions if x not in nodes)]
except StopIteration:
nodes = [self.state.pos] + positions
start_time = time.time()
if self.cached_route is not None:
print(f" Used cached route: {self.cached_route}")
return copy.deepcopy(self.cached_route)
else:
nodes = [self.state.pos] + positions
route = tsp.traveling_salesman_problem(self._position_graph,
nodes=nodes, cycle=True, method=tsp.greedy_tsp)
positions = [x for x in self._env.state[target_identifier].positions if x != c.VALUE_NO_POS]
if self.local_optimization:
nodes = \
[self.state.pos] + \
[x for x in positions if max(abs(np.subtract(x, self.state.pos))) < 3]
try:
while len(nodes) < 7:
nodes += [next(x for x in positions if x not in nodes)]
except StopIteration:
nodes = [self.state.pos] + positions
else:
nodes = [self.state.pos] + positions
route = tsp.traveling_salesman_problem(self._position_graph,
nodes=nodes, cycle=True, method=tsp.greedy_tsp)
self.cached_route = copy.deepcopy(route)
print(f"Cached route: {self.cached_route}")
end_time = time.time()
duration = end_time - start_time
print("TSP calculation took {:.2f} seconds to execute".format(duration))
return route
def _door_is_close(self, state):
@ -144,13 +161,15 @@ class TSPBaseAgent(ABC):
while next_pos == self.state.pos:
if self._static_route:
next_pos = self._static_route.pop(0)
diff = np.subtract(next_pos, self.state.pos)
# Retrieve action based on the pos dif (like in: What do I have to do to get there?)
try:
allowed_directions = [action.name for action in self.state.actions if
action.name in ['north', 'east', 'south', 'west', 'north_east', 'south_east',
'south_west', 'north_west']]
action = next(action for action, pos_diff in MOVEMAP.items() if np.all(diff == pos_diff) and action in allowed_directions)
action = next(action for action, pos_diff in MOVEMAP.items() if
np.all(diff == pos_diff) and action in allowed_directions)
except StopIteration:
print(f"No valid action found for pos diff: {diff}. Using fallback action.")
action = choice(self.state.actions).name
@ -158,3 +177,24 @@ class TSPBaseAgent(ABC):
action = choice(self.state.actions).name
# noinspection PyUnboundLocalVariable
return action
def generate_pos_graph(self):
"""
Generates a point graph based on the agents' allowed movement directions to be used in tsp route calculation.
:return: A graph with nodes that are conneceted as specified by the movement actions.
:rtype: nx.Graph
"""
action_names = {action.name for action in self.state.actions}
if {'north_east', 'south_east', 'south_west', 'north_west'}.issubset(action_names):
# print("All diagonal actions are present")
return points_to_graph(self._env.state.entities.floorlist)
elif {'north', 'east', 'south', 'west'}.issubset(action_names):
# print("All cardinal directions are present")
return points_to_graph(self._env.state.entities.floorlist, allow_euclidean_connections=False)
else:
print("Some actions are missing")
return points_to_graph(self._env.state.entities.floorlist)