Python gym 模块，undo_logger_setup() 实例源码

def test_smoke(env_id):
    """Check that environments start up without errors and that we can extract rewards and observations"""
    gym.undo_logger_setup()
    logging.getLogger().setLevel(logging.INFO)

    env = gym.make(env_id)
    if env.metadata.get('configure.required', False):
        if os.environ.get('FORCE_LATEST_UNIVERSE_DOCKER_RUNTIMES'):  # Used to test universe-envs in CI
            configure_with_latest_docker_runtime_tag(env)
        else:
            env.configure(remotes=1)

    env = wrappers.Unvectorize(env)

    env.reset()
    _rollout(env, timestep_limit=60*30) # Check a rollout

def __init__(self, log):
        """Initialize default configuration."""
        # some libraries think it is a good idea to add handlers by default
        # without documenting that at all, thanks gpy...
        log.propagate = False

        self.log = log
        self.n_jobs = 1
        self.monitor_verbosity = 0

        self._stream_handler = None
        self._file_handler = None
        self._fmt = ('%(process)d - %(asctime)s - %(name)s - %(levelname)s'
                     + ' - %(message)s')
        self._formatter = logging.Formatter(self._fmt)

        try:
            import gym
            gym.undo_logger_setup()
        except:
            pass

def test_nice_vnc_semantics_match(spec, matcher, wrapper):
    # Check that when running over VNC or using the raw environment,
    # semantics match exactly.
    gym.undo_logger_setup()
    logging.getLogger().setLevel(logging.INFO)

    spaces.seed(0)

    vnc_env = spec.make()
    if vnc_env.metadata.get('configure.required', False):
        vnc_env.configure(remotes=1)
    vnc_env = wrapper(vnc_env)
    vnc_env = wrappers.Unvectorize(vnc_env)

    env = gym.make(spec._kwargs['gym_core_id'])

    env.seed(0)
    vnc_env.seed(0)

    # Check that reset observations work
    reset(matcher, env, vnc_env, stage='initial reset')

    # Check a full rollout
    rollout(matcher, env, vnc_env, timestep_limit=50, stage='50 steps')

    # Reset to start a new episode
    reset(matcher, env, vnc_env, stage='reset to new episode')

    # Check that a step into the next episode works
    rollout(matcher, env, vnc_env, timestep_limit=1, stage='1 step in new episode')

    # Make sure env can be reseeded
    env.seed(1)
    vnc_env.seed(1)
    reset(matcher, env, vnc_env, 'reseeded reset')
    rollout(matcher, env, vnc_env, timestep_limit=1, stage='reseeded step')

def test():
    render = False
    filename = 'test.h5'
    resume = False
    # filename = 'pong_gym_keras_mlp_full_batch.h5'
    # resume = True
    # render = True

    gym.undo_logger_setup() # Stop gym logging
    agent = KarpathyPolicyPong(filename, resume=resume)
    game = Game('Pong-v0', agent, render=render, logfile='test.log')
    game.play()

def testA2C():
    render = False
    filename = 'testA2C.h5'
    resume = False
    # resume = True
    # render = True

    gym.undo_logger_setup() # Stop gym logging
    actionSpace = [2, 3]
    agent = A2C_OneGame(2, 1024, actionSpace, filename, resume=resume)
    game = Game('Pong-v0', agent, render=render, logfile='test.log')
    game.play()

def env_factory(cmdl, mode):
    # Undo the default logger and configure a new one.
    gym.undo_logger_setup()
    logger = logging.getLogger()
    logger.setLevel(logging.WARNING)

    print(clr("[Main] Constructing %s environment." % mode, attrs=['bold']))
    env = gym.make(cmdl.env_name)

    if hasattr(cmdl, 'rescale_dims'):
        state_dims = (cmdl.rescale_dims, cmdl.rescale_dims)
    else:
        state_dims = env.observation_space.shape[0:2]

    env_class, hist_len, cuda = cmdl.env_class, cmdl.hist_len, cmdl.cuda

    if mode == "training":
        env = PreprocessFrames(env, env_class, hist_len, state_dims, cuda)
        if hasattr(cmdl, 'reward_clamp') and cmdl.reward_clamp:
            env = SqueezeRewards(env)
        if hasattr(cmdl, 'done_after_lost_life') and cmdl.done_after_lost_life:
            env = DoneAfterLostLife(env)
        print('-' * 50)
        return env

    elif mode == "evaluation":
        if cmdl.eval_env_name != cmdl.env_name:
            print(clr("[%s] Warning! evaluating on a different env: %s"
                      % ("Main", cmdl.eval_env_name), 'red', attrs=['bold']))
            env = gym.make(cmdl.eval_env_name)

        env = PreprocessFrames(env, env_class, hist_len, state_dims, cuda)
        env = EvaluationMonitor(env, cmdl)
        print('-' * 50)
        return env

def setup(exp, single_threaded):
    import gym
    gym.undo_logger_setup()
    from . import policies, tf_util

    config = Config(**exp['config'])
    env = gym.make(exp['env_id'])
    sess = make_session(single_threaded=single_threaded)
    policy = getattr(policies, exp['policy']['type'])(env.observation_space, env.action_space, **exp['policy']['args'])
    tf_util.initialize()

    return config, env, sess, policy

def undo_logger_setup():
    """Undoes the automatic logging setup done by OpenAI Gym. You should call
    this function if you want to manually configure logging
    yourself. Typical usage would involve putting something like the
    following at the top of your script:

    gym.undo_logger_setup()
    logger = logging.getLogger()
    logger.addHandler(logging.StreamHandler(sys.stderr))
    """
    root_logger.removeHandler(handler)
    for logger in _extra_loggers:
        logger.setLevel(logging.NOTSET)