Changeover as event

Source: scheduling/example_08_changeover_as_event.py

What it does

Promotes the changeover to a first-class scheduled event. For every ordered pair (t1, t2) there is an optional interval with its own start, end, and presence:

co_iv[m, t1, t2] = model.new_optional_interval_var(
    co_start[m, t1, t2],
    changeover_time[product_of(t2)],
    co_end[m, t1, t2],
    co_present[m, t1, t2],
    ...,
)

When seq[m, t1, t2] is chosen, the model forces the co interval to be present, sit between the two tasks, and have the right size:

model.add(end[t1] <= co_start[t1, t2]).only_enforce_if(seq[m, t1, t2])
model.add(co_end[t1, t2] <= start[t2]).only_enforce_if(seq[m, t1, t2])
model.add(co_end - co_start == distance).only_enforce_if(seq[m, t1, t2])
model.add(co_present == 1).only_enforce_if(seq[m, t1, t2])
model.add(co_present == 0).only_enforce_if(~seq[m, t1, t2])

Because the changeover is now an interval, it can take part in cumulative constraints (cleaning resource, operator availability, etc.).

Concepts

• Changeover (approach 3: as event)
• Interval variables

Source

from ortools.sat.python import cp_model

# Initiate
model = cp_model.CpModel()

'''
task   product
1       A
2       B
'''

# 1. Data

tasks = {1, 2}
tasks_0 = tasks.union({0})
task_to_product = {0: 'dummy', 1: 'A', 2: 'B'}
processing_time = {'dummy': 0, 'A': 1, 'B': 1}
changeover_time = {'dummy': 0, 'A': 2, 'B': 2}
machines = {0}
machines_starting_products = {0: 'A'}

X = {
    (m, t1, t2)
    for t1 in tasks_0
    for t2 in tasks_0
    for m in machines
    if t1 != t2
}

# This is not yet used
m_cost = {
    (m, t1, t2): 0
    if task_to_product[t1] == task_to_product[t2] or (
            task_to_product[t1] == 'dummy' and task_to_product[t2] == machines_starting_products[m]
    )
    else changeover_time[task_to_product[t2]]
    for (m, t1, t2) in X
}


# 2. Decision variables
max_time = 8

variables_task_ends = {
    task: model.new_int_var(0, max_time, f"task_{task}_end") for task in tasks_0
}

variables_task_starts = {
    task: model.new_int_var(0, max_time, f"task_{task}_end") for task in tasks_0
}

variables_machine_task_starts = {
    (m, t): model.new_int_var(0, max_time, f"start_{m}_{t}")
    for t in tasks_0
    for m in machines
}

variables_machine_task_ends = {
    (m, t): model.new_int_var(0, max_time, f"start_{m}_{t}")
    for t in tasks_0
    for m in machines
}

variables_machine_task_presences = {
    (m, t): model.new_bool_var(f"presence_{m}_{t}")
    for t in tasks_0
    for m in machines
}

variables_machine_task_sequence = {
    (m, t1, t2): model.new_bool_var(f"Machine {m} task {t1} --> task {t2}")
    for (m, t1, t2) in X
}

# intervals
# ! This can replace the end - start = duration constrain
variables_machine_task_intervals = {
    (m, task): model.new_optional_interval_var(
        variables_machine_task_starts[m, task],
        processing_time[task_to_product[task]],
        variables_machine_task_ends[m, task],
        variables_machine_task_presences[m, task],
        name=f"t_interval_{m}_{task}"
    )
    for task in tasks_0
    for m in machines
}


# Add change over-related variables !!!

variables_co_starts = {
    (t1, t2): model.new_int_var(0, max_time, f"co_t{t1}_to_t{t2}_start") for t1 in tasks_0 for t2 in tasks_0 if t1 != t2
}
variables_co_ends = {
    (t1, t2): model.new_int_var(0, max_time, f"co_t{t1}_to_t{t2}_end") for t1 in tasks_0 for t2 in tasks_0 if t1 != t2
}
variables_machine_co_starts = {
    (m, t1, t2): model.new_int_var(0, max_time, f"m{m}_co_t{t1}_to_t{t2}_start")
    for t1 in tasks_0 for t2 in tasks_0 for m in machines if t1 != t2
}
variables_machine_co_ends = {
    (m, t1, t2): model.new_int_var(0, max_time, f"m{m}_co_t{t1}_to_t{t2}_end")
    for t1 in tasks_0 for t2 in tasks_0 for m in machines if t1 != t2
}
variables_machine_co_presences = {
    (m, t1, t2): model.new_bool_var(f"co_presence_m{m}_t{t1}_t{t2}")
    for t1 in tasks_0
    for t2 in tasks_0
    for m in machines
    if t1 != t2
}


variables_machine_co_intervals = {
    (m, t1, t2): model.new_optional_interval_var(
        variables_machine_co_starts[m, t1, t2],
        changeover_time[task_to_product[t2]],
        variables_machine_co_ends[m, t1, t2],
        variables_machine_co_presences[m, t1, t2],
        name=f"co_interval_m{m}_t{t1}_t{t2}"
    )
    for t1 in tasks_0
    for t2 in tasks_0
    for m in machines
    if t1 != t2
}







