2017-03-04 09:50:44 +07:00
{
" cells " : [
{
" cell_type " : " markdown " ,
" metadata " : { } ,
" source " : [
2018-08-05 20:52:03 +07:00
" This notebook was prepared by [Donne Martin](https://github.com/donnemartin). Source and license info is on [GitHub](https://github.com/donnemartin/system-design-primer). "
2017-03-04 09:50:44 +07:00
]
} ,
{
" cell_type " : " markdown " ,
" metadata " : { } ,
" source " : [
" # Design a call center "
]
} ,
{
" cell_type " : " markdown " ,
" metadata " : { } ,
" source " : [
" ## Constraints and assumptions \n " ,
" \n " ,
" * What levels of employees are in the call center? \n " ,
" * Operator, supervisor, director \n " ,
" * Can we assume operators always get the initial calls? \n " ,
" * Yes \n " ,
" * If there is no free operators or the operator can ' t handle the call, does the call go to the supervisors? \n " ,
" * Yes \n " ,
" * If there is no free supervisors or the supervisor can ' t handle the call, does the call go to the directors? \n " ,
" * Yes \n " ,
" * Can we assume the directors can handle all calls? \n " ,
" * Yes \n " ,
" * What happens if nobody can answer the call? \n " ,
" * It gets queued \n " ,
" * Do we need to handle ' VIP ' calls where we put someone to the front of the line? \n " ,
" * No \n " ,
" * Can we assume inputs are valid or do we have to validate them? \n " ,
" * Assume they ' re valid "
]
} ,
{
" cell_type " : " markdown " ,
" metadata " : { } ,
" source " : [
" ## Solution "
]
} ,
{
" cell_type " : " code " ,
" execution_count " : 1 ,
" metadata " : {
" collapsed " : false
} ,
" outputs " : [
{
" name " : " stdout " ,
" output_type " : " stream " ,
" text " : [
" Overwriting call_center.py \n "
]
}
] ,
" source " : [
" %% writefile call_center.py \n " ,
" from abc import ABCMeta, abstractmethod \n " ,
" from collections import deque \n " ,
" from enum import Enum \n " ,
" \n " ,
" \n " ,
" class Rank(Enum): \n " ,
" \n " ,
" OPERATOR = 0 \n " ,
" SUPERVISOR = 1 \n " ,
" DIRECTOR = 2 \n " ,
" \n " ,
" \n " ,
" class Employee(metaclass=ABCMeta): \n " ,
" \n " ,
" def __init__(self, employee_id, name, rank, call_center): \n " ,
" self.employee_id = employee_id \n " ,
" self.name = name \n " ,
" self.rank = rank \n " ,
" self.call = None \n " ,
" self.call_center = call_center \n " ,
" \n " ,
" def take_call(self, call): \n " ,
" \" \" \" Assume the employee will always successfully take the call. \" \" \" \n " ,
" self.call = call \n " ,
" self.call.employee = self \n " ,
" self.call.state = CallState.IN_PROGRESS \n " ,
" \n " ,
" def complete_call(self): \n " ,
" self.call.state = CallState.COMPLETE \n " ,
" self.call_center.notify_call_completed(self.call) \n " ,
" \n " ,
" @abstractmethod \n " ,
" def escalate_call(self): \n " ,
" pass \n " ,
" \n " ,
" def _escalate_call(self): \n " ,
" self.call.state = CallState.READY \n " ,
" call = self.call \n " ,
" self.call = None \n " ,
" self.call_center.notify_call_escalated(call) \n " ,
" \n " ,
" \n " ,
" class Operator(Employee): \n " ,
" \n " ,
" def __init__(self, employee_id, name): \n " ,
" super(Operator, self).__init__(employee_id, name, Rank.OPERATOR) \n " ,
" \n " ,
" def escalate_call(self): \n " ,
" self.call.level = Rank.SUPERVISOR \n " ,
" self._escalate_call() \n " ,
" \n " ,
" \n " ,
" class Supervisor(Employee): \n " ,
" \n " ,
" def __init__(self, employee_id, name): \n " ,
" super(Operator, self).__init__(employee_id, name, Rank.SUPERVISOR) \n " ,
" \n " ,
" def escalate_call(self): \n " ,
" self.call.level = Rank.DIRECTOR \n " ,
" self._escalate_call() \n " ,
" \n " ,
" \n " ,
" class Director(Employee): \n " ,
" \n " ,
" def __init__(self, employee_id, name): \n " ,
" super(Operator, self).__init__(employee_id, name, Rank.DIRECTOR) \n " ,
" \n " ,
" def escalate_call(self): \n " ,
" raise NotImplemented( ' Directors must be able to handle any call ' ) \n " ,
" \n " ,
" \n " ,
" class CallState(Enum): \n " ,
" \n " ,
" READY = 0 \n " ,
" IN_PROGRESS = 1 \n " ,
" COMPLETE = 2 \n " ,
" \n " ,
" \n " ,
" class Call(object): \n " ,
" \n " ,
" def __init__(self, rank): \n " ,
" self.state = CallState.READY \n " ,
" self.rank = rank \n " ,
" self.employee = None \n " ,
" \n " ,
" \n " ,
" class CallCenter(object): \n " ,
" \n " ,
" def __init__(self, operators, supervisors, directors): \n " ,
" self.operators = operators \n " ,
" self.supervisors = supervisors \n " ,
" self.directors = directors \n " ,
" self.queued_calls = deque() \n " ,
" \n " ,
" def dispatch_call(self, call): \n " ,
" if call.rank not in (Rank.OPERATOR, Rank.SUPERVISOR, Rank.DIRECTOR): \n " ,
" raise ValueError( ' Invalid call rank: {} ' .format(call.rank)) \n " ,
" employee = None \n " ,
" if call.rank == Rank.OPERATOR: \n " ,
" employee = self._dispatch_call(call, self.operators) \n " ,
" if call.rank == Rank.SUPERVISOR or employee is None: \n " ,
" employee = self._dispatch_call(call, self.supervisors) \n " ,
" if call.rank == Rank.DIRECTOR or employee is None: \n " ,
" employee = self._dispatch_call(call, self.directors) \n " ,
" if employee is None: \n " ,
" self.queued_calls.append(call) \n " ,
" \n " ,
" def _dispatch_call(self, call, employees): \n " ,
" for employee in employees: \n " ,
" if employee.call is None: \n " ,
" employee.take_call(call) \n " ,
" return employee \n " ,
" return None \n " ,
" \n " ,
" def notify_call_escalated(self, call): # ... \n " ,
" def notify_call_completed(self, call): # ... \n " ,
" def dispatch_queued_call_to_newly_freed_employee(self, call, employee): # ... "
]
}
] ,
" metadata " : {
" kernelspec " : {
" display_name " : " Python 3 " ,
" language " : " python " ,
" name " : " python3 "
} ,
" language_info " : {
" codemirror_mode " : {
" name " : " ipython " ,
" version " : 3
} ,
" file_extension " : " .py " ,
" mimetype " : " text/x-python " ,
" name " : " python " ,
" nbconvert_exporter " : " python " ,
" pygments_lexer " : " ipython3 " ,
" version " : " 3.4.3 "
}
} ,
" nbformat " : 4 ,
" nbformat_minor " : 0
}