数据¶
所提供的带时间窗车辆路径问题 (CVRPTW) 实例来自 Solomon 实例。数据文件的格式如下:
第一行:实例的名称
第五行:车辆数以及它们的共同载重能力
从第十行开始,每一行描述一位客户(从配送中心开始):
客户的编号
x 坐标
y 坐标
需求量
最早到达时间
最晚到达时间
服务时间
建模思路¶
带时间窗容量受限车辆路径问题 (CVRPTW) 的 OptAgent 模型是 CVRP 模型的一种扩展。路径规划部分使用每辆卡车的 list 决策变量、容量约束以及闭环总行驶距离。
由于时间窗较难严格满足,我们将其作为第一优先级目标处理,而非硬约束。如果卡车早于营业时间到达,则需等待客户开门;若晚于最晚到达时间到达,则度量并惩罚其延迟量。问题的第一个目标即为最小化所有客户的总延迟量。当该累计延迟量为零时,相应的解被认为是可行的。
我们使用递归数组计算每辆卡车对每位客户的访问结束时间。对每辆卡车和每位客户,到达时间取以下两者的最大值:
- 上一次访问的结束时间加上行驶时间(在所提供的实例中行驶时间即等于距离)。对于第一次访问,则是自配送中心出发的行驶时间(假设卡车在时刻 0 离开配送中心)。
- 该客户允许的最早到达时间。
结束时间即为该到达时间加上对该客户的服务时间。路径结束时返回配送中心的时刻为最后一次访问的结束时间加上从该处返回配送中心的行驶时间。根据这些结束时间,我们可以方便地计算累计延迟量。
最后,我们按字典序最小化总延迟量、所用车辆数以及总行驶距离。
Python 实现¶
import math
from pathlib import Path
from optagent import OptModel, solve
def read_elements(filename):
return Path(filename).read_text(encoding="utf-8").split()
def main(instance_file, output_file=None, time_limit=20):
data = read_input_cvrptw(instance_file)
nb_customers = data["nb_customers"]
nb_trucks = data["nb_trucks"]
truck_capacity = data["truck_capacity"]
max_horizon = data["max_horizon"]
model = OptModel()
customer_sequences = [model.list(nb_customers) for truck in range(nb_trucks)]
model.constraint(model.partition(customer_sequences))
demands = model.array(data["demands"])
earliest = model.array(data["earliest_start"])
latest = model.array(data["latest_end"])
service_time = model.array(data["service_time"])
distance_matrix = model.array(data["distance_matrix"])
distance_depot = model.array(data["distance_depots"])
trucks_used = [model.count(sequence) > 0 for sequence in customer_sequences]
nb_trucks_used = model.sum(trucks_used)
route_distances = []
route_lateness = []
for truck, sequence in enumerate(customer_sequences):
count = model.count(sequence)
demand_lambda = model.lambda_function(lambda customer: demands[customer // 1])
route_quantity = model.sum(sequence, demand_lambda)
model.constraint(
route_quantity <= truck_capacity,
)
distance_lambda = model.lambda_function(
lambda position: distance_matrix[sequence[(position - 1) // 1], sequence[position // 1]]
)
route_distances.append(
model.sum(model.range(1, count), distance_lambda)
+ model.iif(
count > 0,
distance_depot[sequence[0]] + distance_depot[sequence[(count - 1) // 1]],
0,
)
)
end_time_lambda = model.lambda_function(
lambda position, previous: model.max(
earliest[sequence[position // 1]],
model.iif(
position == 0,
distance_depot[sequence[0]],
previous
+ distance_matrix[
sequence[(position - 1) // 1],
sequence[position // 1],
],
),
)
+ service_time[sequence[position // 1]]
)
end_times = model.array(model.range(0, count), end_time_lambda, 0)
home_lateness = model.iif(
count > 0,
model.max(
0,
end_times[(count - 1) // 1] + distance_depot[sequence[(count - 1) // 1]] - max_horizon,
),
0,
)
late_lambda = model.lambda_function(
lambda position: model.max(
0,
end_times[position // 1] - latest[sequence[position // 1]],
)
)
route_lateness.append(home_lateness + model.sum(model.range(0, count), late_lambda))
total_lateness = model.sum(route_lateness)
total_distance = model.round(100 * model.sum(route_distances)) / 100
model.minimize(total_lateness)
model.minimize(nb_trucks_used)
model.minimize(total_distance)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible routing solution found; Status = {solution.status}")
return solution
routes = [list(sequence.value) for sequence in customer_sequences]
result_lines = [f"{int(nb_trucks_used.value)} {int(total_distance.value)}"]
result_lines.extend(" ".join(str(customer + 1) for customer in route) + " " for route in routes if route)
result_text = "\n".join(result_lines) + "\n"
print(f"Total lateness = {total_lateness.value}; Status = {solution.status}")
print(result_text, end="")
if output_file is not None:
Path(output_file).write_text(result_text, encoding="utf-8")
return solution
# The input files follow the "Solomon" format
def read_input_cvrptw(filename):
file_it = iter(read_elements(filename))
for _ in range(4):
next(file_it)
nb_trucks = int(next(file_it))
truck_capacity = int(next(file_it))
for _ in range(13):
next(file_it)
depot_x = int(next(file_it))
depot_y = int(next(file_it))
for _ in range(2):
next(file_it)
max_horizon = int(next(file_it))
next(file_it)
customers_x = []
customers_y = []
demands = []
earliest_start = []
latest_end = []
service_time = []
while True:
val = next(file_it, None)
if val is None:
break
customers_x.append(int(next(file_it)))
customers_y.append(int(next(file_it)))
demands.append(int(next(file_it)))
ready = int(next(file_it))
due = int(next(file_it))
stime = int(next(file_it))
earliest_start.append(ready)
# in input files due date is meant as latest start time
latest_end.append(due + stime)
service_time.append(stime)
nb_customers = len(customers_x)
# Compute distance matrix
distance_matrix = compute_distance_matrix(customers_x, customers_y)
distance_depots = compute_distance_depots(depot_x, depot_y, customers_x, customers_y)
return {
"nb_customers": nb_customers,
"nb_trucks": nb_trucks,
"truck_capacity": truck_capacity,
"distance_matrix": distance_matrix,
"distance_depots": distance_depots,
"demands": demands,
"service_time": service_time,
"earliest_start": earliest_start,
"latest_end": latest_end,
"max_horizon": max_horizon,
}
# Computes the distance matrix
def compute_distance_matrix(customers_x, customers_y):
nb_customers = len(customers_x)
distance_matrix = [[None for i in range(nb_customers)] for j in range(nb_customers)]
for i in range(nb_customers):
distance_matrix[i][i] = 0
for j in range(nb_customers):
dist = compute_dist(customers_x[i], customers_x[j], customers_y[i], customers_y[j])
distance_matrix[i][j] = dist
distance_matrix[j][i] = dist
return distance_matrix
# Computes the distances to depot
def compute_distance_depots(depot_x, depot_y, customers_x, customers_y):
nb_customers = len(customers_x)
distance_depots = [None] * nb_customers
for i in range(nb_customers):
dist = compute_dist(depot_x, customers_x[i], depot_y, customers_y[i])
distance_depots[i] = dist
return distance_depots
def compute_dist(xi, xj, yi, yj):
return math.sqrt(math.pow(xi - xj, 2) + math.pow(yi - yj, 2))
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_c101_25 = main(INSTANCE_DIR / "C101.25.txt", time_limit=1)