Я знаю, что мои навыки Python определенно могут помочь будь лучше. Я также уверен, что у меня есть несколько действительно глупых ошибок (как логических, так и синтаксических), которые я, вероятно, пропустил в коде, тем более что никто больше не проверял проект. Буду очень признателен за любые отзывы по всем возможным аспектам.
У меня могут быть не только простые ошибки в коде, но и алгоритм PPO может быть реализован неправильно, или выбранная нейронная сеть и ее структура может быть неправильной.
Вот структура проекта:
- Папка PPO: содержит два файла:
actor_critic.py: здесь реализуется нейронная сеть. - ppo.py: реализуется алгоритм оптимизации проксимальной политики (PPO).< /li>
- hyperparameter_optimization .py: Изначально этот файл использовался для случайного поиска лучших гиперпараметров. В настоящий момент он используется для попытки обучения основной сети.
[*]Main.py: просто служит точкой входа для запуска всего проекта.
[*]path Папка: используется для хранения журналы и сети по мере необходимости.
[*]config.yaml: содержит несколько переменных, двумерную сетку с препятствиями (0 для свободного места, 1 для препятствия) и путь для сохранения. различные предметы.
В окружающей среде для расчета энергопотребления используются упрощенные формулы с заранее заданными коэффициентами.
< strong>environment.py:
Код: Выделить всё
import numpy as np
import logging
logging.basicConfig(filename='path where to save logs', level=logging.DEBUG, format='%(asctime)s %(message)s')
def log_data(message):
logging.debug(message)
class AGVEnvironment:
def __init__(self, map_size, start, goal, obstacles, mass_agv=100, mass_cargo=100, max_speed=5, max_acceleration=2, k_att=1.0, k_rep=100.0, rep_range=5.0):
self.map_size = map_size
self.start = start
self.goal = goal
self.obstacles = obstacles
self.mass_agv = mass_agv
self.mass_cargo = mass_cargo
self.k_linear = 0.1
self.k_turn = 0.05
self.k_mass = 0.02
self.k_friction = 0.03
self.k_speed = 0.01
self.k_acceleration = 0.02
self.k_deceleration = 0.02
self.eta_engine = 0.9
self.max_speed = max_speed
self.max_acceleration = max_acceleration
self.k_att = k_att
self.k_rep = k_rep
self.rep_range = rep_range
self.energy_current_step = 0
self.reset()
def reset(self):
self.position = self.start
self.previous_position = self.start
self.angle = 0
self.speed = 0
self.acceleration = 0
self.total_energy = 0
self.steps = 0
self.total_time = 0
self.done = False
return self._get_state()
def _get_state(self):
distance_to_goal = np.linalg.norm(np.array(self.position) - np.array(self.goal))
potential_field = self._calculate_potential_field()
# Converting to arrays
position = np.array(self.position, dtype=float).flatten()
additional_info = np.array([float(self.angle), float(self.speed), float(self.acceleration), float(distance_to_goal), float(self.total_time)]).flatten()
potential_field = np.array(potential_field, dtype=float).flatten()
state = np.hstack((position, additional_info, potential_field))
return state
def _calculate_potential_field(self):
# Attracting force
att_force = self.k_att * (np.array(self.goal) - np.array(self.position))
# Repulsive force
rep_force = np.zeros(2, dtype=np.float64)
radius = 10 # Radius of the area around the agent's position,to reduce the search area
x, y = int(self.position[0]), int(self.position[1])
# Iterate only over cells within the radius
for i in range(max(0, x - radius), min(self.map_size[0], x + radius + 1)):
for j in range(max(0, y - radius), min(self.map_size[1], y + radius + 1)):
if self.obstacles[j][i] == 1:
obstacle_pos = np.array([i, j])
distance = np.linalg.norm(obstacle_pos - np.array(self.position))
if distance == 0:
rep_force = np.array([-100.0, -100.0], dtype=np.float64)
continue # Skip the calculation if the agent has already crashed
if distance < self.rep_range:
rep_force += np.float64(self.k_rep) * (1.0 / distance - 1.0 / np.float64(self.rep_range)) * (1.0 / (distance**2)) * (np.array(self.position, dtype=np.float64) - np.array(obstacle_pos, dtype=np.float64)) / distance
total_force = att_force + rep_force
return total_force
def step(self, action):
if self.done:
raise Exception("Episode has finished. Please reset the environment.")
previous_state = self._get_state()
discrete_action, continuous_action = action
self._apply_discrete_action(discrete_action)
if self.position == self.previous_position: # Check for going out of bounds and return values
reward = np.array([-100.00])
return previous_state, reward, self.done
self._apply_continuous_action(continuous_action)
self.steps += 1
time_step = self._calculate_time()
self.total_time += time_step
self._calculate_energy_consumption(time_step)
reward = self._calculate_reward()
if self.done: # Check if true has already exited from apply_discrete action
return self._get_state(), reward, self.done
self.done = self._check_done()
return self._get_state(), reward, self.done
def _apply_discrete_action(self, action):
x, y = self.position
self.previous_position = self.position
if action == 0: # Up
self.position = (x, y - 1)
elif action == 1: # Down
self.position = (x, y + 1)
elif action == 2: # Left
self.position = (x - 1, y)
elif action == 3: # Right
self.position = (x + 1, y)
elif action == 4: # Up-left
self.position = (x - 1, y - 1)
elif action == 5: # Up-right
self.position = (x + 1, y - 1)
elif action == 6: # Down-left
self.position = (x - 1, y + 1)
elif action == 7: # Down-right
self.position = (x + 1, y + 1)
# Handling map out-of-bounds and collisions with obstacles
if 0 = 3000:
# log_data(f'Model saved Hyperparameters {hyperparams} are effective. {reached_goal_count} goals reached after 100000 episodes.')
# print(f'Hyperparameters {hyperparams} effective. {reached_goal_count} goals reached after 100000 episodes.')
# ppo.save(os.path.join(config['checkpoint_dir'], f'PPO_{episode_reward}.pth'))
if (episode + 1) % 20000 == 0:
if reached_goal_count 0:
avg_time = total_time_sum / reached_goal_count
avg_steps = total_steps_sum / reached_goal_count
avg_energy = total_energy_sum / reached_goal_count
else:
avg_time = avg_steps = avg_energy = 0
print(f'Episode {episode + 1} summary:')
print(f'Total reached goals: {reached_goal_count}')
print(f'Average time to reach goal: {avg_time}')
print(f'Average steps to reach goal: {avg_steps}')
print(f'Average energy consumption to reach goal: {avg_energy}')
log_data(f'Episode {episode + 1} summary:')
log_data(f'Total reached goals: {reached_goal_count}')
log_data(f'Average time to reach goal: {avg_time}')
log_data(f'Average steps to reach goal: {avg_steps}')
log_data(f'Average energy consumption to reach goal: {avg_energy}')
# Reset statistics for the next block of 5000 episodes
reached_goal_count = 0
total_time_sum = 0
total_steps_sum = 0
total_energy_sum = 0
if len(memory.states) >= 10 * hyperparams['batch_size']:
ppo.update(memory, hyperparams['batch_size'])
memory.clear_memory()
# ppo.save(os.path.join(config['checkpoint_dir'], f'PPO_{episode}.pth'))
avg_reward = total_reward / config['num_eval_episodes']
#plt.close(fig)
return avg_reward
def optimize_hyperparameters(num_trials=1000, early_stopping_rounds=50):
config = load_config()
best_hyperparams = None
best_reward = -float('inf')
no_improvement_rounds = 0
for trial in range(num_trials):
hyperparams = random_hyperparameters()
avg_reward = evaluate_hyperparameters(hyperparams, config)
log_data(f'Trial {trial+1}/{num_trials}: Hyperparameters: {hyperparams} \t Average Reward: {avg_reward}')
print(f'Trial {trial+1}/{num_trials}: Hyperparameters: {hyperparams} \t Average Reward: {avg_reward}')
if avg_reward > best_reward:
best_reward = avg_reward
best_hyperparams = hyperparams
no_improvement_rounds = 0
# Save best hyperparameters
with open(os.path.join(config['checkpoint_dir'], 'best_hyperparams.yaml'), 'w') as file:
yaml.dump(best_hyperparams, file)
else:
no_improvement_rounds += 1
if no_improvement_rounds >= early_stopping_rounds:
print(f'Early stopping triggered after {trial+1} trials.')
break
print(f'Best Hyperparameters: {best_hyperparams} \t Best Average Reward: {best_reward}')
return best_hyperparams
if __name__ == "__main__":
optimize_hyperparameters()
Код: Выделить всё
from Train.hyperparameter_optimization import optimize_hyperparameters
if __name__ == "__main__":
num_trials = 10000
early_stopping_rounds = 10000
optimize_hyperparameters(num_trials=num_trials, early_stopping_rounds=early_stopping_rounds)
Код: Выделить всё
map_size: [5, 5]
start: [0, 0]
goal: [4, 4]
obstacles: [[0, 0, 0, 0, 0],
[0, 0, 0, 0, 0],
[0, 0, 1, 0, 0],
[0, 0, 0, 0, 0],
[0, 0, 0, 0, 0]]
mass_agv: 100
mass_cargo: 100
max_speed: 5
max_acceleration: 2
checkpoint_dir: "path to folder path"
max_episodes: 1000
max_timesteps: 300
num_eval_episodes: 100000
log_interval: 10
save_interval: 50
num_trials: 10000
early_stopping_rounds: 10000
Подробнее здесь: https://stackoverflow.com/questions/790 ... ant-figure
