[OR-Tools 스터디] The Job Shop Problem
한가지 흔한 스케줄링 문제는 Jop shop으로, 이는 여러개의 job들을 몇가지 기계들에 동작시키는 것이다.
각 job은 일련의 task들로 구성되며, 주어진 순서에 맞게 반드시 동작되어야 한다.
그리고 각 task들은 특정 기계에서 동작되어야한다.
예로, job은 자동자와 같은 단일 소비재의 제조일 수 있다.
문제는 모든 작업이 완료되는 데 걸리는 시간인 schedule(예정시간)의 길이를 최소화하기 위해 시스템에 task를 schedule(예정시간)하는 것이다.
jop shop 문제에 대해 몇가지 constraint가 있다.
- 이전 job의 task가 완료될 때 까지 다른 job의 task를 시작할 수 없다.
- 기계는 한번에 하나의 task만 동작시킬 수 있다.
- task는 한번 시작하면 반드시 완성될 때 까지 돌아가야 한다.
Example Problem
아래는 jop shop 문제의 간단한 예시로, 각 task가 (m, p) 의 숫자들의 쌍으로 레이블되어져 있으며 여기서 m은 task가 반드시 동작되어야하는 기계들의 숫자이며, p는 task의 동작 시간으로 task에 요구되는 시간의 양이다. (job과 기계들의 순번은 0부터 시작한다.)
- job 0 = [(0, 3), (1, 2), (2, 2)]
- job 1 = [(0, 2), (2, 1), (1, 4)]
- job 2 = [(1, 4), (2, 3)]
이 예시에서, job 0은 3가지 task들을 가진다. 첫번째로 (0, 3)은 0번째 기계에서 3단위의 시간동안 동작해야한다.
두번째로 (1,2)는 1번째 기계에서 2단위의 시간동안 동작해야한다.
다합쳐서 총 8개의 task들이 있다.
A solution for the problem
job shop problem에 대한 해결책은 각 task에 대해 시작 시간을 할당하는 것으로, 이는 위에 주어진 constraint들을 충족시킨다.
아래 diagram은 문제에 대한 한가지 가능한 해결책을 보여준다.

