1.2.1

下载 Notebook 下载完整案例

时间相关带时间窗车辆路径问题 (TDCVRPTW)¶

类别： 路径优化

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

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

问题描述¶

在时间相关带时间窗的容量受限车辆路径问题 (TDCVRPTW) 中，一组具有相同载重能力的配送车辆必须为客户提供服务。各客户具有已知的营业时间以及对单一商品的需求。车辆从同一个配送中心出发并最终返回该配送中心。每位客户必须在其营业时间内由恰好一辆车服务，且每辆车服务的需求总量不得超过其载重能力。客户之间的行驶时间取决于出发时所处的时间段。优化目标依次是最小化延迟、所需车辆数以及总行驶距离。

本例将时间范围划分为清晨、早高峰、白天、晚高峰和夜间五个时段，每个时段使用独立的行驶时间矩阵。

学习要点¶

• 使用 OptAgent 的 list 决策变量表示各车辆的客户访问序列
• 使用递归 lambda 数组计算依赖交通时段的客户访问结束时间
• 使用按优先级排列的多目标，将时间窗延迟作为第一目标

数据¶

所提供的 TDCVRPTW 实例基于 Solomon CVRPTW 实例。数据文件给出车辆数量和容量，以及仓库和各客户的坐标、需求、时间窗和服务时间。

程序将时间范围离散化为五个时段，并根据距离等级调整各时段的行驶时间系数，从而构造非比例的时变行驶时间矩阵。

建模思路¶

每辆车使用一个客户 list，所有 lists 通过 partition 使每位客户恰好由一辆车服务。每条路线的需求总量不得超过车辆容量，路线距离包含仓库往返和相邻客户之间的距离。

递归数组按访问顺序计算服务结束时间：首位客户使用车辆从仓库出发时对应的交通矩阵，后续客户则根据上一位客户的服务结束时间选择交通时段。客户延迟和返仓延迟作为软约束汇总。模型依次最小化总延迟、使用车辆数和总距离。

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 compute_dist(xi, xj, yi, yj):
    return math.sqrt(math.pow(xi - xj, 2) + math.pow(yi - yj, 2))


def get_profile(distance, distance_levels):
    profile = 0
    while profile < len(distance_levels) and distance > distance_levels[profile]:
        profile += 1
    return profile


def compute_distance_matrices(
    customers_x,
    customers_y,
    max_horizon,
    travel_time_profiles,
    time_interval_steps,
    distance_levels,
):
    nb_customers = len(customers_x)
    nb_time_intervals = len(time_interval_steps) - 1
    distance_matrix = [
        [0.0 for _ in range(nb_customers)] for _ in range(nb_customers)
    ]
    travel_time = [
        [
            [0.0 for _ in range(nb_time_intervals)]
            for _ in range(nb_customers)
        ]
        for _ in range(nb_customers)
    ]

    for i in range(nb_customers):
        for j in range(i + 1, nb_customers):
            distance = compute_dist(
                customers_x[i],
                customers_x[j],
                customers_y[i],
                customers_y[j],
            )
            distance_matrix[i][j] = distance
            distance_matrix[j][i] = distance
            profile = get_profile(distance, distance_levels)
            for interval in range(nb_time_intervals):
                local_travel_time = (
                    travel_time_profiles[profile][interval] * distance
                )
                travel_time[i][j][interval] = local_travel_time
                travel_time[j][i][interval] = local_travel_time

    time_to_matrix_idx = [0] * max_horizon
    for interval in range(nb_time_intervals):
        interval_start = round(time_interval_steps[interval] * max_horizon)
        interval_end = round(time_interval_steps[interval + 1] * max_horizon)
        for time in range(interval_start, interval_end):
            time_to_matrix_idx[time] = interval
    return distance_matrix, travel_time, time_to_matrix_idx


def compute_distance_depots(
    depot_x,
    depot_y,
    customers_x,
    customers_y,
    travel_time_profiles,
    nb_time_intervals,
    distance_levels,
):
    distance_depots = []
    travel_time_warehouse = []
    for customer_x, customer_y in zip(customers_x, customers_y):
        distance = compute_dist(depot_x, customer_x, depot_y, customer_y)
        distance_depots.append(distance)
        profile = get_profile(distance, distance_levels)
        travel_time_warehouse.append(
            [
                travel_time_profiles[profile][interval] * distance
                for interval in range(nb_time_intervals)
            ]
        )
    return distance_depots, travel_time_warehouse


