social_dilemmas/envs/map_env.py
def update_moves(self, agent_actions):
        """Converts agent action tuples into a new map and new agent positions.
        Also resolves conflicts over multiple agents wanting a cell.

        This method works by finding all conflicts over a cell and randomly assigning them
       to one of the agents that desires the slot. It then sets all of the other agents
       that wanted the cell to have a move of staying. For moves that do not directly
       conflict with another agent for a cell, but may not be temporarily resolvable
       due to an agent currently being in the desired cell, we continually loop through
       the actions until all moves have been satisfied or deemed impossible.
       For example, agent 1 may want to move from [1,2] to [2,2] but agent 2 is in [2,2].
       Agent 2, however, is moving into [3,2]. Agent-1's action is first in the order so at the
       first pass it is skipped but agent-2 moves to [3,2]. In the second pass, agent-1 will
       then be able to move into [2,2].

        Parameters
        ----------
        agent_actions: dict
            dict with agent_id as key and action as value
        """

        reserved_slots = []
        for agent_id, action in agent_actions.items():
            agent = self.agents[agent_id]
            selected_action = self.all_actions[action]
            # TODO(ev) these two parts of the actions
            if "MOVE" in action or "STAY" in action:
                # rotate the selected action appropriately
                rot_action = self.rotate_action(selected_action, agent.get_orientation())
                new_pos = agent.pos + rot_action
                # allow the agents to confirm what position they can move to
                new_pos = agent.return_valid_pos(new_pos)
                reserved_slots.append((*new_pos, b"P", agent_id))
            elif "TURN" in action:
                new_rot = self.update_rotation(action, agent.get_orientation())
                agent.update_agent_rot(new_rot)

        # now do the conflict resolution part of the process

        # helpful for finding the agent in the conflicting slot
        agent_by_pos = {tuple(agent.pos): agent.agent_id for agent in self.agents.values()}

        # agent moves keyed by ids
        agent_moves = {}

        # lists of moves and their corresponding agents
        move_slots = []
        agent_to_slot = []

        for slot in reserved_slots:
            row, col = slot[0], slot[1]
            if slot[2] == b"P":
                agent_id = slot[3]
                agent_moves[agent_id] = [row, col]
                move_slots.append([row, col])
                agent_to_slot.append(agent_id)

        # cut short the computation if there are no moves
        if len(agent_to_slot) > 0:

            # first we will resolve all slots over which multiple agents
            # want the slot

            # shuffle so that a random agent has slot priority
            shuffle_list = list(zip(agent_to_slot, move_slots))
            np.random.shuffle(shuffle_list)
            agent_to_slot, move_slots = zip(*shuffle_list)
            unique_move, indices, return_count = np.unique(
                move_slots, return_index=True, return_counts=True, axis=0
            )
            search_list = np.array(move_slots)

            # first go through and remove moves that can't possible happen. Three types
            # 1. Trying to move into an agent that has been issued a stay command
            # 2. Trying to move into the spot of an agent that doesn't have a move
            # 3. Two agents trying to walk through one another

            # Resolve all conflicts over a space
            if np.any(return_count > 1):
                for move, index, count in zip(unique_move, indices, return_count):
                    if count > 1:
                        # check that the cell you are fighting over doesn't currently
                        # contain an agent that isn't going to move for one of the agents
                        # If it does, all the agents commands should become STAY
                        # since no moving will be possible
                        conflict_indices = np.where((search_list == move).all(axis=1))[0]
                        all_agents_id = [agent_to_slot[i] for i in conflict_indices]
                        # all other agents now stay in place so update their moves
                        # to reflect this
                        conflict_cell_free = True
                        for agent_id in all_agents_id:
                            moves_copy = agent_moves.copy()
                            # TODO(ev) code duplication, simplify
                            if move.tolist() in self.agent_pos:
                                # find the agent that is currently at that spot and make sure
                                # that the move is possible. If it won't be, remove it.
                                conflicting_agent_id = agent_by_pos[tuple(move)]
                                curr_pos = self.agents[agent_id].pos.tolist()
                                curr_conflict_pos = self.agents[conflicting_agent_id].pos.tolist()
                                conflict_move = agent_moves.get(
                                    conflicting_agent_id, curr_conflict_pos
                                )
                                # Condition (1):
                                # a STAY command has been issued
                                if agent_id == conflicting_agent_id:
                                    conflict_cell_free = False
                                # Condition (2)
                                # its command is to stay
                                # or you are trying to move into an agent that hasn't
                                # received a command
                                elif (
                                    conflicting_agent_id not in moves_copy.keys()
                                    or curr_conflict_pos == conflict_move
                                ):
                                    conflict_cell_free = False

                                # Condition (3)
                                # It is trying to move into you and you are moving into it
                                elif conflicting_agent_id in moves_copy.keys():
                                    if (
                                        agent_moves[conflicting_agent_id] == curr_pos
                                        and move.tolist()
                                        == self.agents[conflicting_agent_id].pos.tolist()
                                    ):
                                        conflict_cell_free = False

                        # if the conflict cell is open, let one of the conflicting agents
                        # move into it
                        if conflict_cell_free:
                            self.agents[agent_to_slot[index]].update_agent_pos(move)
                            agent_by_pos = {
                                tuple(agent.pos): agent.agent_id for agent in self.agents.values()
                            }
                        # ------------------------------------
                        # remove all the other moves that would have conflicted
                        remove_indices = np.where((search_list == move).all(axis=1))[0]
                        all_agents_id = [agent_to_slot[i] for i in remove_indices]
                        # all other agents now stay in place so update their moves
                        # to stay in place
                        for agent_id in all_agents_id:
                            agent_moves[agent_id] = self.agents[agent_id].pos.tolist()

            # make the remaining un-conflicted moves
            while len(agent_moves.items()) > 0:
                agent_by_pos = {tuple(agent.pos): agent.agent_id for agent in self.agents.values()}
                num_moves = len(agent_moves.items())
                moves_copy = agent_moves.copy()
                del_keys = []
                for agent_id, move in moves_copy.items():
                    if agent_id in del_keys:
                        continue
                    if move in self.agent_pos:
                        # find the agent that is currently at that spot and make sure
                        # that the move is possible. If it won't be, remove it.
                        conflicting_agent_id = agent_by_pos[tuple(move)]
                        curr_pos = self.agents[agent_id].pos.tolist()
                        curr_conflict_pos = self.agents[conflicting_agent_id].pos.tolist()
                        conflict_move = agent_moves.get(conflicting_agent_id, curr_conflict_pos)
                        # Condition (1):
                        # a STAY command has been issued
                        if agent_id == conflicting_agent_id:
                            del agent_moves[agent_id]
                            del_keys.append(agent_id)
                        # Condition (2)
                        # its command is to stay
                        # or you are trying to move into an agent that hasn't received a command
                        elif (
                            conflicting_agent_id not in moves_copy.keys()
                            or curr_conflict_pos == conflict_move
                        ):
                            del agent_moves[agent_id]
                            del_keys.append(agent_id)
                        # Condition (3)
                        # It is trying to move into you and you are moving into it
                        elif conflicting_agent_id in moves_copy.keys():
                            if (
                                agent_moves[conflicting_agent_id] == curr_pos
                                and move == self.agents[conflicting_agent_id].pos.tolist()
                            ):
                                del agent_moves[conflicting_agent_id]
                                del agent_moves[agent_id]
                                del_keys.append(agent_id)
                                del_keys.append(conflicting_agent_id)
                    # this move is unconflicted so go ahead and move
                    else:
                        self.agents[agent_id].update_agent_pos(move)
                        del agent_moves[agent_id]
                        del_keys.append(agent_id)

                # no agent is able to move freely, so just move them all
                # no updates to hidden cells are needed since all the
                # same cells will be covered
                if len(agent_moves) == num_moves:
                    for agent_id, move in agent_moves.items():
                        self.agents[agent_id].update_agent_pos(move)
                    break
