arena/arena.py
def run_one_duel(self,
                     mode: str,
                     list_of_agents: List[Agent],
                     starting_agent_id: int = 0,
                     render_game: bool=False,
                     mpi_communicator = None,
                     initial_observation = None)-> GameStatisticsDuels:
        """Runs one game between two agents.
        :param:
        mode: mode of the game (stochastic or deterministic)
        list_of_agents: List of agents to play, they will play in the order given by the list
        starting_agent_id:  Id of the agent who starts the game.
        show_game: If True, GUI will appear showing the game. """

        #prepare the game
        self.env.reset()
        if initial_observation is not None:
            self.env.load_observation(initial_observation)
        if initial_observation is None:
            self.env.set_active_player(starting_agent_id)
            #set players names:
            self.env.set_players_names([agent.name for agent in list_of_agents])
        is_done = False
        #set the initial agent id
        active_agent_id = starting_agent_id
        #set the initial observation
        observation = self.env.show_observation(mode)
        number_of_actions = 0
        results_dict = {}
        #Id if the player who first reaches number of points to win
        first_winner_id = None
        checked_all_players_after_first_winner = False
        previous_actions = [None]
        action = None

        if mpi_communicator is None:
            #local_main_process = True
            mpi_communicator = total_communicator
            local_main_process = total_main_process

        if mpi_communicator is not None:
            local_main_process = mpi_communicator.Get_rank() == 0
            for agent in list_of_agents:
                if agent.multi_process:
                    agent.set_communicator(mpi_communicator)

        if render_game and local_main_process:
            self.env.render()
            time.sleep(GAME_INITIAL_DELAY)

        one_game_obs_list = []

        while  number_of_actions < MAX_NUMBER_OF_MOVES and not is_done:
            #if local_main_process:
                #print('Action number = {}'.format(number_of_actions))
            active_agent = list_of_agents[active_agent_id]
            if active_agent.multi_process == True:
                action = list_of_agents[active_agent_id].choose_action(observation, previous_actions)
            if active_agent.multi_process == False and local_main_process:
                action = list_of_agents[active_agent_id].choose_action(observation, previous_actions)
            previous_actions = [action]

            if local_main_process:
                observation, reward, is_done, info = self.env.step(mode, action)

################## COLECTING STATES IS HERE: ##########################################
                if self.collect_states_mode and  number_of_actions > self.n_min_actions:
                    p = np.random.uniform(0,1)
                    if p <= self.dump_probability:
                        one_game_obs_list.append(observation.recreate_state())

                winner_id = info['winner_id']
            if render_game:
                self.env.render()

            if is_done and local_main_process:
                results_dict[list_of_agents[active_agent_id].my_name_with_id()] = \
                    OneAgentStatistics(reward, self.env.points_of_player_by_id(active_agent_id),
                                       int(winner_id == active_agent_id))
                previous_agent_id = (active_agent_id-1)%len(list_of_agents)
                results_dict[list_of_agents[previous_agent_id].my_name_with_id()] = \
                    OneAgentStatistics(reward*(-1), self.env.points_of_player_by_id(previous_agent_id),
                                   int(previous_agent_id == winner_id))
                if first_winner_id is None:
                    first_winner_id = active_agent_id
                checked_all_players_after_first_winner = active_agent_id == (first_winner_id-1)%len(list_of_agents)
                #print('ALL MOVED = {}'.format(checked_all_players_after_first_winner))

            active_agent_id = (active_agent_id + 1) % len(list_of_agents)
            number_of_actions += 1
            if mpi_communicator is not None:
                is_done = mpi_communicator.bcast(is_done, root = 0)

        if local_main_process:
            self.env.reset()
            one_game_statistics = GameStatisticsDuels(list_of_agents)
            one_game_statistics.register_from_dict(results_dict)
            if winner_id is not None and self.collect_states_mode:
                values = [1 if st.active_player_id == winner_id else -1 for st in one_game_obs_list]
                self.collected_states_df[len(self.collected_states_df)] = {'states' : one_game_obs_list,
                                                                                'values' : values}

        for agent in list_of_agents:
            agent.finish_game()

        return one_game_statistics if local_main_process else None
