HPCC2025/mtkl_sovler.py

import random
import math
import yaml
import json
import numpy as np

# 固定随机种子，便于复现
random.seed(42)

num_iterations = 10000

# ---------------------------
# 参数设置
# ---------------------------
with open('params.yml', 'r', encoding='utf-8') as file:
    params = yaml.safe_load(file)

H = params['H']
W = params['W']
k = params['num_cars']

flight_time_factor = params['flight_time_factor']
comp_time_factor = params['comp_time_factor']
trans_time_factor = params['trans_time_factor']
car_time_factor = params['car_time_factor']
bs_time_factor = params['bs_time_factor']

flight_energy_factor = params['flight_energy_factor']
comp_energy_factor = params['comp_energy_factor']
trans_energy_factor = params['trans_energy_factor']
battery_energy_capacity = params['battery_energy_capacity']

# ---------------------------
# 蒙特卡洛模拟，寻找最佳方案
# ---------------------------
best_T = float('inf')
best_solution = None

for iteration in range(num_iterations):
    # 随机生成分区的行分段数与列分段数
    R = random.randint(0, 5)  # 行分段数
    C = random.randint(0, 5)  # 列分段数

    # 生成随机的行、列分割边界
    horiz = [np.clip(np.floor(random.random() * 10) /10, 0.0, 0.9) for _ in range(R)]
    horiz = sorted(set(horiz))
    horiz = horiz if horiz else []
    row_boundaries = [0] + horiz + [1]
    row_boundaries = [boundary * H for boundary in row_boundaries]

    vert = [np.clip(np.floor(random.random() * 10) /10, 0.0, 0.9) for _ in range(C)]
    vert = sorted(set(vert))
    vert = vert if vert else []
    col_boundaries = [0] + vert + [1]
    col_boundaries = [boundary * W for boundary in col_boundaries]

    # ---------------------------
    # 根据分割边界生成所有矩形任务
    # ---------------------------
    rectangles = []
    valid_partition = True  # 标记此分区是否满足所有约束
    for i in range(len(row_boundaries) - 1):
        for j in range(len(col_boundaries) - 1):
            r1 = row_boundaries[i]
            r2 = row_boundaries[i + 1]
            c1 = col_boundaries[j]
            c2 = col_boundaries[j + 1]
            d = (r2 - r1) * (c2 - c1)  # 任务的照片数量（矩形面积）

            # 求解rho
            rho_time_limit = (flight_time_factor - trans_time_factor) / \
                (comp_time_factor - trans_time_factor)
            rho_energy_limit = (battery_energy_capacity - flight_energy_factor * d - trans_energy_factor * d) / \
                (comp_energy_factor * d - trans_energy_factor * d)
            if rho_energy_limit < 0:
                valid_partition = False
                break
            rho = min(rho_time_limit, rho_energy_limit)
            flight_time = flight_time_factor * d
            comp_time = comp_time_factor * rho * d
            trans_time = trans_time_factor * (1 - rho) * d
            bs_time = bs_time_factor * (1 - rho) * d

            # 计算任务矩形中心，用于后续车辆移动时间计算
            center_r = (r1 + r2) / 2.0
            center_c = (c1 + c2) / 2.0

            rectangles.append({
                'r1': r1, 'r2': r2, 'c1': c1, 'c2': c2,
                'd': d,
                'rho': rho,
                'flight_time': flight_time,
                'comp_time': comp_time,
                'trans_time': trans_time,
                'bs_time': bs_time,
                'center': (center_r, center_c)
            })
        if not valid_partition:
            break

    # 如果分区中存在任务不满足电池约束，则跳过该分区
    if not valid_partition:
        continue

    # ---------------------------
    # 随机将所有矩形任务分配给 k 个系统（车-机-巢）
    # ---------------------------
    system_tasks = {i: [] for i in range(k)}
    for rect in rectangles:
        system = random.randint(0, k - 1)
        system_tasks[system].append(rect)

    # ---------------------------
    # 对于每个系统，计算该系统的总完成时间 T_k：
    # T_k = 所有任务的飞行时间之和 + 车辆的移动时间
    # 车辆移动时间：车辆从区域中心出发，依次经过各任务中心（顺序采用距离区域中心的启发式排序）
    # ---------------------------
    region_center = (H / 2.0, W / 2.0)
    T_k_list = []
    for i in range(k):
        tasks = system_tasks[i]
        tasks.sort(key=lambda r: math.hypot(r['center'][0] - region_center[0],
                                            r['center'][1] - region_center[1]))
        total_flight_time = sum(task['flight_time'] for task in tasks)
        if tasks:
            # 车辆从区域中心到第一个任务中心
            car_time = math.dist(tasks[0]['center'],
                                 region_center) * car_time_factor
            # 依次经过任务中心
            for j in range(len(tasks) - 1):
                prev_center = tasks[j]['center']
                curr_center = tasks[j + 1]['center']
                car_time += math.dist(curr_center,
                                      prev_center) * car_time_factor
            # 回到区域中心
            car_time += math.dist(region_center, curr_center) * car_time_factor
        else:
            car_time = 0

        # 机巢的计算时间
        total_bs_time = sum(task['bs_time'] for task in tasks)
        T_k = max(total_flight_time + car_time, total_bs_time)
        T_k_list.append(T_k)

    T_max = max(T_k_list)  # 整体目标 T 为各系统中最大的 T_k

    # TODO 没有限制系统的总能耗

    if T_max < best_T:
        best_T = T_max
        best_solution = {
            'system_tasks': system_tasks,
            'T_k_list': T_k_list,
            'T_max': T_max,
            'iteration': iteration,
            'R': R,
            'C': C,
            'row_boundaries': row_boundaries,
            'col_boundaries': col_boundaries,
            'car_time': car_time,
            'flight_time': total_flight_time,
            'bs_time': total_bs_time
        }

