Examples

Bike Sharing

import numpy as np
import pandas as pd
import random
from chronon import ProcessManager, EventManager, Process, Resource

STATIONS = ['Station A', 'Station B', 'Station C']
TARGET_OCCUPATION = 0.7
N_CYCLISTS = 7

pm = ProcessManager()

# Defining distances between stations
station_x = np.array([round(random.uniform(0, 30), 2) for s in STATIONS])
distances = abs(station_x.reshape(-1, 1) - station_x)

# Bike process
class IncludeBike(Process):
    def definition(self, bike):
        yield bike.waits(bike.initial_dock)

# Cyclist process
class Cycle(Process):
    def definition(self, cyclist):
        # Request a bike-as-resource
        cyclist.set_checkpoint(f'Requested a bike at {cyclist.from_station}')
        response = yield cyclist.waits(
            self.get_resources(by_properties={'type': 'bike'}),
            which='any',
            having={'station': cyclist.from_station}
        )

        # Get the bike-as-user based on the name of the bike-as-resource obtained by the cyclist-as-user
        bike_resource = response.events[0].resource
        bike = cyclist.get_user(bike_resource.name)

        # Undock bike-as-user from the station of origin
        origin_dock = self.get_resources(by_user=bike)[0]
        cyclist.set_checkpoint(
            f'Got {bike_resource.name} at {origin_dock.station}'
        )
        bike.releases(origin_dock)
        bike.set_checkpoint(
            f'Undocked from {origin_dock.name} at {origin_dock.station}'
        )

        # Cycle to destination
        cycle_duration = distances[
            STATIONS.index(cyclist.from_station), 
            STATIONS.index(cyclist.to_station)
        ]
        bike.set_checkpoint(f'Moving')
        yield bike.waits(cycle_duration)

        # Dock bike-as-user at the destination station
        bike.set_checkpoint(f'Requested a dock at {cyclist.to_station}')
        response = yield bike.waits(
            self.get_resources(by_properties={
                'type': 'dock', 
                'station': cyclist.to_station
            }),
            which='any',
        )
        dest_dock = response.events[0].resource
        bike.set_checkpoint(f'Docked on {dest_dock.name} at {dest_dock.station}')

        # Finish cycle
        cyclist.releases(bike_resource)
        cyclist.set_checkpoint(
            f'Left {bike_resource.name} at {dest_dock.station}'
        )

pm.attach_process(IncludeBike)
pm.attach_process(Cycle)
pm.set_flow(initial_process='IncludeBike')
pm.set_flow(final_process='IncludeBike')
pm.set_flow(initial_process='Cycle')
pm.set_flow(final_process='Cycle')

em = EventManager(pm)

# Custom Resource class for bikes
class BikeResource(Resource):
    @property
    def station(self):
        # Bike-user and bike-resource have the same name
        dock = self.rm.get_resources(by_user=self.name)
        if len(dock) == 0:
            return None
        else:
            return dock[0].station

# Creating station docks as resources
docks = [random.randint(2, 4) for s in STATIONS]
for index, [name, capacity] in enumerate(zip(STATIONS, docks)):
    for d in range(capacity):
        pm.create_resource(f'dock_{index}{d}', type='dock', station=name)

# Creating bikes as resources (for customers) and users (of docks)
bid = 0
dock_resources = pm.get_resources(by_properties={'type': 'dock'})
for dock in dock_resources:
    if random.uniform(0, 1) < TARGET_OCCUPATION:
        pm.create_resource(
            f'bike_{bid}',
            custom_resource=BikeResource,
            type='bike'
        )
        em.create_user(
            f'bike_{bid}',
            initial_process='IncludeBike',
            initial_dock=dock
        )
        bid += 1

# Creating cyclists as users
for c in range(N_CYCLISTS):
    from_station = random.choice(STATIONS)
    to_station = random.choice([s for s in STATIONS if s != from_station])
    user = em.create_user(
        f'cyclist_{c}',
        instant=round(random.uniform(0, 30), 2),
        initial_process='Cycle',
        from_station=from_station,
        to_station=to_station
    )

