HPCC2025/mtkl_sovler.py
2025-03-09 16:53:01 +08:00

307 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import random
import math
import matplotlib.pyplot as plt
import matplotlib.patches as patches
# 固定随机种子,便于复现
random.seed(42)
# ---------------------------
# 参数设置
# ---------------------------
H = 20 # 区域高度网格点之间的距离为25m单位距离
W = 25 # 区域宽度
k = 1 # 系统数量(车-巢-机系统个数)
num_iterations = 1000000 # 蒙特卡洛模拟迭代次数
# 时间系数(单位:秒,每个网格一张照片)
flight_time_factor = 3 # 每张照片对应的飞行时间无人机飞行速度为9.5m/s拍摄照片的时间间隔为3s
comp_uav_factor = 5 # 无人机上每张照片计算时间5s
trans_time_factor = 0.3 # 每张照片传输时间0.3s
car_move_time_factor = 2 * 50 # TODO 汽车每单位距离的移动时间2s加了一个放大因子
comp_bs_factor = 5 # 机巢上每张照片计算时间
# 其他参数
flight_energy_factor = 0.05 # 单位:分钟/张
comp_energy_factor = 0.05 # 计算能耗需要重新估计
trans_energy_factor = 0.0025
battery_capacity = 10 # 无人机只进行飞行续航为30分钟
# bs_energy_factor =
# car_energy_factor =
# ---------------------------
# 蒙特卡洛模拟,寻找最佳方案
# ---------------------------
best_T = float('inf')
best_solution = None
for iteration in range(num_iterations):
# 随机生成分区的行分段数与列分段数
R = random.randint(1, 5) # 行分段数
C = random.randint(1, 5) # 列分段数
# 生成随机的行、列分割边界
row_boundaries = sorted(random.sample(range(1, H), R - 1))
row_boundaries = [0] + row_boundaries + [H]
col_boundaries = sorted(random.sample(range(1, W), C - 1))
col_boundaries = [0] + col_boundaries + [W]
# ---------------------------
# 根据分割边界生成所有矩形任务
# ---------------------------
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) # 任务的照片数量(矩形面积)
# # 每个任务随机生成卸载比率 ρ ∈ [0,1]
# rho = random.random()
# # rho = 0.1
# # 计算各个阶段时间
# flight_time = flight_time_factor * d
# comp_time = comp_uav_factor * rho * d
# trans_time = trans_time_factor * (1 - rho) * d
# comp_bs_time = comp_bs_factor * (1 - rho) * d
# # 检查无人机电池约束:
# # 飞行+计算+传输能耗需不超过电池容量
# flight_energy = flight_energy_factor * d
# comp_energy = comp_energy_factor * rho * d
# trans_energy = trans_energy_factor * (1 - rho) * d
# total_uav_energy = flight_energy + comp_energy + trans_energy
# # 无人机计算与传输时间不超过飞行时间
# if (total_uav_energy > battery_capacity) or (comp_time + trans_time > flight_time):
# # TODO 时间约束的rho上界是个常数0.57如果区域划分定了rho直接取上界即可可以数学证明
# valid_partition = False
# break
# 求解rho
rho_time_limit = (flight_time_factor - trans_time_factor) / \
(comp_uav_factor - trans_time_factor)
rho_energy_limit = (battery_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)
print(rho)
flight_time = flight_time_factor * d
comp_time = comp_uav_factor * rho * d
trans_time = trans_time_factor * (1 - rho) * d
comp_bs_time = comp_bs_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,
'comp_bs_time': comp_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.hypot(tasks[0]['center'][0] - region_center[0],
tasks[0]['center'][1] - region_center[1]) * car_move_time_factor
# 依次经过任务中心
for j in range(1, len(tasks)):
prev_center = tasks[j - 1]['center']
curr_center = tasks[j]['center']
car_time += math.hypot(curr_center[0] - prev_center[0],
curr_center[1] - prev_center[1]) * car_move_time_factor
else:
car_time = 0
# 机巢的计算时间
total_bs_time = sum(task['comp_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
}
# ---------------------------
# 输出最佳方案
# ---------------------------
if best_solution is not None:
print("最佳 T (各系统中最长的完成时间):", best_solution['T_max'])
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}")
else:
print("在给定的模拟次数内未找到满足所有约束的方案。")
# 在输出最佳方案后添加详细信息
if best_solution is not None:
print("\n各系统详细信息:")
region_center = (H / 2.0, W / 2.0)
for system_id, tasks in best_solution['system_tasks'].items():
print(f"\n系统 {system_id} 的任务详情:")
# 按距离区域中心的距离排序任务
tasks_sorted = sorted(tasks, key=lambda r: math.hypot(r['center'][0] - region_center[0],
r['center'][1] - region_center[1]))
if tasks_sorted:
print(
f"轨迹路线: 区域中心({region_center[0]:.1f}, {region_center[1]:.1f})", end="")
current_pos = region_center
total_car_time = 0
total_flight_time = 0
total_flight_energy = 0
total_comp_energy = 0
total_trans_energy = 0
for i, task in enumerate(tasks_sorted, 1):
# 计算车辆移动时间
car_time = math.hypot(task['center'][0] - current_pos[0],
task['center'][1] - current_pos[1]) * car_move_time_factor
total_car_time += car_time
# 更新当前位置
current_pos = task['center']
print(
f" -> 任务{i}({current_pos[0]:.1f}, {current_pos[1]:.1f})", end="")
# 累加各项数据
total_flight_time += task['flight_time']
total_flight_energy += flight_energy_factor * task['d']
total_comp_energy += comp_energy_factor * \
task['rho'] * task['d']
total_trans_energy += trans_energy_factor * \
(1 - task['rho']) * task['d']
print("\n")
print(f"任务数量: {len(tasks_sorted)}")
print(f"车辆总移动时间: {total_car_time:.2f}")
print(f"无人机总飞行时间: {total_flight_time:.2f}")
print(f"能耗统计:")
print(f" - 飞行能耗: {total_flight_energy:.2f} 分钟")
print(f" - 计算能耗: {total_comp_energy:.2f} 分钟")
print(f" - 传输能耗: {total_trans_energy:.2f} 分钟")
print(
f" - 总能耗: {(total_flight_energy + total_comp_energy + total_trans_energy):.2f} 分钟")
print("\n各任务详细信息:")
for i, task in enumerate(tasks_sorted, 1):
print(f"\n任务{i}:")
print(
f" 位置: ({task['center'][0]:.1f}, {task['center'][1]:.1f})")
print(f" 照片数量: {task['d']}")
print(f" 卸载比率(ρ): {task['rho']:.2f}")
print(f" 飞行时间: {task['flight_time']:.2f}")
print(f" 计算时间: {task['comp_time']:.2f}")
print(f" 传输时间: {task['trans_time']:.2f}")
print(f" -- 飞行能耗: {task['d'] * flight_energy_factor:.2f} 分钟")
print(f" -- 计算能耗: {task['d'] * comp_energy_factor:.2f} 分钟")
print(f" -- 传输能耗: {task['d'] * trans_energy_factor:.2f} 分钟")
print(f" 基站计算时间: {task['comp_bs_time']:.2f}")
else:
print("该系统没有分配任务")
print("-" * 50)
if best_solution is not None:
plt.rcParams['font.family'] = ['sans-serif']
plt.rcParams['font.sans-serif'] = ['SimHei']
fig, ax = plt.subplots()
ax.set_xlim(0, W)
ax.set_ylim(0, H)
ax.set_title("区域划分与车-机-巢系统覆盖")
ax.set_xlabel("区域宽度")
ax.set_ylabel("区域高度")
# 定义若干颜色以区分不同系统系统编号从0开始
colors = ['red', 'blue', 'green', 'orange', 'purple', 'cyan', 'magenta']
# 绘制区域中心
region_center = (W / 2.0, H / 2.0) # 注意x对应宽度y对应高度
ax.plot(region_center[0], region_center[1],
'ko', markersize=8, label="区域中心")
# 绘制每个任务区域(矩形)及在矩形中心标注系统编号与卸载比率 ρ
for system_id, tasks in best_solution['system_tasks'].items():
# 重新按车辆行驶顺序排序(启发式:以任务中心距离区域中心的距离排序)
tasks_sorted = sorted(tasks, key=lambda task: math.hypot(
(task['c1'] + (task['c2'] - task['c1']) / 2.0) - region_center[0],
(task['r1'] + (task['r2'] - task['r1']) / 2.0) - region_center[1]
))
for i, task in enumerate(tasks_sorted, 1):
# 绘制矩形:左下角坐标为 (c1, r1),宽度为 (c2 - c1),高度为 (r2 - r1)
rect = patches.Rectangle((task['c1'], task['r1']),
task['c2'] - task['c1'],
task['r2'] - task['r1'],
linewidth=2,
edgecolor=colors[system_id % len(colors)],
facecolor='none')
ax.add_patch(rect)
# 计算矩形中心
center_x = task['c1'] + (task['c2'] - task['c1']) / 2.0
center_y = task['r1'] + (task['r2'] - task['r1']) / 2.0
# 在矩形中心标注:系统编号、执行顺序和卸载比率 ρ
ax.text(center_x, center_y, f"S{system_id}-{i}\nρ={task['rho']:.2f}",
color=colors[system_id % len(colors)],
ha='center', va='center', fontsize=10, fontweight='bold')
# 添加图例
ax.legend()
# 反转 y 轴使得行号从上到下递增(如需,可取消)
ax.invert_yaxis()
plt.show()
else:
print("没有找到满足约束条件的方案,无法进行可视化。")