HPCC2025/env.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 = 3500.0     # 基准时间，通过greedy或者蒙特卡洛计算出来

        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₂

        # 定义动作空间：全部动作均为 1 维连续 [0,1]
        self.action_space = spaces.Box(
            low=0.0, high=1.0, shape=(1,), dtype=np.float32)

        # 定义观察空间为8维向量
        # TODO 返回的状态目前只有位置坐标
        # 阶段 0 状态：前 4 维表示已决策的切分值（未决策部分为 0）
        # 阶段 1 状态：车辆位置 (2D)
        self.observation_space = spaces.Box(
            low=0.0, high=1.0, shape=(self.CUT_NUM + 2 * self.num_cars,), dtype=np.float32)

        # 切分阶段相关变量
        self.col_cuts = []     # 存储竖切位置（c₁, c₂），当值为0时表示不切
        self.row_cuts = []   # 存储横切位置（r₁, r₂）

        self.init_maze_step = 0

        # 路径规划阶段相关变量
        self.MAX_STEPS = 50         # 迷宫走法步数上限
        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
        state = np.concatenate(
            [self.partition_values, np.zeros(np.array(self.car_pos).flatten().shape[0], dtype=np.float32)])
        return state

    def step(self, action):
        # 在所有阶段动作均为 1 维连续动作，取 action[0]
        a = float(action[0])

        if self.phase == 0:
            # 切分阶段：每一步输出一个标量，离散化为 {0, 0.1, ..., 0.9}
            disc_val = np.floor(a * 10) / 10.0
            disc_val = np.clip(disc_val, 0.0, 0.9)
            self.partition_values[self.partition_step] = disc_val
            self.partition_step += 1

            # 构造当前状态：前 partition_step 个为已决策值，其余为 0，再补 7 个 0
            state = np.concatenate(
                [self.partition_values, np.zeros(np.array(self.car_pos).flatten().shape[0], 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.partition_values, np.zeros(np.array(self.car_pos).flatten().shape[0], dtype=np.float32)])
                    return state, reward, True, False, {}
                else:
                    # 进入阶段 1：初始化迷宫
                    self.phase = 1
                    state = np.concatenate(
                        [self.partition_values, np.array(self.car_pos).flatten()])
                    reward = 10
                    
                    # 构建反向索引，方便后续计算
                    self.reverse_rectangles = {v['center']: k for k, v in self.rectangles.items()}
                    return state, reward, False, False, {}

        elif self.phase == 1:
            # 阶段 1：初始化迷宫，让多个车辆从区域中心出发，前往划分区域的中心点
            # 确保 action 的值在 [0, 1]，然后映射到 0~(num_regions-1) 的索引
            num_regions = (len(self.col_cuts) - 1) * \
                (len(self.row_cuts) - 1)
            target_region_index = int(np.floor(a * num_regions))
            target_region_index = np.clip(
                target_region_index, 0, num_regions - 1)
            # 将index映射到笛卡尔坐标
            coord = (target_region_index // (len(self.col_cuts) - 1),
                     target_region_index % (len(self.col_cuts) - 1))
            self.car_pos[self.init_maze_step] = self.rectangles[coord]['center']
            self.car_traj[self.init_maze_step].append(coord)
            self.rectangles[coord]['is_visited'] = True

            # 计数
            self.init_maze_step += 1
            state = np.concatenate(
                [self.partition_values, np.array(self.car_pos).flatten()])
            if self.init_maze_step < self.num_cars:
                return state, 0.0, False, False, {}
            else:
                # 进入阶段 2：走迷宫
                self.phase = 2
                return state, 0.0, False, False, {}

        elif self.phase == 2:
            # 阶段 2：路径规划（走迷宫）
            current_car = self.current_car_index
            # 查表，找出当前车辆所在的网格
            current_row, current_col = self.reverse_rectangles[self.car_pos[current_car]]

            # 当前动作 a 为 1 维连续动作，映射到四个方向
            if a < 0.2:
                move_dir = 'up'
            elif a < 0.4:
                move_dir = 'down'
            elif a < 0.6:
                move_dir = 'left'
            elif a < 0.8:
                move_dir = 'right'
            else:
                move_dir = 'stay'

            # 初始化新的行、列为当前值
            new_row, new_col = current_row, current_col

            if move_dir == 'up' and current_row > 0:
                new_row = current_row - 1
            elif move_dir == 'down' and current_row < len(self.row_cuts) - 2:
                new_row = current_row + 1
            elif move_dir == 'left' and current_col > 0:
                new_col = current_col - 1
            elif move_dir == 'right' and current_col < len(self.col_cuts) - 2:
                new_col = current_col + 1
            # 如果移动不合法，或者动作为stay，则保持原位置
            # TODO 移动不合法，加一些惩罚

            # 更新车辆位置
            self.car_pos[current_car] = self.rectangles[(
                new_row, new_col)]['center']
            if new_row != current_row or new_col != current_col:
                self.car_traj[current_car].append((new_row, new_col))
            self.step_count += 1
            self.current_car_index = (
                self.current_car_index + 1) % self.num_cars

            # 更新访问标记：将新网格标记为已访问
            self.rectangles[(new_row, new_col)]['is_visited'] = True

            # 观察状态
            state = np.concatenate(
                [self.partition_values, np.array(self.car_pos).flatten()])
            reward = 0

            # 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 * 100
            elif done and self.step_count >= self.MAX_STEPS:
                reward += -10000

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