# Running simulation
em.run()

# Reporting
for r in pm.rm.resources:
    print('\n', r)
    print(pm.get_resource(r).usage)

print(em.checkpoints)

print(em.get_state(at=5))

Carwash

As seen in here.

import numpy as np
import random
from datetime import timedelta, datetime
from chronon import ProcessManager, EventManager, Process


RANDOM_SEED = 42
NUM_MACHINES = 2  # Number of machines in the carwash
WASHTIME = 5      # Minutes it takes to clean a car
T_INTER = 3       # Create a car every ~T_INTER minutes
SIM_TIME = 25     # Simulation time in minutes
N_INITIAL = 1     # Number of cars in the car wash at the start


# Instantiate Process Manager
random.seed(RANDOM_SEED)
pm = ProcessManager()

# Create car wash with NUM_MACHINE machines and attach car wash process
pm.create_resource('machines', capacity=NUM_MACHINES)


class CarWash(Process):
    def definition(self, user):
        # Car arrives and requests the use of a machine
        yield user.waits('machines')

        # Car enters the car wash and spends WASHTIME minutes inside
        yield user.waits(timedelta(minutes=WASHTIME))

        # Car leaves the car wash and releases the machine for the next car to use
        user.set_checkpoint(f'Left the carwash. {random.randint(50, 99)}% of dirt removed.')
        user.releases('machines')

pm.attach_process(CarWash)

# Create event stream
em = EventManager(pm)
now = datetime.now()
times = np.concatenate([np.zeros(3), np.random.randint(T_INTER-2, T_INTER+2, int(SIM_TIME/(T_INTER-2)))]).cumsum()
for i, t in enumerate(times):
    em.create_user(f'Car {i}', instant=now+timedelta(minutes=t))

if __name__ == '__main__':
    # Execute!
    em.run(until=now+timedelta(minutes=SIM_TIME))
    print(em.checkpoints)
    print(pm.get_resource('machines').usage)

Bank Renege

As seen in here.

import random
from chronon import ProcessManager, EventManager, Process


# RANDOM_SEED = 42
NEW_CUSTOMERS = 7  # Total number of customers
NUMBER_OF_COUNTERS = 1 # Total number of counters
INTERVAL_CUSTOMERS = 3.0  # Generate new customers roughly every x seconds
MIN_PATIENCE = 1  # Min. customer patience
MAX_PATIENCE = 5  # Max. customer patience


# create process manager
pm = ProcessManager()

# create resource
pm.create_resource('counters', capacity=NUMBER_OF_COUNTERS)

# process definition
class CustomerWaitCounter(Process):
    def definition(self, user):
        # time when customer arrives
        arrive = self.env.now
        user.set_checkpoint('Arrived')
        # amount of patience a customer can have
        patience = random.uniform(MIN_PATIENCE, MAX_PATIENCE)
        # waiting...
        yield user.waits('counters', patience=patience)
        # calculate time waited since customer arrived
        wait = self.env.now - arrive
        if wait < patience:
            # clients go to the counter
            user.set_checkpoint(f'Got to the counter')
            time_discussing = 12.0
            td = random.expovariate(1.0 / time_discussing)
            # yield as long as the customer is discussing at the counter
            yield user.waits(td)
            # release the resource
            user.releases('counters')
            user.set_checkpoint('Finished')
        else:
            # release the resource
            user.releases('counters')
            # customer reneged
            user.set_checkpoint('Reneged')


# store process into process Manager
pm.attach_process(CustomerWaitCounter)

# create event manager
em = EventManager(pm)
for i in range(NEW_CUSTOMERS):
    # create new customer
    t = random.expovariate(1.0 / INTERVAL_CUSTOMERS)
    em.create_user(f'Customer {i}', instant=t)

# random.seed(RANDOM_SEED)

if __name__ == '__main__':
    # execute the program
    em.run()
    print(em.checkpoints)
    print(pm.get_resource('counters').usage)