HPCC2025/GA/sa_finetune.py

import random
import math
import json
import yaml


class SA_FineTuner:
    def __init__(self, initial_row_cuts, initial_col_cuts, car_paths, max_iterations=1000, initial_temp=100, cooling_rate=0.95):
        """
        初始化模拟退火算法
        :param initial_row_cuts: 初始行切分比例列表，例如 [0.1, 0.3, 0.7]
        :param initial_col_cuts: 初始列切分比例列表，例如 [0.2, 0.5, 0.8]
        :param T_function: 目标函数，用于计算切分比例的目标值 T
        :param max_iterations: 最大迭代次数
        :param initial_temp: 初始温度
        :param cooling_rate: 温度下降速率
        """
        # 读取参数
        with open('params3.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.current_row_cuts = initial_row_cuts[:]
        self.current_col_cuts = initial_col_cuts[:]
        self.car_paths = car_paths
        self.current_T = self.T_function(
            self.current_row_cuts, self.current_col_cuts)  # 初始目标值

        self.best_row_cuts = self.current_row_cuts[:]
        self.best_col_cuts = self.current_col_cuts[:]
        self.best_T = self.current_T

        # 模拟退火相关参数
        self.max_iterations = max_iterations
        self.initial_temp = initial_temp
        self.cooling_rate = cooling_rate
        self.temperature = initial_temp

    def fine_tune(self, cuts):
        """
        微调切分比例
        :param cuts: 当前切分比例列表，例如 [0.1, 0.3, 0.7]
        :return: 微调后的切分比例列表
        """
        new_cuts = cuts[:]
        if not new_cuts:
            return new_cuts

        # 随机选择一个切分点进行微调
        index = random.randint(1, len(new_cuts) - 2)    # 左闭右闭！！！
        adjustment = random.choice([-0.01, 0.01])  # 固定步长调整
        new_cuts[index] += adjustment

        # 确保切分比例合法
        if new_cuts[index] >= new_cuts[index + 1]:
            new_cuts[index] = new_cuts[index + 1] - 0.01
        elif new_cuts[index] <= new_cuts[index - 1]:
            new_cuts[index] = new_cuts[index - 1] + 0.01
        else:
            pass
        new_cuts[index] = max(0.01, min(0.99, new_cuts[index]))

        return new_cuts

    def accept_solution(self, current_T, new_T, temperature):
        """
        判断是否接受新解
        :param current_T: 当前解的目标值
        :param new_T: 新解的目标值
        :param temperature: 当前温度
        :return: 是否接受新解（True/False）
        """
        if new_T < current_T:
            return True
        else:
            # 根据模拟退火的概率公式接受较差解
            delta = new_T - current_T
            acceptance_probability = math.exp(-delta / temperature)
            return random.random() < acceptance_probability

    def T_function(self, row_cuts, col_cuts):
        """
        计算切分比例的目标值 T（占位函数）
        :param row_cuts: 行切分比例
        :param col_cuts: 列切分比例
        :return: 目标值 T
        """
        rectangles = []
        for i in range(len(row_cuts) - 1):
            for j in range(len(col_cuts) - 1):
                d = (col_cuts[j+1] - col_cuts[j]) * self.W * \
                    (row_cuts[i+1] - 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:
                    return float('inf')
                rho = min(rho_time_limit, rho_energy_limit)

                flight_time = self.flight_time_factor * d
                bs_time = self.bs_time_factor * (1 - rho) * d

                rectangles.append({
                    'flight_time': flight_time,
                    'bs_time': bs_time,
                    'center': ((row_cuts[i] + row_cuts[i+1]) / 2.0 * self.H,
                               (col_cuts[j] + col_cuts[j+1]) / 2.0 * self.W)
                })

        mortorcade_time_lt = []
        for idx in range(self.num_cars):
            car_path = self.car_paths[idx]

            flight_time = sum(rectangles[point]['flight_time']
                              for point in car_path)
            bs_time = sum(rectangles[point]['bs_time'] for point in car_path)

            car_time = 0
            for i in range(len(car_path) - 1):
                first_point = car_path[i]
                second_point = car_path[i + 1]
                car_time += math.dist(
                    rectangles[first_point]['center'], rectangles[second_point]['center']) * self.car_time_factor
            car_time += math.dist(rectangles[0]['center'],
                                  [self.H / 2, self.W / 2]) * self.car_time_factor
            car_time += math.dist(rectangles[-1]['center'],
                                  [self.H / 2, self.W / 2]) * self.car_time_factor
            mortorcade_time_lt.append(max(car_time + flight_time, bs_time))

        return max(mortorcade_time_lt)

    def run(self):
        """
        运行模拟退火算法
        :return: 最优行切分比例、最优列切分比例、最小目标值 T
        """
        for iteration in range(self.max_iterations):
            # 随机选择行或列进行微调
            if random.random() < 0.5:
                new_row_cuts = self.fine_tune(self.current_row_cuts)
                new_col_cuts = self.current_col_cuts[:]
            else:
                new_row_cuts = self.current_row_cuts[:]
                new_col_cuts = self.fine_tune(self.current_col_cuts)

            # 计算新解的目标值
            new_T = self.T_function(new_row_cuts, new_col_cuts)

            # 判断是否接受新解
            if self.accept_solution(self.current_T, new_T, self.temperature):
                self.current_row_cuts = new_row_cuts
                self.current_col_cuts = new_col_cuts
                self.current_T = new_T

                # 更新最优解
                if new_T < self.best_T:
                    self.best_row_cuts = new_row_cuts
                    self.best_col_cuts = new_col_cuts
                    self.best_T = new_T

            # 降低温度
            self.temperature *= self.cooling_rate

            # 打印进度（可选）
            if iteration % 100 == 0:
                print(
                    f"Iteration {iteration}: Best T = {self.best_T}, Temperature = {self.temperature}")

        return self.best_row_cuts, self.best_col_cuts, self.best_T


def load_initial_solution(file_path):
    """
    从 JSON 文件加载初始解
    :param file_path: JSON 文件路径
    :return: 行切分比例、列切分比例
    """
    with open(file_path, 'r', encoding='utf-8') as file:
        data = json.load(file)
    row_cuts = data['row_boundaries']
    col_cuts = data['col_boundaries']
    car_paths = data['car_paths']
    return row_cuts, col_cuts, car_paths


# 示例调用
if __name__ == "__main__":
    random.seed(42)

    # ---------------------------
    # 需要修改的超参数
    # ---------------------------
    file_path = r"solutions\trav_ga_params2_parallel.json"
    max_iterations=10000
    initial_temp=100
    cooling_rate=0.95

    initial_row_cuts, initial_col_cuts, car_paths = load_initial_solution(
        file_path)

    sa = SA_FineTuner(initial_row_cuts, initial_col_cuts, car_paths, max_iterations, initial_temp, cooling_rate)
    best_row_cuts, best_col_cuts, best_T = sa.run()

    # 输出结果
    print("Best row cuts:", best_row_cuts)
    print("Best col cuts:", best_col_cuts)
    print("Best T:", best_T)