Using simpy to simulate a workflow

simpy-process.py

import simpy
import mesa

import random

# A claim arrives every 3 minutes. We'll model this with 
# # an exponential distribution and lambda = 1/3

# One team for intake and assignment.
# - Each employee can only work 1 claim at a time.
# Two teams of handlers: simple, and complex.
# - Each simple team employee can work 2 at a time.
# - Each complex team employee can work 4 at a a time.

# Assume 80% of claims are expedited to simple team.
CLAIM_ARRIVAL_MEAN = 3
ARRIVALS = [] # list that holds arrival time of each claim by id.

N_EMP_INTAKE = 3
EMP_INTAKE_CAPACITY = 1
EMP_INTAKE_AHT_MEAN = 8 # average time (minutes) spent in intake process
EMP_INTAKE_AHT_STD = 3 # std of time (minutes) spent in intake process

N_EMP_HANDLE_SIMPLE = 3
EMP_HANDLE_SIMPLE_CAPACITY = 2
EMP_HANDLE_SIMPLE_AHT_MEAN = 20 # average time (minutes) spent handling a simple claim.
EMP_HANDLE_SIMPLE_AHT_STD = 5 # std time (minutes) spent handling a simple claim.

N_EMP_HANDLE_COMPLEX = 3
EMP_HANDLE_COMPLEX_CAPACITY = 4
EMP_HANDLE_COMPLEX_AHT_MEAN = 40 # average time (minutes) spent handling a complex claim.
EMP_HANDLE_COMPLEX_AHT_STD = 8 # std time (minutes) spent handling a complex claim.

PCT_EXPEDITED_MEAN = 0.8 # these go directly to the SIMPLE Handler team.

# Pre-generate all claims arriving to the simulation so that if we change the configuration
# we have a consistent set of arrivals to compare configurations against.
random.seed(315)
N_CLAIMS = 3
CLAIMS = [ random.expovariate(1 / CLAIM_ARRIVAL_MEAN) for claim in range(N_CLAIMS)]

def claim_arrival(env, claim_intake_team, claim_handle_simple_team, claim_handle_complex_team):
    """
    Simulate a claim arriving every CLAIM_ARRIVAL_MEAN minutes.

    This is the top-level simpy event for simulation: all other
    events originate from a claim arriving.
    """
    next_claim_id = 0
    while True:
        # Take the next claim from the list CLAIMS generated above.
        # if len(CLAIMS) > 0:
        next_claim = random.expovariate(1 / CLAIM_ARRIVAL_MEAN)
        ARRIVALS.append({"claim_id": next_claim_id, "time_arrival_delta": next_claim})
        
        # next_claim = CLAIMS.pop()
        print("Next claim arriving in {} minutes.".format(next_claim))
        
        # Wait for the claim to arrive
        yield env.timeout(next_claim)

        next_claim_id += 1

        # while len(CLAIMS) > 0:
        env.process(intake_claim(env, 0, claim_intake_team, claim_handle_simple_team, claim_handle_complex_team))

def intake_claim(env, claims_processed, intake_team, simple_claim_handler_team, complex_claim_handler_team):
    intake_queue_begin = env.now

    # Assign claim randomly to one of the team employees.
    random_employee_idx = int(random.randint(0, (len(intake_team) - 1)))
    intake_employee = intake_team[random_employee_idx]

    with intake_employee.request() as req:
        yield req
        intake_queue_end = env.now
        
        # Work the intake process
        print("claim intake start")
        intake_begin = env.now
        yield env.timeout(random.gauss(EMP_INTAKE_AHT_MEAN, EMP_INTAKE_AHT_STD))
        intake_end = env.now
        print("claim intake done")

        # Randomly determine if this claim is expedited (simple team) or complex.
        if random.random() < PCT_EXPEDITED_MEAN:
            # simple path
            env.process(handle_simple_claim(env, simple_claim_handler_team))
        else:
            # complex path
            env.process(handle_complex_claim(env, complex_claim_handler_team))

def handle_simple_claim(env, team_employees):
    handle_simple_queue_begin = env.now

    # Assign claim randomly to one of the team employees.
    random_employee_idx = int(random.randint(0, (len(team_employees) - 1)))
    simple_claim_handler_employee = team_employees[random_employee_idx]

    with simple_claim_handler_employee.request() as req:
        yield req
        handle_simple_queue_end = env.now

        # Work the claim
        print("simple claim handle start at {}".format(env.now))
        yield env.timeout(random.gauss(EMP_HANDLE_SIMPLE_AHT_MEAN, EMP_HANDLE_SIMPLE_AHT_STD))
        print("simple claim handle done at {}".format(env.now))

def handle_complex_claim(env, team_employees):
    handle_complex_queue_begin = env.now

    # Assign claim randomly to one of the team employees.
    random_employee_idx = int(random.randint(0, (len(team_employees) - 1)))    
    complex_claim_handler_employee = team_employees[random_employee_idx]

    with complex_claim_handler_employee.request() as req:
        yield req
        handle_complex_queue_end = env.now

        # Work the claim
        print("complex claim handle start at {}".format(env.now))
        yield env.timeout(random.gauss(EMP_HANDLE_COMPLEX_AHT_MEAN, EMP_HANDLE_COMPLEX_AHT_STD))
        print("complex claim handle end at {}".format(env.now))

if __name__ == "__main__":
    # Initialize the simulation environment.
    env = simpy.Environment()

    # Initialize the teams with employees.
    claim_intake_employees = [ simpy.Resource(env, capacity = EMP_INTAKE_CAPACITY) for employee in range(N_EMP_INTAKE) ]
    claim_handle_simple_employees = [ simpy.Resource(env, capacity = EMP_HANDLE_SIMPLE_CAPACITY) for employee in range(N_EMP_HANDLE_SIMPLE) ]
    claim_handle_complex_employees = [ simpy.Resource(env, capacity = EMP_HANDLE_COMPLEX_CAPACITY) for employee in range(N_EMP_HANDLE_COMPLEX) ]

    env.process(
        claim_arrival(env, 
            claim_intake_employees, 
            claim_handle_simple_employees, 
            claim_handle_complex_employees)
        )
    
    env.run(until = 200)
    print("ARRIVALS", ARRIVALS)