Only allow intervals to be produced on machines during shifts cp-sat or-tools

Question

I want to minimize the end points of intervals on machines. However, specific employees can only work on specific machines. And specific employees only work specific hours. I don't know how to include that constraint. Here is the code I have. First we set the shifts and the duration of the one task we want to execute.

from ortools.sat.python import cp_model

# Create a CP-SAT model instance
model = cp_model.CpModel()

# Define the time horizon for scheduling
horizon = 600

# Define working hours and associated employees for each shift
# Each tuple represents a shift's start and end time, and the list contains the names of employees working that shift
employees_per_shift = {
    (0, int(horizon/3)-1): ["Person_1"],
    (int(horizon/3), int((2/3)*horizon)): ["Person_1", "Person_2"],
    (int((2*horizon)/3)-1, int(horizon)): ["Person_1", "Person_2"],
}

# Define machine options available for each employee
# Each employee can have one or more sets of machines they can operate
machineset_options_per_employee = {
    "Person_1": [["GAG-Berlin"]],
    "Person_2": [["GAG-Rom"], ["GAG-Rom", "GAG-Berlin"]]
}

# Define the duration each machine is occupied for
duration_per_machine = {'GAG-Berlin': 300, 'GAG-Rom': 200}

# List of all machines
all_machines = ['GAG-Berlin', "GAG-Rom"]

Then we define the task:

# Define the start, end, and duration of the interval for machine operation
start = model.NewIntVar(0, horizon, 'start0')
end = model.NewIntVar(0, horizon, 'end0')
duration = model.NewIntVar(min(duration_per_machine.values()), max(duration_per_machine.values()), 'duration0')  
interval = model.NewIntervalVar(start, duration, end, 'interval0')

# Initialize dictionaries for storing intervals per machine, and booleans indicating task completion per machine
intervals_per_machine = {machine: [] for machine in all_machines}
bool_per_task = []
bools_per_task_per_machine = {machine: [] for machine in all_machines}

# Create intervals and booleans for each machine
for machine in all_machines:
    m_suffix = f"machine_{machine}"
    m_bool = model.NewBoolVar(f'bool_{m_suffix}')
    m_duration_int = duration_per_machine[machine]
    m_start = model.NewIntVar(0, horizon, f'start_{m_suffix}')
    m_end = model.NewIntVar(0, horizon, f'end_{m_suffix}')
    m_interval = model.NewOptionalIntervalVar(m_start, m_duration_int, m_end, m_bool, f'interval_{m_suffix}')
    
    # Link the machine usage to the interval variables
    model.Add(start == m_start).OnlyEnforceIf(m_bool)
    model.Add(end == m_end).OnlyEnforceIf(m_bool)
    model.Add(duration == m_duration_int).OnlyEnforceIf(m_bool)

    # Store the created interval and boolean for each machine
    intervals_per_machine[machine] += [m_interval]
    bool_per_task.append(m_bool)
    bools_per_task_per_machine[machine] += [m_bool]

# Ensure exactly one machine is selected for the task
model.AddExactlyOne(bool_per_task)

Next we define the booleans that indicate which machineset is available to produced on during which shift:

# Initialize a dictionary to store boolean variables for each person's machine set options per shift
person_machineset_vars = {person: {} for person in machineset_options_per_employee}
for work_hours, employees in employees_per_shift.items():
    start_shift = work_hours[0]
    for person in employees:
        # Create boolean variables for each machine set option indicating whether the person starts a shift with that set
        person_machineset_vars[person].update({
            start_shift: [
                model.NewBoolVar(f"{person}_shift_start_{start_shift}_machineset_{i}")
                for i, _ in enumerate(machineset_options_per_employee[person])
            ]
        })