Converts agent action tuples into a new map and new agent positions.
        Also resolves conflicts over multiple agents wanting a cell.

        This method works by finding all conflicts over a cell and randomly assigning them
       to one of the agents that desires the slot. It then sets all of the other agents
       that wanted the cell to have a move of staying. For moves that do not directly
       conflict with another agent for a cell, but may not be temporarily resolvable
       due to an agent currently being in the desired cell, we continually loop through
       the actions until all moves have been satisfied or deemed impossible.
       For example, agent 1 may want to move from [1,2] to [2,2] but agent 2 is in [2,2].
       Agent 2, however, is moving into [3,2]. Agent-1's action is first in the order so at the
       first pass it is skipped but agent-2 moves to [3,2]. In the second pass, agent-1 will
       then be able to move into [2,2].

        Parameters
        ----------
        agent_actions: dict
            dict with agent_id as key and action as value
        
Converts agent action tuples into a new map and new agent positions.
Also resolves conflicts over multiple agents wanting a cell.

This method works by finding all conflicts over a cell and randomly assigning them
to one of the agents that desires the slot. It then sets all of the other agents
that wanted the cell to have a move of staying. For moves that do not directly
conflict with another agent for a cell, but may not be temporarily resolvable
due to an agent currently being in the desired cell, we continually loop through
the actions until all moves have been satisfied or deemed impossible.
For example, agent 1 may want to move from [1,2] to [2,2] but agent 2 is in [2,2].
Agent 2, however, is moving into [3,2]. Agent-1's action is first in the order so at the
first pass it is skipped but agent-2 moves to [3,2]. In the second pass, agent-1 will
then be able to move into [2,2].

