314 lines
15 KiB
Python
314 lines
15 KiB
Python
import gymnasium as gym
|
||
from gymnasium import spaces
|
||
import numpy as np
|
||
import yaml
|
||
import math
|
||
|
||
|
||
class PartitionMazeEnv(gym.Env):
|
||
"""
|
||
自定义环境，分为两阶段：
|
||
阶段 0：区域切分（共 4 步，每一步输出一个标量，用于确定竖切和横切位置）。
|
||
切分顺序为：第一步输出 c₁，第二步输出 c₂，第三步输出 r₁，第四步输出 r₂。
|
||
离散化后取值仅为 {0, 0.1, 0.2, …, 0.9}（其中 0 表示不切）。
|
||
阶段 1：车辆路径规划（走迷宫），车辆从区域中心出发，在九宫格内按照上下左右移动，
|
||
直到所有目标格子被覆盖或步数上限达到。
|
||
"""
|
||
|
||
def __init__(self, config=None):
|
||
super(PartitionMazeEnv, self).__init__()
|
||
# 车队参数设置
|
||
with open('params.yml', 'r', encoding='utf-8') as file:
|
||
params = yaml.safe_load(file)
|
||
|
||
self.H = params['H']
|
||
self.W = params['W']
|
||
self.num_cars = params['num_cars']
|
||
|
||
self.flight_time_factor = params['flight_time_factor']
|
||
self.comp_time_factor = params['comp_time_factor']
|
||
self.trans_time_factor = params['trans_time_factor']
|
||
self.car_time_factor = params['car_time_factor']
|
||
self.bs_time_factor = params['bs_time_factor']
|
||
|
||
self.flight_energy_factor = params['flight_energy_factor']
|
||
self.comp_energy_factor = params['comp_energy_factor']
|
||
self.trans_energy_factor = params['trans_energy_factor']
|
||
self.battery_energy_capacity = params['battery_energy_capacity']
|
||
|
||
##############################
|
||
# 可能需要手动修改的超参数
|
||
##############################
|
||
self.CUT_NUM = 4 # 横切一半，竖切一半
|
||
self.BASE_LINE = 4000 # 基准时间，通过greedy或者蒙特卡洛计算出来
|
||
self.MAX_STEPS = 50 # 迷宫走法步数上限
|
||
|
||
self.phase = 0 # 阶段控制，0：区域划分阶段，1：迷宫初始化阶段，2：走迷宫阶段
|
||
self.partition_step = 0 # 区域划分阶段步数，范围 0~4
|
||
self.partition_values = np.zeros(
|
||
self.CUT_NUM, dtype=np.float32) # 存储 c₁, c₂, r₁, r₂
|
||
|
||
# 定义动作空间：长度为 15 的离散动作空间
|
||
# 前 10 个表示切分动作 {0, 0.1, ..., 0.9}，后 5 个表示上下左右移动和保持不动
|
||
self.action_space = spaces.Discrete(15)
|
||
|
||
# 定义观察空间为8维向量
|
||
# 阶段 0 状态：前 4 维表示已决策的切分值（未决策部分为 0）
|
||
# 阶段 1 状态：车辆位置 (2D)
|
||
max_regions = (self.CUT_NUM // 2 + 1) ** 2
|
||
self.observation_space = spaces.Box(
|
||
low=0.0, high=100.0, shape=(1 + self.CUT_NUM + max_regions,), dtype=np.float32)
|
||
|
||
# 切分阶段相关变量
|
||
self.col_cuts = [] # 存储竖切位置（c₁, c₂），当值为0时表示不切
|
||
self.row_cuts = [] # 存储横切位置（r₁, r₂）
|
||
|
||
self.init_maze_step = 0
|
||
|
||
# 路径规划阶段相关变量
|
||
self.step_count = 0
|
||
self.rectangles = {}
|
||
self.car_pos = [(self.H / 2, self.W / 2) for _ in range(self.num_cars)]
|
||
self.car_traj = [[] for _ in range(self.num_cars)]
|
||
self.current_car_index = 0
|
||
|
||
def reset(self, seed=None, options=None):
|
||
# 重置所有变量，回到切分阶段（phase 0）
|
||
self.phase = 0
|
||
self.partition_step = 0
|
||
self.partition_values = np.zeros(self.CUT_NUM, dtype=np.float32)
|
||
self.col_cuts = []
|
||
self.row_cuts = []
|
||
self.init_maze_step = 0
|
||
self.region_centers = []
|
||
self.step_count = 0
|
||
self.rectangles = {}
|
||
self.car_pos = [(self.H / 2, self.W / 2) for _ in range(self.num_cars)]
|
||
self.car_traj = [[] for _ in range(self.num_cars)]
|
||
self.current_car_index = 0
|
||
# 状态：前 4 维为 partition_values，其余补 0
|
||
max_regions = (self.CUT_NUM // 2 + 1) ** 2
|
||
state = np.concatenate([
|
||
[self.phase],
|
||
self.partition_values,
|
||
np.zeros(max_regions, dtype=np.float32)
|
||
])
|
||
return state
|
||
|
||
def step(self, action):
|
||
# 在所有阶段动作均为离散动作
|
||
if self.phase == 0:
|
||
# 切分阶段：前 10 个动作对应 {0, 0.1, ..., 0.9}
|
||
disc_val = action * 0.1 # 修正为动作直接映射到切分比例
|
||
self.partition_values[self.partition_step] = disc_val
|
||
self.partition_step += 1
|
||
|
||
# 构造当前状态：前 partition_step 个为已决策值，其余为 0，再补 7 个 0
|
||
state = np.concatenate(
|
||
[[self.phase], self.partition_values, np.zeros(
|
||
(self.CUT_NUM // 2 + 1) ** 2, dtype=np.float32)]
|
||
)
|
||
|
||
# 如果未完成 4 步，则仍处于切分阶段，不发奖励，done 为 False
|
||
if self.partition_step < self.CUT_NUM:
|
||
return state, 0.0, False, False, {}
|
||
else:
|
||
# 完成 4 步后，计算切分边界
|
||
# 过滤掉 0，并去重后排序
|
||
vert = sorted(set(v for v in self.partition_values[:len(
|
||
self.partition_values) // 2] if v > 0))
|
||
horiz = sorted(set(v for v in self.partition_values[len(
|
||
self.partition_values) // 2:] if v > 0))
|
||
vertical_cuts = vert if vert else []
|
||
horizontal_cuts = horiz if horiz else []
|
||
|
||
# 边界：始终包含 0 和 1
|
||
self.col_cuts = [0.0] + vertical_cuts + [1.0]
|
||
self.row_cuts = [0.0] + horizontal_cuts + [1.0]
|
||
|
||
# 判断分区是否合理，并计算各个分区的任务卸载率ρ
|
||
valid_partition = True
|
||
for i in range(len(self.row_cuts) - 1):
|
||
for j in range(len(self.col_cuts) - 1):
|
||
d = (self.col_cuts[j+1] - self.col_cuts[j]) * self.W * \
|
||
(self.row_cuts[i+1] - self.row_cuts[i]) * self.H
|
||
rho_time_limit = (self.flight_time_factor - self.trans_time_factor) / \
|
||
(self.comp_time_factor - self.trans_time_factor)
|
||
rho_energy_limit = (self.battery_energy_capacity - self.flight_energy_factor * d - self.trans_energy_factor * d) / \
|
||
(self.comp_energy_factor * d -
|
||
self.trans_energy_factor * d)
|
||
if rho_energy_limit < 0:
|
||
valid_partition = False
|
||
break
|
||
rho = min(rho_time_limit, rho_energy_limit)
|
||
|
||
flight_time = self.flight_time_factor * d
|
||
bs_time = self.bs_time_factor * (1 - rho) * d
|
||
|
||
self.rectangles[(i, j)] = {
|
||
'center': ((self.row_cuts[i] + self.row_cuts[i+1]) * self.H / 2, (self.col_cuts[j+1] + self.col_cuts[j]) * self.W / 2),
|
||
'flight_time': flight_time,
|
||
'bs_time': bs_time,
|
||
'is_visited': False
|
||
}
|
||
if not valid_partition:
|
||
break
|
||
|
||
if not valid_partition:
|
||
reward = -10000
|
||
state = np.concatenate(
|
||
[[self.phase], self.partition_values, np.zeros(
|
||
(self.CUT_NUM // 2 + 1) ** 2, dtype=np.float32)]
|
||
)
|
||
return state, reward, True, False, {}
|
||
else:
|
||
# 初始化迷宫
|
||
self.phase = 1
|
||
reward = 10
|
||
|
||
# 构建反向索引，方便后续计算
|
||
self.reverse_rectangles = {
|
||
v['center']: k for k, v in self.rectangles.items()}
|
||
|
||
# 阶段 1：初始化迷宫，让多个车辆从区域中心出发，前往最近的几个区域中心点
|
||
region_centers = [
|
||
(i, j, self.rectangles[(i, j)]['center'])
|
||
for i in range(len(self.row_cuts) - 1)
|
||
for j in range(len(self.col_cuts) - 1)
|
||
]
|
||
# 按照与区域中心的距离从近到远排序
|
||
region_centers.sort(
|
||
key=lambda x: math.dist(x[2], (self.H / 2, self.W / 2))
|
||
)
|
||
|
||
# 分配最近的区域给每辆车
|
||
for idx in range(self.num_cars):
|
||
i, j, center = region_centers[idx]
|
||
self.car_pos[idx] = center
|
||
self.car_traj[idx].append((i, j))
|
||
self.rectangles[(i, j)]['is_visited'] = True
|
||
|
||
# 进入阶段 2：走迷宫
|
||
self.phase = 2
|
||
|
||
# 构造访问状态向量
|
||
max_regions = (self.CUT_NUM // 2 + 1) ** 2
|
||
visit_status = np.zeros(max_regions, dtype=np.float32)
|
||
|
||
# 将实际区域的访问状态填入向量
|
||
for i in range(len(self.row_cuts) - 1):
|
||
for j in range(len(self.col_cuts) - 1):
|
||
idx = i * (len(self.col_cuts) - 1) + j
|
||
visit_status[idx] = float(
|
||
self.rectangles[(i, j)]['is_visited'])
|
||
for i in range(idx + 1, max_regions):
|
||
visit_status[i] = 100
|
||
state = np.concatenate(
|
||
[[self.phase], self.partition_values, visit_status])
|
||
return state, reward, False, False, {}
|
||
|
||
elif self.phase == 2:
|
||
# 阶段 2：路径规划（走迷宫）
|
||
reward = 0
|
||
|
||
# 后 4 个动作对应上下左右移动
|
||
current_car = self.current_car_index
|
||
current_row, current_col = self.reverse_rectangles[self.car_pos[current_car]]
|
||
# 初始化新的行、列为当前值
|
||
new_row, new_col = current_row, current_col
|
||
|
||
if action == 10 and current_row > 0: # 上
|
||
new_row = current_row - 1
|
||
elif action == 11 and current_row < len(self.row_cuts) - 2: # 下
|
||
new_row = current_row + 1
|
||
elif action == 12 and current_col > 0: # 左
|
||
new_col = current_col - 1
|
||
elif action == 13 and current_col < len(self.col_cuts) - 2: # 右
|
||
new_col = new_col + 1
|
||
else:
|
||
# 无效动作，保持原地
|
||
pass
|
||
# 检查是否移动
|
||
car_moved = (new_row != current_row or new_col != current_col)
|
||
# 更新车辆位置
|
||
self.car_pos[current_car] = self.rectangles[(
|
||
new_row, new_col)]['center']
|
||
if car_moved:
|
||
self.car_traj[current_car].append((new_row, new_col))
|
||
# 更新访问标记：将新网格标记为已访问
|
||
self.rectangles[(new_row, new_col)]['is_visited'] = True
|
||
|
||
# 记录所有车辆一轮中是否移动
|
||
if self.current_car_index == 0:
|
||
# 如果是新一轮的开始，初始化移动标记
|
||
self.cars_moved = [False] * self.num_cars
|
||
self.cars_moved[current_car] = car_moved
|
||
# 如果一轮结束，检查是否所有车辆都没有移动
|
||
if self.current_car_index == (self.num_cars - 1) and not any(self.cars_moved):
|
||
reward -= 10 # 扣除 10 分奖励
|
||
|
||
self.step_count += 1
|
||
self.current_car_index = (
|
||
self.current_car_index + 1) % self.num_cars
|
||
max_regions = (self.CUT_NUM // 2 + 1) ** 2
|
||
visit_status = np.zeros(max_regions, dtype=np.float32)
|
||
|
||
# 将实际区域的访问状态填入向量
|
||
for i in range(len(self.row_cuts) - 1):
|
||
for j in range(len(self.col_cuts) - 1):
|
||
idx = i * (len(self.col_cuts) - 1) + j
|
||
visit_status[idx] = float(
|
||
self.rectangles[(i, j)]['is_visited'])
|
||
for i in range(idx + 1, max_regions):
|
||
visit_status[i] = 100
|
||
state = np.concatenate(
|
||
[[self.phase], self.partition_values, visit_status])
|
||
|
||
# Episode 终止条件：所有网格均被访问或步数达到上限
|
||
done = all([value['is_visited'] for _, value in self.rectangles.items()]) or (
|
||
self.step_count >= self.MAX_STEPS)
|
||
if done and all([value['is_visited'] for _, value in self.rectangles.items()]):
|
||
# 区域覆盖完毕，根据轨迹计算各车队的执行时间
|
||
T = max([self._compute_motorcade_time(idx)
|
||
for idx in range(self.num_cars)])
|
||
# print(T)
|
||
# print(self.partition_values)
|
||
# print(self.car_traj)
|
||
reward += self.BASE_LINE / T * 1000
|
||
# print(reward)
|
||
elif done and self.step_count >= self.MAX_STEPS:
|
||
reward += -1000
|
||
|
||
return state, reward, done, False, {}
|
||
|
||
def _compute_motorcade_time(self, idx):
|
||
flight_time = sum(self.rectangles[tuple(point)]['flight_time']
|
||
for point in self.car_traj[idx])
|
||
bs_time = sum(self.rectangles[tuple(point)]['bs_time']
|
||
for point in self.car_traj[idx])
|
||
|
||
# 计算车的移动时间，首先在轨迹的首尾添加上大区域中心
|
||
car_time = 0
|
||
for i in range(len(self.car_traj[idx]) - 1):
|
||
first_point = self.car_traj[idx][i]
|
||
second_point = self.car_traj[idx][i + 1]
|
||
car_time += math.dist(self.rectangles[first_point]['center'], self.rectangles[second_point]['center']) * \
|
||
self.car_time_factor
|
||
car_time += math.dist(self.rectangles[self.car_traj[idx][0]]['center'], [
|
||
self.H / 2, self.W / 2]) * self.car_time_factor
|
||
car_time += math.dist(self.rectangles[self.car_traj[idx][-1]]['center'], [
|
||
self.H / 2, self.W / 2]) * self.car_time_factor
|
||
|
||
return max(float(car_time) + flight_time, bs_time)
|
||
|
||
def render(self):
|
||
if self.phase == 1:
|
||
print("Phase 1: Initialize maze environment.")
|
||
print(f"Partition values so far: {self.partition_values}")
|
||
print(f"Motorcade positon: {self.car_pos}")
|
||
# input('1111')
|
||
elif self.phase == 2:
|
||
print("Phase 2: Play maze.")
|
||
print(f'Motorcade trajectory: {self.car_traj}')
|
||
# input('2222')
|