1.2.2

下载 Notebook 下载完整案例

带时间窗的取送货问题 (PDPTW)¶

类别： 路径优化

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

在带时间窗的取送货问题 (PDPTW) 中，一组具有相同容量的配送车辆必须根据客户的需求和营业时间收取并交付物品。客户是成对出现的：在每对中，一个客户对应一个取货点（需求为正），另一个对应一个送货点（需求为负）。每个客户必须恰好由一辆车服务。此外，成对客户必须由同一辆卡车服务，且每个取货点必须在其关联的送货点之前被处理。车辆从同一个仓库出发并返回该仓库，且其载荷在路径的任何时刻都不得超过其容量。目标包括最小化车队规模以及总行驶距离。

建模要点¶

• 添加 list 决策变量 来建模卡车的客户访问序列
• 使用 find 算子确保成对客户由同一辆卡车服务
• 使用 递归 lambda 函数 定义一个数组来计算卡车随时间的载荷以及客户的访问时间

数据¶

我们提供的带时间窗的取送货问题 (PDPTW) 实例来自 Li & Lim 基准。数据文件的格式如下：

• 第一行给出车辆数量、容量以及速度（未使用）

• 从第二行开始，对每个客户（从仓库开始）：

• 客户索引

• x 坐标

• y 坐标

• 需求

• 最早到达时间

• 最晚到达时间

• 服务时间

• 对应取货订单的索引（若该订单为取货则为 0）

• 对应送货订单的索引（若该订单为送货则为 0）

建模思路¶

带时间窗的取送货问题 (PDPTW) 的 OptAgent 模型是 带时间窗的容量受限车辆路径问题 (CVRPTW) 模型的扩展。因此，对于问题的路径规划和时间窗方面，我们请读者参考该模型。

成对客户必须由同一辆卡车访问。使用 find 算子，我们分别检索包含每对取货点和送货点的两个 list 变量。然后我们可以将这两个 list 约束为相同。此外，必须先取货再送货。使用 index 算子，我们确保取货点在 list 中位于送货点之前。

每辆卡车的载荷沿路径变化：取货时增加，送货时减少。我们使用 递归数组 计算卡车随时间的载荷：访问客户后的载荷等于访问上一客户后的载荷加上当前客户的需求（取货时为正，送货时为负）。然后我们使用可变参数 ‘and’ 算子确保路径上任何时刻都满足容量约束。

最后，目标与 带时间窗的容量受限车辆路径问题 (CVRPTW) 相同：我们最小化总延迟、使用的卡车数量以及总行驶距离。

Python 实现¶

from pathlib import Path
import math

from optagent import OptModel, solve


def read_elem(filename):
    return Path(filename).read_text(encoding="utf-8").split()


class _OptAgentRunner:
    def __init__(self, time_limit):
        self.model = OptModel()
        self.time_limit = time_limit

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        return False

    def solve(self):
        return solve(self.model, time_limit_s=float(self.time_limit))


