1.2.1

下载 Notebook 下载完整案例

带准备时间的柔性作业车间调度问题 (FJSP-SDST)¶

类别： 排程

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

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

问题描述¶

在带序列相关准备时间的柔性作业车间调度问题 (FJSP-SDST) 中,一组作业必须在车间内的机器上加工。每个作业由一组有序的任务(称为工序)组成:每道工序只能在其前一道工序结束后才能开始。每道工序都有一组兼容的机器,且必须由其中一台机器进行加工。工序的加工时间取决于执行该工序的机器。机器一次只能加工一个任务,且在两个连续任务之间必须进行准备。这些准备时间既取决于任务,也取决于所选机器。目标是最小化最大完工时间(makespan),即所有作业完成的时间。

学到的要点¶

• 添加 区间决策变量 来建模任务
• 添加 列表决策变量 来建模任务在机器上的分配及其加工顺序
• 定义 lambda 函数 将区间变量和列表变量关联起来

数据¶

我们提供的实例来自柔性作业车间调度问题的 Fattahi [1] 数据集,并附加了随机生成的准备时间。其格式如下:

• 第一行:作业数、机器数、每道工序的平均机器数(不需要)

• 从第二行起,对每个作业:

• 该作业中的工序数

• 对每道工序:

• 与该工序兼容的机器数

• 对每台兼容机器:机器索引及在该机器上的加工时间

• 对每台机器和每道工序:

• 在该机器上该工序与每个其他工序之间的准备时间。

建模思路¶

带序列相关准备时间的柔性作业车间调度问题 (FJSP-SDST) 的 OptAgent 模型使用区间决策变量来建模工序的时间范围,并使用列表决策变量来表示每台机器上调度的任务顺序。

利用 partition 算子,我们确保每个任务被分配到恰好一台机器。对于每道工序,我们使用 contains 算子过滤掉不兼容的机器。使用 find 算子,我们可以获取被选中处理每个任务的机器索引。这使我们能够推导出每道工序的加工时间(取决于所选机器),并相应地对每个区间的长度进行约束。

优先关系约束很容易写出:对于每个作业,其每道工序必须在前一道工序结束后才开始。析取资源约束可以表述如下:对所有 i,在位置 i+1 上加工的任务必须等到在位置 i 上加工的任务结束后(加上这两个任务之间的准备时间)才能开始。为了建模这个约束,我们定义一个 lambda 函数 来表达两个相邻活动之间的关系。该函数随后在一个可变参 and_ 算子内对每台机器上加工的所有任务求值。注意,这些 and_ 表达式中项的数量在搜索过程中是变化的,列表的大小(每台机器上分配的任务数)也随之变化。

目标是最小化最大完工时间(makespan),即所有任务完成的时间。

[1] P. Fattahi, M.S. Mehrabad, F. Jolai. (2007). Mathematical modeling and heuristic approaches to flexible job shop scheduling problems. Journal of Intelligent Manufacturing, 18(3), 331–342.

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve



# Constant for incompatible machines
INFINITE = 1000000


def read_instance(filename):
    lines = Path(filename).read_text().splitlines()

    first_line = lines[0].split()
    # Number of jobs
    nb_jobs = int(first_line[0])
    # Number of machines
    nb_machines = int(first_line[1])

    # Number of operations for each job
    nb_operations = [int(lines[j + 1].split()[0]) for j in range(nb_jobs)]

    # Number of tasks
    nb_tasks = sum(nb_operations[j] for j in range(nb_jobs))

    # Processing time for each task, for each machine
    task_processing_time = [[INFINITE] * nb_machines for _ in range(nb_tasks)]

    # For each job, for each operation, the corresponding task id
    job_operation_task = [[0] * nb_operations[j] for j in range(nb_jobs)]

    # Setup time between every two consecutive tasks, for each machine
    task_setup_time = [[[-1] * nb_tasks for _ in range(nb_tasks)] for _ in range(nb_machines)]

    task_id = 0
    for j in range(nb_jobs):
        line = lines[j + 1].split()
        field_offset = 0
        for o in range(nb_operations[j]):
            nb_machines_operation = int(line[field_offset + o + 1])
            for i in range(nb_machines_operation):
                machine = int(line[field_offset + o + 2 * i + 2]) - 1
                time = int(line[field_offset + o + 2 * i + 3])
                task_processing_time[task_id][machine] = time
            job_operation_task[j][o] = task_id
            task_id += 1
            field_offset += 2 * nb_machines_operation

    id_line = nb_jobs + 2
    max_setup = 0
    for machine in range(nb_machines):
        for before in range(nb_tasks):
            task_setup_time[machine][before] = list(map(int, lines[id_line].split()))
            max_setup = max(
                max_setup,
                max(setup if setup != INFINITE else 0 for setup in task_setup_time[machine][before]),
            )
            id_line += 1

    # Trivial upper bound for the end times of the tasks
    max_sum_processing_times = sum(
        max(task_processing_time[i][m] for m in range(nb_machines) if task_processing_time[i][m] != INFINITE)
        for i in range(nb_tasks)
    )
    max_end = max_sum_processing_times + nb_tasks * max_setup

    return (
        nb_jobs,
        nb_machines,
        nb_tasks,
        task_processing_time,
        job_operation_task,
        nb_operations,
        task_setup_time,
        max_end,
    )