# 3. Objectives

make_span = model.new_int_var(0, max_time, "make_span")

model.add_max_equality(
    make_span,
    [variables_task_ends[task] for task in tasks]
)

model.minimize(make_span)


# 4. Constraints

# One task to one machine.
for task in tasks:
    task_candidate_machines = machines
    tmp = [
        variables_machine_task_presences[m, task]
        for m in task_candidate_machines
    ]
    # this task is only present in one machine
    model.add_exactly_one(tmp)


# task level link to machine-task level
for task in tasks_0:
    task_candidate_machines = machines
    for m in task_candidate_machines:
        model.add(
            variables_task_starts[task] == variables_machine_task_starts[m, task]
        ).only_enforce_if(variables_machine_task_presences[m, task])

        model.add(
            variables_task_ends[task] == variables_machine_task_ends[m, task]
        ).only_enforce_if(variables_machine_task_presences[m, task])


# co level link to machine-co level
for t1 in tasks_0:
    for t2 in tasks_0:
        if t1 != t2:

            for m in machines:
                model.add(
                    variables_co_starts[t1, t2] == variables_machine_co_starts[m, t1, t2]
                ).only_enforce_if(variables_machine_co_presences[m, t1, t2])

                model.add(
                    variables_co_ends[t1, t2] == variables_machine_co_ends[m, t1, t2]
                ).only_enforce_if(variables_machine_co_presences[m, t1, t2])


# for dummies: Force task 0 (dummy) starts at 0 and is present on all machines
model.add(variables_task_starts[0] == 0)
for m in machines:
    model.add(variables_machine_task_presences[m, 0] == 1)


# Sequence
for m in machines:
    arcs = list()
    for t1 in tasks_0:
        for t2 in tasks_0:
            # arcs
            if t1 != t2:
                arcs.append([
                        t1,
                        t2,
                        variables_machine_task_sequence[(m, t1, t2)]
                ])
                distance = m_cost[m, t1, t2]
                # cannot require the time index of task 0 to represent the first and the last position
                if t2 != 0:
                    # to schedule tasks and c/o
                    model.add(
                        variables_task_ends[t1] <= variables_co_starts[t1, t2]
                    ).only_enforce_if(variables_machine_task_sequence[(m, t1, t2)])

                    model.add(
                        variables_co_ends[t1, t2] <= variables_task_starts[t2]
                    ).only_enforce_if(variables_machine_task_sequence[(m, t1, t2)])

                    model.add(
                        variables_co_ends[t1, t2] - variables_co_starts[t1, t2] == distance
                    ).only_enforce_if(variables_machine_task_sequence[(m, t1, t2)])

                    # ensure intervals are consistent so we can apply resource constraints later
                    model.add(
                        variables_machine_co_presences[m, t1, t2] == 1
                    ).only_enforce_if(variables_machine_task_sequence[(m, t1, t2)])

                    model.add(
                        variables_machine_co_presences[m, t1, t2] == 0
                    ).only_enforce_if(~variables_machine_task_sequence[(m, t1, t2)])


    for task in tasks:
        arcs.append([
            task, task, ~variables_machine_task_presences[(m, task)]
        ])
    model.add_circuit(arcs)


# Solve


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


# Post-process

if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
    for task in tasks:
        print(f'Task {task} ',
              solver.value(variables_task_starts[task]), solver.value(variables_task_ends[task])
              )

    print('Make-span:', solver.value(make_span))

    for m in machines:

        print(f'------------\nMachine {m}')
        print(f'Starting dummy product: {machines_starting_products[m]}')
        for t1 in tasks_0:
            for t2 in tasks_0:
                if t1 != t2:
                    value = solver.value(variables_machine_task_sequence[(m, t1, t2)])
                    if value == 1 and t2 != 0:
                        print(f'{t1} --> {t2}   {task_to_product[t1]} >> {task_to_product[t2]}  cost: {m_cost[m, t1, t2]}')
                        print('variables_machine_task_sequence[t1, t2]', solver.value(variables_machine_task_sequence[m, t1, t2]))
                        print('variables_co_starts[t1, t2]', solver.value(variables_co_starts[t1, t2]))
                        print('variables_co_ends[t1, t2]', solver.value(variables_co_ends[t1, t2]))
                        print('variables_machine_co_presences[m, t1, t2]', solver.value(variables_machine_co_presences[m, t1, t2]))
                        print('variables_machine_co_starts[m, t1, t2]', solver.value(variables_machine_co_starts[m, t1, t2]))
                        print('variables_machine_co_ends[m, t1, t2]', solver.value(variables_machine_co_ends[m, t1, t2]))
                        #print('variables_machine_co_intervals[m, t1, t2]', variables_machine_co_intervals[m, t1, t2])

                    if value == 1 and t2 == 0:
                        print(f'{t1} --> {t2}   Closing')


elif status == cp_model.INFEASIBLE:
    print("Infeasible")
elif status == cp_model.MODEL_INVALID:
    print("Model invalid")
else:
    print(status)