Runs one game between two agents.
        :param:
        mode: mode of the game (stochastic or deterministic)
        list_of_agents: List of agents to play, they will play in the order given by the list
        starting_agent_id:  Id of the agent who starts the game.
        show_game: If True, GUI will appear showing the game. 
Runs one game between two agents.
Runs one game between two agents .
def run_one_duel(self,
                     mode: str,
                     list_of_agents: List[Agent],
                     starting_agent_id: int = 0,
                     render_game: bool=False,
                     mpi_communicator = None,
                     initial_observation = None)-> GameStatisticsDuels:
        self.env.reset()
        if initial_observation is not None:
            self.env.load_observation(initial_observation)
        if initial_observation is None:
            self.env.set_active_player(starting_agent_id)
            self.env.set_players_names([agent.name for agent in list_of_agents])
        is_done = False
        active_agent_id = starting_agent_id
        observation = self.env.show_observation(mode)
        number_of_actions = 0
        results_dict = {}
        first_winner_id = None
        checked_all_players_after_first_winner = False
        previous_actions = [None]
        action = None
        if mpi_communicator is None:
            mpi_communicator = total_communicator
            local_main_process = total_main_process
        if mpi_communicator is not None:
            local_main_process = mpi_communicator.Get_rank() == 0
            for agent in list_of_agents:
                if agent.multi_process:
                    agent.set_communicator(mpi_communicator)
        if render_game and local_main_process:
            self.env.render()
            time.sleep(GAME_INITIAL_DELAY)
        one_game_obs_list = []
        while  number_of_actions < MAX_NUMBER_OF_MOVES and not is_done:
            active_agent = list_of_agents[active_agent_id]
            if active_agent.multi_process == True:
                action = list_of_agents[active_agent_id].choose_action(observation, previous_actions)
            if active_agent.multi_process == False and local_main_process:
                action = list_of_agents[active_agent_id].choose_action(observation, previous_actions)
            previous_actions = [action]
            if local_main_process:
                observation, reward, is_done, info = self.env.step(mode, action)
                if self.collect_states_mode and  number_of_actions > self.n_min_actions:
                    p = np.random.uniform(0,1)
                    if p <= self.dump_probability:
                        one_game_obs_list.append(observation.recreate_state())
                winner_id = info['winner_id']
            if render_game:
                self.env.render()
            if is_done and local_main_process:
                results_dict[list_of_agents[active_agent_id].my_name_with_id()] = \
                    OneAgentStatistics(reward, self.env.points_of_player_by_id(active_agent_id),
                                       int(winner_id == active_agent_id))
                previous_agent_id = (active_agent_id-1)%len(list_of_agents)
                results_dict[list_of_agents[previous_agent_id].my_name_with_id()] = \
                    OneAgentStatistics(reward*(-1), self.env.points_of_player_by_id(previous_agent_id),
                                   int(previous_agent_id == winner_id))
                if first_winner_id is None:
                    first_winner_id = active_agent_id
                checked_all_players_after_first_winner = active_agent_id == (first_winner_id-1)%len(list_of_agents)
            active_agent_id = (active_agent_id + 1) % len(list_of_agents)
            number_of_actions += 1
            if mpi_communicator is not None:
                is_done = mpi_communicator.bcast(is_done, root = 0)
        if local_main_process:
            self.env.reset()
            one_game_statistics = GameStatisticsDuels(list_of_agents)
            one_game_statistics.register_from_dict(results_dict)
            if winner_id is not None and self.collect_states_mode:
                values = [1 if st.active_player_id == winner_id else -1 for st in one_game_obs_list]
                self.collected_states_df[len(self.collected_states_df)] = {'states' : one_game_obs_list,
                                                                                'values' : values}
        for agent in list_of_agents:
            agent.finish_game()
        return one_game_statistics if local_main_process else None
