Call Center Code

from abc import ABCMeta, abstractmethod
from collections import deque
from enum import Enum


class Rank(Enum):

    OPERATOR = 0
    SUPERVISOR = 1
    DIRECTOR = 2


class Employee(metaclass=ABCMeta):

    def __init__(self, employee_id, name, rank):
        self.employee_id = employee_id
        self.name = name
        self.rank = rank
        self.call = None
        self.call_center = None

    def take_call(self, call):
        """Assume the employee will always successfully take the call."""
        self.call = call
        self.call.employee = self
        self.call.state = CallState.IN_PROGRESS

    def complete_call(self):
        assert self.call_center is not None
        self.call.state = CallState.COMPLETE
        self.call_center.notify_call_completed(self.call)

    @abstractmethod
    def escalate_call(self):
        pass

    def _escalate_call(self):
        assert self.call_center is not None
        self.call.state = CallState.READY
        call = self.call
        self.call = None
        self.call_center.notify_call_escalated(call)

    def set_call_center(self, call_center):
        self.call_center = call_center


class Operator(Employee):

    def __init__(self, employee_id, name):
        super(Operator, self).__init__(employee_id, name, Rank.OPERATOR)

    def escalate_call(self):
        self.call.level = Rank.SUPERVISOR
        self._escalate_call()


class Supervisor(Employee):

    def __init__(self, employee_id, name):
        super(Supervisor, self).__init__(employee_id, name, Rank.SUPERVISOR)

    def escalate_call(self):
        self.call.level = Rank.DIRECTOR
        self._escalate_call()


class Director(Employee):

    def __init__(self, employee_id, name):
        super(Director, self).__init__(employee_id, name, Rank.DIRECTOR)

    def escalate_call(self):
        raise NotImplementedError('Directors must be able to handle any call')


class CallState(Enum):

    READY = 0
    IN_PROGRESS = 1
    COMPLETE = 2


class Call(object):

    def __init__(self, rank):
        self.state = CallState.READY
        self.rank = rank
        self.employee = None


class CallCenter(object):

    def __init__(self, operators, supervisors, directors):
        self.operators = []
        self.supervisors = []
        self.directors = []
        self.queued_calls = deque()
        for employee in operators + supervisors + directors:
            self.add_employee(employee)

    def dispatch_call(self, call):
        if call.rank not in (Rank.OPERATOR, Rank.SUPERVISOR, Rank.DIRECTOR):
            raise ValueError('Invalid call rank: {}'.format(call.rank))
        employee = None
        if call.rank == Rank.OPERATOR:
            employee = self._dispatch_call(call, self.operators)
        if call.rank == Rank.SUPERVISOR or employee is None:
            employee = self._dispatch_call(call, self.supervisors)
        if call.rank == Rank.DIRECTOR or employee is None:
            employee = self._dispatch_call(call, self.directors)

        if employee is None:
            self.queued_calls.append(call)

    def _dispatch_call(self, call, employees):
        for employee in employees:
            if employee.call is None:
                employee.take_call(call)
                return employee
        return None

    def add_employee(self, employee: Employee):
        if isinstance(employee, Operator):
            self.operators.append(employee)
        elif isinstance(employee, Supervisor):
            self.supervisors.append(employee)
        elif isinstance(employee, Director):
            self.directors.append(employee)
        else:
            raise ValueError('Invalid employee type: {}'.format(type(employee)))
        employee.set_call_center(self)


    def notify_call_escalated(self, call):
        pass

    def notify_call_completed(self, call):
        pass

    def dispatch_queued_call_to_newly_freed_employee(self, call, employee):
        pass

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook was prepared by [Donne Martin](https://github.com/donnemartin). Source and license info is on [GitHub](https://github.com/ido777/system-design-primer-update)."
   ]
  },
  {
   "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 available operators or the operator can't handle the call, does the call go to the supervisors?\n",
    "    * Yes\n",
    "* If there is no available 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.rank = 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.rank = 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
}

# test_call_center.py - Generated by CodiumAI

import pytest
from call_center.call_center import Operator, Supervisor, Director, Call, CallCenter, Rank

"""
Code Analysis:
- The CallCenter class is used to manage calls in a call center. 
- It takes three parameters: operators, supervisors, and directors. 
- The queued_calls attribute is a deque object, which is a double-ended queue. 
- The dispatch_call() method is used to assign a call to an employee based on the call's rank. 
- If an employee is not available, the call is added to the queued_calls deque. 
- The _dispatch_call() method is used to assign a call to an employee. 
- The notify_call_escalated() and notify_call_completed() methods are used to notify the call center when a call is escalated or completed. 
- The dispatch_queued_call_to_newly_freed_employee() method is used to assign a queued call to an employee who has just become available.
"""

"""
Test Plan:
- test_dispatch_call(): tests that the dispatch_call() method assigns the call to the correct employee based on the call's rank
- test_dispatch_call_with_no_available_employee(): tests that the dispatch_call() method adds the call to the queued_calls deque if no employee is available
- test_add_employee(): tests that the add_employee() method adds the employee to the correct list based on the employee's type

"""

class TestCallCenter():

    def setup_method(self, method):
        self.operators = [Operator(1, 'John'), Operator(2, 'Jane')]
        self.supervisors = [Supervisor(3, 'Bob')]
        self.directors = [Director(4, 'Alice')]
        self.call_center = CallCenter(self.operators, self.supervisors, self.directors)

    def test_dispatch_call(self):
        call = Call(Rank.OPERATOR)
        self.call_center.dispatch_call(call)
        assert self.operators[0].call == call

        call = Call(Rank.SUPERVISOR)
        self.call_center.dispatch_call(call)
        assert self.supervisors[0].call == call

        call = Call(Rank.DIRECTOR)
        self.call_center.dispatch_call(call)
        assert self.directors[0].call == call

    def test_dispatch_call_with_no_available_employee(self):
        for employee in self.operators + self.supervisors + self.directors:
            employee.take_call(Call(Rank.OPERATOR))
            assert employee.call is not None

        call = Call(Rank.OPERATOR)
        self.call_center.dispatch_call(call)
        assert len(self.call_center.queued_calls) == 1


    def test_add_employee(self):
        operator = Operator(5, 'Tom')
        supervisor = Supervisor(6, 'Jerry')
        director = Director(7, 'Sally')

        self.call_center.add_employee(operator)
        assert operator in self.call_center.operators

        self.call_center.add_employee(supervisor)
        assert supervisor in self.call_center.supervisors

        self.call_center.add_employee(director)
        assert director in self.call_center.directors

Comments