social_dilemmas/envs/map_env.py
def step(self, actions):
        """Takes in a dict of actions and converts them to a map update

        Parameters
        ----------
        actions: dict {agent-id: int}
            dict of actions, keyed by agent-id that are passed to the agent. The agent
            interprets the int and converts it to a command

        Returns
        -------
        observations: dict of arrays representing agent observations
        rewards: dict of rewards for each agent
        dones: dict indicating whether each agent is done
        info: dict to pass extra info to gym
        """

        self.beam_pos = []
        agent_actions = {}
        for agent_id, action in actions.items():
            agent_action = self.agents[agent_id].action_map(action)
            agent_actions[agent_id] = agent_action

        # Remove agents from color map
        for agent in self.agents.values():
            row, col = agent.pos[0], agent.pos[1]
            self.single_update_world_color_map(row, col, self.world_map[row, col])

        self.update_moves(agent_actions)

        for agent in self.agents.values():
            pos = agent.pos
            new_char = agent.consume(self.world_map[pos[0], pos[1]])
            self.single_update_map(pos[0], pos[1], new_char)

        # execute custom moves like firing
        self.update_custom_moves(agent_actions)

        # execute spawning events
        self.custom_map_update()

        map_with_agents = self.get_map_with_agents()
        # Add agents to color map
        for agent in self.agents.values():
            row, col = agent.pos[0], agent.pos[1]
            # Firing beams have priority over agents and should cover them
            if self.world_map[row, col] not in [b"F", b"C"]:
                self.single_update_world_color_map(row, col, agent.get_char_id())

        observations = {}
        rewards = {}
        dones = {}
        info = {}
        for agent in self.agents.values():
            agent.full_map = map_with_agents
            rgb_arr = self.color_view(agent)
            # concatenate on the prev_actions to the observations
            if self.return_agent_actions:
                prev_actions = np.array(
                    [actions[key] for key in sorted(actions.keys()) if key != agent.agent_id]
                ).astype(np.uint8)
                visible_agents = self.find_visible_agents(agent.agent_id)
                observations[agent.agent_id] = {
                    "curr_obs": rgb_arr,
                    "other_agent_actions": prev_actions,
                    "visible_agents": visible_agents,
                    "prev_visible_agents": agent.prev_visible_agents,
                }
                agent.prev_visible_agents = visible_agents
            else:
                observations[agent.agent_id] = {"curr_obs": rgb_arr}
            rewards[agent.agent_id] = agent.compute_reward()
            dones[agent.agent_id] = agent.get_done()

        if self.use_collective_reward:
            collective_reward = sum(rewards.values())
            for agent in rewards.keys():
                rewards[agent] = collective_reward

        dones["__all__"] = np.any(list(dones.values()))
        return observations, rewards, dones, info
Takes in a dict of actions and converts them to a map update

        Parameters
        ----------
        actions: dict {agent-id: int}
            dict of actions, keyed by agent-id that are passed to the agent. The agent
            interprets the int and converts it to a command

        Returns
        -------
        observations: dict of arrays representing agent observations
        rewards: dict of rewards for each agent
        dones: dict indicating whether each agent is done
        info: dict to pass extra info to gym
        
Takes in a dict of actions and converts them to a map update
Parameters

dict {agent-id: int}
dict of actions, keyed by agent-id that are passed to the agent. The agent
interprets the int and converts it to a command

Returns

