예제 소스

LunarLander-hgf.py

from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
import os
import cv2
import numpy as np
from tqdm import tqdm
import torch

# SARSA 알고리즘 파라미터
alpha = 0.1  # 학습률
gamma = 0.99  # 할인율
epsilon = 0.1  # 탐험 확률
num_episodes = 1000

# Create the environment
env = make_vec_env('LunarLander-v3', n_envs=1)

# Create the PPO model with adjusted hyperparameters
model = PPO('MlpPolicy', env, verbose=1, learning_rate=0.0003, n_steps=2048, batch_size=64, gamma=0.99, gae_lambda=0.95)

# Train the model
model.learn(total_timesteps=100000)

# Save the model
model_dir = "./save_model_LunarLander_hgf"
os.makedirs(model_dir, exist_ok=True)
model.save(os.path.join(model_dir, "ppo_lunarlander"))

# Load the model
model = PPO.load(os.path.join(model_dir, "ppo_lunarlander"))

# Evaluate the model over multiple episodes
scores = []

for episode in tqdm(range(num_episodes)):
    obs = env.reset()  # VecEnv API 사용
    done = False
    score = 0

    while not done:
        try:
            action, _states = model.predict(obs, deterministic=True)
            obs, reward, done, info = env.step(action)

            # Modify reward
            # Add penalty for moving away from the target
            if np.abs(obs[0][0]) > 0.1:  # Example threshold for moving away
                reward -= 10

            # Add penalty for moving too horizontally
            if np.abs(obs[0][1]) < 0.05 and np.abs(obs[0][3]) < 0.05:
                reward -= 10

            # Add penalty for moving too high
            if obs[0][1] > 1.0:  # Example threshold for moving too high
                reward -= 10

            score += reward

            # Render the environment
            img = env.render(mode='rgb_array')
            if img is not None:
                cv2.imshow('LunarLander', img)
                cv2.waitKey(1)
        except Exception as e:
            print(f"Error in episode {episode + 1}: {e}")
    
    scores.append(score)
    # tqdm.write(f"Episode {episode + 1}: Score = {score}")

cv2.destroyAllWindows()
env.close()

# Print average score
average_score = np.mean(scores)
print(f"Average score over {num_episodes} episodes: {average_score}")