Merge pull request #4 from rodrigo-arenas/develop

RELEASE 0.1.0

.coveragerc

@@ -0,0 +1,8 @@
+[run]
+omit =
+    ./tests/*
+    ./venv/*
+    ./docs/*
+[report]
+precision = 2
+show_missing = True

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+#Pycharm
+.idea

README.md

@@ -1,2 +1,27 @@
 # pyworkforce
-Common tools for workforce management and production schedule and optimization
+Common tools for workforce management, schedule and optimization problems.
+
+
+### Queue systems:
+
+It can be used for solving the required number of positions to manage a number of transactions,
+under some systems pre-defined parameters and goals.
+
+
+#### Example:
+
+```python
+from pyworkforce.queuing.erlang import ErlangC
+
+erlang = ErlangC(transactions=100, asa=20/60, aht=3, interval=30, max_occupancy=0.85, shrinkage=0.3)
+
+positions_requirements = erlang.required_positions(service_level=0.8)
+print("positions_requirements: ", positions_requirements)
+
+
+>> positions_requirements:  {'raw_positions': 14, 
+                             'positions': 20, 
+                             'service_level': 0.8883500191794669, 
+                             'occupancy': 0.7142857142857143, 
+                             'waiting_probability': 0.1741319335950498}
+```

dev-requirements.txt

@@ -0,0 +1,4 @@
+pytest==6.2.2
+codecov==2.1.11
+pytest-cov==2.11.1
+twine==2.4.7

examples/queing/erlangc.py

@@ -0,0 +1,30 @@
+"""
+Example taken from: https://www.callcentrehelper.com/erlang-c-formula-example-121281.htm
+
+Requirement: Find the number of agents required to manage call center transactions
+under the following parameters:
+
+Number of calls: 100
+In a period of minutes: 30
+Average Handling Time (seconds): 180 (3 minutes)
+Required Service Level: 80%
+Target Answer Time (Seconds): 20
+Maximum Occupancy: 85%
+Shrinkage: 30%
+"""
+
+from pyworkforce.queuing.erlang import ErlangC
+
+erlang = ErlangC(transactions=100, asa=20/60, aht=3, interval=30, max_occupancy=0.85, shrinkage=0.3)
+
+positions_requirements = erlang.required_positions(service_level=0.8)
+print("positions_requirements: ", positions_requirements)
+
+achieved_service_level = erlang.service_level(positions=positions_requirements['raw_positions'])
+print("achieved_service_level: ", achieved_service_level)
+
+waiting_probability = erlang.waiting_probability(positions=positions_requirements['raw_positions'])
+print("waiting_probability: ", waiting_probability)
+
+achieved_occupancy = erlang.achieved_occupancy(positions=positions_requirements['raw_positions'])
+print("achieved_occupancy: ", achieved_occupancy)

pyworkforce/queuing/erlang.py

@@ -0,0 +1,98 @@
+from math import exp, ceil, floor
+
+
+class ErlangC:
+    def __init__(self, transactions: float, asa: float, aht: float,
+                 interval: int = None, shrinkage=0.0, max_occupancy=1.0,
+                 **kwargs):
+        """
+        Computes the number of positions required fo attend a number of transactions in a queue system based on ErlangC
+        Implementation based on: https://lucidmanager.org/data-science/call-centre-workforce-planning-erlang-c-in-r/
+
+        :param transactions: number of total transactions that comes in
+        :param aht: average handling time of a transaction (minutes)
+        :param interval: Interval length (minutes)
+        :param shrinkage: Percentage of time that an operator unit is not available
+        :param max_occupancy: Maximum percentage of time that an attending position can be occupied
+        """
+        self.n_transactions = transactions
+        self.asa = asa
+        self.aht = aht
+        self.interval = interval
+        self.intensity = (self.n_transactions / self.interval) * self.aht
+        self.shrinkage = shrinkage
+        self.max_occupancy = max_occupancy
+
+    def waiting_probability(self, positions, scale_positions=False):
+        """
+        :param positions: Number of positions to attend the transactions
+        :param scale_positions: True if the positions where calculated using shrinkage
+        :return: the probability of a transaction waits in queue
+        """
+        if scale_positions:
+            productive_positions = floor((1 - self.shrinkage) * positions)
+        else:
+            productive_positions = positions
+
+        erlang_b_inverse = 1
+        for position in range(1, productive_positions + 1):
+            erlang_b_inverse = 1 + (erlang_b_inverse * position / self.intensity)
+
+        erlang_b = 1 / erlang_b_inverse
+        return productive_positions * erlang_b / (productive_positions - self.intensity * (1 - erlang_b))
+
+    def service_level(self, positions, scale_positions=False):
+        """
+        :param positions: Number of positions attending
+        :param scale_positions: True if the positions where calculated using shrinkage
+        :return: achieved service level
+        """
+        if scale_positions:
+            productive_positions = floor((1 - self.shrinkage) * positions)
+        else:
+            productive_positions = positions
+
+        probability_wait = self.waiting_probability(productive_positions, scale_positions=False)
+        exponential = exp(-(productive_positions - self.intensity) * (self.asa / self.aht))
+        return max(0, 1 - (probability_wait * exponential))
+
+    def achieved_occupancy(self, positions, scale_positions=False):
+        """
+        :param positions: Number of raw positions
+        :param scale_positions: True if the positions where calculated using shrinkage
+        :return: Expected occupancy of positions
+        """
+        if scale_positions:
+            productive_positions = floor((1 - self.shrinkage) * positions)
+        else:
+            productive_positions = positions
+
+        return self.intensity/productive_positions
+
+    def required_positions(self, service_level):
+        """
+        :param service_level: Target service level
+        :return: Number of positions needed to ensure the required service level
+        """
+        positions = round(self.intensity + 1)
+        achieved_service_level = self.service_level(positions, scale_positions=False)
+        while achieved_service_level < service_level:
+            positions += 1
+            achieved_service_level = self.service_level(positions, scale_positions=False)
+
+        achieved_occupancy = self.achieved_occupancy(positions, scale_positions=False)
+
+        if achieved_occupancy > self.max_occupancy:
+            raw_positions = ceil(self.intensity / self.max_occupancy)
+            achieved_occupancy = self.achieved_occupancy(raw_positions)
+            achieved_service_level = self.service_level(raw_positions)
+
+        raw_positions = ceil(positions)
+        waiting_probability = self.waiting_probability(positions=raw_positions)
+        positions = ceil(raw_positions / (1 - self.shrinkage))
+
+        return {"raw_positions": raw_positions,
+                "positions": positions,
+                "service_level": achieved_service_level,
+                "occupancy": achieved_occupancy,
+                "waiting_probability": waiting_probability}

setup.py

@@ -0,0 +1,28 @@
+import pathlib
+from setuptools import setup
+
+# The directory containing this file
+HERE = pathlib.Path(__file__).parent
+
+# The text of the README file
+README = (HERE / "README.md").read_text()
+
+# This call to setup() does all the work
+setup(
+    name="pyworkforce",
+    version="0.1.0",
+    description="Common tools for workforce management, schedule and optimization problems",
+    long_description=README,
+    long_description_content_type="text/markdown",
+    url="https://github.com/rodrigo-arenas/pyworkforce",
+    author="Rodrigo Arenas",
+    author_email="[email protected]",
+    license="MIT",
+    classifiers=[
+        "License :: MIT License",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.8",
+    ],
+    packages=["pyworkforce"],
+    include_package_data=True,
+)

tests/test_queing.py

@@ -0,0 +1,16 @@
+from pyworkforce.queuing.erlang import ErlangC
+
+
+def test_expected_erlangc_results():
+    erlang = ErlangC(transactions=100, asa=0.33, aht=3, interval=30, shrinkage=0.3)
+    results = erlang.required_positions(service_level=0.8)
+    raw_positions = results['raw_positions']
+    positions = results['positions']
+    service_level = results['service_level']
+    occupancy = results['occupancy']
+
+    assert raw_positions == 14
+    assert positions == 20
+    assert round(service_level, 3) == 0.888
+    assert round(occupancy, 3) == 0.714
+