# ---------------------------
# 输出最佳方案
# ---------------------------
if best_solution is not None:
    print("最佳 T (各系统中最长的完成时间):", best_solution['T_max'])
    print(best_solution['iteration'], "次模拟后找到最佳方案:")
    print("分区情况:")
    print("行分段数:", best_solution['R'])
    print("列分段数:", best_solution['C'])
    print("行分割边界:", best_solution['row_boundaries'])
    print("列分割边界:", best_solution['col_boundaries'])
    print("每辆车的运行轨迹情况:")
    car_paths = {}
    for i in range(k):
        num_tasks = len(best_solution['system_tasks'][i])
        print(
            f"系统 {i}: 完成时间 T = {best_solution['T_k_list'][i]}, 飞行任务数量: {num_tasks}")
        tasks = best_solution['system_tasks'][i]
        tasks.sort(key=lambda r: math.hypot(r['center'][0] - region_center[0],
                                            r['center'][1] - region_center[1]))
        if tasks:
            print(
                f"轨迹路线: 区域中心({region_center[0]:.1f}, {region_center[1]:.1f})", end="")
            current_pos = region_center
            car_path = []
            for j, task in enumerate(tasks, 1):
                current_pos = task['center']
                car_path.append(current_pos)
                print(
                    f" -> 任务{j}({current_pos[0]:.1f}, {current_pos[1]:.1f})", end="")
            print(" -> 区域中心")
            car_paths[i] = car_path

    # 保存分区边界和车辆轨迹到JSON文件
    output_data = {
        'row_boundaries': [boundary / H for boundary in best_solution['row_boundaries']],
        'col_boundaries': [boundary / W for boundary in best_solution['col_boundaries']],
        'car_paths': car_paths
    }
    with open('./solutions/best_solution_mtkl.json', 'w', encoding='utf-8') as f:
        json.dump(output_data, f, ensure_ascii=False, indent=4)
else:
    print("在给定的模拟次数内未找到满足所有约束的方案。")