98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
|
|
### DO NOT EDIT THIS FILE ###
|
|
|
|
from pypdevs.DEVS import AtomicDEVS
|
|
import random
|
|
import dataclasses
|
|
|
|
# The reason for annotating the *State-classes as 'dataclass', is because this automatically generates a nice __repr__-function, so that if the simulator is set to verbose, you can actually see what the state is.
|
|
|
|
class Ship:
|
|
def __init__(self, size, creation_time):
|
|
self.size = size
|
|
self.creation_time = creation_time
|
|
|
|
# useful in verbose mode:
|
|
def __repr__(self):
|
|
return f"Ship(size={self.size},created={self.creation_time})"
|
|
|
|
@dataclasses.dataclass
|
|
class GeneratorState:
|
|
current_time: float
|
|
time_until_next_ship: float
|
|
to_generate: int
|
|
random: random.Random
|
|
|
|
def __init__(self, seed=0, gen_num=1000):
|
|
self.current_time = 0.0 # for statistics only
|
|
self.time_until_next_ship = 0.0
|
|
self.to_generate = gen_num
|
|
self.random = random.Random(seed)
|
|
|
|
class Generator(AtomicDEVS):
|
|
def __init__(self,
|
|
seed=0, # random seed
|
|
lambd=1.0/60.0, # how often to generate a ship - in this example, once per minute
|
|
gen_types=[1,1,2], # ship sizes to generate, will be sampled uniformly - in this example, size 1 is twice as likely as size 2.
|
|
gen_num=1000, # number of ships total to generate
|
|
):
|
|
super().__init__("Generator")
|
|
|
|
# State (for everything that is mutable)
|
|
self.state = GeneratorState(seed=seed, gen_num=gen_num)
|
|
|
|
# I/O
|
|
self.out_ship = self.addOutPort("out_event")
|
|
|
|
# Parameters (read-only)
|
|
self.lambd = lambd
|
|
self.gen_types = gen_types
|
|
|
|
def timeAdvance(self):
|
|
return self.state.time_until_next_ship
|
|
|
|
def outputFnc(self):
|
|
size = self.state.random.choice(self.gen_types) # uniformly sample from gen_types
|
|
# watch out: outputFnc is called *before* intTransition!
|
|
creation = self.state.current_time + self.state.time_until_next_ship
|
|
return { self.out_ship: Ship(size, creation) }
|
|
|
|
def intTransition(self):
|
|
self.state.current_time += self.state.time_until_next_ship
|
|
self.state.to_generate -= 1
|
|
if self.state.to_generate > 0:
|
|
self.state.time_until_next_ship = self.state.random.expovariate(self.lambd)
|
|
else:
|
|
# stop generating
|
|
self.state.time_until_next_ship = float('inf')
|
|
return self.state
|
|
|
|
@dataclasses.dataclass
|
|
class SinkState:
|
|
current_time: float
|
|
ships: list
|
|
|
|
def __init__(self):
|
|
self.current_time = 0.0
|
|
self.ships = []
|
|
|
|
class Sink(AtomicDEVS):
|
|
def __init__(self):
|
|
super().__init__("Sink")
|
|
self.state = SinkState()
|
|
|
|
# On this input port, the Sink expects to receive a *list* of Ships. This is because a Lock can contain more than one Ship, and the Lock can send them all at once to the Sink (with a single event).
|
|
self.in_ships = self.addInPort("in_ships")
|
|
|
|
def extTransition(self, inputs):
|
|
self.state.current_time += self.elapsed
|
|
if self.in_ships in inputs:
|
|
ships = inputs[self.in_ships]
|
|
for ship in ships:
|
|
ship.finished_time = self.state.current_time
|
|
# amount of time spent in the system:
|
|
ship.queueing_duration = ship.finished_time - ship.creation_time
|
|
self.state.ships.extend(ships)
|
|
return self.state
|
|
|
|
### DO NOT EDIT THIS FILE ###
|