def main(instance_file, output_file=None, time_limit=20):
    (
        nb_jobs,
        nb_machines,
        nb_tasks,
        task_processing_time_data,
        job_operation_task,
        nb_operations,
        task_setup_time_data,
        max_end,
    ) = read_instance(instance_file)

    model = OptModel()

    # Sequence of tasks assigned to each machine.
    jobs_order = [model.list(nb_tasks) for machine in range(nb_machines)]
    machines = model.array(jobs_order)
    # Each task is scheduled on a machine
    model.constraint(model.partition(machines))

    # Only compatible machines can be selected for a task
    for task in range(nb_tasks):
        for machine in range(nb_machines):
            if task_processing_time_data[task][machine] == INFINITE:
                model.constraint(model.not_(jobs_order[machine].contains(task)))
    # For each task, the selected machine
    task_machine = [model.find(machines, task) for task in range(nb_tasks)]
    task_processing_time = model.array(task_processing_time_data)
    task_setup_time = model.array(task_setup_time_data)
    # Interval decisions: time range of each task
    tasks = [model.interval(0, max_end) for _ in range(nb_tasks)]

    # The task duration depends on the selected machine
    durations = [task_processing_time.at(task, task_machine[task]) for task in range(nb_tasks)]
    for task in range(nb_tasks):
        model.constraint(tasks[task].length() == durations[task])

    task_array = model.array(tasks)

    # Precedence constraints between the operations of a job
    for job in range(nb_jobs):
        for operation in range(nb_operations[job] - 1):
            before = job_operation_task[job][operation]
            after = job_operation_task[job][operation + 1]
            model.constraint(tasks[before] < tasks[after])

    # Disjunctive resource constraints between the tasks on a machine
    for machine, sequence in enumerate(jobs_order):
        adjacent_positions = model.range(0, sequence.count() - 1)
        setup_before_next = model.lambda_function(
            lambda i: task_array[sequence[i + 1]].start()
            >= task_array[sequence[i]].end()
            + task_setup_time.at(machine, sequence[i], sequence[i + 1])
        )
        model.constraint(model.and_(adjacent_positions, setup_before_next))
    # Minimize the makespan: end of the last task
    makespan = model.max([task.end() for task in tasks])
    model.minimize(makespan)

    solution = solve(model, time_limit_s=float(time_limit))
    if not solution.feasible:
        print(f"No feasible schedule found; Status = {solution.feasible}")
        return solution

    expressions = {"makespan": makespan}
    for task in range(nb_tasks):
        expressions[f"task_{task}_machine"] = task_machine[task]
        expressions[f"task_{task}_start"] = tasks[task].start()
        expressions[f"task_{task}_end"] = tasks[task].end()
    values = {name: expression.value for name, expression in expressions.items()}

    schedule_rows = []
    for job in range(nb_jobs):
        for operation in range(nb_operations[job]):
            task = job_operation_task[job][operation]
            schedule_rows.append(
                (
                    job + 1,
                    operation + 1,
                    values[f"task_{task}_machine"] + 1,
                    values[f"task_{task}_start"],
                    values[f"task_{task}_end"],
                )
            )

    print(f"Makespan = {values['makespan']}; Status = {solution.feasible}")
    print("Job\tOperation\tMachine\tStart\tEnd")
    for row in schedule_rows:
        print("\t".join(map(str, row)))
    if output_file is not None:
        Path(output_file).write_text(
            "\n".join("\t".join(map(str, row)) for row in schedule_rows) + "\n",
            encoding="utf-8",
        )
    return solution

本地运行¶

INSTANCE_DIR = Path.cwd() / "instances"

solution = main(INSTANCE_DIR / "Fattahi_setup_01.fjs", time_limit=10)