def read_input_cvrptw(filename):
    elements = iter(read_elements(filename))
    for _ in range(4):
        next(elements)
    nb_trucks = int(next(elements))
    truck_capacity = int(next(elements))

    for _ in range(13):
        next(elements)
    depot_x = int(next(elements))
    depot_y = int(next(elements))
    for _ in range(2):
        next(elements)
    max_horizon = int(next(elements))
    next(elements)

    customers_x = []
    customers_y = []
    demands = []
    earliest_start = []
    latest_end = []
    service_time = []
    while next_customer := next(elements, None):
        int(next_customer)
        customers_x.append(int(next(elements)))
        customers_y.append(int(next(elements)))
        demands.append(int(next(elements)))
        ready = int(next(elements))
        due = int(next(elements))
        duration = int(next(elements))
        earliest_start.append(ready)
        latest_end.append(due + duration)
        service_time.append(duration)

    travel_time_profiles = [
        [1.00, 2.50, 1.75, 2.50, 1.00],
        [1.00, 2.00, 1.50, 2.00, 1.00],
        [1.00, 1.60, 1.10, 1.60, 1.00],
    ]
    distance_levels = [10, 25]
    time_interval_steps = [0.0, 0.2, 0.4, 0.6, 0.8, 1.0]
    nb_time_intervals = len(time_interval_steps) - 1

    distance_matrix, travel_time, time_to_matrix_idx = (
        compute_distance_matrices(
            customers_x,
            customers_y,
            max_horizon,
            travel_time_profiles,
            time_interval_steps,
            distance_levels,
        )
    )
    distance_depots, travel_time_warehouse = compute_distance_depots(
        depot_x,
        depot_y,
        customers_x,
        customers_y,
        travel_time_profiles,
        nb_time_intervals,
        distance_levels,
    )
    return {
        "nb_customers": len(customers_x),
        "nb_trucks": nb_trucks,
        "truck_capacity": truck_capacity,
        "distance_matrix": distance_matrix,
        "travel_time": travel_time,
        "time_to_matrix_idx": time_to_matrix_idx,
        "distance_depots": distance_depots,
        "travel_time_warehouse": travel_time_warehouse,
        "demands": demands,
        "service_time": service_time,
        "earliest_start": earliest_start,
        "latest_end": latest_end,
        "max_horizon": max_horizon,
    }


def main(input_file, output_file=None, time_limit=20):
    data = read_input_cvrptw(input_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"])
    travel_time = model.array(data["travel_time"])
    time_to_matrix_idx = model.array(data["time_to_matrix_idx"])
    distance_depots = model.array(data["distance_depots"])
    travel_time_warehouse = model.array(data["travel_time_warehouse"])

    trucks_used = [sequence.count() > 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 = sequence.count()
        route_quantity = model.sum(
            sequence, model.lambda_function(lambda customer: demands[customer])
        )
        model.constraint(
            route_quantity <= truck_capacity
        )

        distance_lambda = model.lambda_function(
            lambda position: distance_matrix[
                sequence[position - 1], sequence[position]
            ]
        )
        route_distances.append(
            model.sum(model.range(1, count), distance_lambda)
            + model.iif(
                count > 0,
                distance_depots[sequence[0]]
                + distance_depots[sequence[count - 1]],
                0,
            )
        )

        end_time_lambda = model.lambda_function(
            lambda position, previous: model.max(
                earliest[sequence[position]],
                model.iif(
                    position == 0,
                    travel_time_warehouse[
                        sequence[0], time_to_matrix_idx[0]
                    ],
                    previous
                    + travel_time[
                        sequence[position - 1],
                        sequence[position],
                        time_to_matrix_idx[model.round(previous)],
                    ],
                ),
            )
            + service_time[sequence[position]]
        )
        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]
                + travel_time_warehouse[
                    sequence[count - 1],
                    time_to_matrix_idx[model.round(end_times[count - 1])],
                ]
                - max_horizon,
            ),
            0,
        )
        late_lambda = model.lambda_function(
            lambda position: model.max(
                0, end_times[position] - latest[sequence[position]]
            )
        )
        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

    output_lines = [
        f"{int(nb_trucks_used.value)} {int(total_distance.value)}"
    ]
    for sequence in customer_sequences:
        if sequence.value:
            output_lines.append(
                " ".join(str(customer + 1) for customer in sequence.value)
            )

    result_text = "\n".join(output_lines)
    print(
        f"Total lateness = {total_lateness.value}; Status = {solution.status}\n"
        + result_text
    )
    if output_file is not None:
        Path(output_file).write_text(result_text + "\n", encoding="utf-8")
    return solution

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

solution_c101_25 = main(
    INSTANCE_DIR / "C101.25.txt",
    time_limit=1,
)

solution_r101_25 = main(
    INSTANCE_DIR / "R101.25.txt",
    time_limit=1,
)

solution_rc101_25 = main(
    INSTANCE_DIR / "RC101.25.txt",
    time_limit=1,
)