[OR-Tools 스터디] Employee Scheduling

다교대 시스템으로 일하는 고용자들이 있는 조직들에서는 매일 충분한 노동자들이 schedule될 필요가 있다. 

전형적으로, schedule은 "직원들은 행으로 2교대로 일하면 안된다"와 같은 제약들을 가지고 있다. 

 

모든 제약조건들을 만족하는 schedule을 찾는 것은 계산적으로 어려울 수 있다. 

 

따르는 섹션에서는 직원 scheduling 문제의 두가지 예시를 보여주고, CP-SAT solver를 사용하여 그것들을 어떻게 해결할 것인지를 보여준다.

 

A nurse scheduling problem

다음 예제에서는 병원 관리인들이 3일 동안 4명의 간호사들의 일정을 짜야할 필요가 있다. 그들은 다음의 조건들을 따른다. 

• 각 하루를 8시간씩 3파트로 나눈다
• 매일 모든 교대는 한명의 간호사에게 할당되며, 어떠한 간호사들도 두번 이상 교대를 수행해서는 안된다.
• 각 간호사들은 3일동안 최소 두번의 교대가 할당된다.

 

from ortools.sat.python import cp_model

#register a solutions callback
#각 해결책을 부를 때 solver에 callback을 등록할 필요가 있다.


def EmployeeScheduling():
    num_nurses = 4
    num_shifts = 3
    num_days = 3
    all_nurses = range(num_nurses)
    all_shifts = range(num_shifts)
    all_days = range(num_days)

    #create the model
    model = cp_model.CpModel()

    #create the variables
    shifts = {}
    for n in all_nurses:
        for d in all_days:
            for s in all_shifts:
                shifts[(n, d, s)] = model.NewBoolVar('shift_n%id%is%i' % (n, d, s))
                #만약 교대s를 간호사 n에게 day d에 부여했다면 1이고 아니면 0

    #Assign nurses to shifts
    #다음으로 따르는 제약들에 따라 간호사들에게 어떻게 교대를 배정할지를 보여준다.
    #1. 각 교대는 매일 한명의 간호사에게 배정된다
    #2. 각 간호사들은 매일 최대 한번만 교대 가능하다
    for d in all_days:
        for s in all_shifts:
            model.AddExactlyOne(shifts[(n, d, s)] for n in all_nurses)
            #마지막 줄은 매번 교대 때 마다 간호사들의 sum에 shift를 1로 부여한다.

    #각 간호사들은 매일 최대 한번씩 교대하도록하는 코드이다.
    for n in all_nurses:
        for d in all_days:
            model.AddAtMostOne(shifts[(n, d, s)] for s in all_shifts)
    #각 간호사들에 대해, 교대의 합은 최대 1로 부여된다.

    #Assign shifts evenly
    #다음으로 간호사들을 가능한 고르게 교대를 배정하는 방법을 보여준다.
    #3일동안 총 9번의 교대가 있기 때문에, 4명의 간호사별로 한 명 당 두번의 교대를 배정할 수 있다.
    #그 후에는 어떤 간호사에게도 배정될 수 있는 교대 근무가 한 번 남아 있을 것이다.
    #따르는 코드는 각 간호사들이 3일 동안 최소 두번의 교대근무를 보장해주는 것이다.

    min_shifts_per_nurse = (num_shifts * num_days) // num_nurses #몫 계산
    #num_shifts * num_days가 매 스케줄 기간에 전체 교대이기 때문에, 최소 (num_shifts * num_days) // num_nurses 만큼 각 간호사들에게
    # 할당될 수 있다. 한 번의 교대는 남게 된다.
    if num_shifts * num_days % num_nurses == 0:
        max_shifts_per_nurse = min_shifts_per_nurse
    else:
        max_shifts_per_nurse = min_shifts_per_nurse + 1
    for n in all_nurses:
        shifts_worked = []
        for d in all_days:
            for s in all_shifts:
                shifts_worked.append(shifts[(n, d, s)])
        model.Add(min_shifts_per_nurse <= sum(shifts_worked))
        model.Add(sum(shifts_worked) <= max_shifts_per_nurse)

    #update solver parameters
    # non-optimization model의 경우, 모든 해결책들에 대해 검색이 가능하다.
    solver = cp_model.CpSolver()
    solver.parameters.linearization_level = 0
    #enumerate all solutions
    solver.parameters.enumerate_all_solutions = True

    class NursesPartialSolutionPrinter(cp_model.CpSolverSolutionCallback):
        def __init__(self, shifts, num_nurses, num_days, num_shifts, limit):
            cp_model.CpSolverSolutionCallback.__init__(self)
            self._shifts = shifts
            self._num_nurses = num_nurses
            self._num_days = num_days
            self._num_shifts = num_shifts
            self._solution_count = 0
            self._solution_limit = limit

        def on_solution_callback(self):
            self._solution_count += 1
            print('Solution %i' % self._solution_count)
            for d in range(self._num_days):
                print('Day %i' % d)
                for n in range(self._num_nurses):
                    is_working = False
                    for s in range(self._num_shifts):
                        if self.Value(self._shifts[(n, d, s)]):
                            is_working = True
                            print(' Nurse %i works shift %i' % (n, s))
                    if not is_working:
                        print('    Nurse {} does not work'.format(n))
            if self._solution_count >= self._solution_limit:
                print('Stop search after %i solutions' % self._solution_limit)
                self.StopSearch()

        def solution_count(self):
            return self._solution_count
    solution_limit = 5
    solution_printer = NursesPartialSolutionPrinter(shifts, num_nurses, num_days, num_shifts, solution_limit)

    #invoke the solver
    solver.Solve(model, solution_printer)


