Environments

Base Environment

envs.freeciv_base_env.FreecivBaseEnv

Bases: Env, EzPickle

Basic CivRealm environment

Source code in src/civrealm/envs/freeciv_base_env.py

class FreecivBaseEnv(gymnasium.Env, utils.EzPickle):
    """ Basic CivRealm environment """
    metadata = {'render_modes': ['human']}

    def __init__(
            self, username: str = fc_args['username'],
            client_port=None,
            is_minitask=False):
        self.username = username
        self.is_minitask = is_minitask
        # Record whether the env is currently running.
        self.running = False

        # Create dummy controller to retrieve action_space and observation_space.
        self.civ_controller = CivController(username=self.username,
                                            visualize=fc_args['debug.take_screenshot'] or fc_args['debug.get_webpage_image'],
                                            is_minitask=self.is_minitask)
        self._action_space = self.civ_controller.action_space
        self._observation_space = self.civ_controller.observation_space
        self.set_up_recording()
        utils.EzPickle.__init__(self, self.username,
                                client_port, self.is_minitask)

    def set_up_screenshots(self):
        self._screenshot_step_count = 0
        curr_date_time = str(datetime.date.today()) + "_" + \
            str(datetime.datetime.now().time().strftime('%H-%M-%S.%f'))
        self.screenshot_dir = os.path.join(
            os.path.dirname(fc_logger.handlers[0].baseFilename),
            'screenshots', self.username + "_" + str(self.get_port()) + "_" + curr_date_time)
        os.makedirs(self.screenshot_dir, exist_ok=True)

    def set_up_recording(self):
        # For recording purposes. self.record_step_count only increases when recording is enabled.
        self._record_step_count = 0
        self.recording_dir = os.path.join(
            os.path.dirname(fc_logger.handlers[0].baseFilename),
            'recordings', fc_args['username'])
        os.makedirs(self.recording_dir, exist_ok=True)

    @property
    def action_space(self):
        self._action_space = self.civ_controller.action_space
        return self._action_space

    @property
    def observation_space(self):
        self._observation_space = self.civ_controller.observation_space
        return self._observation_space

    def _take_screenshot(self):
        if not fc_args['debug.take_screenshot']:
            return
        turn = self.civ_controller.get_turn()
        screenshot_filename = os.path.join(
            self.screenshot_dir, f'turn_{turn:03d}_step_{self._screenshot_step_count:04d}.png')
        self.civ_controller.take_screenshot(screenshot_filename)
        self._screenshot_step_count += 1

    def _record_to_file(self, name, content, default_json_encoder=None):
        if fc_args['debug.record_action_and_observation'] is False:
            return

        turn = self.civ_controller.get_turn()
        self._recording_base_filename = os.path.join(
            self.recording_dir, f'turn_{turn:03d}_step_{self._record_step_count:04d}')
        with open(f'{self._recording_base_filename}_{name}.json', 'w') as f:
            json.dump(content, f, skipkeys=True, sort_keys=True,
                      default=default_json_encoder)

    def _record_observation(self, observation):
        self._record_to_file('state', observation, lambda x: x.get_bitvector_in_ascii()
                             if isinstance(x, BitVector) else x.tolist())

    def _record_action(self, available_actions, action):
        self._record_to_file('available_action',
                             available_actions, lambda x: x.encode_to_json())
        if action:
            self._record_to_file('chosen_action', action,
                                 lambda x: x.encode_to_json())
        self._record_step_count += 1

    def _get_info_and_observation(self):
        info, observation = self.civ_controller.get_info_and_observation()
        self._record_observation(observation)
        return info, observation

    def _get_reward(self):
        return self.civ_controller.get_reward()

    def _get_terminated(self):
        return self.civ_controller.game_has_terminated()

    def _get_truncated(self):
        return self.civ_controller.game_has_truncated()

    def step(self, action):
        self.civ_controller.perform_action(action)
        try:
            info, observation = self._get_info_and_observation()
            reward = self._get_reward()
            terminated = self._get_terminated()
            truncated = self._get_truncated()

            available_actions = info['available_actions']
            self._record_action(available_actions, action)
            self._take_screenshot()
        except (ServerTimeoutException, BeginTurnTimeoutException) as server_problem:
            fc_logger.error(repr(server_problem))
            reward = 0
            info = {}
            observation = {}
            terminated = False
            truncated = True

        except Exception as e:
            # print(traceback.format_exc())
            # fc_logger.error(repr(e))
            # reward = 0
            # info = None
            # observation = None
            # terminated = False
            # truncated = True
            raise e

        return observation, reward, terminated, truncated, info

    def get_port(self):
        return self.civ_controller.client_port

    def get_username(self):
        return self.civ_controller.clstate.username

    def get_playerid(self):
        return self.civ_controller.player_ctrl.my_player_id

    def reset(self, seed=None, options=None, client_port=None, **kwargs):
        # If call reset when the env is still running, we close it first.
        if self.running:
            print('Close running environment before reset.')
            self.close()
        if client_port is None:
            client_port = Ports.get()
        print(f'Reset with port: {client_port}')
        fc_logger.debug(f'Reset with port: {client_port}')
        # self.civ_controller = CivController(username=self.username, client_port=client_port, visualize=fc_args['debug.take_screenshot'], is_minitask=self.is_minitask)
        # self._action_space = self.civ_controller.action_space
        # self._observation_space = self.civ_controller.observation_space
        self.civ_controller.reset_civ_controller(client_port)

        if fc_args['debug.take_screenshot']:
            self.set_up_screenshots()

        if seed is not None:
            fc_args['debug.randomly_generate_seeds'] = False
            fc_args['debug.mapseed'] = seed
            fc_args['debug.agentseed'] = seed

        # fc_logger.debug(f'begin_logged: {self.civ_controller.clstate.begin_logged}, turn_active: {self.civ_controller.turn_manager.turn_active}')
        # Log in and get the first info and observation
        self.civ_controller.init_network()
        info, observation = self._get_info_and_observation()
        # Log in success, set running as True
        self.running = True
        return observation, info

    def get_game_results(self):
        game_results = self.civ_controller.game_ctrl.game_results
        return dict(sorted(game_results.items()))

    def evaluate_game(self):
        game_scores = self.civ_controller.request_scorelog()
        return self.civ_controller.game_ctrl.get_game_scores(game_scores)

    def plot_game_scores(self):
        players, tags, turns, evaluations = self.evaluate_game()
        if evaluations is None:
            return

        plot_game_scores_folder = (f"plot_game_scores/{time.strftime('%Y-%m-%d-%H-%M-%S')}-"
                                   f"{self.civ_controller.client_port}")
        if not os.path.exists(plot_game_scores_folder):
            os.makedirs(plot_game_scores_folder)

        player_colors = self.civ_controller.player_ctrl.get_player_colors()
        for ptag in EVALUATION_TAGS:
            if ptag not in evaluations:
                continue

            plt.figure()
            for player_id in evaluations[ptag].keys():
                scores = evaluations[ptag][player_id]
                x_1 = players[player_id]['start_turn']
                x_axis = range(x_1, x_1 + len(scores))
                plt.plot(
                    x_axis, scores, color=player_colors[player_id], label='player' + '_' + str(player_id))

            plt.legend()
            pfile = os.path.join(plot_game_scores_folder, ptag + '.png')
            plt.savefig(pfile)
            plt.close()

    def get_final_score(self):
        _, _, _, evaluations = self.evaluate_game()
        score = {}
        if evaluations != None:
            for tag in EVALUATION_TAGS:
                score[tag] = evaluations[tag][self.civ_controller.player_ctrl.my_player_id][-1]
        return score

    def render(self):
        """Render the environment based on freeciv-web.
        """
        pass

    def close(self):
        # fc_logger.info(f'Env port: {self.get_port()} closes ....')
        self.civ_controller.close()
        self.running = False

