Python gym 模块,monitoring() 实例源码
我们从Python开源项目中,提取了以下11个代码示例,用于说明如何使用gym.monitoring()。
def test_only_complete_episodes_written():
with helpers.tempdir() as temp:
env = gym.make('CartPole-v0')
env.monitor.start(temp, video_callable=False)
env.reset()
d = False
while not d:
_, _, d, _ = env.step(env.action_space.sample())
env.reset()
env.step(env.action_space.sample())
env.monitor.close()
# Only 1 episode should be written
results = monitoring.load_results(temp)
assert len(results['episode_lengths']) == 1, "Found {} episodes written; expecting 1".format(len(results['episode_lengths']))
def __init__(self, env_name, record_video=True, video_schedule=None, log_dir=None, record_log=True):
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.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.monitor.start(log_dir, video_schedule, force=True) # add 'force=True' if want overwrite dirs
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
def terminate(self):
if self.monitoring:
self.env.monitor.close()
if self._log_dir is not None:
print("""
***************************
Training finished! You can upload results to OpenAI Gym by running the following command:
python scripts/submit_gym.py %s
***************************
""" % self._log_dir)
def __init__(self, env, type="origin"):
self.env = env
self.type = type
self.video_schedule = None
if not pms.record_movie:
self.video_schedule = NoVideoSchedule()
else:
if self.video_schedule is not None:
self.video_schedule = CappedCubicVideoSchedule()
self.env.monitor.start("log/trpo" ,self.video_schedule, force=True)
self.monitoring = True
def test_video_callable_false_does_not_record():
with helpers.tempdir() as temp:
env = gym.make('CartPole-v0')
env.monitor.start(temp, video_callable=False)
env.reset()
env.monitor.close()
results = monitoring.load_results(temp)
assert len(results['videos']) == 0
def test_video_callable_records_videos():
with helpers.tempdir() as temp:
env = gym.make('CartPole-v0')
env.monitor.start(temp)
env.reset()
env.monitor.close()
results = monitoring.load_results(temp)
assert len(results['videos']) == 1, "Videos: {}".format(results['videos'])
def __init__(self, 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(), 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
def reset(self, **kwargs):
if self._force_reset and self.monitoring:
recorder = self.env._monitor.stats_recorder
if recorder is not None:
recorder.done = True
return self.env.reset()
def terminate(self):
if self.monitoring:
self.env._close()
if self._log_dir is not None:
print("""
***************************
Training finished! You can upload results to OpenAI Gym by running the following command:
python scripts/submit_gym.py %s
***************************
""" % self._log_dir)
def _start(self, directory, video_callable=None, force=False, resume=False,
write_upon_reset=False, uid=None, mode=None):
"""Start monitoring.
Args:
directory (str): A per-training run directory where to record stats.
video_callable (Optional[function,False]): function that takes in the index of the episode and outputs a boolean,indicating whether we should record a video on this episode. The default (for video_callable is None) is to take perfect cubes,capped at 1000. False disables video recording.
force (bool): Clear out existing training data from this directory (by deleting every file prefixed with "openaigym.").
resume (bool): Retain the training data already in this directory,which will be merged with our new data
write_upon_reset (bool): Write the manifest file on each reset. (This is currently a JSON file,so writing it is somewhat expensive.)
uid (Optional[str]): A unique id used as part of the suffix for the file. By default,uses os.getpid().
mode (['evaluation','training']): Whether this is an evaluation or training episode.
"""
if self.env.spec is None:
logger.warning("Trying to monitor an environment which has no 'spec' set. This usually means you did not create it via 'gym.make',and is recommended only for advanced users.")
env_id = '(unkNown)'
else:
env_id = self.env.spec.id
if not os.path.exists(directory):
logger.info('Creating monitor directory %s', directory)
if six.PY3:
os.makedirs(directory, exist_ok=True)
else:
os.makedirs(directory)
if video_callable is None:
video_callable = capped_cubic_video_schedule
elif video_callable == False:
video_callable = disable_videos
elif not callable(video_callable):
raise error.Error('You must provide a function,None,or False for video_callable,not {}: {}'.format(type(video_callable), video_callable))
self.video_callable = video_callable
# Check on whether we need to clear anything
if force:
clear_monitor_files(directory)
elif not resume:
training_manifests = detect_training_manifests(directory)
if len(training_manifests) > 0:
raise error.Error('''Trying to write to monitor directory {} with existing monitor files: {}.
You should use a unique directory for each training run,or use 'force=True' to automatically clear prevIoUs monitor files.'''.format(directory, ','.join(training_manifests[:5])))
self._monitor_id = monitor_closer.register(self)
self.enabled = True
self.directory = os.path.abspath(directory)
# We use the 'openai-gym' prefix to determine if a file is
# ours
self.file_prefix = FILE_PREFIX
self.file_infix = '{}.{}'.format(self._monitor_id, uid if uid else os.getpid())
self.stats_recorder = stats_recorder.StatsRecorder(directory, '{}.episode_batch.{}'.format(self.file_prefix, self.file_infix), autoreset=self.env_semantics_autoreset, env_id=env_id)
if not os.path.exists(directory): os.mkdir(directory)
self.write_upon_reset = write_upon_reset
if mode is not None:
self._set_mode(mode)