rahul_code

 #Environment setup -static obstacles


 1. Static Environment Setup-Fully Centralized TD3


class StaticEnvironment:
    def __init__(self, num_agents, state_dim, action_dim, obstacle_positions, goal_positions):
        self.num_agents = num_agents
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.obstacle_positions = obstacle_positions
        self.goal_positions = goal_positions
        self.agent_states = np.random.uniform(low=0, high=10, size=(num_agents, state_dim))

    def reset(self):
        self.agent_states = np.random.uniform(low=0, high=10, size=(self.num_agents, self.state_dim))
        return self.agent_states

    def step(self, actions):
        rewards = []
        new_states = []
        dones = []

        for i, action in enumerate(actions):
            state = self.agent_states[i]

            # Update state based on action
            new_state = state + action  # Simple dynamics: new position = current + action
            self.agent_states[i] = new_state

            # Reward: Encourage movement toward goal, penalize collision
            distance_to_goal = np.linalg.norm(new_state - self.goal_positions[i])
            collision = any(np.linalg.norm(new_state - obs) < 1.0 for obs in self.obstacle_positions)

            reward = -distance_to_goal
            if collision:
                reward -= 100  # Heavy penalty for collision

            rewards.append(reward)
            new_states.append(new_state)
            dones.append(distance_to_goal < 1.0 or collision)

        return np.array(new_states), np.array(rewards), np.array(dones)

    def render(self):
        pass  # Add visualization if needed


 2.For  Centralized TD3 Implementation


# TD3 Agent for Static Environment
class StaticTD3Agent(TD3Agent):
    def __init__(self, state_dim, action_dim, num_agents):
        super().__init__(state_dim * num_agents, action_dim * num_agents)  # Centralized state and action spaces
        self.num_agents = num_agents

    def select_action(self, states):
        """
        States include the concatenated states of all agents.
        """
        centralized_state = np.concatenate(states)
        centralized_action = super().select_action(centralized_state)
        return np.split(centralized_action, self.num_agents)  # Split actions for each agent


3.Workflow Integration
   1.Initialize the Environment and Agent:
# Define static environment
num_agents = 3
state_dim = 4  # Example: [x, y, vx, vy]
action_dim = 2  # Example: [vx, vy]
obstacle_positions = [np.array([5, 5]), np.array([8, 2])]
goal_positions = [np.array([10, 10]), np.array([10, 0]), np.array([0, 10])]

env = StaticEnvironment(num_agents, state_dim, action_dim, obstacle_positions, goal_positions)

# Initialize TD3 Agent
agent = StaticTD3Agent(state_dim, action_dim, num_agents)


2.Training Loop:
 # Training Loop
num_episodes = 500
replay_buffer = ReplayBuffer()

for episode in range(num_episodes):
    states = env.reset()
    episode_reward = 0

    for step in range(200):  # Max steps per episode
        actions = agent.select_action(states)
        next_states, rewards, dones = env.step(actions)

        # Store experience in replay buffer
        replay_buffer.store(np.concatenate(states), np.concatenate(actions), 
                            np.mean(rewards), np.concatenate(next_states), 
                            np.any(dones))

        # Update the TD3 agent
        agent.update(replay_buffer)

        states = next_states
        episode_reward += np.mean(rewards)

        if np.any(dones):
            break

    print(f"Episode {episode + 1}, Reward: {episode_reward}")







 ##Environment Setup for UAV (Dynamic Obstacles)



import numpy as np

class UAVDynamicEnvironment:
    def __init__(self, state_dim, action_dim, goal_position, obstacle_positions, obstacle_velocities):
        self.state_dim = state_dim  # e.g., [x, y, vx, vy]
        self.action_dim = action_dim  # e.g., [vx, vy]
        self.goal_position = goal_position
        self.obstacle_positions = np.array(obstacle_positions)
        self.obstacle_velocities = np.array(obstacle_velocities)
        self.state = np.random.uniform(low=0, high=10, size=(state_dim,))
    
    def reset(self):
        self.state = np.random.uniform(low=0, high=10, size=(self.state_dim,))
        return self.state
    
    def step(self, action):
        # Update UAV's state
        self.state[:2] += self.state[2:] + action  # Simple dynamics: position = velocity + action
        
        # Update obstacles' positions based on their velocities
        self.obstacle_positions += self.obstacle_velocities
        
        # Calculate reward
        distance_to_goal = np.linalg.norm(self.state[:2] - self.goal_position)
        collision = any(np.linalg.norm(self.state[:2] - obs) < 1.0 for obs in self.obstacle_positions)
        
        reward = -distance_to_goal  # Encourage moving toward the goal
        if collision:
            reward -= 100  # Heavy penalty for collision
        
        done = distance_to_goal < 1.0 or collision  # Done if goal reached or collision
        
        return self.state, reward, done


##SAC (Soft Actor-Critic) Implementation For dynamic obstacles :


import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

# Define Q-network (Critic)
class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(QNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim + action_dim, 256)
        self.fc2 = nn.Linear(256, 256)
        self.fc3 = nn.Linear(256, 1)

    def forward(self, state, action):
        x = torch.cat([state, action], dim=-1)
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

