|
这次 我们用ALNS来做 MDHVRPTW。 ALNS的理论请先参考其他文章。
把启发式算法之前几次的题 稍为改一改：
根据猜测， 虽然每条线路之间的时间 越小越好，但这个耗时 一部分是共线 路程距离的，所以部分程度上可以忽略。 ?
而车辆数越少这个，好像改为 车辆使用费用（指空载满载时的耗油 过路费等使用车时产生的费用） 更合适
如果货物点因为同时被 时间、车辆数二者限制 ?而得不到被运送，应该得到惩罚。现在假设车辆数 总会是够的，但是记录了车辆费用，用的车越多 自然越不好。这样绕过了 得不到运送的情况。
所以 最终目标 ?就是费用越小越好。
import random
import pickle
from collections import namedtuple, defaultdict
from itertools import product
from functools import wraps
import pandas as pd
import numpy as np
from scipy.spatial.distance import pdist, squareform
import matplotlib.pyplot as plt
# ***** begin 配置信息 *****
# **** 题目数据配置
generate_new_flag = True
# depos_no = 4 # 为简便起见 放在正方形四顶点位置 不可配置
max_car_no = 999
cargo_site_total = 80
site_location_range = tuple(range(-100, 101, 2))
depos_location_edge_ratio = 1 / 8
site_time_range = [(480, 720), (720, 1080), (1080, 1260)] # 8:00~12:00 12:00~18:00 18:00~21:00
car_carrying_range = (100, 150, 200) # 小 中 大车的载重
car_empty_cost = (0.7, 0.8, 0.9) # 小 中 大车 空载时 每公里费用 (载重更多 每公里费用更小)
every_weight_cost = (0.15, 0.125, 0.1)
car_speeds = (1.08, 1.05, 1.0)
move_time_ratio = 0.5 # 装卸货时间比例,与货物重量 相乘得时间
max_cargo_weight = 100
distance_weight = 0.5
car_cost_weight = 1 - distance_weight
# **** 算法配置
max_iter = 5000
clear_adap_iter = 1000 # 每迭代多少次，清理一下自适应信息
volatil_factor = 0.4 # 挥发速度
path_destroy_min = 1
path_destroy_max = 3
max_try_times = 200 # 设得越大 单次搜索时间加大 但搜到feasible solution可能性也越大 太小了可能 单次总是搜不到feasible solution
accept_scores = (1.5, 1.2, 0.8, 0.5)
# **** 画图配置
# plot_flag = False # windows下运行 可设为True
# ***** end 配置信息 *****
# 配置计算 及 其他定义
path_destroy_range = list(range(path_destroy_min, path_destroy_max + 1))
car_type_ix_dict = {"S": 0, "M": 1, "L": 2}
bigger_car_type = {"S": ("M", "L"), "M": ("L",), "L": tuple()}
deposite_info = namedtuple("deposite_info", ("location_x", "location_y", "car_list",
"start_time", "end_time"))
cargo_site_info = namedtuple("cargo_site_info", ("location_x", "location_y", "cargo_weight",
"start_time", "end_time"))
# 正常情况 应该把Utils写成一个文件或文件夹 而不是一个类..
class Utils:
@staticmethod
def pickle_dump(data, write_file):
pickle_file = open(write_file, 'wb')
pickle.dump(data, pickle_file)
pickle_file.close()
return None
@staticmethod
def pickle_load(read_file):
pickle_file = open(read_file, 'rb')
result = pickle.load(pickle_file)
pickle_file.close()
return result
def while_max_times(retry_times=50, reach_max_return=None):
def retry_decorator(func):
@wraps(func)
def wrapper_func(*args, **kwargs):
count = 0
while count < retry_times:
ret = func(*args, **kwargs)
if not ret:
count += 1
if count >= retry_times:
return reach_max_return
continue
else:
return ret
return wrapper_func
return retry_decorator
class Car:
def __init__(self, type='M'):
self.type = type
if self.type == 'S': # small
self.type_ix = 0
self.max_carriage = car_carrying_range[0]
self.empty_cost = car_empty_cost[0]
elif self.type == 'M': # middle
self.type_ix = 1
self.max_carriage = car_carrying_range[1]
self.empty_cost = car_empty_cost[1]
else:
self.type_ix = 2
self.max_carriage = car_carrying_range[-1]
self.empty_cost = car_empty_cost[-1]
self.load_weight = 0
# self.car_consume = 0
@property
def loading_weight(self):
return self.load_weight
@loading_weight.setter
def loading_weight(self, newadd_cargo_weight):
self.load_weight += newadd_cargo_weight
class Solution:
def __init__(self, vrp_data_ins, route_info=(), score=None):
self.vrp_data = vrp_data_ins
if not route_info:
self.get_a_possible_solution()
else:
if not self.route_info_auth_ok(route_info):
raise ValueError("the route_info is not feasible!")
self.route_info = route_info
if score:
self.obj = score
else:
self.obj = self.cal_score()
@classmethod
def cp_from_ins(cls, sol_ins):
new_sol_ins = cls(sol_ins.vrp_data, sol_ins.route_info, sol_ins.obj)
return new_sol_ins
def cal_score(self):
consumes = []
for dep_ix, paths in self.route_info:
consumes.append(self.cal_paths_score(paths, dep_ix))
return sum(consumes)
def cal_paths_score(self, paths, dep_ix):
consume_paths = []
for path in paths:
consume_paths.append(self.cal_path_score(path, dep_ix))
return sum(consume_paths)
def cal_path_score(self, path, dep_ix):
weights = 0
consume = 0
last_ix = 0
car_type_ix = car_type_ix_dict[path[-1]]
only_cargo_path = path[:-1]
for ix, cargo_site_ix in enumerate(only_cargo_path):
if ix == 0:
dis = self.vrp_data.cargo_site_dep_dis[(cargo_site_ix, dep_ix)]
consume = car_empty_cost[car_type_ix] * dis
else:
if ix == len(only_cargo_path) - 1:
dis = self.vrp_data.cargo_site_dep_dis[(cargo_site_ix, dep_ix)]
else:
dis = self.vrp_data.cargo_site_dis[(last_ix, cargo_site_ix)]
weights += self.vrp_data.cargo_site_dict[cargo_site_ix].cargo_weight
consume += weights * every_weight_cost[car_type_ix] * dis
last_ix = cargo_site_ix
return consume
def cal_add_time(self, cargo_ix, handle_time, last_cargo_ix, car_type='', car=None):
if last_cargo_ix:
if not car_type:
if not car:
raise ValueError("the car_type and car can not be empty simultaneously!")
car_type = car.type
car_speed = car_speeds[car_type_ix_dict[car_type]]
drive_time = self.vrp_data.cargo_site_dis[(last_cargo_ix, cargo_ix)] / car_speed
return handle_time + drive_time
return handle_time
def get_a_possible_solution(self):
dep_2_cargo_sites = self.vrp_data.get_dep_2_cargo_sites()
route_info = []
for dep_ix, car_id_list in dep_2_cargo_sites.items():
# todo
car_dis_list = [(self.vrp_data.cargo_site_dict[cargo_ix],
self.vrp_data.cargo_site_dep_dis[(cargo_ix, dep_ix)], cargo_ix)
for cargo_ix in car_id_list]
# car_dis_list.sort(key=lambda car_dis: (car_dis[0].start_time, car_dis[0].cargo_weight, car_dis[1]))
car_dis_list.sort(key=lambda car_dis: (car_dis[0].start_time, car_dis[1], car_dis[0].cargo_weight))
# 总是假设 车辆可以在最早时间到达第一个货物点
elapse_time = 0
# path_dis = sum(x[1] for x in car_dis_list)
paths = []
path = []
car = None
last_cargo_ix = 0
for ix, (cargo_site, dis, cargo_ix) in enumerate(car_dis_list):
if not car:
car = Car('M')
elapse_time = 0
last_cargo_ix = 0
# print(cargo_ix, cargo_site.cargo_weight, car.loading_weight)
handle_time = round(move_time_ratio * cargo_site.cargo_weight, 1)
if elapse_time + handle_time < cargo_site.end_time:
if car.loading_weight + cargo_site.cargo_weight < car.max_carriage:
car.loading_weight = cargo_site.cargo_weight
path.append(cargo_ix)
elapse_time += self.cal_add_time(cargo_ix, handle_time, last_cargo_ix, car=car)
last_cargo_ix = cargo_ix
else:
path.append(car.type)
paths.append(tuple(path))
path = [cargo_ix]
car = Car('M')
car.loading_weight = cargo_site.cargo_weight
elapse_time = handle_time
if ix == len(car_dis_list) - 1:
path.append(car.type)
paths.append(tuple(path))
else:
path.append(car.type)
paths.append(tuple(path))
path = [cargo_ix]
car = Car('M')
car.loading_weight = cargo_site.cargo_weight
elapse_time = handle_time
# paths 样例： [[6, 26, 1, 44, 68, 50, 'L'], [20, 12, 22, 52, 36, 67, 'L'], [7, 0, 30, 'L']]
route_info.append((dep_ix, tuple(paths)))
self.route_info = tuple(route_info)
self.obj = self.cal_score()
return None
def path_auth_ok(self, path, car_type='', only_load=False, only_tw=False):
# 返回值 (整体鉴权结果, 载重鉴权结果, 时间窗鉴权结果) 默认为False
if only_tw:
tw_result = self.path_auth_tw_ok(path, car_type)
return tw_result, False, tw_result
load_result = self.path_auth_load_ok(path, car_type)
if load_result:
if only_load:
return True, True, False
tw_result = self.path_auth_tw_ok(path, car_type)
return tw_result, load_result, tw_result
return False, False, False
def get_only_car_path_and_car_type(self, path, car_type=''):
# car_type 如果为空 则从path最后一项取
# path 可能的值为 (50, 20, 12, 'L') 或 (50, 20, 12)
if not car_type:
car_type = path[-1]
only_cargo_path = path[:-1]
else:
only_cargo_path = path
return only_cargo_path, car_type
def path_auth_tw_ok(self, path, car_type=''):
# 主要检查 time_window是否OK
only_cargo_path, car_type = self.get_only_car_path_and_car_type(path, car_type)
elapse_time = 0
last_cargo_ix = 0
for cargo_ix in only_cargo_path:
cur_handle_time = round(move_time_ratio * self.vrp_data.cargo_site_dict[cargo_ix].cargo_weight, 1)
if elapse_time + cur_handle_time > self.vrp_data.cargo_site_dict[cargo_ix].end_time:
return False
elapse_time += self.cal_add_time(cargo_ix, cur_handle_time, last_cargo_ix, car_type)
last_cargo_ix = cargo_ix
return True
def path_auth_load_ok(self, path, car_type=''):
# 主要检查 车辆载重 是否OK
only_cargo_path, car_type = self.get_only_car_path_and_car_type(path, car_type)
car_max_carriage = Car(car_type).max_carriage
if sum([self.vrp_data.cargo_site_dict[x].cargo_weight for x in only_cargo_path]) > car_max_carriage:
return False
return True
def paths_auth_ok(self, paths):
# 载重鉴权的计算量小 故先鉴权载重 再鉴权时间窗
for path in paths:
if not self.path_auth_load_ok(path):
print(" path load auth failed! %s " % (path,)) # 元组直接打印有问题 改成(path,)
return False
for path in paths:
if not self.path_auth_tw_ok(path):
print(" path time window auth failed! %s " % path)
return False
return True
def route_info_auth_ok(self, route_info, depos_no=4):
if len(route_info) != depos_no:
print(" the length of route_info %s does not equal depos_no %s" % (len(route_info), depos_no))
return False
temp_cargo_sites = [cargo_site for x in route_info for path in x[1] for cargo_site in path[:-1]]
if (len(temp_cargo_sites) != len(self.vrp_data.cargo_site_dict) or
len(temp_cargo_sites) != len(set(temp_cargo_sites))):
print(" the length of cargo_sites %s is not correct! cargo_site is %s, cargo_sites set is %s"
% (len(temp_cargo_sites), len(self.vrp_data.cargo_site_dict), len(set(temp_cargo_sites))))
return False
for dep_ix, paths in route_info:
if not self.paths_auth_ok(paths):
return False
return True
class QuestionDataHandle:
def __init__(self):
pass
def get_data(self, generate_new=False, pick_file='mdhvrptw_question.pick'):
if not generate_new:
try:
print("will get data by file!")
# 目前的实现 配置信息不能在导出后有变化。 如需优化 需要将题目数据配置信息 同时导出导入
return Utils.pickle_load(pick_file)
except Exception as e:
print("!! Attention ！！ can not load from vrp_question.pick. will generate new! \n")
# 报错后，继续用新生成的数据 当然 这里也可以报错后 退出处理
return self.generate_data_process(pick_file)
def generate_data_process(self, pick_file):
deposite_list, cargo_site_dict = self.generate_data()
Utils.pickle_dump((deposite_list, cargo_site_dict), pick_file)
return deposite_list, cargo_site_dict
def generate_data(self, depos_no=4):
while True:
deposite_list = self.get_deposite_list(depos_no)
cargo_site_list = [cargo_site_info(random.choice(site_location_range),
random.choice(site_location_range),
random.randint(0, max_cargo_weight),
*random.choice(site_time_range))
for _ in range(cargo_site_total)]
if self.auth(deposite_list, cargo_site_list):
break
return deposite_list, {ix: cargo_sit_info for ix, cargo_sit_info in enumerate(cargo_site_list)}
def auth(self, deposite_list, cargo_site_list):
# 偶尔有生成的坐标点 xy坐标都一样的情况，此时重新再生成
location_list = [(cargo_s.location_x, cargo_s.location_y)
for cargo_s in cargo_site_list]
location_list.extend([(dep.location_x, dep.location_y)
for dep in deposite_list])
if len(set(location_list)) == len(location_list):
return True
return False
def get_deposite_list(self, depos_no=4):
deposite_list = []
deposite_loc_range = self.get_deposite_location_range(depos_no)
# 这里简单地 平均分配 小中大三种车型
avg_type_no = int(max_car_no / 3)
car_list = [Car('S')] * avg_type_no + [Car('M')] * avg_type_no + [Car('L')] * avg_type_no
for dep_ix in range(depos_no):
deposite_list.append(deposite_info(deposite_loc_range[dep_ix][0],
deposite_loc_range[dep_ix][1],
car_list, 0, 1440))
return deposite_list
def get_deposite_location_range(self, depos_no=4):
# 剪掉过于旁边和中心的，让depsite 分布在三环左右(假设总共五环)
desire_loc = (site_location_range[-1] - site_location_range[0]) / depos_no
return list(map(lambda x: [y * desire_loc for y in x], product([1, -1], repeat=2)))
class VRPData:
def __init__(self, deposite_list, cargo_site_dict):
self.deposite_list = deposite_list
self.cargo_site_dict = cargo_site_dict
# self.deposite_location = [(dep.location_x, dep.location_y) for dep in deposite_list]
self.cargo_site_loc = [(cargo_site.location_x, cargo_site.location_y) for cargo_site in cargo_site_dict.values()]
self.cargo_site_dep_dis = {}
self.cargo_site_dis = {}
self.dis_cal()
def dis_cal(self):
for cargo_ix, cargo_site in self.cargo_site_dict.items():
for dep_ix, dep in enumerate(self.deposite_list):
self.cargo_site_dep_dis[(cargo_ix, dep_ix)] = np.linalg.norm(
np.array([dep.location_x, dep.location_y]) - np.array([cargo_site.location_x, cargo_site.location_y]))
distance_matrix = squareform(pdist(np.array(self.cargo_site_loc)))
for row_ix, row_dis_list in enumerate(distance_matrix):
for col_ix, col_dis in enumerate(row_dis_list):
self.cargo_site_dis[(row_ix, col_ix)] = col_dis
def get_dep_2_cargo_sites(self):
cargo_nearst_dep = {}
cargo_deps_dis = []
max_dep_ix = len(self.deposite_list) - 1
for (cargo_ix, dep_ix), distance in self.cargo_site_dep_dis.items():
cargo_deps_dis.append((dep_ix, distance))
if dep_ix == max_dep_ix: # cargo_ix对应一轮所有的dep数据 结束
cargo_deps_dis.sort(key=lambda x: x[1])
cargo_nearst_dep[cargo_ix] = cargo_deps_dis[0][0]
cargo_deps_dis = []
dep_2_cargo_sites = defaultdict(list)
[dep_2_cargo_sites[dep_ix].append(cargo_ix) for cargo_ix, dep_ix in cargo_nearst_dep.items()]
return dep_2_cargo_sites
class Alns:
def __init__(self, deposite_list, cargo_site_dict):
self.vrp_data = VRPData(deposite_list, cargo_site_dict)
self.solution = Solution(self.vrp_data)
self.best_solution = Solution.cp_from_ins(self.solution)
self.destroy_method = ["inner_path", "between_path", "by_car_type", "by_cargo_quant_type"]
self.repair_method = ["random", "greedy"]
self.clear_adaptive_info()
print("the initial solution score is %s and route info is %s \n" %
(self.best_solution.obj, [x[1] for x in self.solution.route_info]))
def clear_adaptive_info(self, temperature_max=1000):
self.destroy_weight = np.array([1.0 for _ in self.destroy_method])
self.destroy_score = np.array([1.0 for _ in self.destroy_method])
self.destroy_select_times = np.array([0.0 for _ in self.destroy_method])
self.repair_weight = np.array([1.0 for _ in self.repair_method])
self.repair_score = np.array([1.0 for _ in self.repair_method])
self.repair_select_times = np.array([0.0 for _ in self.repair_method])
self.temperature = temperature_max
print("reset the adaptive info finished ! ")
def __call__(self):
for ix in range(max_iter):
temp_solution, destroy_ix, repair_ix = self.destroy_and_repair_handle(self.solution)
accept, add_score = self.accept_handle(temp_solution)
if accept:
self.solution = Solution.cp_from_ins(temp_solution)
self.destroy_score[destroy_ix] += add_score
self.repair_score[repair_ix] += add_score
if temp_solution.obj <= self.best_solution.obj:
self.best_solution = Solution.cp_from_ins(temp_solution)
self.update_dr_weight(destroy_ix, repair_ix)
if ix % 10 == 0:
print("run times %s, the best score is: %s" % (ix, self.best_solution.obj))
if ix % clear_adap_iter == 0:
self.clear_adaptive_info()
print("Alns: the best score is: %s, and route_info is \n%s "
% (self.best_solution.obj, [x[1] for x in self.best_solution.route_info]))
print(self.best_solution.route_info_auth_ok(self.best_solution.route_info))
def accept_handle(self, temp_solution, scores=accept_scores, decay=0.99):
if temp_solution.obj <= self.best_solution.obj:
self.temperature *= decay
return True, scores[0]
elif temp_solution.obj <= self.solution.obj:
self.temperature *= decay
return True, scores[1]
else:
# the simulated annealing acceptance criteria
if np.random.rand() < np.exp(- temp_solution.obj) / self.temperature:
self.temperature *= decay
return True, scores[3]
else:
self.temperature *= decay
return False, scores[-1]
def destroy_and_repair_handle(self, solution):
destory_ix, destroy_method, repair_ix, repair_method = self.choose_and_update_select()
# print("tttttttt ", destroy_method, repair_method)
if destroy_method == "inner_path":
return self.destroy_repair_inner_path(solution, repair_method), destory_ix, repair_ix
if destroy_method == "between_path":
return self.destroy_repair_between_path(solution, repair_method), destory_ix, repair_ix
if destroy_method == "by_car_type":
return self.destroy_repair_by_car_type(solution, repair_method), destory_ix, repair_ix
if destroy_method == "by_cargo_quant_type":
return self.destroy_repair_by_cargo_quant_type(solution, repair_method), destory_ix, repair_ix
# return self.destroy_repair_inner_path(solution), 0, 0
# return self.destroy_repair_between_path(solution), 0, 0
# return self.destroy_repair_by_car_type(solution), 0, 0
# return self.destroy_repair_by_cargo_quant_type(solution), 0, 0
def choose_and_update_select(self):
destory_ix, destroy_method = self.choose_destroy_method()
repair_ix, repair_method = self.choose_repair_method()
self.destroy_select_times[destory_ix] += 1
self.repair_select_times[repair_ix] += 1
return destory_ix, destroy_method, repair_ix, repair_method
def choose_destroy_method(self):
probs = (self.destroy_weight / sum(self.destroy_weight)).cumsum()
index_need = np.where(probs > np.random.rand())[0][0]
return index_need, self.destroy_method[index_need]
def choose_repair_method(self):
probs = (self.repair_weight / sum(self.repair_weight)).cumsum()
index_need = np.where(probs > np.random.rand())[0][0]
return index_need, self.repair_method[index_need]
def destroy_repair_inner_path(self, solution, repair_method='greedy', paths_no_ratio=0.5):
# solution.route_info内为元组 无需copy
new_route_info = []
for dep_ix, paths in solution.route_info:
new_paths = []
# 选择一定数量的paths 做后续处理
if paths_no_ratio < 1:
paths_destroy_index = set(random.sample(range(len(paths)), int(len(paths) * paths_no_ratio)))
for path_ix, path in enumerate(paths):
# len(path) == 2 意味着只有一个货物点
if len(path) == 2 or (paths_no_ratio < 1 and (path_ix not in paths_destroy_index)):
new_paths.append(path)
continue
path_destroy_no = random.randint(path_destroy_min,
min(path_destroy_max, int((len(path) - 1) * 0.5), path_destroy_min))
if repair_method == 'random':
new_path = self.random_repair_inner_path(solution, path, path_destroy_no)
else:
new_path = self.greedy_repair_inner_path(solution, path, path_destroy_no)
new_paths.append(tuple(new_path))
new_route_info.append((dep_ix, tuple(new_paths)))
return Solution(self.vrp_data, tuple(new_route_info))
def random_destroy_between_path(self):
pass
def random_repair_inner_path(self, solution, path, path_destroy_no):
new_path = list(path[:-1])
car_type = path[-1]
@while_max_times(retry_times=max_try_times, reach_max_return=tuple())
def random_repair_inner_path_func():
remove_insert_index = random.sample(range(len(new_path)), path_destroy_no * 2)
last_ix = 0
for ix, ri_idx in enumerate(remove_insert_index):
if ix % 2 == 1: # 从0开始的序列，ix为奇数时 实际为第偶数个
new_path[last_ix], new_path[ri_idx] = new_path[ri_idx], new_path[last_ix]
else:
last_ix = ri_idx
if solution.path_auth_ok(new_path, car_type)[0]:
new_path.append(car_type)
return tuple(new_path)
return tuple()
return random_repair_inner_path_func()
def greedy_repair_inner_path(self, solution, path, path_destroy_no):
car_type = path[-1]
@while_max_times(retry_times=max_try_times, reach_max_return=tuple())
def greedy_repair_inner_path_func():
new_path = list(path[:-1])
remove_index = random.sample(range(len(new_path)), path_destroy_no)
for rmv_ix in remove_index:
distances = [(ix, self.vrp_data.cargo_site_dis[(rmv_ix, cargo_ix)])
for ix, cargo_ix in enumerate(new_path) if ix != rmv_ix]
insert_ix = min(distances, key=lambda x: x[1])[0]
new_path.insert(insert_ix, new_path.pop(rmv_ix))
if solution.path_auth_ok(new_path, car_type)[0]:
new_path.append(car_type)
return tuple(new_path)
return tuple()
return greedy_repair_inner_path_func()
def get_paths_remove_list(self, paths):
# 固定取一半paths中的cargo_site 插入到另外未被选中的paths中
path_destroy_index = set(random.sample(range(len(paths)), int(len(paths) * 0.5)))
path_remove_list = []
[path_remove_list.append((path_ix,
random.sample(path[:-1], min(path_destroy_max, int((len(path) - 1) * 0.5)))))
for path_ix, path in enumerate(paths) if path_ix in path_destroy_index]
return path_remove_list
def get_rmv_insert_info(self, paths):
path_remove_list = self.get_paths_remove_list(paths)
rmv_path_ixs = []
rmv_list = []
insert_path_ix = []
for rmv_path_ix, path_rmv_list in path_remove_list:
rmv_path_ixs.append(rmv_path_ix)
rmv_list.extend(path_rmv_list)
insert_path_ix.extend(
random.sample([x for x in range(len(paths)) if x != rmv_path_ix], len(path_rmv_list)))
return self.get_paths_after_rmv(paths, path_remove_list, rmv_path_ixs
), insert_path_ix, self.get_insert_info(insert_path_ix, rmv_list)
def get_paths_after_rmv(self, paths, path_remove_list, rmv_path_ixs):
paths_after_rmv = []
path_ix_rmv_dict = {path_ix: set(rmv_list) for path_ix, rmv_list in path_remove_list}
for path_ix, path in enumerate(paths):
if path_ix in path_ix_rmv_dict.keys():
paths_after_rmv.append(tuple([x for x in path if x not in path_ix_rmv_dict[path_ix]]))
else:
paths_after_rmv.append(path)
# print(" nnnnnnnnnnnn ", paths_after_rmv)
return paths_after_rmv
def get_insert_info(self, insert_path_ix, rmv_list):
insert_path_content = defaultdict(list)
{insert_path_content[path_ix].append(rmv) for path_ix, rmv in zip(insert_path_ix, rmv_list)}
return insert_path_content
def get_repair_new_path(self, solution, rmv_cargo_site_list, path, repair_method):
if repair_method == 'random':
new_path = self.random_repair_path(solution, rmv_cargo_site_list, path)
else:
new_path = self.greedy_repair_path(solution, rmv_cargo_site_list, path)
return new_path
def destroy_repair_between_path(self, solution, repair_method='greedy'):
# solution.route_info内为元组 无需copy
new_route_info = []
for dep_ix, paths in solution.route_info:
new_paths = []
count = 0
while count < max_try_times:
new_paths = []
paths_after_rmv, insert_path_ix, insert_path_content = self.get_rmv_insert_info(paths)
insert_path_ix_set = set(insert_path_ix)
for path_ix, path in enumerate(paths_after_rmv):
if path_ix not in insert_path_ix_set:
new_paths.append(path)
continue
new_path = self.get_repair_new_path(solution, insert_path_content[path_ix], path, repair_method)
if new_path:
new_paths.append(tuple(new_path))
else:
break
if len(new_paths) == len(paths):
break
count += 1
# if sum([len(path[:-1]) for path in paths]) != sum([len(path[:-1]) for path in new_paths]):
if len(new_paths) != len(paths):
new_paths = paths
# print("can not get a new paths after trying max times, will use the old one.")
new_route_info.append((dep_ix, tuple(new_paths)))
return Solution(self.vrp_data, tuple(new_route_info))
def get_bigger_type(self, car_type, method='random', solution=None, old_path=(), dep_ix=-1):
bigger_types = bigger_car_type[car_type]
if len(bigger_types) > 0:
if len(bigger_types) == 1:
return bigger_types[0]
else:
if method == "random":
return random.choice(bigger_types)
else:
scores_info = self.cal_new_car_type_score_info(solution, bigger_types, old_path, dep_ix)
if scores_info:
return scores_info[0][0]
return ''
def switch_path_order_by_tw(self, only_cargo_path):
tw_info = [(cargo_site_ix, self.vrp_data.cargo_site_dict[cargo_site_ix].start_time)
for cargo_site_ix in only_cargo_path]
tw_info.sort(key=lambda x: x[1])
tw_df = pd.DataFrame(tw_info, columns=["cargo_site_ix", "start_time"])
new_path = []
for s_time, group in tw_df.groupby("start_time"):
tmp = group["cargo_site_ix"].values
np.random.shuffle(tmp)
new_path.extend(tmp)
return new_path
def cal_new_car_type_score_info(self, solution, possible_types, old_path, dep_ix, auth_new_path=True):
scores = []
score_old = solution.cal_path_score(old_path, dep_ix)
for new_car_type in possible_types:
if (not auth_new_path) or solution.path_auth_ok(old_path[:-1], new_car_type, only_load=True)[0]:
score_new = solution.cal_path_score(list(old_path[:-1]) + [new_car_type], dep_ix)
if score_new < score_old:
scores.append((new_car_type, score_new))
if scores:
scores.sort(key=lambda x: x[1])
return scores
def random_repair_path(self, solution, to_insert_cargo_site_list, path, change_car_type=False, change_tw=False):
car_type = path[-1]
def get_new_path():
new_path = list(path[:-1])
[new_path.insert(random.choice(range(len(new_path) - 1) if len(new_path) != 1 else [0, 1]),
insert_cargo_site)
for insert_cargo_site in to_insert_cargo_site_list]
return new_path
@while_max_times(retry_times=max_try_times, reach_max_return=tuple())
def random_repair_path_func():
only_cargo_new_path = get_new_path()
result, load_result, tw_result = solution.path_auth_ok(only_cargo_new_path, car_type)
if result:
return tuple(only_cargo_new_path + [car_type])
elif change_car_type and (not load_result):
bigger_type = self.get_bigger_type(car_type)
if bigger_type and solution.path_auth_ok(only_cargo_new_path, bigger_type)[0]:
return tuple(only_cargo_new_path + [bigger_type])
elif change_tw and (not tw_result): # todo 优化
new_new_path = self.switch_path_order_by_tw(only_cargo_new_path)
if solution.path_auth_ok(new_new_path, car_type)[0]:
return tuple(new_new_path + [car_type])
return tuple()
return random_repair_path_func()
def greedy_repair_path(self, solution, to_insert_cargo_site_list, path, dep_ix=-1,
change_car_type=False, change_tw=False):
car_type = path[-1]
def get_new_path():
new_path = list(path[:-1])
for insert_cargo_site in to_insert_cargo_site_list:
distances = [(ix, self.vrp_data.cargo_site_dis[(insert_cargo_site, cargo_ix)])
for ix, cargo_ix in enumerate(new_path)]
insert_ix = min(distances, key=lambda x: x[1])[0]
new_path.insert(insert_ix, insert_cargo_site)
return new_path
@while_max_times(retry_times=max_try_times, reach_max_return=tuple())
def greedy_repair_path_func():
only_cargo_new_path = get_new_path()
result, load_result, tw_result = solution.path_auth_ok(only_cargo_new_path, car_type)
if result:
return tuple(only_cargo_new_path + [car_type])
elif change_car_type and (not load_result):
bigger_type = self.get_bigger_type(car_type, "greedy", solution,
only_cargo_new_path + [car_type], dep_ix)
if bigger_type and solution.path_auth_ok(only_cargo_new_path, bigger_type)[0]:
return tuple(only_cargo_new_path + [bigger_type])
elif change_tw and (not tw_result): # todo 优化
new_new_path = self.switch_path_order_by_tw(only_cargo_new_path)
if solution.path_auth_ok(new_new_path, car_type)[0]:
return tuple(new_new_path + [car_type])
return tuple()
return greedy_repair_path_func()
def destroy_repair_by_car_type(self, solution, repair_method='random'):
new_route_info = []
for dep_ix, paths in solution.route_info:
new_paths = []
for path_ix, path in enumerate(paths):
if repair_method == 'random':
new_path = self.random_repair_by_car_type(solution, path)
else:
new_path = self.greedy_repair_by_car_type(solution, path, dep_ix)
new_paths.append(tuple(new_path))
new_route_info.append((dep_ix, tuple(new_paths)))
if solution.route_info != tuple(new_route_info):
return Solution(self.vrp_data, tuple(new_route_info))
return solution
def random_repair_by_car_type(self, solution, old_path):
old_car_type = old_path[-1]
possible_car_type = [x for x in car_type_ix_dict.keys() if x != old_car_type]
new_car_type = random.choice(possible_car_type)
if solution.path_auth_ok(old_path[:-1], new_car_type, only_load=True)[0]:
return list(old_path[:-1]) + [new_car_type]
return old_path
def greedy_repair_by_car_type(self, solution, old_path, dep_ix):
old_car_type = old_path[-1]
possible_car_type = [x for x in car_type_ix_dict.keys() if x != old_car_type]
scores_info = self.cal_new_car_type_score_info(solution, possible_car_type, old_path, dep_ix)
if scores_info:
return list(old_path[:-1]) + [scores_info[0][0]]
return old_path
def destroy_repair_by_cargo_quant_type(self, solution=None, repair_method='random'):
# 路径货物点 少于2个或很多的情况 移到其他路径上去
new_route_info = []
for dep_ix, paths in solution.route_info:
new_paths = []
count = 0
while count < max_try_times:
new_paths = []
paths_after_rmv, insert_path_ix, insert_path_content = self.get_rmv_insert_cargo_quant(paths)
insert_path_ix_set = set(insert_path_ix)
for path_ix, path in enumerate(paths_after_rmv):
if path_ix not in insert_path_ix_set:
new_paths.append(path)
continue
new_path = self.get_repair_new_path_cargo_quant(solution, insert_path_content[path_ix],
path, repair_method, dep_ix)
if new_path:
new_paths.append(tuple(new_path))
else:
break
if len(new_paths) == len(paths):
break
count += 1
# if sum([len(path[:-1]) for path in paths]) != sum([len(path[:-1]) for path in new_paths]):
if len(new_paths) != len(paths):
new_paths = paths
# print("can not get a new paths after trying max times, will use the old one.")
new_route_info.append((dep_ix, tuple(new_paths)))
return Solution(self.vrp_data, tuple(new_route_info))
def get_repair_new_path_cargo_quant(self, solution, rmv_cargo_site_list, path, repair_method, dep_ix=-1):
if repair_method == 'random':
new_path = self.random_repair_path(solution, rmv_cargo_site_list, path,
change_car_type=True, change_tw=True)
else:
new_path = self.greedy_repair_path(solution, rmv_cargo_site_list, path, dep_ix,
change_car_type=True, change_tw=True)
return new_path
def get_paths_remove_cargo_quant(self, paths):
path_cargo_site_no = [(path_ix, len(path) - 1) for path_ix, path in enumerate(paths)]
path_cargo_site_no.sort(key=lambda x: x[1])
too_short_tobe_del = [x for x in path_cargo_site_no if x[1] <= 2]
too_long = path_cargo_site_no[-path_destroy_max:]
paths_destroy_index = set(random.sample([x[0] for x in too_short_tobe_del + too_long],
random.choice(path_destroy_range)))
path_remove_list = []
path_delete_list = []
for path_ix, path in enumerate(paths):
if path_ix in paths_destroy_index:
if len(path) <= 3: # 货物点数 <= 2..
path_delete_list.append(path_ix)
else:
path_remove_list.append((path_ix,
random.sample(path[:-1], random.choice(path_destroy_range))))
return path_remove_list, path_delete_list
def get_paths_after_rmv_cargo_quant(self, paths, path_remove_list, path_delete_list):
paths_after_rmv = []
path_ix_rmv_dict = {path_ix: set(rmv_list) for path_ix, rmv_list in path_remove_list}
after_rmv_can_insert = []
for path_ix, path in enumerate(paths):
if path_ix in path_ix_rmv_dict.keys():
tmp_path = tuple([x for x in path if x not in path_ix_rmv_dict[path_ix]])
if len(tmp_path) > 1:
paths_after_rmv.append(tmp_path)
after_rmv_can_insert.append(False)
elif path_ix not in path_delete_list:
paths_after_rmv.append(path)
after_rmv_can_insert.append(True)
# print(" nnnnnnnnnnnn ", any([len(x) <= 1 for x in paths_after_rmv]))
return paths_after_rmv, after_rmv_can_insert
def get_rmv_insert_cargo_quant(self, paths):
path_remove_list, path_delete_list = self.get_paths_remove_cargo_quant(paths)
paths_after_rmv, after_rmv_can_insert = self.get_paths_after_rmv_cargo_quant(paths, path_remove_list, path_delete_list)
can_insert_path = [ix for ix, flag in enumerate(after_rmv_can_insert) if flag]
rmv_list = []
insert_path_ix = []
for _, path_rmv_list in path_remove_list:
rmv_list.extend(path_rmv_list)
# todo 前面的同步？
if len(can_insert_path) > len(path_rmv_list):
insert_path_ix.extend(random.sample(can_insert_path, len(path_rmv_list)))
elif len(can_insert_path) == len(path_rmv_list):
insert_path_ix.extend(can_insert_path)
else:
multi, remain = divmod(len(path_rmv_list), len(can_insert_path))
tmp_list = can_insert_path * multi + random.sample(can_insert_path, remain)
insert_path_ix.extend(tmp_list)
return paths_after_rmv, insert_path_ix, self.get_insert_info(insert_path_ix, rmv_list)
def update_dr_weight(self, destroy_ix, repair_ix):
self.destroy_weight[destroy_ix] = volatil_factor * self.destroy_weight[destroy_ix] + (1 - volatil_factor) * (
self.destroy_score[destroy_ix] / self.destroy_select_times[destroy_ix])
self.repair_weight[repair_ix] = volatil_factor * self.repair_weight[repair_ix] + (1 - volatil_factor) * (
self.repair_score[repair_ix] / self.repair_select_times[repair_ix])
def run():
deposite_list, cargo_site_dict = QuestionDataHandle().get_data(generate_new=generate_new_flag)
Alns(deposite_list, cargo_site_dict)()
if __name__ == '__main__':
run()
自适应的过程 用到模拟退火思想。
里面用到了while循环，为了限制尝试次数过多 及代码简洁， ?用了@while_max_times装饰器
被破坏而未修复的middle_solution 个人觉得 可以不用显式存在 成这样：
middle_solution, destroy_ix = self.destroy(self.best_solution)
temp_solution, repair_ix = self.repair(middle_solution)
为了方便直接破坏和修改 (比如直接交换位置)我用的：
temp_solution, destroy_ix, repair_ix = self.destroy_and_repair_handle(self.solution)
修复的方式 除了random greedy 还有regret repair, 我这里暂时没用。也可以为每一种破坏方式单独搞一个修复权重
另外，在使用启发式搜索时，一定要注意运行效率，因为实际情况很可能是非常非常复杂的。一个小优化，在demo中省下来的只有一小点可忽略不计的性能开销，在实际问题中很可能会节省很多计算时间。
例如 在计算两两距离时，也可以很潇洒的用?
[[np.linalg.norm(np.array(p1) - np.array(p2)) for p2 in site_location] for p1 in site_location]，但是这个重复计算了一半的次数。我选用了 pdist, squareform? ?应该好一点；
每次根据cargo_site编号取cargo_site信息时，一般常想到的方法 可能是把所有的cargo_site放在一个cargo_site_list中，使用下标索引访问。我转成字典了，cargo_site_dict。这个增加的空间，应该是有限的，但频繁查找会快很多。
另外 用于加速的一些工具 如lru_cache ?numba.jit ? 也可以试一试
cargo_path 本来应该跟 car_type一一对应 两个list，使用时用zip成对读取。为了看着方便，把car_type放在最后了?
还有其他的待优化的
?
|