HPCC2025/env_dis.py

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')