Creating a discrete-event simulator¶
Let’s imagine a factory where a certain resource arrives, a certain server processes it and sends it to another station and leaves the factory.
Rules¶
Arrivals have a quantity of components and a interarraival time.
Stations process the components, one at time.
When an input enters to the station, updates the processing time, and adds the element to process.
When processing time is completed, extracts the part from the elements to process and start to process another if it is possible.
Exit counts the procesed entities.
Creating stations¶
Stations are DiscreteEventModels, so you can extend the class and implement the abstract methods to work with them.
[1]:
import sys
if sys.version_info >= (3, 8):
from typing import TypedDict
else:
from typing_extensions import TypedDict
from typing import Dict
from gsf.core.expressions import Expression
from gsf.core.types import Time
from gsf.dynamic_system.dynamic_systems import DiscreteEventDynamicSystem
from gsf.models.models import DiscreteEventModel
class StationState(TypedDict):
"""State definition of the station"""
parts: int
remaining_time: Time
class Station(DiscreteEventModel):
"""Station of the simulator
It process the inputs that receives. Its state has the number of parts that currently are inside the
station and the remaining time to finish to process one of that parts.
Attributes:
_processing_time(Expression): time to process one part.
"""
_processing_time: Expression
def __init__(self, dynamic_system: DiscreteEventDynamicSystem, processing_time: Expression):
"""
Args:
dynamic_system (DiscreteEventDynamicSystem): factory where stations belongs.
processing_time (Expression): time to process one part.
"""
super().__init__(dynamic_system, state={
'parts': 0,
'remaining_time': -1
})
self._processing_time = processing_time
def _internal_state_transition_function(self, state: StationState) -> StationState:
"""Removes one part from processing, and schedules and event to process a new one.
"""
state["parts"] = max(state["parts"] - 1, 0)
self.schedule(self.get_time())
return state
def _external_state_transition_function(self, state: StationState, inputs: Dict[str, int],
event_time: Time) -> StationState:
"""Adds parts to process
"""
values = inputs.values()
state["remaining_time"] = state["remaining_time"] - event_time
for number_of_parts in values:
if state["parts"] > 0:
state["parts"] = state["parts"] + number_of_parts
elif state["parts"] == 0:
state["parts"] = number_of_parts
self.schedule(self.get_time())
return state
def _time_advance_function(self, state: StationState) -> Time:
"""Obtains the time of the next processed entity.
"""
if state["parts"] < 1:
state["remaining_time"] = Time(-1)
else:
state["remaining_time"] = Time(self._processing_time.evaluate())
return state["remaining_time"]
def _output_function(self, state: StationState) -> int:
"""Returns a part.
"""
if state["parts"] > 0:
return 1
return 0
def __str__(self):
return self.get_id()
Stations process the parts for a _processing_time.
When time ends, an autonomous event is emitted, so the framework runs the _internal_state_transition_function, where the station removes one part from processing, and schedules an event to process a new one. When a part comes from another station, the framework runs the _external_state_transition_function where the station adds new parts to process.
The time is Time(-1) if there is no parts to process, and evaluates an expression if there are parts. An expression is a datatype of the framework that allows include almost any value.
Creating the generator¶
The generator creates parts given an interarrival time, and the number of pieces to create at that arrival.
[2]:
from typing import List, Union
from gsf.core.expressions import Expression
from gsf.core.types import Time
from gsf.dynamic_system.dynamic_systems import DiscreteEventDynamicSystem
from gsf.models.core.base_model import ModelState
from gsf.models.models import DiscreteEventModel
GeneratorState = Union[Expression, List[int]]
class Generator(DiscreteEventModel):
"""Generator of parts
Creates parts given an interarrival time, and the number of pieces to create at that arrival
Attributes:
_interarrival_time (Expression): interarrival time of pieces.
"""
_interarrival_time: Expression
def __init__(self, dynamic_system: DiscreteEventDynamicSystem, pieces: GeneratorState,
interarrival_time: Expression):
"""Args:
dynamic_system(DiscreteEventDynamicSystem): factory where stations belongs.
pieces(GeneratorState): number of pieces to create when there is an arrival.
interarrival_time(Expression): interarrival time of pieces.
"""
super().__init__(dynamic_system, state=pieces)
self.schedule(Time(0))
self._interarrival_time = interarrival_time
def _internal_state_transition_function(
self, state: GeneratorState) -> ModelState:
"""Generates a part"""
if isinstance(state, list):
state.pop(0)
self.schedule(self.get_time())
return state
def _time_advance_function(self, state: GeneratorState) -> Time:
"""Calculates the time of the creation of next part"""
if isinstance(state, list):
return self._interarrival_time.evaluate() if len(state) > 0 else Time(-1)
else:
return self._interarrival_time.evaluate()
def _output_function(self, state: GeneratorState):
"""Get the created part"""
if isinstance(state, list):
return state[0]
else:
return state.evaluate()
def __str__(self):
return "Generator"
The generator state could be a list of integers or an expression. If it is a list, the state represent the number of parts per arrival for each arrival, so the parts[0] equals to the number of parts at first arrival, parts[1] to the second arrival, and so on. On the other hand, if it is an expression, it generates the number of parts that the expression gives during infinite arrivals.
If the state is a list, the output of an event should be the first element in the list, in turn, if it is a expression, it will evaluate the expression.
If the parts list is empy, the generator stops of schedule autonomous events.
Creating the exit¶
[3]:
from typing import Dict
from gsf.core.types import Time
from gsf.dynamic_system.dynamic_systems import DiscreteEventDynamicSystem
from gsf.models.core.base_model import ModelState
from gsf.models.models import DiscreteEventModel
class Exit(DiscreteEventModel):
"""Exit
Sink of a factory. Here come all the processed part of the factory.
"""
def __init__(self, dynamic_system: DiscreteEventDynamicSystem):
"""Args:
dynamic_system(DiscreteEventDynamicSystem): factory where stations belongs.
"""
super().__init__(dynamic_system, state=0)
def _external_state_transition_function(self, state: int, inputs: Dict[str, int],
event_time: Time) -> int:
"""Receives the parts"""
return state + sum(inputs.values())
def _time_advance_function(self, state: ModelState) -> Time:
"""Prevents to execute an autonomous event"""
return Time(-1)
def _output_function(self, state: ModelState) -> int:
"""Returns the number of parts processed."""
return state
def __str__(self):
return "Exit"
Creating the factory¶
Is the dynamic system where all the stations process parts.
[4]:
from gsf.core.expressions import Expression
from gsf.dynamic_system.dynamic_systems import DiscreteEventDynamicSystem
class FactorySystem(DiscreteEventDynamicSystem):
"""Factory system
Is the dynamic system where all the stations process parts.
Attributes:
generator (Generator): Generator of entities of the factory.
exit (Exit): Sink of the factory,
"""
generator: Generator
exit: Exit
def __init__(self, pieces: GeneratorState, interarrival_time: Expression):
"""Args:
pieces(GeneratorState): number of pieces to create when there is an arrival.
interarrival_time(Expression): interarrival time of pieces.
"""
super().__init__()
self.generator = Generator(self, pieces, interarrival_time)
self.exit = Exit(self)
It creates the generator and exit. The stations will have to be defined externally, so is possible to define multiple types of factories.
Running the simulation¶
Let’s make a simple simulation where one part arrive at time t=0 and two parts at t=1, with processing times of one and two seconds.
[5]:
from gsf.experiments.experiment_builders import DiscreteEventExperiment
from gsf.core.mathematics.values import Value
factory = FactorySystem([1, 2], Value(1)) # 1 and 2 parts every second
station_1 = Station(factory, Value(1)) # 1 second processing time
station_2 = Station(factory, Value(2)) # 2 seconds processing time
# build the network
factory.generator.add(station_1) # from generator to station_1
station_1.add(station_2) # from station_1 to station_2
station_2.add(factory.exit) # from station_2 to exit
experiment = DiscreteEventExperiment(factory)
experiment.simulation_control.start(stop_time=10)
experiment.simulation_control.wait()
print(factory.exit.get_output())
print(experiment.simulation_report.generate_report())
3
+------------------------------------+
| Simulation report |
+------+--------+-----------+--------+
| time | model3 | Generator | model2 |
+------+--------+-----------+--------+
| 0 | - | 1 | - |
| 1 | - | 2 | 1 |
| 2 | - | - | 1 |
| 3 | 1 | - | 1 |
| 5 | 1 | - | - |
| 7 | 1 | - | - |
+------+--------+-----------+--------+
You can see the network usign factory.show()
[6]:
factory.show()
[6]:
Add expressions¶
You can use framework’s predefined expressions
[7]:
from gsf.core.mathematics.distributions import PoissonDistribution, ExponentialDistribution, TriangularDistribution
from gsf.experiments.experiment_builders import DiscreteEventExperiment
factory = FactorySystem(PoissonDistribution(5), ExponentialDistribution(0.5))
station_1 = Station(factory, TriangularDistribution(1, 2, 5))
station_2 = Station(factory, TriangularDistribution(1, 4, 5))
factory.generator.add(station_1)
station_1.add(station_2)
station_2.add(factory.exit)
experiment = DiscreteEventExperiment(factory)
experiment.simulation_control.start(stop_time=15)
experiment.simulation_control.wait()
print(factory.exit.get_output())
print(experiment.simulation_report.generate_report())
3
+-------------------------------------------+
| Simulation report |
+-------------+-----------+--------+--------+
| time | Generator | model6 | model7 |
+-------------+-----------+--------+--------+
| 0 | 6 | - | - |
| 0.96332559 | 4 | - | - |
| 2.09312678 | 10 | - | - |
| 2.35056324 | 3 | - | - |
| 2.39291718 | 4 | - | - |
| 3.16931019 | 9 | - | - |
| 3.47817454 | 7 | - | - |
| 3.74797734 | 6 | - | - |
| 4.01101600 | - | 1 | - |
| 4.13636268 | 7 | - | - |
| 4.28584944 | 4 | - | - |
| 4.66146672 | 6 | - | - |
| 4.97266491 | 5 | - | - |
| 5.35416232 | 5 | - | - |
| 6.74599293 | - | - | 1 |
| 7.02375607 | 5 | - | - |
| 7.25656051 | - | 1 | - |
| 7.27027466 | 7 | - | - |
| 7.27831129 | 6 | - | - |
| 7.41882874 | 6 | - | - |
| 7.97573887 | 3 | - | - |
| 8.07425179 | 7 | - | - |
| 8.34431883 | 7 | - | - |
| 8.41782964 | 4 | - | - |
| 8.99115356 | 2 | - | - |
| 9.15874705 | - | - | 1 |
| 9.46516163 | 5 | - | - |
| 10.14813636 | - | 1 | - |
| 10.50405463 | 6 | - | - |
| 11.21912148 | 5 | - | - |
| 11.47218174 | 2 | - | - |
| 11.65776092 | 5 | - | - |
| 11.91477662 | 3 | - | - |
| 12.18615471 | 7 | - | - |
| 12.66111431 | 7 | - | - |
| 13.04349449 | 5 | - | - |
| 13.07539413 | 2 | - | - |
| 13.19808818 | 3 | - | - |
| 13.71377193 | - | - | 1 |
| 13.96347672 | - | 1 | - |
| 14.04346664 | 4 | - | - |
| 14.53036266 | 10 | - | - |
| 14.63644353 | 9 | - | - |
+-------------+-----------+--------+--------+
Icon theme "Numix-Circle-Light" not found.