uav_obstacle_avoiding_drl

In this decentralized architecture, the UAV interacts with multiple obstacle types simultaneously. The training loop follows these steps:

1. Observation: Use apf.calculateDynamicState(q) to get observations for all spheres, cylinders, and cones.
2. Action Selection: For each obstacle, call ddpgNet.get_action(obs, noise_scale=noise). During early episodes (e.g., episode <= 30), actions are sampled randomly from act_bound to encourage exploration.
3. Environment Step: Use apf.getqNext(...) to calculate the next position based on the combined influence of the chosen actions (represented by eta values).
4. Reward & Storage: Calculate reward via getReward(...) and store the transition using ddpgReplayBufferStore.
5. Decentralized Update: Every update_every steps, each agent independently samples a batch from its own replay_buffer and calls .update(data=batch) if the buffer size meets the batch_size requirement.

To train a centralized Twin Delayed Deep Deterministic Policy Gradient (TD3) agent for UAV obstacle avoidance in static environments, use the main.py entrypoint. The workflow involves initializing an APF (Artificial Potential Field) environment, setting up an AgentTD3 controller, and managing a ReplayBuffer for experience replay.

Key steps in the training loop:

1. Initialize Environment: Create an APF instance and use Arguments(apf) to retrieve observation and action dimensions.
2. Initialize Agent: Instantiate AgentTD3() and call .init(hidden_dim, obs_dim, act_dim).
3. Interaction Loop:
   • Calculate observations by concatenating state information from spheres, cylinders, and cones into a single vector.
   • Select actions using centralizedContriller.select_action(obs) and transform them via transformAction.
   • Decompose the continuous action vector into specific components for action_sphere, action_cylinder, and action_cone based on the obstacle counts in the apf object.
   • Step the environment using apf.getqNext(...).
   • Calculate rewards using getReward(...).
4. Training: Periodically call centralizedContriller.update_net(buffer, update_every, batch_size, 1) to update the neural networks.
5. Model Saving: The script automatically saves the trained actor model to TrainedModel/centralizedActor.pkl when a new maximum reward is achieved.

from TD3Model import AgentTD3, ReplayBuffer
from Static_obstacle_avoidance.ApfAlgorithm import APF
from Static_obstacle_avoidance.Method import Arguments, transformAction, getReward, setup_seed

# Setup
setup_seed(1)
apf = APF()
args = Arguments(apf)

# Initialize Agent
centralizedContriller = AgentTD3()
centralizedContriller.init(256, args.obs_dim, args.act_dim)

# Initialize Buffer
buffer = ReplayBuffer(int(1e6), args.obs_dim, args.act_dim, False, True)

# Training loop logic (simplified)
# ... interaction with apf.getqNext and centralizedContriller.update_net ...

The main.py script serves as the entrypoint for training a Deep Deterministic Policy Gradient (DDPG) controller to navigate a UAV through dynamic environments using the IIFDS (Improved Integrated Force Field and Distance Smoothing) method.

Training Workflow:

1. Initialization: Sets a random seed using setup_seed(5), initializes the Config object, and instantiates the IIFDS environment and DDPG controller.
2. Weight Loading: If conf.if_load_weights is enabled and weight files exist in TrainedModel/, it loads the actor (ac_weights.pkl) and target actor (ac_tar_weights.pkl) states.
3. Exploration Strategy:
   • For the first 30 episodes, the agent uses pure random actions (random.uniform(-1, 1)).
   • After 30 episodes, it uses the trained actor with decaying noise (noise_scale).
4. Environment Interaction Loop:
   • Obtains observations via iifds.updateObs() and iifds.calDynamicState().
   • Maps actions from the $[-1, 1]$ range to physical bounds using transformAction().
   • Updates the environment state using iifds.getqNext().
   • Calculates rewards via getReward().
   • Stores transitions in the dynamicController.replay_buffer.
5. Model Updates: Every update_every steps, if the buffer size exceeds batch_size, the controller performs updates using sampled batches.
6. Evaluation & Saving: After each episode, test_multiple() evaluates the current policy. If the average reward reaches a new historical maximum (after half of MAX_EPISODE), the model weights are saved to TrainedModel/.

