add starting point for mosis 2024 assignment

assignment/environment.py

@@ -0,0 +1,96 @@
+
+### 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 ###