I am looking to assign tasks to people. Each successive task should be started after the first one is finished and should be started at the same location as the preceding task. While the time constraint works fine, when using circuit constraints for the location constraint, I am not getting the right results. I am looking to allocate all tasks to as few people as possible.
Tasks 1-3-4 are a possible sequence. So is 1-3-5, 2-3-4, etc. It is possible for 3 employees to cover all tasks here. I only get 5 employees as the optimal answer
What am I doing wrong?
from collections import defaultdict
from ortools.sat.python import cp_model
model = cp_model.CpModel()
solver = cp_model.CpSolver()
tasks = {
1: [2, 3, 90, 165], # task_id: start_location, end_location, start_time, end_time
2: [2, 3, 160, 265],
3: [3, 1, 280, 355],
4: [1, 2, 380, 455],
5: [1, 2, 810, 885]
}
employees = range(5)
assigns = {}
employee_task_intervals = defaultdict(list)
for e in employees:
for t in tasks.keys():
assigns[e, t] = model.NewBoolVar(f'{t}_{e}')
start = model.NewIntVar(0, 1440, f'task_starts_at_{t}_{e}')
end = model.NewIntVar(0, 1440, f'task_ends_at_{t}_{e}')
duration = tasks[t][3] - tasks[t][2]
interval = model.NewOptionalIntervalVar(start, duration, end, assigns[e,t], f'task_interval_{t}_{e}')
employee_task_intervals[e].append(interval)
model.AddNoOverlap(employee_task_intervals[e])
for t in tasks.keys():
model.Add(sum(assigns[e,t] for e in employees) == 1)
for e in employees:
arcs = []
#if no tasks assigned to employee then add a self loop for employee
any_active = model.NewBoolVar(f'any_active_task_in_{e}')
model.Add( sum(assigns[e, t] for t in tasks.keys()) == 0).OnlyEnforceIf(any_active.Not())
model.Add( sum(assigns[e, t] for t in tasks.keys()) > 0).OnlyEnforceIf(any_active)
arcs.append([0, 0, any_active.Not()])
for t in range(len(employee_task_intervals[e])):
first = model.NewBoolVar(f'first_{e}_{t}')
last = model.NewBoolVar(f'last_{e}_{t}')
arcs.append( [0, t+1, first] )
arcs.append( [t+1, 0, last] )
model.AddImplication(assigns[e,t+1].Not(), first.Not())
model.AddImplication(assigns[e,t+1].Not(), last.Not())
for i in range(len(employee_task_intervals[e])):
if i == t:
arcs.append([t+1, t+1, assigns[e,t+1].Not()])
continue
task_follows = model.NewBoolVar(f'task_follows_{i+1}_{t+1}_{e}')
arcs.append([t+1, i+1, task_follows])
#next task starts at same location as current stop
model.Add(tasks[i+1][0] == tasks[t+1][1]).OnlyEnforceIf(task_follows)
#if the next stop does not start at same locations, then only one of the two tasks are true for this employee
model.Add(assigns[e, t+1]+assigns[e, i+1] <= 1).OnlyEnforceIf(task_follows.Not())
model.AddCircuit(arcs)
employees_worked = {}
for e in employees:
employee_has_task = model.NewBoolVar(f'used_{e}')
model.Add( sum ( assigns[e, t] for t in tasks.keys() ) > 0 ).OnlyEnforceIf(employee_has_task)
model.Add( sum ( assigns[e, t] for t in tasks.keys() ) == 0 ).OnlyEnforceIf(employee_has_task.Not())
employees_worked[e] = employee_has_task
model.Minimize(sum(employees_worked[e] for e in employees))
status = solver.Solve(model)
assert status in [cp_model.FEASIBLE, cp_model.OPTIMAL]
for e in employees:
for t in tasks.keys():
if solver.BooleanValue(assigns[e, t]):
print(f'Employee: {e}, Task: {t}')
Updated code that fixed the problem. I had not sequenced the tasks by time.
for e in employees:
arcs = []
#if no tasks assigned to employee then add a self loop for employee
any_active = model.NewBoolVar(f'any_active_task_in_{e}')
model.Add( sum(assigns[e, t] for t in tasks.keys()) == 0).OnlyEnforceIf(any_active.Not())
model.Add( sum(assigns[e, t] for t in tasks.keys()) > 0).OnlyEnforceIf(any_active)
arcs.append([0, 0, any_active.Not()])
for t in range(len(employee_task_intervals[e])):
first = model.NewBoolVar(f'first_{e}_{t}')
last = model.NewBoolVar(f'last_{e}_{t}')
arcs.append( [0, t+1, first] )
arcs.append( [t+1, 0, last] )
model.AddImplication(assigns[e,t+1].Not(), first.Not())
model.AddImplication(assigns[e,t+1].Not(), last.Not())
for i in range(len(employee_task_intervals[e])):
if i == t:
arcs.append([t+1, t+1, assigns[e,t+1].Not()])
continue
task_follows = model.NewBoolVar(f'task_follows_{i+1}_{t+1}_{e}')
arcs.append([t+1, i+1, task_follows])
#task must start after previous task
model.Add(employee_task_intervals[e][i].StartExpr() >= employee_task_intervals[e][t].EndExpr()).OnlyEnforceIf(task_follows)
#task must start at the same location as previous task
model.Add(tasks[i+1][0] == tasks[t+1][1]).OnlyEnforceIf(task_follows)
model.AddImplication(task_follows, assigns[e,t+1])
model.AddImplication(task_follows, assigns[e,i+1])
model.AddCircuit(arcs)