EmployeeScheduling()

 

 

Scheduling with shift requests

 

이번 섹션에서는, 이전 예시와 특정 교대에 대해 간호사의 요청을 더하였다. 우리는 스케줄에 대해 그것들을 충족시키는 요청들의 수를 최대화하고자 한다.

대부분의 스케줄링 문제에 있어서 모든 가능한 스케줄을 출력하는 것이 일반적으로 실용적이지 않기에, objective함수를 최적화하는 것이 좋다.

 

이 예제는 이전의 예시와 동일한 constraint들을 가지고 있다.  

 

from ortools.sat.python import cp_model

def Opt_EmployeeScheduling():
    num_nurses = 5
    num_shifts = 3
    num_days = 7
    all_nurses = range(num_nurses)
    all_shifts = range(num_shifts)
    all_days = range(num_days)
    shift_requests = [[[0, 0, 1], [0, 0, 0], [0, 0, 0], [0, 0, 0], [0, 0, 1],
                       [0, 1, 0], [0, 0, 1]],
                      [[0, 0, 0], [0, 0, 0], [0, 1, 0], [0, 1, 0], [1, 0, 0],
                       [0, 0, 0], [0, 0, 1]],
                      [[0, 1, 0], [0, 1, 0], [0, 0, 0], [1, 0, 0], [0, 0, 0],
                       [0, 1, 0], [0, 0, 0]],
                      [[0, 0, 1], [0, 0, 0], [1, 0, 0], [0, 1, 0], [0, 0, 0],
                       [1, 0, 0], [0, 0, 0]],
                      [[0, 0, 0], [0, 0, 1], [0, 1, 0], [0, 0, 0], [1, 0, 0],
                       [0, 1, 0], [0, 0, 0]]]

    #create the model
    model = cp_model.CpModel()

    #create the variables
    #이전 에제에서 변수들을 더한 것에서, 데이터는 또한 각 날마다 3번의 교대에 따라 3인 1조로 포함시켰다.
    #3인의 각 요소들은 교대 요청 여부에 따라 0 또는 1로 표기되어있다.
    #예를 들어, 첫번째 행의 5번째 포지션에 위치한 [0, 0, 1]과 같은 경우는 간호사 1번이 5일째에 교대 3을 요청했음을 나타낸다.
    shifts = {}
    for n in all_nurses:
        for d in all_days:
            for s in all_shifts:
                shifts[(n, d, s)] = model.NewBoolVar('shift_n%id%is%i' % (n, d, s))

    #create the constraints
    for d in all_days:
        for s in all_shifts:
            model.AddExactlyOne(shifts[(n, d, s)] for n in all_nurses)
    for n in all_nurses:
        for d in all_days:
            model.AddAtMostOne(shifts[(n, d, s)] for s in all_shifts)
    min_shifts_per_nurse = (num_shifts * num_days) // num_nurses
    if num_shifts * num_days % num_nurses == 0:
        max_shifts_per_nurse = min_shifts_per_nurse
    else:
        max_shifts_per_nurse = min_shifts_per_nurse + 1
    for n in all_nurses:
        num_shifts_worked = 0
        for d in all_days:
            for s in all_shifts:
                num_shifts_worked += shifts[(n, d, s)]
        model.Add(min_shifts_per_nurse <= num_shifts_worked)
        model.Add(num_shifts_worked <= max_shifts_per_nurse)

    #objective for the example
    model.Maximize(
        sum(shift_requests[n][d][s] * shifts[(n, d, s)] for n in all_nurses
            for d in all_days for s in all_shifts)
    )
    # shift_requests[n][d][s] * shifts[(n, d, s)]는 만약 shift s가 간호가 n에게 d day에 배정되었고, 간호사가 교대를 요청했다면 1이 된다.
    # objective는 요청에 충족되는 교대 배정의 수이다.

    #invoke the solver
    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    if status == cp_model.OPTIMAL:
        print('Solution:')
        for d in all_days:
            print('Day', d)
            for n in all_nurses:
                for s in all_shifts:
                    if solver.Value(shifts[(n, d, s)]) == 1:
                        if shift_requests[n][d][s] == 1:
                            print('Nurse', n, 'works shift', s, '(requested).')
                        else:
                            print('Nurse', n, 'works shift', s, '(not requested).')
            print()
        print(f'Number of shift requests met = {solver.ObjectiveValue()}',
              f'(out of {num_nurses * min_shifts_per_nurse})')
    else:
        print('No optimal solution found!')
Opt_EmployeeScheduling()

Reference : 

https://developers.google.com/optimization/scheduling/employee_scheduling