Proof-of-concept OR

schedule.py

def job_shop():
    machines = [
        Machine(cmm_id="1"),
        Machine(cmm_id="2"),
    ]

    jobs = [
        Job(job_id="431", instances=3, duration_min=20),
        Job(job_id="459", instances=1, duration_min=100), # ???
        Job(job_id="449", instances=10, duration_min=40),
        Job(job_id="461", instances=1, duration_min=100), # ???
        Job(job_id="457", instances=1, duration_min=100), # ???
        Job(job_id="426", instances=5, duration_min=95),
        Job(job_id="419", instances=11, duration_min=50),
        Job(job_id="458", instances=3, duration_min=90),
        Job(job_id="467", instances=108, duration_min=15),
    ]

    deadline = hours(11)

    model = cp_model.CpModel()

    ############################################################################
    # Decision Variables
    ############################################################################

    makespan = model.NewIntVar(0, deadline, "makespan")
    machine_intervals = defaultdict(list)
    running = {}
    matrix = {}
    for j, m in itertools.product(jobs, machines):
        beg = model.NewIntVar(0, deadline, f"beg_{j.job_id}_{m.cmm_id}")
        end = model.NewIntVar(0, deadline, f"end_{j.job_id}_{m.cmm_id}")
        interval = model.NewIntervalVar(beg, j.duration_min, end, f"interval_{j.job_id}")

        matrix[(j.job_id, m.cmm_id)] = {
            "beg": beg,
            "end": end,
            "interval": interval,
        }
        machine_intervals[m.cmm_id].append(interval)
        running[(j.job_id, m.cmm_id)] = model.NewBoolVar(f"running_{j.job_id}_{m.cmm_id}")

    ############################################################################
    # Constraints
    ############################################################################

    # Sensible makespan
    for k, v in matrix.items():
        model.Add(makespan >= v["end"])

    # No overlapping intervals *per machine*
    for _, xs in machine_intervals.items():
        model.AddNoOverlap(xs)

    # Run each job exactly once
    for j in jobs:
        model.Add(
            sum(
                v
                for k, v in running.items()
                if k[0] == j.job_id
            ) == 1
        )

    # Ensure four jobs get assigned to M1 (just to illustrate the lack of parallelization)
    for j in jobs:
        model.Add(
            sum(
                v
                for k, v in running.items()
                if k[1] == "1"
            ) == 4
        )

    ############################################################################
    # Objective
    ############################################################################

    model.Minimize(makespan)

    # solve
    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    if status not in {cp_model.OPTIMAL}:
        print("No solution found")
    else:
        results = []
        for k, v in matrix.items():
            beg, end = solver.Value(v["beg"]), solver.Value(v["end"])
            is_run = solver.Value(running[k])
            if is_run:
                results.append((k[0], k[1], beg, end))

        results = sorted(results, key=lambda x: (x[1], x[2]))

        print(
            tui_gantt(
                [(x[0], x[1], x[2], x[3]) for x in results],
                scale=0.1,
            )
        )