def main(instance_file, output_file=None, time_limit=20):
    #
    # Read instance data
    #
    nb_customers, nb_trucks, truck_capacity, dist_matrix_data, dist_depot_data, \
        demands_data, service_time_data, earliest_start_data, latest_end_data, \
        pick_up_index, delivery_index, max_horizon = read_input_pdptw(instance_file)

    with _OptAgentRunner(time_limit) as optimizer:
        #
        # Declare the optimization model
        #
        model = optimizer.model

        # Sequence of customers visited by each truck
        customers_sequences = [
            model.list(nb_customers)
            for k in range(nb_trucks)
        ]

        # All customers must be visited by exactly one truck
        model.constraint(model.partition(customers_sequences))

        # /Create Hexaly arrays to be able to access them with "at" operators
        demands = model.array(demands_data)
        earliest = model.array(earliest_start_data)
        latest = model.array(latest_end_data)
        service_time = model.array(service_time_data)
        dist_matrix = model.array(dist_matrix_data)
        dist_depot = model.array(dist_depot_data)

        dist_routes = [None] * nb_trucks
        end_time = [None] * nb_trucks
        home_lateness = [None] * nb_trucks
        lateness = [None] * nb_trucks

        # A truck is used if it visits at least one customer
        trucks_used = [(model.count(customers_sequences[k]) > 0) for k in range(nb_trucks)]
        nb_trucks_used = model.sum(trucks_used)

        # Pickups and deliveries
        customers_sequences_array = model.array(customers_sequences)
        for i in range(nb_customers):
            if pick_up_index[i] == -1:
                pick_up_list_index = model.find(customers_sequences_array, i)
                delivery_list_index = model.find(customers_sequences_array, delivery_index[i])
                model.constraint(pick_up_list_index == delivery_list_index)
                pick_up_list = model.at(customers_sequences_array, pick_up_list_index)
                delivery_list = model.at(customers_sequences_array, delivery_list_index)
                model.constraint(model.index(pick_up_list, i) < model.index(delivery_list, delivery_index[i]))

        for k in range(nb_trucks):
            sequence = customers_sequences[k]
            c = model.count(sequence)

            # The quantity needed in each route must not exceed the truck capacity at any
            # point in the sequence
            demand_lambda = model.lambda_function(
                lambda i, prev: prev + demands[sequence[i // 1]])
            route_quantity = model.array(model.range(0, c), demand_lambda, 0)

            quantity_lambda = model.lambda_function(
                lambda i: route_quantity[i // 1] <= truck_capacity)
            model.constraint(model.and_(model.range(0, c), quantity_lambda))

            # Distance traveled by each truck
            dist_lambda = model.lambda_function(
                lambda i: model.at(
                    dist_matrix, sequence[(i - 1) // 1], sequence[i // 1]
                ))
            dist_routes[k] = model.sum(model.range(1, c), dist_lambda) \
                + model.iif(c > 0, dist_depot[sequence[0]] + dist_depot[sequence[c - 1]], 0)

            # End of each visit
            end_lambda = model.lambda_function(
                lambda i, prev:
                    model.max(
                        earliest[sequence[i // 1]],
                        model.iif(
                            i == 0,
                            dist_depot[sequence[0]],
                            prev + model.at(
                                dist_matrix, sequence[(i - 1) // 1], sequence[i // 1]
                            )))
                    + service_time[sequence[i // 1]])

            end_time[k] = model.array(model.range(0, c), end_lambda, 0)

            # Arriving home after max_horizon
            home_lateness[k] = model.iif(
                trucks_used[k],
                model.max(
                    0,
                    end_time[k][c - 1] + dist_depot[sequence[c - 1]] - max_horizon),
                0)

            # Completing visit after latest_end
            late_selector = model.lambda_function(
                lambda i: model.max(
                    0, end_time[k][i // 1] - latest[sequence[i // 1]]
                ))
            lateness[k] = home_lateness[k] + model.sum(model.range(0, c), late_selector)

        # Total lateness (must be 0 for the solution to be valid)
        total_lateness = model.sum(lateness)

        # Total distance traveled
        total_distance = model.div(model.round(100 * model.sum(dist_routes)), 100)

        # Objective: minimize the number of trucks used, then minimize the distance traveled
        model.minimize(total_lateness)
        model.minimize(nb_trucks_used)
        model.minimize(total_distance)

        solution = optimizer.solve()
        if not solution.feasible:
            print(f"No feasible schedule found; Status = {solution.feasible}")
            return solution

        #
        # Write the solution in a file with the following format:
        #  - number of trucks used and total distance
        #  - for each truck the customers visited (omitting the start/end at the depot)
        #
        if output_file is not None:
            with open(output_file, 'w', encoding="utf-8") as f:
                f.write("%d %.2f\n" % (nb_trucks_used.value, total_distance.value))
                for k in range(nb_trucks):
                    if trucks_used[k].value != 1:
                        continue
                    # Values in sequence are in 0...nbCustomers. +2 is to put it back in
                    # 2...nbCustomers+2 as in the data files (1 being the depot)
                    for customer in customers_sequences[k].value:
                        f.write("%d " % (customer + 1))
                    f.write("\n")
        print(
            f"Trucks used = {nb_trucks_used.value}; "
            f"Distance = {total_distance.value}; Status = {solution.feasible}"
        )
        return solution


# The input files follow the "Li & Lim" format
def read_input_pdptw(filename):
    file_it = iter(read_elem(filename))

    nb_trucks = int(next(file_it))
    truck_capacity = int(next(file_it))
    next(file_it)

    next(file_it)

    depot_x = int(next(file_it))
    depot_y = int(next(file_it))

    for i in range(2):
        next(file_it)

    max_horizon = int(next(file_it))

    for i in range(3):
        next(file_it)

    customers_x = []
    customers_y = []
    demands = []
    earliest_start = []
    latest_end = []
    service_time = []
    pick_up_index = []
    delivery_index = []

    while True:
        val = next(file_it, None)
        if val is None:
            break
        i = int(val) - 1
        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))
        pick = int(next(file_it))
        delivery = 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)
        pick_up_index.append(pick - 1)
        delivery_index.append(delivery - 1)

    nb_customers = i + 1

    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_trucks, truck_capacity, distance_matrix, distance_depots, \
            demands, service_time, earliest_start, latest_end, pick_up_index, \
            delivery_index, max_horizon


# Compute 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


# Compute the distances to the 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))

本地运行¶

在 notebook 所在目录执行以下 cell，即可调用一个小型取送货时间窗实例。

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_pdptw = main(INSTANCE_DIR / "lr101.txt", time_limit=1)