Parameters

dict
dict with agent_id as key and action as value
Converts agent action tuples into a new map and new agent positions . Also resolves conflicts over multiple agents wanting a cell . This method works by finding all conflicts over a cell and randomly assigning them to one of the agents that desires the slot . It then sets all of the other agents that wanted the cell to have a move of staying . For moves that do not directly conflict with another agent for a cell but may not be temporarily resolvable due to an agent currently being in the desired cell we continually loop through the actions until all moves have been satisfied or deemed impossible . For example agent 1 may want to move from [ 1 2 ] to [ 2 2 ] but agent 2 is in [ 2 2 ] . Agent 2 however is moving into [ 3 2 ] . Agent - 1 ' s action is first in the order so at the first pass it is skipped but agent - 2 moves to [ 3 2 ] . In the second pass agent - 1 will then be able to move into [ 2 2 ] . Parameters dict dict with agent_id as key and action as value
def update_moves(self, agent_actions):
        reserved_slots = []
        for agent_id, action in agent_actions.items():
            agent = self.agents[agent_id]
            selected_action = self.all_actions[action]
            if "MOVE" in action or "STAY" in action:
                rot_action = self.rotate_action(selected_action, agent.get_orientation())
                new_pos = agent.pos + rot_action
                new_pos = agent.return_valid_pos(new_pos)
                reserved_slots.append((*new_pos, b"P", agent_id))
            elif "TURN" in action:
                new_rot = self.update_rotation(action, agent.get_orientation())
                agent.update_agent_rot(new_rot)
        agent_by_pos = {tuple(agent.pos): agent.agent_id for agent in self.agents.values()}
        agent_moves = {}
        move_slots = []
        agent_to_slot = []
        for slot in reserved_slots:
            row, col = slot[0], slot[1]
            if slot[2] == b"P":
                agent_id = slot[3]
                agent_moves[agent_id] = [row, col]
                move_slots.append([row, col])
                agent_to_slot.append(agent_id)
        if len(agent_to_slot) > 0:
            shuffle_list = list(zip(agent_to_slot, move_slots))
            np.random.shuffle(shuffle_list)
            agent_to_slot, move_slots = zip(*shuffle_list)
            unique_move, indices, return_count = np.unique(
                move_slots, return_index=True, return_counts=True, axis=0
            )
            search_list = np.array(move_slots)
            if np.any(return_count > 1):
                for move, index, count in zip(unique_move, indices, return_count):
                    if count > 1:
                        conflict_indices = np.where((search_list == move).all(axis=1))[0]
                        all_agents_id = [agent_to_slot[i] for i in conflict_indices]
                        conflict_cell_free = True
                        for agent_id in all_agents_id:
                            moves_copy = agent_moves.copy()
                            if move.tolist() in self.agent_pos:
                                conflicting_agent_id = agent_by_pos[tuple(move)]
                                curr_pos = self.agents[agent_id].pos.tolist()
                                curr_conflict_pos = self.agents[conflicting_agent_id].pos.tolist()
                                conflict_move = agent_moves.get(
                                    conflicting_agent_id, curr_conflict_pos
                                )
                                if agent_id == conflicting_agent_id:
                                    conflict_cell_free = False
                                elif (
                                    conflicting_agent_id not in moves_copy.keys()
                                    or curr_conflict_pos == conflict_move
                                ):
                                    conflict_cell_free = False
                                elif conflicting_agent_id in moves_copy.keys():
                                    if (
                                        agent_moves[conflicting_agent_id] == curr_pos
                                        and move.tolist()
                                        == self.agents[conflicting_agent_id].pos.tolist()
                                    ):
                                        conflict_cell_free = False
                        if conflict_cell_free:
                            self.agents[agent_to_slot[index]].update_agent_pos(move)
                            agent_by_pos = {
                                tuple(agent.pos): agent.agent_id for agent in self.agents.values()
                            }
                        remove_indices = np.where((search_list == move).all(axis=1))[0]
                        all_agents_id = [agent_to_slot[i] for i in remove_indices]
                        for agent_id in all_agents_id:
                            agent_moves[agent_id] = self.agents[agent_id].pos.tolist()
            while len(agent_moves.items()) > 0:
                agent_by_pos = {tuple(agent.pos): agent.agent_id for agent in self.agents.values()}
                num_moves = len(agent_moves.items())
                moves_copy = agent_moves.copy()
                del_keys = []
                for agent_id, move in moves_copy.items():
                    if agent_id in del_keys:
                        continue
                    if move in self.agent_pos:
                        conflicting_agent_id = agent_by_pos[tuple(move)]
                        curr_pos = self.agents[agent_id].pos.tolist()
                        curr_conflict_pos = self.agents[conflicting_agent_id].pos.tolist()
                        conflict_move = agent_moves.get(conflicting_agent_id, curr_conflict_pos)
                        if agent_id == conflicting_agent_id:
                            del agent_moves[agent_id]
                            del_keys.append(agent_id)
                         received a command
                        elif (
                            conflicting_agent_id not in moves_copy.keys()
                            or curr_conflict_pos == conflict_move
                        ):
                            del agent_moves[agent_id]
                            del_keys.append(agent_id)
                        elif conflicting_agent_id in moves_copy.keys():
                            if (
                                agent_moves[conflicting_agent_id] == curr_pos
                                and move == self.agents[conflicting_agent_id].pos.tolist()
                            ):
                                del agent_moves[conflicting_agent_id]
                                del agent_moves[agent_id]
                                del_keys.append(agent_id)
                                del_keys.append(conflicting_agent_id)
                    else:
                        self.agents[agent_id].update_agent_pos(move)
                        del agent_moves[agent_id]
                        del_keys.append(agent_id)
                if len(agent_moves) == num_moves:
                    for agent_id, move in agent_moves.items():
                        self.agents[agent_id].update_agent_pos(move)
                    break