# Basic execution flow logic
from DDPGModel import DDPG
from Dynamic_obstacle_avoidance.IIFDS import IIFDS
from Dynamic_obstacle_avoidance.Method import setup_seed, transformAction, getReward, test_multiple
from Dynamic_obstacle_avoidance.config import Config

setup_seed(5)
conf = Config()
iifds = IIFDS()
dynamicController = DDPG(conf.obs_dim, conf.act_dim)

# Training loop logic follows...

This script implements a Fully Centralized Deep Deterministic Policy Gradient (DDPG) approach for UAV obstacle avoidance. It uses an Artificial Potential Field (APF) environment where a single centralized controller receives the concatenated states of all obstacles (spheres, cylinders, and cones) to determine actions.

Key workflow:

1. Initialize Environment: Create an APF instance to manage obstacle states and dynamics.
2. Define Dimensions: Calculate obs_dim and act_dim based on the number of obstacles (spheres, cylinders, and cones) provided by the APF instance.
3. Initialize Controller: Instantiate DDPG with the calculated dimensions and action bounds.
4. Training Loop:
   • Collect observations by concatenating obstacle states into a single vector.
   • Generate actions using centralizedContriller.get_action(obs, noise_scale=noise).
   • Decompose the action vector back into specific obstacle actions (action_sphere, action_cylinder, action_cone).
   • Interact with the environment using apf.getqNext(...).
   • Store transitions in the replay buffer using centralizedContriller.replay_buffer.store(...).
   • Periodically update the model using centralizedContriller.update(data=batch).
5. Model Saving: If the reward exceeds the historical maximum after 2/3 of the training episodes, the actor model is saved to TrainedModel/centralizedActor.pkl.

from DDPGModel import DDPG
from Static_obstacle_avoidance.ApfAlgorithm import APF
from Static_obstacle_avoidance.Method import getReward, setup_seed

# Setup
setup_seed(11)
apf = APF()
obs_dim = 6 * (apf.numberOfSphere + apf.numberOfCylinder + apf.numberOfCone)
act_dim = 1 * (apf.numberOfSphere + apf.numberOfCylinder + apf.numberOfCone)
act_bound = [0.1, 3]

# Initialize Controller
centralizedContriller = DDPG(obs_dim, act_dim, act_bound)

# Training loop logic (simplified)
# ... loop episodes and steps ...
# action = centralizedContriller.get_action(obs, noise_scale=noise)
# qNext = apf.getqNext(apf.epsilon0, action_sphere, action_cylinder, action_cone, q, qBefore)
# centralizedContriller.replay_buffer.store(obs, action, reward, obs_next, done)
# centralizedContriller.update(data=batch)

The train(arglist) function is the primary entrypoint for training Multi-Agent Deep Deterministic Policy Gradient (MADDPG) agents to manage obstacle avoidance using an Artificial Potential Field (APF) environment.

Workflow:

1. Environment Setup: Initializes an APF environment which contains spheres, cylinders, and cones as obstacles.
2. Agent Initialization: Creates MLPActor and MLPQFunction networks for each obstacle. Each obstacle acts as an agent that learns to adjust its 'gravitational factor' (action) to influence the UAV's path.
3. Memory: Uses a ReplayBuffer to store transitions (obs, action, reward, next_obs, done).
4. Training Loop:
   • For a specified number of episodes, the UAV moves through the environment.
   • Actions are sampled randomly until arglist.actor_begin_work is reached, after which the trained actors take control with added exploration noise (var).
   • Experiences are stored in the buffer.
   • agents_train is called to perform gradient updates on both actors and critics.
5. Model Saving: Automatically saves the best performing actor models to the TrainedModel/ directory based on the highest reward achieved.
6. Visualization: Uses a Painter class to log and plot reward data.

from Static_obstacle_avoidance.MADDPG.main import train
from arguments import parse_args

# Initialize arguments from CLI
arglist = parse_args()

# Start the training process
train(arglist)