각 job에 대한 task들이 겹치지 않는 시간 간격들로 문제에 주어진 순서에 맞게 스케줄되어있는 것을 확인할 수 있다.
이 해결책의 길이는 12로, 모든 3가지 job들을 완성했을 때의 첫 시간이다.
그러나, 이 문제에서 이 해결책은 최적이 아니라는 것을 아래에서 볼 수 있을 것이다.
Variables and constraints for the problem
이 섹션은 이 문제에 대해 어떻게 변수들과 constraint들을 설정하는지를 보여준다.
첫번째는, task(i, j)가 job i에 대한 순서에서 j번째 task임을 나타낸다.
예시로, task(0, 2)는 job 0의 두번째 task임을 의미한다.
다음으로 t_i,j는 task(i,j)의 시작시간을 의미한다. t_i,j는 job shop 문제에서 변수들이다.
해결책을 찾는다는 것은 문제의 요구사항을 충족하는 여러 변수들에 대한 값을 찾는 것도 포함한다.
jop shop 문제에 대한 constraint들의 두가지 종류가 있다.
- Precedence constraints : 이들은 같은 job에 있어서 어느 두가지 연속적인 task들이 있는 상태일 때 발생하는데, 두번째 것이 시작할 수 있기 전에 첫번째 것이 반드시 완료되어야 한다. 예시로, task(0, 2)와 task(0, 3)이 job 0에 대해 연속적인 task들이다. tas(0, 2)에 대한 처리 시간은 2이기 때문에, task(0, 3)의 시작시간은 반드시 task2가 시작된 후인 최소 2단위여야 한다. 결과적으로, 다음의 constraint를 얻을 수 있다. - t_0,2 + 2 <= t_0,3
- No overlap constraints : 이들은 기계가 동시에 두 가지 task들을 수행할 수 없다는 규제에서 발생한 것이다. 예를 들어, task(0, 2)와 task(2,1)은 둘 다 기계 1에서 동작된다. 이들의 처리 속도는 각각 2와 4이기 때문에, 따르는 constraint들 중 하나는 반드시 지켜야한다.
- t_0,2 + 2 <= t_2,1(task(0, 2)가 task(2,1) 전에 예약된 경우)
- t_2,1 + 4 <= t_0,2(task(2, 1)이 task(0,2) 이전에 예약된 경우)
Objective for the problem
job shop 문제의 objective는 makespan을 최소화하기 위한 것이다 : job의 처음 시작 시점부터 마지막 끝나는 시간까지의 총 시간 길이
A Program solution
import collections
from ortools.sat.python import cp_model
def JopShopProblem():
#define the data
jobs_data = [
#task = (machine_id, processing_time).
[(0, 3), (1, 2), (2,2)], #Job 0
[(0, 2), (2, 1), (1, 4)], #Job 1
[(1, 4), (2, 3)] #Job 2
]
machines_count = 1 + max(task[0] for job in jobs_data for task in job) #3
# [(0, 3), (1, 2), (2, 2)] : job[0]
# task[0]
# 0
# 1
# 2
# [(0, 2), (2, 1), (1, 4)] : job[0]
# task[0]
# 0
# 2
# 1
# [(1, 4), (2, 3)] : job[0]
# task[0]
# 1
# 2
all_machines = range(machines_count)
#computes horizon dynamically as the sum of all durations.
horizon = sum(task[1] for job in jobs_data for task in job) #21. 모든 job들의 모든 task들이 처리되는 총 시간
#horizon은 다음의 이유들로 인해 모든 task들이 완료되기 위해 충분히 커야함.
# : 만약 겹치지 않는 시간 interval들로 task들을 예약한다면(최적화 안된 솔루션), 예약 총 길이는 확실히 horizon이다.
#따라서 최적화된 솔루션의 기간은 어떠한 horizon보다 클 수 없다.
#declare the model
model = cp_model.CpModel()
#define the variables
#named tuple to store information about created variables.
task_type = collections.namedtuple('task_type', 'start end interval')
#named tuple to manipulate solution information
assigned_task_type = collections.namedtuple('assigned_task_type',
'start job index duration')
#creates job intervals and add to the corresponding machine lists.
all_tasks = {}
machine_to_intervals = collections.defaultdict(list)
for job_id, job in enumerate(jobs_data):
for task_id, task in enumerate(job):
machine = task[0]
duration = task[1]
suffix = '_%i_%i' % (job_id, task_id)
start_var = model.NewIntVar(0, horizon, 'start' + suffix) #task의 시작 시간
end_var = model.NewIntVar(0, horizon, 'end' + suffix) #task의 끝나는 시간
# NewIntervalVar: interval variable을 만드는 메소드이다. 각 task들에 대한 가변 시간 간격이다.
interval_var = model.NewIntervalVar(start_var, duration, end_var, 'interval' + suffix) #(task의 시작 시간 변수, task의 시간 간격 길이, task의 끝나는 시간 변수, 간격 변수의 이름)
#결과적으로 어떠한 정답이건, end_var - start_var = duration이 되어야 한다.
all_tasks[job_id, task_id] = task_type(start = start_var, end = end_var, interval=interval_var)
machine_to_intervals[machine].append(interval_var)
#Define the constraints
#create and add disjunctive constraints
for machine in all_machines:
model.AddNoOverlap(machine_to_intervals[machine])
#이는 동일한 기계들의 시간이 겹치지 않도록 no overlap constraint들을 설정하는 메소드이다.
#precedences inside a job
#precedence constraint : 시간을 겹쳐서 같은 job에 대해 연속적으로 task를 수행하는 것을 막는 것이다 .
for job_id, job in enumerate(jobs_data):
for task_id in range(len(job) - 1):
#task의 끝나는 시간은 job의 다음 task 시작 시간보다 이전이어야 한다.
model.Add(all_tasks[job_id, task_id + 1].start
>= all_tasks[job_id, task_id].end)
#define the objective
#makespan objective
obj_var = model.NewIntVar(0, horizon, 'makespan')
#변수 obj_var를 생성하는데, 이 값은 모든 job들에 대한 끝나는 시간의 최대값이다. 즉, makespan이다.
model.AddMaxEquality(obj_var, [
all_tasks[job_id, len(job) - 1].end
for job_id, job in enumerate(jobs_data)
])
model.Minimize(obj_var)
#invoke the solver
solver = cp_model.CpSolver()
status = solver.Solve(model)
if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
print('Solution:')
# 각 기계에 대해 할당된 task들의 하나의 리스트를 만든다.
assigned_jobs = collections.defaultdict(list)
for job_id, job in enumerate(jobs_data):
for task_id, task in enumerate(job):
machine = task[0]
assigned_jobs[machine].append(
assigned_task_type(start = solver.Value(
all_tasks[job_id, task_id].start),
job=job_id,
index = task_id,
duration = task[1]
)
)
#create per machine output lines
output = ''
for machine in all_machines:
#sort by starting time
assigned_jobs[machine].sort()
sol_line_tasks ='Machine ' + str(machine) + ':'
sol_line = ' '
for assigned_task in assigned_jobs[machine]:
name = 'job_%i_task_%i' % (assigned_task.job, assigned_task.index)
#Add spaces to output to align columns.
sol_line_tasks += '%-15s' % name
start = assigned_task.start
duration = assigned_task.duration
sol_tmp = '[%i, %i]' % (start, start + duration)
#add spaces to output to align columns.
sol_line += '%-15s' % sol_tmp
sol_line += '\n'
sol_line_tasks += '\n'
output += sol_line_tasks
output += sol_line
# Finally print the solution found.
print(f'Optimal Schedule Length: {solver.ObjectiveValue()}')
print(output)
else:
print('No solution found.')
JopShopProblem()
Reference :
https://developers.google.com/optimization/scheduling/job_shop