.py
c:\Users\heman\Desktop\Notes\PPO.py:344: UserWarning: Creating a tensor from a list of numpy.ndarrays is extremely slow. Please consider converting the list to a single numpy.ndarray with numpy.array() before converting to a tensor. (Triggered internally at C:\actions-runner\_work\pytorch\pytorch\builder\windows\pytorch\torch\csrc\utils\tensor_new.cpp:281.)
  state_tensor = th.FloatTensor(state).unsqueeze(0).to(self.device)
Traceback (most recent call last):
  File "c:\Users\heman\Desktop\Notes\PPO.py", line 485, in <module>
    main()
  File "c:\Users\heman\Desktop\Notes\PPO.py", line 481, in main
    agent.run()
  File "c:\Users\heman\Desktop\Notes\PPO.py", line 409, in run
    action, log_prob, value = self.select_action(state)
                              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\heman\Desktop\Notes\PPO.py", line 344, in select_action
    state_tensor = th.FloatTensor(state).unsqueeze(0).to(self.device)
                   ^^^^^^^^^^^^^^^^^^^^^
ValueError: expected sequence of length 4 at dim 1 (got 0)
PS C:\Users\heman\Desktop\Notes> 
import warnings
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union

import gym
import numpy as np
import torch as th
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.distributions import Categorical, Normal
from gym.spaces import Box, Discrete
from collections import deque
import random
import time

# Type Aliases
Tensor = th.Tensor
Device = Union[th.device, str]
Schedule = Callable[[float], float]


def explained_variance(y_pred: np.ndarray, y_true: np.ndarray) -> float:
    """
    Computes the fraction of variance that y_pred explains about y_true.

    :param y_pred: Predicted values
    :param y_true: True values
    :return: Explained variance
    """
    var_y = np.var(y_true)
    if var_y == 0:
        return float('nan')
    return float(1 - np.var(y_true - y_pred) / var_y)


def constant_fn(val: float) -> Schedule:
    """
    Returns a function that returns a constant value.

    :param val: Constant value
    :return: Function that ignores its input and returns val
    """
    return lambda _: val


def get_schedule_fn(value_schedule: Union[Schedule, float]) -> Schedule:
    """
    Transforms a value or schedule into a callable schedule.

    :param value_schedule: A float or a callable schedule
    :return: A callable schedule
    """
    if isinstance(value_schedule, (float, int)):
        return constant_fn(float(value_schedule))
    elif callable(value_schedule):
        return value_schedule
    else:
        raise TypeError("value_schedule must be a float, int, or a callable")


class ActorCritic(nn.Module):
    """
    Actor-Critic Network for PPO.

    :param observation_space: Gym observation space
    :param action_space: Gym action space
    :param hidden_sizes: Sizes of hidden layers
    """

    def __init__(
        self,
        observation_space: gym.Space,
        action_space: gym.Space,
        hidden_sizes: Tuple[int, ...] = (64, 64),
    ):
        super(ActorCritic, self).__init__()

        # Determine the number of inputs based on the observation space
        if isinstance(observation_space, Box):
            self.obs_dim = observation_space.shape[0]
        else:
            raise NotImplementedError("Only Box observation spaces are supported.")

        # Determine the number of outputs based on the action space
        if isinstance(action_space, Discrete):
            self.action_dim = action_space.n
            self.is_discrete = True
        elif isinstance(action_space, Box):
            self.action_dim = action_space.shape[0]
            self.is_discrete = False
        else:
            raise NotImplementedError("Only Discrete and Box action spaces are supported.")

        # Build shared layers
        layers = []
        input_dim = self.obs_dim
        for size in hidden_sizes:
            layers.append(nn.Linear(input_dim, size))
            layers.append(nn.Tanh())
            input_dim = size
        self.shared_layers = nn.Sequential(*layers)

        # Policy head
        if self.is_discrete:
            self.policy_head = nn.Linear(input_dim, self.action_dim)
        else:
            self.policy_mean = nn.Linear(input_dim, self.action_dim)
            self.policy_log_std = nn.Parameter(th.zeros(self.action_dim))

        # Value head
        self.value_head = nn.Linear(input_dim, 1)

    def forward(self, x: Tensor) -> Tuple[Tensor, Tensor]:
        """
        Forward pass through shared layers and value head.

        :param x: Input tensor
        :return: Tuple of policy logits/means and value estimates
        """
        shared = self.shared_layers(x)
        if self.is_discrete:
            logits = self.policy_head(shared)
            return logits, self.value_head(shared).squeeze(-1)
        else:
            mean = self.policy_mean(shared)
            log_std = self.policy_log_std.expand_as(mean)
            std = th.exp(log_std)
            return mean, std, self.value_head(shared).squeeze(-1)

    def get_action_dist(self, x: Tensor):
        """
        Returns the action distribution based on the current policy.

        :param x: Input tensor
        :return: Torch distribution
        """
        if self.is_discrete:
            logits, _ = self.forward(x)
            return Categorical(logits=logits)
        else:
            mean, std, _ = self.forward(x)
            return Normal(mean, std)

    def act(self, x: Tensor) -> Tuple[Tensor, Tensor]:
        """
        Samples an action from the policy and gets the value estimate.

        :param x: Input tensor
        :return: Action and value
        """
        if self.is_discrete:
            logits, value = self.forward(x)
            dist = Categorical(logits=logits)
            action = dist.sample()
            return action, value
        else:
            mean, std, value = self.forward(x)
            dist = Normal(mean, std)
            action = dist.sample()
            return action, value