render()

Render the environment based on freeciv-web.

Source code in src/civrealm/envs/freeciv_base_env.py

def render(self):
    """Render the environment based on freeciv-web.
    """
    pass

MiniTask Environment

envs.freeciv_minitask_env.FreecivMinitaskEnv

Bases: FreecivBaseEnv

CivRealm environment for mini-game.

Source code in src/civrealm/envs/freeciv_minitask_env.py

class FreecivMinitaskEnv(FreecivBaseEnv):
    """ CivRealm environment for mini-game. """

    def __init__(self, username: str = fc_args["minitask_username"], client_port: int = fc_args['client_port']):
        super().__init__(username=username, client_port=client_port, is_minitask=True)
        fc_args['username'] = username
        self.filename = None
        self.task_type = None
        fc_args['debug.autosave'] = False
        self._last_minitask_score = None
        self.overall_mini_score = 0

    @staticmethod
    def get_minitask(name, minitask_pattern, max_id):
        if not isinstance(minitask_pattern, dict):
            minitask_pattern = dict()

        minitask_id = minitask_pattern.get('id', random.randint(0, max_id))
        minitask_level = minitask_pattern.get(
            'level', random.choice(MinitaskDifficulty.list()))
        minitask_type = minitask_pattern.get(
            'type', random.choice(MinitaskType.list()))

        if isinstance(minitask_type, list):
            minitask_type = random.choice(minitask_type)

        if minitask_type in FULLGAME_MINITASK_LIST and fc_args['username'] == "minitask":
            minitask_level = "normal"
            minitask_id = 0

        if minitask_type not in MinitaskType.list():
            raise ValueError(
                f"Not supported type as {minitask_pattern}. The suppported list is {MinitaskType.list()}!")
        if minitask_id > max_id or minitask_id < 0:
            raise ValueError(
                f"Not supported id as {minitask_id}. The suppported range is [0, {max_id}]!")
        if minitask_level not in MinitaskDifficulty.list():
            raise ValueError(
                f"Not supported diffculty as {minitask_level}. The suppported list is {MinitaskDifficulty.list()}!")

        minitask = '{}_T1_task_{}_level_{}_id_{}'.format(
            name, minitask_type, minitask_level, minitask_id)
        set_logging_file('minitask', minitask)
        fc_logger.warning(f"Randomly selected minitask {minitask}!")
        return minitask

    def _get_info_and_observation(self):
        info, observation = super()._get_info_and_observation()
        if 'player' in info['available_actions'] and self.task_type in BATTLE_MINITASK_LIST:
            del info['available_actions']['player']
        return info, observation

    def _set_minitask_info(self, info):
        info['minitask'] = {
            'status': self._get_game_status(),
            'success': self._get_success(),
        }
        info['minitask'].update(self._get_detail())
        return

    def reset(self, seed=None, options=None, minitask_pattern=None, max_id=MAX_ID):
        """
        Reset the mini-game environment as fully random game or specific game.

        Parameters
        ----------
        seed : int
            Random seed for game.
        options : dict
            Env configuration.
        minitask_pattern : dict
            Assignment the following fields to return a specified game:\n
            `type`: the type of mini-game, see the available options MinitaskType;\n
            `level`: the difficulty of mini-game, see the available options MinitaskDifficulty;\n
            `id`: the id of mini-game, the available range is 0 to MAX_ID.\n
            If a field is not assigned a value, the field will be randomly selected within the feasible domain.
        max_id : int
            The max id of mini-game.
        """
        self.overall_mini_score = 0
        self.set_minitask(seed, minitask_pattern, max_id)
        observations, info = super().reset(seed, options)
        self._set_minitask_info(info)
        self._last_minitask_score = None
        return observations, info

    def set_minitask(self, seed, minitask_pattern, max_id):
        random.seed(seed)
        minitask = self.get_minitask(
            fc_args['username'], minitask_pattern, max_id)
        self.filename = minitask
        self.task_type = re.match(
            r"{}_T\d+_task_([0-9a-z_]+)_level".format(fc_args['username']), minitask)[1]
        if self.task_type in FULLGAME_MINITASK_LIST and fc_args['username'] == "minitask":
            self.civ_controller.set_parameter('debug.take_player', "AI*1")
        self.civ_controller.set_parameter('debug.load_game', minitask)
        return

    def minitask_has_terminated(self):
        """
        In addition to the game termination judgment of the full game, 
        the mini-game has additional conditions for the end of the game process.
        """
        minitask_info = self.civ_controller.get_turn_message()
        if any([msg.get("status") == MinitaskGameStatus.MGS_END_GAME.value for msg in minitask_info]):
            return True
        return False

    def _get_terminated(self):
        return self.civ_controller.game_has_terminated() or self.minitask_has_terminated()

    def _get_step_msg(self, key):
        minitask_results = self.civ_controller.get_turn_message()
        for msg in minitask_results[::-1]:
            if key in msg:
                if key == 'metrics':
                    return msg[key][-1]
                return msg[key]
        return

    def _get_reward(self):
        metrics = self._get_step_msg('metrics')
        current_score = 0.0
        if metrics is None:
            return current_score
        if self._last_minitask_score is None:
            self._last_minitask_score = metrics['mini_score']
        current_score = metrics['mini_score'] - self._last_minitask_score
        self._last_minitask_score = metrics['mini_score']
        self.overall_mini_score = metrics['mini_score']
        return current_score

    def _get_game_status(self):
        status = self._get_step_msg('status')
        if status is None:
            return MinitaskGameStatus.MGS_IN_GAME.value
        return status

    def _get_success(self):
        metrics = self._get_step_msg('metrics')
        if metrics is None:
            return MinitaskPlayerStatus.MPS_UNKNOWN.value
        return metrics.get('is_mini_success')

    def _get_detail(self):
        metrics = self._get_step_msg('metrics')
        if metrics is None:
            return dict()
        return metrics

    def step(self, action):
        fc_logger.debug(f"mini-env step action: {action}")
        observation, reward, terminated, truncated, info = super().step(action)
        self._set_minitask_info(info)
        return observation, reward, terminated, truncated, info

    def get_game_results(self):
        game_results = self.civ_controller.game_ctrl.game_results
        minitask_results = self.civ_controller.get_turn_message()
        results = dict(sorted(game_results.items()))
        results.update({"minitask_sav": self.filename})
        results.update({"minitask_type": self.task_type})
        results.update(dict(minitask=minitask_results))
        return results

    def get_final_score(self):
        score = {}
        score['mini_score'] = self.overall_mini_score
        return score

