joeriPDEVS/assignment/environment.py

### 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()
        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 ###