class RolloutBuffer:
    """
    Buffer to store trajectories for PPO.

    :param buffer_size: Maximum number of timesteps to store
    :param n_envs: Number of parallel environments
    """

    def __init__(self, buffer_size: int, n_envs: int):
        self.buffer_size = buffer_size
        self.n_envs = n_envs
        self.ptr = 0
        self.full = False

        # Initialize buffers
        self.observations = np.zeros((buffer_size, n_envs, 4), dtype=np.float32)  # Assuming observation dim=4
        self.actions = np.zeros((buffer_size, n_envs), dtype=np.int64) if isinstance(n_envs, int) else np.zeros((buffer_size, n_envs, 1), dtype=np.float32)
        self.log_probs = np.zeros((buffer_size, n_envs), dtype=np.float32)
        self.rewards = np.zeros((buffer_size, n_envs), dtype=np.float32)
        self.dones = np.zeros((buffer_size, n_envs), dtype=np.bool_)
        self.values = np.zeros((buffer_size, n_envs), dtype=np.float32)
        self.advantages = np.zeros((buffer_size, n_envs), dtype=np.float32)
        self.returns = np.zeros((buffer_size, n_envs), dtype=np.float32)

    def reset(self) -> None:
        """
        Resets the buffer.
        """
        self.ptr = 0
        self.full = False
        for key in self.__dict__:
            if isinstance(self.__dict__[key], np.ndarray):
                self.__dict__[key].fill(0)

    def add(
        self,
        obs: np.ndarray,
        actions: np.ndarray,
        log_probs: np.ndarray,
        rewards: np.ndarray,
        dones: np.ndarray,
        values: np.ndarray,
    ) -> None:
        """
        Adds a step to the buffer.

        :param obs: Observation
        :param actions: Action taken
        :param log_probs: Log probability of the action
        :param rewards: Reward received
        :param dones: Done flag
        :param values: Value estimate
        """
        assert self.ptr < self.buffer_size, "Rollout buffer is full."
        self.observations[self.ptr] = obs
        self.actions[self.ptr] = actions
        self.log_probs[self.ptr] = log_probs
        self.rewards[self.ptr] = rewards
        self.dones[self.ptr] = dones
        self.values[self.ptr] = values
        self.ptr += 1
        if self.ptr == self.buffer_size:
            self.full = True

    def compute_returns_and_advantages(
        self, last_value: np.ndarray, dones: np.ndarray, gamma: float, gae_lambda: float
    ) -> None:
        """
        Computes returns and advantages using GAE.

        :param last_value: Value estimate for the last state
        :param dones: Done flags for the last state
        :param gamma: Discount factor
        :param gae_lambda: GAE lambda
        """
        self.returns[-1] = last_value * (1 - dones)
        self.advantages[-1] = self.returns[-1] - self.values[-1]

        for step in reversed(range(self.buffer_size - 1)):
            self.returns[step] = (
                self.rewards[step]
                + gamma * self.returns[step + 1] * (1 - self.dones[step])
            )
            delta = (
                self.rewards[step]
                + gamma * self.values[step + 1] * (1 - self.dones[step])
                - self.values[step]
            )
            self.advantages[step] = delta + gamma * gae_lambda * (1 - self.dones[step]) * self.advantages[step + 1]

    def get_batches(self, batch_size: int) -> List[Dict[str, Any]]:
        """
        Generates mini-batches for training.

        :param batch_size: Size of each mini-batch
        :return: List of mini-batches
        """
        obs_shape = self.observations.shape
        batch_size_total = self.buffer_size * self.n_envs
        indices = np.random.permutation(batch_size_total)
        batches = [
            {
                "obs": self.observations.reshape(-1, obs_shape[-1])[indices[i:i + batch_size]],
                "actions": self.actions.flatten()[indices[i:i + batch_size]],
                "log_probs": self.log_probs.flatten()[indices[i:i + batch_size]],
                "returns": self.returns.flatten()[indices[i:i + batch_size]],
                "advantages": self.advantages.flatten()[indices[i:i + batch_size]],
                "values": self.values.flatten()[indices[i:i + batch_size]],
            }
            for i in range(0, batch_size_total, batch_size)
        ]
        return batches