minitask_has_terminated()

In addition to the game termination judgment of the full game, the mini-game has additional conditions for the end of the game process.

Source code in src/civrealm/envs/freeciv_minitask_env.py

def minitask_has_terminated(self):
    """
    In addition to the game termination judgment of the full game, 
    the mini-game has additional conditions for the end of the game process.
    """
    minitask_info = self.civ_controller.get_turn_message()
    if any([msg.get("status") == MinitaskGameStatus.MGS_END_GAME.value for msg in minitask_info]):
        return True
    return False

reset(seed=None, options=None, minitask_pattern=None, max_id=MAX_ID)

Reset the mini-game environment as fully random game or specific game.

Parameters:

Name	Type	Description	Default
seed	int	Random seed for game.	None
options	dict	Env configuration.	None
minitask_pattern	dict	Assignment the following fields to return a specified game: type: the type of mini-game, see the available options MinitaskType; level: the difficulty of mini-game, see the available options MinitaskDifficulty; id: the id of mini-game, the available range is 0 to MAX_ID. If a field is not assigned a value, the field will be randomly selected within the feasible domain.	None
max_id	int	The max id of mini-game.	MAX_ID

Source code in src/civrealm/envs/freeciv_minitask_env.py

def reset(self, seed=None, options=None, minitask_pattern=None, max_id=MAX_ID):
    """
    Reset the mini-game environment as fully random game or specific game.

    Parameters
    ----------
    seed : int
        Random seed for game.
    options : dict
        Env configuration.
    minitask_pattern : dict
        Assignment the following fields to return a specified game:\n
        `type`: the type of mini-game, see the available options MinitaskType;\n
        `level`: the difficulty of mini-game, see the available options MinitaskDifficulty;\n
        `id`: the id of mini-game, the available range is 0 to MAX_ID.\n
        If a field is not assigned a value, the field will be randomly selected within the feasible domain.
    max_id : int
        The max id of mini-game.
    """
    self.overall_mini_score = 0
    self.set_minitask(seed, minitask_pattern, max_id)
    observations, info = super().reset(seed, options)
    self._set_minitask_info(info)
    self._last_minitask_score = None
    return observations, info

