Hi, I've implemented a DQN Agent and I want you to implement a code to test my implemented DQN on TWO environments:‘CartPole-v1’ and any other
complex environment of your choice from OpenAI LunarLander, OpenAI MountainCar, OpenAI
Atari Breakout (Choose the easy one). Provide performance results including
reward dynamics (total reward per episode). Goal : ‘CartPole-v1’ environment is considered to be solved if it is getting an average
reward of more than 470 points over 100 consecutive episodes during evaluation. That being said , Here is the DQN implementation:
class DQNAgentPytorch:
    def __init__(self, env, memory_size=2000, epsilon=1.0, gamma=0.99, learning_rate=0.001):
        self.env = env
        self.observation_space = env.observation_space
        self.action_space = env.action_space
        self.memory = deque(maxlen=memory_size)
        self.epsilon = epsilon
        self.gamma = gamma
        self.lr = learning_rate
        self.model = self.build_model()
        self.target_model = self.build_model()
        self.optimizer = optim.Adam(self.model.parameters(), lr=self.lr)
        self.criterion = nn.MSELoss()
        self.update_target_net()

    def build_model(self):
        model = nn.Sequential(
            nn.Linear(self.observation_space.n, 16),
            nn.ReLU(),
            nn.Linear(16, 16),
            nn.ReLU(),
            nn.Linear(16, self.action_space.n)
        )
        return model

    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))

    def choose_action(self, state, test=False):
        if test or random.random() > self.epsilon:
            with torch.no_grad():
                state_oh = self.state_to_onehot(state)
                q_values = self.model(state_oh)
                return torch.argmax(q_values).item()
        else:
            return np.random.randint(self.action_space.n)

    def state_to_onehot(self, state):
        return torch.tensor(np.identity(self.observation_space.n)[state].reshape(1, -1), dtype=torch.float32)

    def train(self, batch_size):
        if len(self.memory) < batch_size:
            return

        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = self.target_model(self.state_to_onehot(state)).squeeze(0)
            if done:
                target[action] = reward
            else:
                next_q_values = self.target_model(self.state_to_onehot(next_state)).squeeze(0)
                target[action] = reward + self.gamma * torch.max(next_q_values)

            inputs = self.state_to_onehot(state)
            targets = target.unsqueeze(0)

            self.optimizer.zero_grad()

            outputs = self.model(inputs)
            loss = self.criterion(outputs, targets)
            loss.backward()

            self.optimizer.step()

    def update_target_net(self):
        self.target_model.load_state_dict(self.model.state_dict())

    def update_epsilon(self, new_epsilon):
        self.epsilon = new_epsilon

    def save_weights(self, file_name):
        torch.save(self.model.state_dict(), file_name)

    def load_weights(self, file_name):
        self.epsilon = 0
        self.model.load_state_dict(torch.load(file_name))                                                                                                                                                                                                                            def train_dqn(agent, episodes, batch_size, epsilon_decay=0.995, min_epsilon=0.01):
    rewards = []
    steps_list = []
    epsilons = []
    target_update_freq = 25

    for episode_i in range(1, episodes + 1):
        state = agent.env.reset()
        done = False
        episode_reward = 0
        steps = 0
        while not done:
            steps += 1
            action = agent.choose_action(state)
            next_state, reward, done, _ = agent.env.step(action)
            agent.remember(state, action, reward, next_state, done)
            agent.train(batch_size)

            state = next_state
            episode_reward += reward

            if steps % target_update_freq == 0:
                agent.update_target_net()

        rewards.append(episode_reward)
        steps_list.append(steps)
        epsilons.append(agent.epsilon)
        agent.update_epsilon(max(agent.epsilon * epsilon_decay, min_epsilon))

        print(f"Episode: {episode_i}, Total Reward: {episode_reward}, Epsilon: {agent.epsilon}, Steps: {steps}")

    return rewards, epsilons, steps_list

def plot_results(rewards, epsilons):
    plt.subplot(2, 1, 1)
    plt.plot(rewards)
    plt.title('DQN GridWorld Rewards')
    plt.subplot(2, 1, 2)
    plt.plot(epsilons)
    plt.title('Epsilon Decay')
    plt.xlabel('Episodes')
    plt.tight_layout()
    plt.show()env = GridWorldDeterministic()
agent = DQNAgentPytorch(env)

rewards, epsilons, steps_list = train_dqn(agent, episodes=100, batch_size=32)
plot_results(rewards, epsilons)

agent.save_weights('aboda_assignment2_part2_dqn_gridworld.h5')                                                                                                                                                                                            
def evaluate(agent, env, episodes=5, max_steps=10):
    test_rewards = []
    
    for episode_i in range(1, episodes + 1):
        state = env.reset()
        done = False
        total_reward = 0
        step = 0

        while not done and step < max_steps:
            action = agent.choose_action(state, test=True)
            next_state, reward, done, _ = env.step(action)
            env.render()
            print("Action taken:", action)
            print("New state:", state)
            print("Reward received:", reward)
            print("Done?", done)
            print("="*20)
            state = next_state
            total_reward += reward
            step += 1
        
        test_rewards.append(total_reward)
        print(f'Test Episode: {episode_i}, Total Reward: {total_reward}, Steps: {step}')
    
    return test_rewards

env = GridWorldDeterministic()
agent = DQNAgentPytorch(env)
agent.load_weights('aboda_assignment2_part2_dqn_gridworld.h5')

test_rewards = evaluate(agent, env)

# Plot test reward graph
plt.plot(test_rewards)
plt.xlabel('Test Episodes')
plt.ylabel('Rewards')
plt.title('DQN GridWorld Test Rewards')
plt.grid()
plt.show()