def run_one_duel ( self , mode : str , list_of_agents : List [ Agent ] , starting_agent_id : int = 0 , render_game : bool = False , mpi_communicator = None , initial_observation = None ) -> GameStatisticsDuels : self . env . reset ( ) if initial_observation is not None : self . env . load_observation ( initial_observation ) if initial_observation is None : self . env . set_active_player ( starting_agent_id ) self . env . set_players_names ( [ agent . name for agent in list_of_agents ] ) is_done = False active_agent_id = starting_agent_id observation = self . env . show_observation ( mode ) number_of_actions = 0 results_dict = { } first_winner_id = None checked_all_players_after_first_winner = False previous_actions = [ None ] action = None if mpi_communicator is None : mpi_communicator = total_communicator local_main_process = total_main_process if mpi_communicator is not None : local_main_process = mpi_communicator . Get_rank ( ) == 0 for agent in list_of_agents : if agent . multi_process : agent . set_communicator ( mpi_communicator ) if render_game and local_main_process : self . env . render ( ) time . sleep ( GAME_INITIAL_DELAY ) one_game_obs_list = [ ] while number_of_actions < MAX_NUMBER_OF_MOVES and not is_done : active_agent = list_of_agents [ active_agent_id ] if active_agent . multi_process == True : action = list_of_agents [ active_agent_id ] . choose_action ( observation , previous_actions ) if active_agent . multi_process == False and local_main_process : action = list_of_agents [ active_agent_id ] . choose_action ( observation , previous_actions ) previous_actions = [ action ] if local_main_process : observation , reward , is_done , info = self . env . step ( mode , action ) if self . collect_states_mode and number_of_actions > self . n_min_actions : p = np . random . uniform ( 0 , 1 ) if p <= self . dump_probability : one_game_obs_list . append ( observation . recreate_state ( ) ) winner_id = info [ 'winner_id' ] if render_game : self . env . render ( ) if is_done and local_main_process : results_dict [ list_of_agents [ active_agent_id ] . my_name_with_id ( ) ] = OneAgentStatistics ( reward , self . env . points_of_player_by_id ( active_agent_id ) , int ( winner_id == active_agent_id ) ) previous_agent_id = ( active_agent_id - 1 ) % len ( list_of_agents ) results_dict [ list_of_agents [ previous_agent_id ] . my_name_with_id ( ) ] = OneAgentStatistics ( reward * ( - 1 ) , self . env . points_of_player_by_id ( previous_agent_id ) , int ( previous_agent_id == winner_id ) ) if first_winner_id is None : first_winner_id = active_agent_id checked_all_players_after_first_winner = active_agent_id == ( first_winner_id - 1 ) % len ( list_of_agents ) active_agent_id = ( active_agent_id + 1 ) % len ( list_of_agents ) number_of_actions += 1 if mpi_communicator is not None : is_done = mpi_communicator . bcast ( is_done , root = 0 ) if local_main_process : self . env . reset ( ) one_game_statistics = GameStatisticsDuels ( list_of_agents ) one_game_statistics . register_from_dict ( results_dict ) if winner_id is not None and self . collect_states_mode : values = [ 1 if st . active_player_id == winner_id else - 1 for st in one_game_obs_list ] self . collected_states_df [ len ( self . collected_states_df ) ] = { 'states' : one_game_obs_list , 'values' : values } for agent in list_of_agents : agent . finish_game ( ) return one_game_statistics if local_main_process else None
Runs one game between two agents.
Runs one game between two agents .
"""Runs one game between two agents.
        :param:
        mode: mode of the game (stochastic or deterministic)
        list_of_agents: List of agents to play, they will play in the order given by the list
        starting_agent_id:  Id of the agent who starts the game.
        show_game: If True, GUI will appear showing the game. """ #prepare the game #set players names: #set the initial agent id #set the initial observation #Id if the player who first reaches number of points to win #local_main_process = True #if local_main_process: #print('Action number = {}'.format(number_of_actions)) ################## COLECTING STATES IS HERE: ########################################## #print('ALL MOVED = {}'.format(checked_all_players_after_first_winner))
self
mode
str
list_of_agents
List[Agent]
starting_agent_id
int
render_game
bool
mpi_communicator
initial_observation