飞机降落问题¶

类别： 排程

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

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

问题描述¶

在飞机降落问题中,需要为一组飞机安排降落时间。每架飞机可以在目标时间附近的一个预定时间窗口内降落,且相邻两架飞机之间必须保持一定的间隔时间。每架飞机都需要支付提前或延迟降落的惩罚费用。目标是在满足所有约束条件的前提下,最小化总惩罚费用。

学习要点¶

使用列表决策变量表示降落顺序
使用 lambda 表达式递归地定义飞机的降落时间
使用非线性算子,如三元条件表达式

数据¶

我们提供来自 OR Library 的算例。数据文件的格式如下:

第一行:飞机数量与冻结时间
从第二行起,每架飞机 i 的信息:
出现时间
最早降落时间
目标降落时间
最晚降落时间
早于目标时间降落时,每单位时间的惩罚费用
晚于目标时间降落时,每单位时间的惩罚费用
对于每架飞机 j:在飞机 i 降落后,飞机 j 降落所需的间隔时间

我们仅考虑问题的静态版本,忽略出现时间与冻结时间。

建模思路¶

飞机降落问题的 OptAgent 模型使用列表决策变量表示降落顺序。列表中的第 i 个元素对应第 i 个降落的飞机的索引。为了确保所有飞机均被调度,我们对列表变量施加排列约束。

我们还引入了另一个决策变量:每架飞机的偏好降落时间。当一架飞机在前一架飞机之后必须等待才能降落时,它只能在目标时间之后降落。因此,一架飞机的偏好降落时间在最早降落时间与目标时间之间。

递归 lambda 函数将每架飞机的降落时间定义为偏好降落时间与该飞机在不违反与前机间隔时间的前提下所能降落的最早时间之间的最大值。

目标是最小化由每架飞机提前或延迟降落所产生的惩罚费用。为了计算每架飞机的该费用,我们使用三元条件表达式,根据降落时间与目标时间的差值,选择提前费用或延迟费用。

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve




def read_elem(filename):
    with open(filename, encoding="utf-8") as f:
        return [str(elem) for elem in f.read().split()]


def read_instance(instance_file):
    file_it = iter(read_elem(instance_file))
    nb_planes = int(next(file_it))
    next(file_it)  # Skip freezeTime value
    earliest_time_data = []
    target_time_data = []
    latest_time_data = []
    earliness_cost_data = []
    tardiness_cost_data = []
    separation_time_data = []

    for p in range(nb_planes):
        next(file_it)  # Skip appearanceTime values
        earliest_time_data.append(int(next(file_it)))
        target_time_data.append(int(next(file_it)))
        latest_time_data.append(int(next(file_it)))
        earliness_cost_data.append(float(next(file_it)))
        tardiness_cost_data.append(float(next(file_it)))
        separation_time_data.append([0] * nb_planes)

        for pp in range(nb_planes):
            separation_time_data[p][pp] = int(next(file_it))

    return (
        nb_planes,
        earliest_time_data,
        target_time_data,
        latest_time_data,
        earliness_cost_data,
        tardiness_cost_data,
        separation_time_data,
    )


def main(instance_file, output_file=None, time_limit=20):
    nb_planes, earliest_time_data, target_time_data, latest_time_data, earliness_cost_data, tardiness_cost_data, separation_time_data = read_instance(instance_file)
    model = OptModel()

    # List variable: landing_order[p] is the index of the p-th plane to land
    landing_order = model.list(nb_planes)

    # All planes must be scheduled
    model.constraint(
        model.count(landing_order) == nb_planes
    )

    # Create OptAgent arrays to be able to access them with an "at" operator
    target_time = model.array(target_time_data)
    latest_time = model.array(latest_time_data)
    earliness_cost = model.array(earliness_cost_data)
    tardiness_cost = model.array(tardiness_cost_data)
    separation_time = model.array(separation_time_data)

    # Int variable: preferred landing time for each plane
    preferred_time_vars = [
        model.int(
            lb=earliest_time_data[p],
            ub=target_time_data[p],
        )
        for p in range(nb_planes)
    ]
    preferred_time = model.array(preferred_time_vars)

    # Landing time for each plane (recursive: respects separation with the previous plane)
    landing_time_lambda = model.lambda_function(
        lambda p, prev: model.max(
            model.at(preferred_time, model.at(landing_order, p)),
            model.iif(
                p > 0,
                prev
                + model.at(
                    separation_time,
                    model.at(landing_order, p - 1),
                    model.at(landing_order, p),
                ),
                0,
            ),
        )
    )
    landing_time = model.array(model.range(0, nb_planes), landing_time_lambda)

    # Landing times must respect separation from every previous plane
    for p in range(1, nb_planes):
        last_separation_end = [
            landing_time[previous_plane]
            + model.at(
                separation_time,
                model.at(landing_order, previous_plane),
                model.at(landing_order, p),
            )
            for previous_plane in range(p)
        ]
        model.constraint(landing_time[p] >= model.max(last_separation_end))

    total_cost_terms = []
    for p in range(nb_planes):
        plane_index = model.at(landing_order, p)

        # Constraint on latest landing time
        model.constraint(
            model.at(landing_time, p) <= model.at(latest_time, plane_index),
        )

        # Cost for each plane
        difference_to_target_time = model.abs(
            model.at(landing_time, p) - model.at(target_time, plane_index)
        )
        unit_cost = model.iif(
            model.at(landing_time, p) < model.at(target_time, plane_index),
            model.at(earliness_cost, plane_index),
            model.at(tardiness_cost, plane_index),
        )
        total_cost_terms.append(unit_cost * difference_to_target_time)

    total_cost = model.sum(total_cost_terms)

    # Minimize the total cost
    model.minimize(total_cost)

    solution = solve(model, time_limit_s=float(time_limit))
    result_values = {'landing_order': landing_order.value, 'total_cost': total_cost.value}

    landing_sequence = [int(item) for item in result_values["landing_order"]]
    print(
        f"Nb planes = {nb_planes}; Total cost = {int(result_values['total_cost'])}; "
        f"Status = {solution.feasible}"
    )
    print("Landing order:", " ".join(str(p) for p in landing_sequence))

    if output_file is not None:
        result_text = (
            f"{int(result_values['total_cost'])}\n"
            + " ".join(str(p) for p in landing_sequence)
            + "\n"
        )
        Path(output_file).write_text(result_text, encoding="utf-8")
    return solution

本地运行¶

Notebook 直接调用 main 并显式传入实例路径。以下代码格相互独立，可以按需要单独运行；调整 time_limit 可以控制每个实例的求解时间。

from pathlib import Path

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

solution_airland1 = main(INSTANCE_DIR / "airland1.txt", time_limit=1)

solution_airland5 = main(INSTANCE_DIR / "airland5.txt", time_limit=1)

solution_airland10 = main(INSTANCE_DIR / "airland10.txt", time_limit=1)