class PPOAgent:
    """
    Proximal Policy Optimization Agent.

    :param env_id: Gym environment ID
    :param hidden_sizes: Sizes of hidden layers
    :param lr: Learning rate
    :param gamma: Discount factor
    :param gae_lambda: GAE lambda
    :param clip_epsilon: Clipping parameter epsilon
    :param epochs: Number of epochs per update
    :param batch_size: Mini-batch size
    :param max_grad_norm: Maximum gradient norm for clipping
    :param total_timesteps: Total training timesteps
    :param update_timestep: Timesteps per update
    """

    def __init__(
        self,
        env_id: str,
        hidden_sizes: Tuple[int, ...] = (64, 64),
        lr: float = 3e-4,
        gamma: float = 0.99,
        gae_lambda: float = 0.95,
        clip_epsilon: float = 0.2,
        epochs: int = 10,
        batch_size: int = 64,
        max_grad_norm: float = 0.5,
        total_timesteps: int = 1e6,
        update_timestep: int = 2048,
    ):
        # Environment
        self.env = gym.make(env_id)
        self.envs = self.env  # For simplicity, using single environment

        # Device
        self.device = th.device('cuda' if th.cuda.is_available() else 'cpu')

        # Network
        self.policy = ActorCritic(self.env.observation_space, self.env.action_space, hidden_sizes).to(self.device)
        self.optimizer = optim.Adam(self.policy.parameters(), lr=lr)

        # Hyperparameters
        self.gamma = gamma
        self.gae_lambda = gae_lambda
        self.clip_epsilon = clip_epsilon
        self.epochs = epochs
        self.batch_size = batch_size
        self.max_grad_norm = max_grad_norm
        self.total_timesteps = int(total_timesteps)
        self.update_timestep = update_timestep

        # Buffer
        self.buffer = RolloutBuffer(buffer_size=self.update_timestep, n_envs=1)

        # Tracking
        self.timesteps = 0
        self.episode_rewards = deque(maxlen=100)
        self.episode_lengths = deque(maxlen=100)

    def select_action(self, state: np.ndarray) -> Tuple[int, float, float]:
        """
        Selects an action based on the current policy.

        :param state: Current state
        :return: Action, log probability, value estimate
        """
        state_tensor = th.FloatTensor(state).unsqueeze(0).to(self.device)
        dist = self.policy.get_action_dist(state_tensor)
        action = dist.sample()
        log_prob = dist.log_prob(action).sum(dim=-1)
        if self.policy.is_discrete:
            action = action.item()
            log_prob = log_prob.item()
            value = self.policy(state_tensor)[1].item()
        else:
            action = action.cpu().numpy()[0]
            log_prob = log_prob.item()
            value = self.policy(state_tensor)[2].item()
        return action, log_prob, value

    def train(self) -> None:
        """
        Trains the PPO agent using data from the buffer.
        """
        batches = self.buffer.get_batches(self.batch_size)
        for _ in range(self.epochs):
            for batch in batches:
                obs = th.FloatTensor(batch["obs"]).to(self.device)
                actions = th.FloatTensor(batch["actions"]).to(self.device)
                returns = th.FloatTensor(batch["returns"]).to(self.device)
                advantages = th.FloatTensor(batch["advantages"]).to(self.device)
                old_log_probs = th.FloatTensor(batch["log_probs"]).to(self.device)
                values = th.FloatTensor(batch["values"]).to(self.device)

                dist = self.policy.get_action_dist(obs)
                new_log_probs = dist.log_prob(actions)
                if self.policy.is_discrete:
                    new_log_probs = new_log_probs.sum(dim=-1)
                else:
                    new_log_probs = new_log_probs.sum(dim=-1)
                entropy = dist.entropy().sum(dim=-1).mean()

                ratio = th.exp(new_log_probs - old_log_probs)
                surr1 = ratio * advantages
                surr2 = th.clamp(ratio, 1.0 - self.clip_epsilon, 1.0 + self.clip_epsilon) * advantages
                policy_loss = -th.min(surr1, surr2).mean()
                value_loss = F.mse_loss(self.policy(obs)[-1], returns)

                loss = policy_loss + 0.5 * value_loss - 0.01 * entropy

                self.optimizer.zero_grad()
                loss.backward()
                nn.utils.clip_grad_norm_(self.policy.parameters(), self.max_grad_norm)
                self.optimizer.step()

    def run(self) -> None:
        """
        Runs the training loop.
        """
        state = self.env.reset()
        done = False
        episode_reward = 0
        episode_length = 0

        while self.timesteps < self.total_timesteps:
            for _ in range(self.update_timestep):
                if done:
                    state = self.env.reset()
                    episode_reward = 0
                    episode_length = 0

                action, log_prob, value = self.select_action(state)
                next_state, reward, done, _ = self.env.step(action)

                self.buffer.add(
                    obs=state,
                    actions=action,
                    log_probs=log_prob,
                    rewards=reward,
                    dones=done,
                    values=value,
                )

                state = next_state
                episode_reward += reward
                episode_length += 1
                self.timesteps += 1

                if done:
                    self.episode_rewards.append(episode_reward)
                    self.episode_lengths.append(episode_length)

                if self.timesteps >= self.total_timesteps:
                    break

            # Compute returns and advantages
            with th.no_grad():
                last_value = self.policy(th.FloatTensor(state).unsqueeze(0).to(self.device))
                if self.policy.is_discrete:
                    last_value = last_value[1].item()
                else:
                    last_value = last_value[-1].item()

            self.buffer.compute_returns_and_advantages(
                last_value=last_value,
                dones=done,
                gamma=self.gamma,
                gae_lambda=self.gae_lambda,
            )

            # Train the policy
            self.train()

            # Reset buffer
            self.buffer.reset()

            # Logging
            if len(self.episode_rewards) > 0:
                print(
                    f"Timesteps: {self.timesteps}, "
                    f"Average Reward: {np.mean(self.episode_rewards):.2f}, "
                    f"Average Length: {np.mean(self.episode_lengths):.2f}"
                )


def main():
    """
    Main function to train the PPO agent.
    """
    env_id = "CartPole-v1"
    agent = PPOAgent(
        env_id=env_id,
        hidden_sizes=(64, 64),
        lr=3e-4,
        gamma=0.99,
        gae_lambda=0.95,
        clip_epsilon=0.2,
        epochs=10,
        batch_size=64,
        max_grad_norm=0.5,
        total_timesteps=1e5,
        update_timestep=2048,
    )
    agent.run()


if __name__ == "__main__":
    main()