我们从Python开源项目中，提取了以下17个代码示例，用于说明如何使用gym.wrappers()。
def run_episode(self, env: gym.Env): meta_wrapper = get_wrapper(env, chi.rl.wrappers.Wrapper) done = False ob = env.reset() a, meta = self.act(ob) rs = [] while not done: if meta_wrapper: meta_wrapper.set_meta(meta) # send meta information to wrappers ob, r, done, info = env.step(a) a, meta = self.act(ob, r, done, info) rs.append(r) return sum(rs)
def _make_env(self, name): import ppaquette_gym_super_mario from ppaquette_gym_super_mario import wrappers name_split = name.split('-') if name_split[0] != 'meta': prefix, world, level = name_split[:3] author, prefix = prefix.split('/') suffix = '-'.join(name_split[3:]) self._env_name = '/'.join([author, '-'.join(['meta', prefix, suffix])]) self._env_level = (int(world) - 1) * 4 + int(level) - 1 else: self._env_name = name self._env_level = None env = gym.make(self._env_name) # modewrapper = wrappers.SetPlayingMode('algo') return GymNintendoWrapper(env)
def __init__(self, env_name, record_video=True, video_schedule=None, log_dir=None, record_log=True, force_reset=False): if log_dir is None: if logger.get_snapshot_dir() is None: logger.log("Warning: skipping Gym environment monitoring since snapshot_dir not configured.") else: log_dir = os.path.join(logger.get_snapshot_dir(), "gym_log") Serializable.quick_init(self, locals()) env = gym.envs.make(env_name) self.env = env self.env_id = env.spec.id assert not (not record_log and record_video) if log_dir is None or record_log is False: self.monitoring = False else: if not record_video: video_schedule = NoVideoSchedule() else: if video_schedule is None: video_schedule = CappedCubicVideoSchedule() self.env = gym.wrappers.Monitor(self.env, log_dir, video_callable=video_schedule, force=True) self.monitoring = True self._observation_space = convert_gym_space(env.observation_space) logger.log("observation space: {}".format(self._observation_space)) self._action_space = convert_gym_space(env.action_space) logger.log("action space: {}".format(self._action_space)) self._horizon = env.spec.tags['wrapper_config.TimeLimit.max_episode_steps'] self._log_dir = log_dir self._force_reset = force_reset
def reset(self): if self._force_reset and self.monitoring: from gym.wrappers.monitoring import Monitor assert isinstance(self.env, Monitor) recorder = self.env.stats_recorder if recorder is not None: recorder.done = True return self.env.reset()
def test_no_double_wrapping(): temp = tempfile.mkdtemp() try: env = gym.make("FrozenLake-v0") env = wrappers.Monitor(env, temp) try: env = wrappers.Monitor(env, temp) except error.DoubleWrapperError: pass else: assert False, "Should not allow double wrapping" env.close() finally: shutil.rmtree(temp)
def simpleSSBMEnv(act_every=3, **kwargs): env = SSBMEnv(**kwargs) # TODO: make this a wrapper env.action_space = spaces.Discrete(len(ssbm.simpleControllerStates)) env.realController = lambda action: ssbm.simpleControllerStates[action].realController() from .box_wrapper import BoxWrapper env = BoxWrapper(env) from gym.wrappers import SkipWrapper return SkipWrapper(3)(env)
def make_env(env_name, monitor_dir): env = HistoryWrapper(HISTORY_STEPS)(gym.make(env_name)) if monitor_dir: env = gym.wrappers.Monitor(env, monitor_dir) return env
def make_env(env_name, monitor_dir=None, wrappers=()): """ Make gym environment with optional monitor :param env_name: name of the environment to create :param monitor_dir: optional directory to save monitor results :param wrappers: list of optional Wrapper object instances :return: environment object """ env = gym.make(env_name) for wrapper in wrappers: env = wrapper(env) if monitor_dir: env = gym.wrappers.Monitor(env, monitor_dir) return env
def __init__(self, env_name, record_video=True, video_schedule=None, log_dir=None, record_log=True, force_reset=False): if log_dir is None: if logger.get_snapshot_dir() is None: logger.log( "Warning: skipping Gym environment monitoring since snapshot_dir not configured.") else: log_dir = os.path.join(logger.get_snapshot_dir(), "gym_log") Serializable.quick_init(self, locals()) env = gym.envs.make(env_name) self.env = env self.env_id = env.spec.id assert not (not record_log and record_video) if log_dir is None or record_log is False: self.monitoring = False else: if not record_video: video_schedule = NoVideoSchedule() else: if video_schedule is None: video_schedule = CappedCubicVideoSchedule() self.env = gym.wrappers.Monitor( self.env, log_dir, video_callable=video_schedule, force=True) self.monitoring = True self._observation_space = convert_gym_space(env.observation_space) logger.log("observation space: {}".format(self._observation_space)) self._action_space = convert_gym_space(env.action_space) logger.log("action space: {}".format(self._action_space)) self._horizon = env.spec.tags['wrapper_config.TimeLimit.max_episode_steps'] self._log_dir = log_dir self._force_reset = force_reset
def reset(self): if self._force_reset and self.monitoring: from gym.wrappers.monitoring import _Monitor assert isinstance(self.env, _Monitor) recorder = self.env.stats_recorder if recorder is not None: recorder.done = True return self.env.reset()
def __init__(self, name, dump_dir=None, force_dump=False, state_mode='DEFAULT'): super().__init__() with get_env_lock(): self._gym = self._make_env(name) if dump_dir: io.mkdir(dump_dir) self._gym = gym.wrappers.Monitor(self._gym, dump_dir, force=force_dump) assert state_mode in ('DEFAULT', 'RENDER', 'BOTH') self._state_mode = state_mode
def _restart(self, *args, **kwargs): super()._restart(*args, **kwargs) self._action_list.clear() # https://github.com/ppaquette/gym-super-mario/blob/master/ppaquette_gym_super_mario/wrappers/action_space.py
def __init__(self, env_name, record_video=True, video_schedule=None, log_dir=None, record_log=True, force_reset=False): if log_dir is None: if logger.get_snapshot_dir() is None: logger.log("Warning: skipping Gym environment monitoring since snapshot_dir not configured.") else: log_dir = os.path.join(logger.get_snapshot_dir(), "gym_log") Serializable.quick_init(self, locals()) env = gym.envs.make(env_name) self.env = env self.env_id = env.spec.id monitor_manager.logger.setLevel(logging.WARNING) assert not (not record_log and record_video) if log_dir is None or record_log is False: self.monitoring = False else: if not record_video: video_schedule = NoVideoSchedule() else: if video_schedule is None: video_schedule = CappedCubicVideoSchedule() self.env = gym.wrappers.Monitor(self.env, log_dir, video_callable=video_schedule, force=True) self.monitoring = True self._observation_space = convert_gym_space(env.observation_space) self._action_space = convert_gym_space(env.action_space) self._horizon = env.spec.timestep_limit self._log_dir = log_dir self._force_reset = force_reset