# Dictionary to store the boolean variables for machine usage per shift, organized by machine
bools_per_shift_per_machine = {machine: {work_hours: [] for work_hours, _ in employees_per_shift.items()} for machine in all_machines}
for work_hours, employees in employees_per_shift.items():
    start_shift = work_hours[0]
    for person in employees:
        for i, _ in enumerate(machineset_options_per_employee[person]):
            for machine in machineset_options_per_employee[person][i]:
                # Add the boolean variables to the corresponding machine and shift
                bools_per_shift_per_machine[machine][work_hours].append(person_machineset_vars[person][start_shift][i])

Then we would need to add the constraint that links the two. An interval can only be produced on a machine if there is an employee who is working on a shift at that time.

I tried many things. However, I didn't got to the right result. For example this only says that simply one shift should overlap:

# This simply says (at least) one is overlapping
for machine in bools_per_shift_per_machine:
    print(machine)
    tmp_machine_shift_bools = []
    for work_hours, machine_selected_bool in bools_per_shift_per_machine[machine].items():
        print(work_hours)
        for machine_interval in intervals_per_machine[machine]:
            print(machine_interval)

            # If the task on the machine is selected, then machine should start before the shift ends and end after the shift starts
            # It should have overlap with at least one shift
            tmp_machine_shift_bool = model.NewBoolVar(f"{machine}_shift_bool_{work_hours[0]}_{work_hours[1]}")
            print(tmp_machine_shift_bool)
            model.Add(machine_interval.StartExpr() < work_hours[1]).OnlyEnforceIf(tmp_machine_shift_bool)
            model.Add(machine_interval.EndExpr() >= work_hours[0]).OnlyEnforceIf(tmp_machine_shift_bool)
            tmp_machine_shift_bools.append(tmp_machine_shift_bool)
            
    model.Add(sum(tmp_machine_shift_bools)>=1).OnlyEnforceIf(bools_per_task_per_machine[machine][0])
    model.Add(sum(tmp_machine_shift_bools)==0).OnlyEnforceIf(bools_per_task_per_machine[machine][0].Not())
    model.AddImplication(bools_per_task_per_machine[machine][0], machine_selected_bool[0])
    model.AddImplication(bools_per_task_per_machine[machine][0].Not(), machine_selected_bool[0].Not())

In the end I print:

# Minimize when its done
model.Minimize(sum([end]))
# Create a solver and solve the model
solver = cp_model.CpSolver()
status = solver.Solve(model)

if (status == cp_model.FEASIBLE) or (status==cp_model.OPTIMAL):
    print("Solution found:")
    print("Objective value", solver.ObjectiveValue())
    print("Interval 1: Start =", solver.Value(start), ", End =", solver.Value(end))
    for machine in bools_per_task_per_machine:
        if solver.Value(bools_per_task_per_machine[machine][0]):
            print(f"Machine {machine} is used")
    for machine in bools_per_shift_per_machine:
        for shift, machine_selected_bools in bools_per_shift_per_machine[machine].items():
            for machine_selected_bool in machine_selected_bools:
                if solver.Value(machine_selected_bool):
                    print('yes', machine, shift, machine_selected_bool)
                else:
                    print('no',machine,shift, machine_selected_bool)
else:
    raise ValueError("No solution found.")

And I expect to see:

Solution found:
Objective value 300.0
Interval 1: Start = 0 , End = 300
Machine GAG-Berlin is used
yes GAG-Berlin (0, 199) Person_1_shift_start_0_machineset_0
no GAG-Berlin (200, 400) Person_1_shift_start_200_machineset_0
yes GAG-Berlin (200, 400) Person_2_shift_start_200_machineset_1
no GAG-Berlin (399, 600) Person_1_shift_start_399_machineset_0
no GAG-Berlin (399, 600) Person_2_shift_start_399_machineset_1
no GAG-Rom (200, 400) Person_2_shift_start_200_machineset_0
no GAG-Rom (200, 400) Person_2_shift_start_200_machineset_1
no GAG-Rom (399, 600) Person_2_shift_start_399_machineset_0
no GAG-Rom (399, 600) Person_2_shift_start_399_machineset_1