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)