cpmpy中Cumulative约束的性能优化实践与解决方案

import cpmpy as cp
import logging
from typing import List

# 配置日志，便于观察运行信息
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')

class CumulativeTestModel:
    """
    一个使用cpmpy的Cumulative约束来调度任务并最小化所需机器数量的模型。
    """
    def __init__(self, task_duration: int, nb_tasks: int, end_date: int):
        self.model: cp.Model = cp.Model()

        # 定义变量
        # objective: 最小化所需的机器数量，范围从0到任务总数
        self.objective: cp.IntVar = cp.intvar(0, nb_tasks, name="machines")
        # starts: 每个任务的开始时间，范围从0到end_date
        starts: List[cp.IntVar] = [cp.intvar(0, end_date, name=f"start_{i}") for i in range(nb_tasks)]
        # durations: 所有任务的持续时间相同
        durations: List[int] = [task_duration] * nb_tasks
        # ends: 每个任务的结束时间，范围从0到end_date
        ends: List[cp.IntVar] = [cp.intvar(0, end_date, name=f"end_{i}") for i in range(nb_tasks)]
        # demands: 每个任务的需求资源量，此处每个任务需要1台机器
        demands: List[int] = [1] * nb_tasks

        # 添加Cumulative约束到模型
        # 该约束确保在任何时间点，所有活跃任务的资源需求之和不超过self.objective（即机器数量）
        self.model += cp.Cumulative(
            start=starts,
            duration=durations,
            end=ends,
            demand=demands,
            capacity=self.objective,
        )

        # 最小化目标变量，即所需机器数量
        self.model.minimize(self.objective)
        logging.info(f"Model created with {nb_tasks} tasks.")

    def run(self):
        """
        运行模型并打印结果。
        """
        # 使用ortools作为求解器
        solver = cp.model.SolverLookup.get("ortools", self.model)
        has_solution = solver.solve()

        if not has_solution:
            logging.info("No solution found.")
        else:
            logging.info(f"Solution found: {solver.status()} -> {self.objective.value()} in {solver.solve_time} seconds")


if __name__ == "__main__":
    # 示例用法：观察不同任务数量下的性能
    logging.info("--- Testing with ortools solver ---")
    CumulativeTestModel(task_duration=10, nb_tasks=3, end_date=15).run()
    CumulativeTestModel(task_duration=10, nb_tasks=5, end_date=25).run()
    CumulativeTestModel(task_duration=10, nb_tasks=7, end_date=35).run()
    CumulativeTestModel(task_duration=10, nb_tasks=9, end_date=45).run()
    CumulativeTestModel(task_duration=10, nb_tasks=11, end_date=55).run()
    # 对于13个任务，在未优化前可能无法完成
    # CumulativeTestModel(task_duration=10, nb_tasks=13, end_date=65).run()

    logging.info("\n--- Testing with minizinc:chuffed solver (example for comparison) ---")
    # 假设使用Minizinc作为后端，并指定Chuffed求解器
    # 注意：cpmpy的SolverLookup默认查找已安装的求解器，此处仅作示意
    # 如果要使用Minizinc后端，可能需要额外的配置或安装
    # solver_chuffed = cp.model.SolverLookup.get("minizinc:chuffed", CumulativeTestModel(10, 11, 55).model)
    # solver_chuffed.solve()

INFO: Model created with 3 tasks.
INFO: Solution found: ExitStatus.OPTIMAL -> 3 in 0.009132 seconds
INFO: Model created with 11 tasks.
INFO: Solution found: ExitStatus.OPTIMAL -> 3 in 0.002023 seconds
INFO: Model created with 13 tasks.
INFO: Solution found: ExitStatus.OPTIMAL -> 3 in 0.000835 seconds
INFO: Model created with 21 tasks.
INFO: Solution found: ExitStatus.OPTIMAL -> 3 in 0.001112 seconds