Python gym 模块,wrappers() 实例源码
我们从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, 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 test_no_double_wrapping():
temp = tempfile.mkdtemp()
try:
env = gym.make("FrozenLake-v0")
env = wrappers.Monitor(env, "Should not allow double wrapping"
env.close()
finally:
shutil.rmtree(temp)
def __init__(self,
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(), 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, 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 __init__(self, 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, 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, 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
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。