Learn by example
In this section we will focus on a standard use-case of using Drops in the context of discrete-event simulations. The following example will also play another role in the advanced section where we will describe how to combine Drops with Flask to make a live web-application that shows the current state of the simulation in a UI.
Minimal Example
Create a python file called hello.py with the following code:
from drops import DEvent, Drops, Event
def foo(e: Event):
print("Hello " + e.body.get("greeter"))
return DEvent(msg="say_goodbye", delay_s=1, body={"name": "Alice"})
def bar(e: Event):
print("Goodbye " + e.body.get("name"))
if __name__ == "__main__":
drops = Drops()
drops.register_callback(foo, "say_hello")
drops.register_callback(bar, "say_goodbye")
# Publish a message with no delay and a given body into the queue.
# After the broker receives this message, follow-up events will be
# triggered until everything inside the queue is consumed.
drops.enqueue(msg="say_hello", delay_s=0, body={"greeter": "Alice"})
drops.run()
And then run your application from your terminal
Goodbye Alice
Car Washing Line
Problem statement and simulation model
No discrete-event simulation tutorial would be considered comprehensive without the classic example of a car washing line. Here's the scenario:
Imagine a scenario where cars arrive at a car wash in an unpredictable pattern from a finite source. At the facility, there's a single washing line that takes approximately \(d \in \mathbb{N}\) minutes to complete a cycle for each car.
Once a car finishes its wash, the next car in line is permitted to enter the washing process.
A naive approach to model this in a flow-chart could look like this:
---
title: Here needs a basic flowChart
---
graph LR;
A[Start] --> B{Car Arrives};
B -- Yes --> C(Washing Position Empty?);
C -- Yes --> D[Start Wash];
C -- No --> E[Put Car in Waiting Line];
D --> F{Next Car Waiting?};
F -- Yes --> G[Car Arrives];
F -- No --> H[End];
E --> B;
G --> B;
H --> I[End Wash];
I --> J{Waiting Line Not Empty?};
J -- Yes --> K[Take Car From Waiting Line];
J -- No --> L[End];
K --> B;
Implementation of Drops-Components
In Drops we call everything that can be registered a Component. This is a pure function, a class-instance, a module, a generator-function or a closure.
Entities and value-objects
During our implementation we will be as explicit as possible and use type annotations, aka. Annotated type-hints from the typing.py
module.
The first and last entity we are going to model, is the car.
from typing import NamedTuple
class Car(NamedTuple):
car_id: int # Positive integer greater than 0
dirt: float # Real number between 0 and 1
As we can see, our car only has two immutable attributes, namely an Id that will serve as a plain counter and a percentage value dirt that tells us how dirty the car is.
Sources
Now we need something that can send a car in a specific time period to our washing line. We will call this thing a source. In our case, we will model our source as a finite source sending cars every 5 minutes to our washing line.
from itertools import count
from random import random
from drops import DEvent, Event
from tests.examples.car_wash.car import Car
def finite_car_source_maker(num_cars: int):
counter = count()
def car_source(e: Event) -> DEvent:
car_id = next(counter)
while num_cars - car_id > 0:
return DEvent(
msg="car_arrives",
delay_s=5,
body={"car": Car(car_id=car_id, dirt=random())}
)
return car_source
Most developers would have started using a class-based approach to model our source. But since Drops is capable of registering we decide on using a one to model our car-source. This will make our code more concise and remove some unnecessary boilerplate as well.
First we initialize the outer function finite_car_source_maker with a maximum number of cars we want to send to the washing line. Then
we instantiate an infinite counter from the itertools module. After that we define the actual car_source.
In this function the counter from the outer-namespace serves as the Id for the car we want to send to the washing line and
as a counter in the while-loop. There we loop as long as we have not send enough cars which is \(n_{cars} - n_{send} > 0\).
The return-type of our source is an instance of DelayedEvent. Drops will use this entity to create a concrete Event instance
in its mechanics that will be dispatched to every Component in the next iteration of the dispatching-algorithm.
Classes
The next Component we are going to model is the actual WashingLine aka. the star of our application.
import random
from collections import deque
from dataclasses import dataclass
from drops.core import DEvent as DEvent
from drops.core import Event
@dataclass
class WashingLine:
consumptions = (
"car_arrives",
"start_wash",
"end_wash"
)
collected_dirt = 0
waiting_line = deque()
washing_position = deque(maxlen=1)
time_washing = 8
def car_arrives(self, e: Event) -> DEvent:
car = e.body.get("car")
print(f"t={e.time} arriving {car}")
if len(self.washing_position) == 0:
print(f"t={e.time} getting {car} from the event at start_wash and append to washing position")
self.washing_position.append(car)
return DEvent(msg="start_wash", delay_s=1, body=e.body)
else:
print(f"t={e.time} putting {car} in waiting line")
self.waiting_line.append(car)
def start_wash(self, e: Event) -> DEvent:
return DEvent(msg="end_wash", delay_s=self.time_washing, body={"car": e.body.get("car")})
def end_wash(self, e: Event) -> DEvent:
clean_car = self.washing_position.popleft()
print(f"t={e.time} leaving system {clean_car}")
removed_dirt = round(random.uniform(0, clean_car.dirt), 3)
self.collected_dirt += removed_dirt
if self.waiting_line:
next_car = self.waiting_line.popleft()
return DEvent(msg="car_arrives", delay_s=1, body={"car": next_car})