dict of arrays representing agent observations
rewards: dict of rewards for each agent
dones: dict indicating whether each agent is done
info: dict to pass extra info to gym
Takes in a dict of actions and converts them to a map update Parameters dict { agent - id : int } dict of actions keyed by agent - id that are passed to the agent . The agent interprets the int and converts it to a command Returns dict of arrays representing agent observations rewards : dict of rewards for each agent dones : dict indicating whether each agent is done info : dict to pass extra info to gym
def step(self, actions):
        self.beam_pos = []
        agent_actions = {}
        for agent_id, action in actions.items():
            agent_action = self.agents[agent_id].action_map(action)
            agent_actions[agent_id] = agent_action
        for agent in self.agents.values():
            row, col = agent.pos[0], agent.pos[1]
            self.single_update_world_color_map(row, col, self.world_map[row, col])
        self.update_moves(agent_actions)
        for agent in self.agents.values():
            pos = agent.pos
            new_char = agent.consume(self.world_map[pos[0], pos[1]])
            self.single_update_map(pos[0], pos[1], new_char)
        self.update_custom_moves(agent_actions)
        self.custom_map_update()
        map_with_agents = self.get_map_with_agents()
        for agent in self.agents.values():
            row, col = agent.pos[0], agent.pos[1]
            if self.world_map[row, col] not in [b"F", b"C"]:
                self.single_update_world_color_map(row, col, agent.get_char_id())
        observations = {}
        rewards = {}
        dones = {}
        info = {}
        for agent in self.agents.values():
            agent.full_map = map_with_agents
            rgb_arr = self.color_view(agent)
            if self.return_agent_actions:
                prev_actions = np.array(
                    [actions[key] for key in sorted(actions.keys()) if key != agent.agent_id]
                ).astype(np.uint8)
                visible_agents = self.find_visible_agents(agent.agent_id)
                observations[agent.agent_id] = {
                    "curr_obs": rgb_arr,
                    "other_agent_actions": prev_actions,
                    "visible_agents": visible_agents,
                    "prev_visible_agents": agent.prev_visible_agents,
                }
                agent.prev_visible_agents = visible_agents
            else:
                observations[agent.agent_id] = {"curr_obs": rgb_arr}
            rewards[agent.agent_id] = agent.compute_reward()
            dones[agent.agent_id] = agent.get_done()
        if self.use_collective_reward:
            collective_reward = sum(rewards.values())
            for agent in rewards.keys():
                rewards[agent] = collective_reward
        dones["__all__"] = np.any(list(dones.values()))
        return observations, rewards, dones, info
def step ( self , actions ) : self . beam_pos = [ ] agent_actions = { } for agent_id , action in actions . items ( ) : agent_action = self . agents [ agent_id ] . action_map ( action ) agent_actions [ agent_id ] = agent_action for agent in self . agents . values ( ) : row , col = agent . pos [ 0 ] , agent . pos [ 1 ] self . single_update_world_color_map ( row , col , self . world_map [ row , col ] ) self . update_moves ( agent_actions ) for agent in self . agents . values ( ) : pos = agent . pos new_char = agent . consume ( self . world_map [ pos [ 0 ] , pos [ 1 ] ] ) self . single_update_map ( pos [ 0 ] , pos [ 1 ] , new_char ) self . update_custom_moves ( agent_actions ) self . custom_map_update ( ) map_with_agents = self . get_map_with_agents ( ) for agent in self . agents . values ( ) : row , col = agent . pos [ 0 ] , agent . pos [ 1 ] if self . world_map [ row , col ] not in [ b"F" , b"C" ] : self . single_update_world_color_map ( row , col , agent . get_char_id ( ) ) observations = { } rewards = { } dones = { } info = { } for agent in self . agents . values ( ) : agent . full_map = map_with_agents rgb_arr = self . color_view ( agent ) if self . return_agent_actions : prev_actions = np . array ( [ actions [ key ] for key in sorted ( actions . keys ( ) ) if key != agent . agent_id ] ) . astype ( np . uint8 ) visible_agents = self . find_visible_agents ( agent . agent_id ) observations [ agent . agent_id ] = { "curr_obs" : rgb_arr , "other_agent_actions" : prev_actions , "visible_agents" : visible_agents , "prev_visible_agents" : agent . prev_visible_agents , } agent . prev_visible_agents = visible_agents else : observations [ agent . agent_id ] = { "curr_obs" : rgb_arr } rewards [ agent . agent_id ] = agent . compute_reward ( ) dones [ agent . agent_id ] = agent . get_done ( ) if self . use_collective_reward : collective_reward = sum ( rewards . values ( ) ) for agent in rewards . keys ( ) : rewards [ agent ] = collective_reward dones [ "__all__" ] = np . any ( list ( dones . values ( ) ) ) return observations , rewards , dones , info
Takes in a dict of actions and converts them to a map update
Parameters
Takes in a dict of actions and converts them to a map update Parameters
"""Takes in a dict of actions and converts them to a map update

        Parameters
        ----------
        actions: dict {agent-id: int}
            dict of actions, keyed by agent-id that are passed to the agent. The agent
            interprets the int and converts it to a command

        Returns
        -------
        observations: dict of arrays representing agent observations
        rewards: dict of rewards for each agent
        dones: dict indicating whether each agent is done
        info: dict to pass extra info to gym
        """ # Remove agents from color map # execute custom moves like firing # execute spawning events # Add agents to color map # Firing beams have priority over agents and should cover them # concatenate on the prev_actions to the observations
self
actions