# Define Policy Network (Actor)
class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(PolicyNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim, 256)
        self.fc2 = nn.Linear(256, 256)
        self.fc3 = nn.Linear(256, action_dim)
        self.log_std = nn.Parameter(torch.zeros(action_dim))  # Log std for exploration

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        x = torch.relu(self.fc2(x))
        mean = self.fc3(x)
        std = torch.exp(self.log_std)
        return mean, std

# Define SAC Agent
class SAC:
    def __init__(self, state_dim, action_dim, alpha=0.2, gamma=0.99, tau=0.005, lr=3e-4):
        self.state_dim = state_dim
        self.action_dim = action_dim
        self.alpha = alpha
        self.gamma = gamma
        self.tau = tau
        
        # Initialize Q networks and policy network
        self.q1_network = QNetwork(state_dim, action_dim)
        self.q2_network = QNetwork(state_dim, action_dim)
        self.policy_network = PolicyNetwork(state_dim, action_dim)
        
        # Target Q networks
        self.target_q1_network = QNetwork(state_dim, action_dim)
        self.target_q2_network = QNetwork(state_dim, action_dim)
        self.target_q1_network.load_state_dict(self.q1_network.state_dict())
        self.target_q2_network.load_state_dict(self.q2_network.state_dict())
        
        # Optimizers
        self.q1_optimizer = optim.Adam(self.q1_network.parameters(), lr=lr)
        self.q2_optimizer = optim.Adam(self.q2_network.parameters(), lr=lr)
        self.policy_optimizer = optim.Adam(self.policy_network.parameters(), lr=lr)

    def select_action(self, state):
        state = torch.tensor(state, dtype=torch.float32).unsqueeze(0)
        mean, std = self.policy_network(state)
        dist = torch.distributions.Normal(mean, std)
        action = dist.sample()  # Sample from the distribution
        log_prob = dist.log_prob(action).sum(dim=-1, keepdim=True)
        return action.detach().numpy(), log_prob.detach()

    def update(self, replay_buffer, batch_size=256):
        # Sample a batch from the replay buffer
        states, actions, rewards, next_states, dones = replay_buffer.sample(batch_size)
        
        states = torch.tensor(states, dtype=torch.float32)
        actions = torch.tensor(actions, dtype=torch.float32)
        rewards = torch.tensor(rewards, dtype=torch.float32).unsqueeze(1)
        next_states = torch.tensor(next_states, dtype=torch.float32)
        dones = torch.tensor(dones, dtype=torch.float32).unsqueeze(1)
        
        # Compute Q-targets
        with torch.no_grad():
            next_action, next_log_prob = self.select_action(next_states)
            target_q1 = self.target_q1_network(next_states, next_action)
            target_q2 = self.target_q2_network(next_states, next_action)
            target_q = torch.min(target_q1, target_q2) - self.alpha * next_log_prob

        # Update Q-networks
        q1_loss = torch.mean((self.q1_network(states, actions) - (rewards + self.gamma * (1 - dones) * target_q)) ** 2)
        q2_loss = torch.mean((self.q2_network(states, actions) - (rewards + self.gamma * (1 - dones) * target_q)) ** 2)

        self.q1_optimizer.zero_grad()
        q1_loss.backward()
        self.q1_optimizer.step()

        self.q2_optimizer.zero_grad()
        q2_loss.backward()
        self.q2_optimizer.step()

        # Update policy network
        mean, std = self.policy_network(states)
        dist = torch.distributions.Normal(mean, std)
        action = dist.sample()
        log_prob = dist.log_prob(action).sum(dim=-1, keepdim=True)

        q1_value = self.q1_network(states, action)
        q2_value = self.q2_network(states, action)
        q_value = torch.min(q1_value, q2_value)

        policy_loss = torch.mean(self.alpha * log_prob - q_value)
        
        self.policy_optimizer.zero_grad()
        policy_loss.backward()
        self.policy_optimizer.step()

        # Soft update for target networks
        for target_param, param in zip(self.target_q1_network.parameters(), self.q1_network.parameters()):
            target_param.data.copy_(self.tau * param.data + (1.0 - self.tau) * target_param.data)

        for target_param, param in zip(self.target_q2_network.parameters(), self.q2_network.parameters()):
            target_param.data.copy_(self.tau * param.data + (1.0 - self.tau) * target_param.data)



##Training Loop for UAV Obstacle Avoidance
# Initialize environment and agent
state_dim = 4  # [x, y, vx, vy]
action_dim = 2  # [vx, vy]
goal_position = np.array([10, 10])
obstacle_positions = [np.array([5, 5]), np.array([8, 2])]
obstacle_velocities = [np.array([0.1, 0.1]), np.array([-0.05, 0.05])]  # Dynamic obstacles

env = UAVDynamicEnvironment(state_dim, action_dim, goal_position, obstacle_positions, obstacle_velocities)
agent = SAC(state_dim, action_dim)

# Training loop
num_episodes = 500
replay_buffer = ReplayBuffer()  # Make sure to implement this

for episode in range(num_episodes):
    state = env.reset()
    episode_reward = 0
    for step in range(200):
        action, log_prob = agent.select_action(state)
        next_state, reward, done = env.step(action)
        
        replay_buffer.store(state, action, reward, next_state, done)
        
        agent.update(replay_buffer)

        state = next_state
        episode_reward += reward
        
        if done:
            break
    print(f"Episode {episode + 1}, Reward: {episode_reward}")