Tensor Environment

envs.freeciv_tensor_env.FreecivTensorEnv

Bases: Wrapper

CivRealm environment with Tensor actions

Source code in src/civrealm/envs/freeciv_tensor_env.py

class FreecivTensorEnv(Wrapper):
    """CivRealm environment with Tensor actions"""

    metadata = FreecivBaseEnv.metadata

    def __init__(
        self,
        username: str = fc_args["username"],
        client_port: int = fc_args["client_port"],
        config: dict = default_tensor_config,
    ):
        tensor_env = GameOverScoreInfo(
            TensorWrapper(
                env=PenalizeConsecutiveTurnDoneReward(
                    FreecivBaseEnv(username=username, client_port=client_port),
                    penalty=-1,
                ),
                config=config,
            )
        )
        super().__init__(tensor_env)
        self._cached_reset_result = super().reset()
        # reset during init to get valid obs space
        self.first_reset = True

    def reset(self, **kwargs):
        if self.first_reset and len(kwargs) == 0:
            # use cached reset during init for first reset
            obs, info = self._cached_reset_result
            self.first_reset = False
            return obs, info
        return super().reset(**kwargs)

LLM Environment

envs.freeciv_llm_env.FreecivLLMEnv

Bases: Wrapper

CivRealm environment with llm actions

Source code in src/civrealm/envs/freeciv_llm_env.py

class FreecivLLMEnv(Wrapper):
    """CivRealm environment with llm actions"""

    metadata = FreecivBaseEnv.metadata

    def __init__(self,
                 username: str = fc_args["username"],
                 client_port: int = fc_args["client_port"]):

        llm_env = LLMWrapper(FreecivBaseEnv(
            username=username, client_port=client_port))
        super().__init__(llm_env)

    def reset(self, **kwargs):
        return self.env.reset(**kwargs)