In the SimPy documentation they describe a car process which involves a car which transitions between parking and driving. From being parked it takes 5 units of time to change to being in a state of driving. Similarly, it takes 2 units of time to go from being in a state of driving to a state of parked. Here is a state diagram representing this system:
stateDiagram
[*] --> Parked
Parked --> Driving: 5
Driving --> Parked: 2
Parked --> [*]
Driving --> [*]For something this simple we don’t really need discrete event simulation software, but it can be helpful for the sake of illustration to do it anyway. Let’s use idiomatic Ciw to make an equivalent simulation.
Let our queueing network have two nodes, each representing either of the two states “Parked” or “Driving”. The car will go between these two nodes. When the car arrives at a node that will represent that it has begun whatever process is involved, and when the car finishes service at that node we will interpret that to mean it is now changing state.
Since there will be only one car in the whole network, we do not need to give any consideration to service discipline. If the car is on the queue, then serve it. Otherwise, don’t.
We do need to pay attention to arrival times however. We need the car to arrive at the first node at the start of the simulation and for no additional cars to arrive thereafter. This can be achieved with a sequential distribution available in Ciw:
"""Demonstrate Ciw's ``Sequential`` distribution for a single arrival at time zero."""
import ciw
ciw.dists.Sequential([0, float('inf')])The class ciw.dists.Sequential takes an iterable (usually a list) representing the sequence of arrival times to be done. When it has gotten through the list it will start again at the start of the list. But here’s the catch: we have set the second time to \(\infty\), which it will never reach. So there will be a single arrival at \(t=0\) and none thereafter. We need to specify that no arrivals occur at the second node, which we can do using None:
"""Arrival distributions for the car example: one car at ``t=0``, none thereafter."""
import ciw
arrival_dists = [ciw.dists.Sequential([0, float('inf')]), None]For service times we know that the car will start driving after 5 units of time, then be parked for 2 units of time, and repeat. These are constant random variables which can be specified in Ciw using ciw.dists.Deterministic:
"""Deterministic service distributions for parking (5) and driving (2)."""
import ciw
service_dists = [ciw.dists.Deterministic(5), ciw.dists.Deterministic(2)]We don’t need to, but let’s add some fluff so that we get a similar printout. To do that let’s write a wrapper distribution.
from typing import NoReturn
import ciw
class PrintDistWrapper(ciw.dists.Distribution):
"""
A wrapper class for a probability distribution that adds a print statement
before sampling from the underlying distribution.
Parameters:
- dist (ciw.dists.Distribution): The underlying probability distribution to be wrapped.
- message (str): A custom message to be printed before sampling.
Note: This class inherits from ciw.dists.Distribution.
Example:
```
underlying_dist = SomeDistributionClass(parameters)
wrapper_dist = PrintDistWrapper(underlying_dist, "Sampling from the distribution:")
sample_result = wrapper_dist.sample()
```
"""
def __init__(self, dist: ciw.dists.Distribution, message: str) -> NoReturn:
"""
Initialize the PrintDistWrapper instance.
Parameters:
- dist (ciw.dists.Distribution): The underlying probability distribution to be wrapped.
- message (str): A custom message to be printed before sampling.
"""
super().__init__()
self.dist = dist
self.message = message
def sample(self, t: float = None, ind: ciw.Individual = None) -> float:
"""
Generate a sample from the underlying distribution, printing a custom message.
Parameters:
- t: Optional parameter (if applicable to the underlying distribution).
- ind: Optional parameter (if applicable to the underlying distribution).
Returns:
- The sampled value from the underlying distribution.
Note: This method overrides the sample method of ciw.dists.Distribution.
"""
print(self.message, t)
return self.dist.sample(t, ind)The PrintDistWrapper class allows us to print whatever message we like when the distribution is sampled from. Now we can just wrap our service distributions like this:
"""Wrap the deterministic service distributions so that each sample is announced."""
import ciw
from print_dist_wrapper import PrintDistWrapper
service_dists = [
PrintDistWrapper(
ciw.dists.Deterministic(5),
'Start parking at'),
PrintDistWrapper(
ciw.dists.Deterministic(2),
'Start driving at')
]We only need one server at each node since there could only ever be one car either there, or not there, requiring service. So we can just set those number of servers in a list like this:
"""One server at each node, which is sufficient for the single-car network."""
num_servers = [1, 1]Since we have two nodes, we need to specify the routing behaviour. It is simple in this case, and stationary. Whenever the car is leaving node one, go to node two. Whenever the car is leaving node two, go to node one. We can specify this as the following routing matrix:
\[\begin{bmatrix} 0 & 1 \\ 1 & 0 \end{bmatrix}\]
In Ciw we can take our routing matrix to be:
"""Routing matrix: whenever the car leaves a node it goes to the other one."""
R = [[0.0, 1.0], [1.0, 0.0]]We can now put together our simulation object:
"""Create the two-node Ciw network that models the parked/driving car."""
import ciw
from arrival_distributions import arrival_dists
from number_of_servers import num_servers
from routing_matrix import R
from wrapped_service_distributions import service_dists
network = ciw.create_network(
arrival_distributions = arrival_dists,
service_distributions = service_dists,
number_of_servers = num_servers,
routing=R
)And put together our simulation object:
"""Instantiate a Ciw simulation for the car network."""
import ciw
from create_network import network
simulation = ciw.Simulation(network)And simulate (for 15 units of time)!
"""Run the car simulation for 15 units of time."""
from create_simulation import simulation
simulation.simulate_until_max_time(15)Start parking at 0
Start driving at 5
Start parking at 7
Start driving at 12
Start parking at 14Happy simulating! :)