def update_moves ( self , agent_actions ) : reserved_slots = [ ] for agent_id , action in agent_actions . items ( ) : agent = self . agents [ agent_id ] selected_action = self . all_actions [ action ] if "MOVE" in action or "STAY" in action : rot_action = self . rotate_action ( selected_action , agent . get_orientation ( ) ) new_pos = agent . pos + rot_action new_pos = agent . return_valid_pos ( new_pos ) reserved_slots . append ( ( * new_pos , b"P" , agent_id ) ) elif "TURN" in action : new_rot = self . update_rotation ( action , agent . get_orientation ( ) ) agent . update_agent_rot ( new_rot ) agent_by_pos = { tuple ( agent . pos ) : agent . agent_id for agent in self . agents . values ( ) } agent_moves = { } move_slots = [ ] agent_to_slot = [ ] for slot in reserved_slots : row , col = slot [ 0 ] , slot [ 1 ] if slot [ 2 ] == b"P" : agent_id = slot [ 3 ] agent_moves [ agent_id ] = [ row , col ] move_slots . append ( [ row , col ] ) agent_to_slot . append ( agent_id ) if len ( agent_to_slot ) > 0 : shuffle_list = list ( zip ( agent_to_slot , move_slots ) ) np . random . shuffle ( shuffle_list ) agent_to_slot , move_slots = zip ( * shuffle_list ) unique_move , indices , return_count = np . unique ( move_slots , return_index = True , return_counts = True , axis = 0 ) search_list = np . array ( move_slots ) if np . any ( return_count > 1 ) : for move , index , count in zip ( unique_move , indices , return_count ) : if count > 1 : conflict_indices = np . where ( ( search_list == move ) . all ( axis = 1 ) ) [ 0 ] all_agents_id = [ agent_to_slot [ i ] for i in conflict_indices ] conflict_cell_free = True for agent_id in all_agents_id : moves_copy = agent_moves . copy ( ) if move . tolist ( ) in self . agent_pos : conflicting_agent_id = agent_by_pos [ tuple ( move ) ] curr_pos = self . agents [ agent_id ] . pos . tolist ( ) curr_conflict_pos = self . agents [ conflicting_agent_id ] . pos . tolist ( ) conflict_move = agent_moves . get ( conflicting_agent_id , curr_conflict_pos ) if agent_id == conflicting_agent_id : conflict_cell_free = False elif ( conflicting_agent_id not in moves_copy . keys ( ) or curr_conflict_pos == conflict_move ) : conflict_cell_free = False elif conflicting_agent_id in moves_copy . keys ( ) : if ( agent_moves [ conflicting_agent_id ] == curr_pos and move . tolist ( ) == self . agents [ conflicting_agent_id ] . pos . tolist ( ) ) : conflict_cell_free = False if conflict_cell_free : self . agents [ agent_to_slot [ index ] ] . update_agent_pos ( move ) agent_by_pos = { tuple ( agent . pos ) : agent . agent_id for agent in self . agents . values ( ) } remove_indices = np . where ( ( search_list == move ) . all ( axis = 1 ) ) [ 0 ] all_agents_id = [ agent_to_slot [ i ] for i in remove_indices ] for agent_id in all_agents_id : agent_moves [ agent_id ] = self . agents [ agent_id ] . pos . tolist ( ) while len ( agent_moves . items ( ) ) > 0 : agent_by_pos = { tuple ( agent . pos ) : agent . agent_id for agent in self . agents . values ( ) } num_moves = len ( agent_moves . items ( ) ) moves_copy = agent_moves . copy ( ) del_keys = [ ] for agent_id , move in moves_copy . items ( ) : if agent_id in del_keys : continue if move in self . agent_pos : conflicting_agent_id = agent_by_pos [ tuple ( move ) ] curr_pos = self . agents [ agent_id ] . pos . tolist ( ) curr_conflict_pos = self . agents [ conflicting_agent_id ] . pos . tolist ( ) conflict_move = agent_moves . get ( conflicting_agent_id , curr_conflict_pos ) if agent_id == conflicting_agent_id : del agent_moves [ agent_id ] del_keys . append ( agent_id ) elif ( conflicting_agent_id not in moves_copy . keys ( ) or curr_conflict_pos == conflict_move ) : del agent_moves [ agent_id ] del_keys . append ( agent_id ) elif conflicting_agent_id in moves_copy . keys ( ) : if ( agent_moves [ conflicting_agent_id ] == curr_pos and move == self . agents [ conflicting_agent_id ] . pos . tolist ( ) ) : del agent_moves [ conflicting_agent_id ] del agent_moves [ agent_id ] del_keys . append ( agent_id ) del_keys . append ( conflicting_agent_id ) else : self . agents [ agent_id ] . update_agent_pos ( move ) del agent_moves [ agent_id ] del_keys . append ( agent_id ) if len ( agent_moves ) == num_moves : for agent_id , move in agent_moves . items ( ) : self . agents [ agent_id ] . update_agent_pos ( move ) break
Converts agent action tuples into a new map and new agent positions.
Converts agent action tuples into a new map and new agent positions .
"""Converts agent action tuples into a new map and new agent positions.
        Also resolves conflicts over multiple agents wanting a cell.

        This method works by finding all conflicts over a cell and randomly assigning them
       to one of the agents that desires the slot. It then sets all of the other agents
       that wanted the cell to have a move of staying. For moves that do not directly
       conflict with another agent for a cell, but may not be temporarily resolvable
       due to an agent currently being in the desired cell, we continually loop through
       the actions until all moves have been satisfied or deemed impossible.
       For example, agent 1 may want to move from [1,2] to [2,2] but agent 2 is in [2,2].
       Agent 2, however, is moving into [3,2]. Agent-1's action is first in the order so at the
       first pass it is skipped but agent-2 moves to [3,2]. In the second pass, agent-1 will
       then be able to move into [2,2].

        Parameters
        ----------
        agent_actions: dict
            dict with agent_id as key and action as value
        """ # TODO(ev) these two parts of the actions # rotate the selected action appropriately # allow the agents to confirm what position they can move to # now do the conflict resolution part of the process # helpful for finding the agent in the conflicting slot # agent moves keyed by ids # lists of moves and their corresponding agents # cut short the computation if there are no moves # first we will resolve all slots over which multiple agents # want the slot # shuffle so that a random agent has slot priority # first go through and remove moves that can't possible happen. Three types # 1. Trying to move into an agent that has been issued a stay command # 2. Trying to move into the spot of an agent that doesn't have a move # 3. Two agents trying to walk through one another # Resolve all conflicts over a space # check that the cell you are fighting over doesn't currently # contain an agent that isn't going to move for one of the agents # If it does, all the agents commands should become STAY # since no moving will be possible # all other agents now stay in place so update their moves # to reflect this # TODO(ev) code duplication, simplify # find the agent that is currently at that spot and make sure # that the move is possible. If it won't be, remove it. # Condition (1): # a STAY command has been issued # Condition (2) # its command is to stay # or you are trying to move into an agent that hasn't # received a command # Condition (3) # It is trying to move into you and you are moving into it # if the conflict cell is open, let one of the conflicting agents # move into it # ------------------------------------ # remove all the other moves that would have conflicted # all other agents now stay in place so update their moves # to stay in place # make the remaining un-conflicted moves # find the agent that is currently at that spot and make sure # that the move is possible. If it won't be, remove it. # Condition (1): # a STAY command has been issued # Condition (2) # its command is to stay # or you are trying to move into an agent that hasn't received a command # Condition (3) # It is trying to move into you and you are moving into it # this move is unconflicted so go ahead and move # no agent is able to move freely, so just move them all # no updates to hidden cells are needed since all the # same cells will be covered
self
agent_actions