LI, HI and HO tests for PythonPDEVS

devstone/pythonpdevs/devstone.py

@@ -44,16 +44,35 @@ class DelayedAtomic(AtomicDEVS):
         return "active"
 
 
+class DelayedAtomicStats(DelayedAtomic):
+    def __init__(self, name: str, int_delay: float, ext_delay: float, add_out_port: bool = False, prep_time=0):
+        super().__init__(name, int_delay, ext_delay, add_out_port, prep_time)
+
+        self.int_count = 0
+        self.ext_count = 0
+
+    def intTransition(self):
+        self.int_count += 1
+        return super().intTransition()
+
+    def extTransition(self, inputs):
+        self.ext_count += 1
+        return super().extTransition(inputs)
+
+
 class DEVStoneWrapper(CoupledDEVS, ABC):
 
     def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float,
-                 add_atomic_out_ports: bool = False):
+                 add_atomic_out_ports: bool = False, prep_time=0, stats=False):
         super().__init__(name)
 
         self.depth = depth
         self.width = width
         self.int_delay = int_delay
         self.ext_delay = ext_delay
+        self.prep_time = prep_time
+        self.stats = stats
+        self.add_atomic_out_ports = add_atomic_out_ports
 
         self.i_in = self.addInPort("i_in")
         self.o_out = self.addOutPort("o_out")
@@ -68,7 +87,11 @@ class DEVStoneWrapper(CoupledDEVS, ABC):
             raise ValueError("Invalid ext_delay")
 
         if depth == 1:
-            atomic = DelayedAtomic("Atomic_0_0", int_delay, ext_delay, add_out_port=True)
+            if self.stats:
+                atomic = DelayedAtomicStats("Atomic_0_0", int_delay, ext_delay, add_out_port=True, prep_time=prep_time)
+            else:
+                atomic = DelayedAtomic("Atomic_0_0", int_delay, ext_delay, add_out_port=True, prep_time=prep_time)
+
             self.addSubModel(atomic)
 
             self.connectPorts(self.i_in, atomic.i_in)
@@ -80,8 +103,12 @@ class DEVStoneWrapper(CoupledDEVS, ABC):
             self.connectPorts(coupled.o_out, self.o_out)
 
             for idx in range(width - 1):
-                atomic = DelayedAtomic("Atomic_%d_%d" % (depth - 1, idx), int_delay, ext_delay,
-                                       add_out_port=add_atomic_out_ports)
+                if self.stats:
+                    atomic = DelayedAtomicStats("Atomic_%d_%d" % (depth - 1, idx), int_delay, ext_delay,
+                                                add_out_port=add_atomic_out_ports, prep_time=prep_time)
+                else:
+                    atomic = DelayedAtomic("Atomic_%d_%d" % (depth - 1, idx), int_delay, ext_delay,
+                                           add_out_port=add_atomic_out_ports, prep_time=prep_time)
                 self.addSubModel(atomic)
 
     @abstractmethod
@@ -92,21 +119,24 @@ class DEVStoneWrapper(CoupledDEVS, ABC):
 
 class LI(DEVStoneWrapper):
 
-    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float):
-        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=False)
+    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float, prep_time=0, stats=False):
+        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=False, prep_time=prep_time,
+                         stats=stats)
 
         for idx in range(1, len(self.component_set)):
             assert isinstance(self.component_set[idx], AtomicDEVS)
             self.connectPorts(self.i_in, self.component_set[idx].i_in)
 
     def gen_coupled(self):
-        return LI("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay)
+        return LI("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay,
+                  prep_time=self.prep_time, stats=self.stats)
 
 
 class HI(DEVStoneWrapper):
 
-    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float):
-        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=True)
+    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float, prep_time=0, stats=False):
+        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=True, prep_time=prep_time,
+                         stats=stats)
 
         if len(self.component_set) > 1:
             assert isinstance(self.component_set[-1], AtomicDEVS)
@@ -118,13 +148,15 @@ class HI(DEVStoneWrapper):
             self.connectPorts(self.i_in, self.component_set[idx].i_in)
 
     def gen_coupled(self):
-        return HI("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay)
+        return HI("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay,
+                  prep_time=self.prep_time, stats=self.stats)
 
 
 class HO(DEVStoneWrapper):
 
-    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float):
-        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=False)
+    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float, prep_time=0, stats=False):
+        super().__init__(name, depth, width, int_delay, ext_delay, add_atomic_out_ports=True, prep_time=prep_time,
+                         stats=stats)
 
         self.i_in2 = self.addInPort("i_in2")
         self.o_out2 = self.addOutPort("o_out2")
@@ -145,7 +177,8 @@ class HO(DEVStoneWrapper):
             self.connectPorts(self.component_set[idx].o_out, self.o_out2)
 
     def gen_coupled(self):
-        return HO("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay)
+        return HO("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay,
+                  prep_time=self.prep_time, stats=self.stats)
 
 
 class HOmod(CoupledDEVS):
@@ -190,7 +223,8 @@ class HOmod(CoupledDEVS):
                 for i in range(width):
                     min_row_idx = 0 if i < 2 else i - 1
                     for j in range(min_row_idx, width - 1):
-                        atomic = DelayedAtomic("Atomic_%d_%d_%d" % (depth - 1, i, j), int_delay, ext_delay, add_out_port=True)
+                        atomic = DelayedAtomic("Atomic_%d_%d_%d" % (depth - 1, i, j), int_delay, ext_delay,
+                                               add_out_port=True)
                         self.addSubModel(atomic)
                         atomics[i].append(atomic)
 
@@ -208,11 +242,12 @@ class HOmod(CoupledDEVS):
                     self.connectPorts(atomics[1][i].o_out, atomics[0][i].i_in)
                 for i in range(2, width):  # Rest of rows
                     for j in range(len(atomics[i])):
-                        self.connectPorts(atomics[i][j].o_out, atomics[i-1][j+1].i_in)
+                        self.connectPorts(atomics[i][j].o_out, atomics[i - 1][j + 1].i_in)
 
 
 if __name__ == '__main__':
     import sys
+
     sys.setrecursionlimit(10000)
     root = HOmod("Root", 4, 3, 0, 0)
     sim = Simulator(root)

devstone/pythonpdevs/test_devstone.py

@@ -0,0 +1,280 @@
+from abc import abstractmethod
+from unittest import TestCase
+import random
+
+from pypdevs.DEVS import AtomicDEVS, CoupledDEVS
+from pypdevs.simulator import Simulator
+
+from devstone import LI, DelayedAtomic, DelayedAtomicStats, HI, HO
+from generator import Generator
+from main import DEVStoneEnvironment
+
+
+class Utils:
+
+    @staticmethod
+    def count_atomics(coupled):
+        """
+        :return: Number of atomic components in a coupled model
+        """
+        atomic_count = 0
+        for comp in coupled.component_set:
+            if isinstance(comp, AtomicDEVS):
+                atomic_count += 1
+            elif isinstance(comp, CoupledDEVS):
+                atomic_count += Utils.count_atomics(comp)
+            else:
+                raise RuntimeError("Unrecognized type of component")
+
+        return atomic_count
+
+    @staticmethod
+    def count_ic(coupled):
+        """
+        :return: Number of ic couplings in a coupled model
+        """
+        ic_count = 0
+        for comp in coupled.component_set:
+            for src_port in comp.OPorts:
+                for dst_port in src_port.outline:
+                    if dst_port.host_DEVS != coupled:
+                        ic_count += 1
+
+        for comp in coupled.component_set:
+            if isinstance(comp, CoupledDEVS):
+                ic_count += Utils.count_ic(comp)
+            elif not isinstance(comp, AtomicDEVS):
+                raise RuntimeError("Unrecognized type of component")
+
+        return ic_count
+
+    @staticmethod
+    def count_eic(coupled):
+        """
+        :return: Number of eic couplings in a coupled model
+        """
+        eic_count = sum([len(port.outline) for port in coupled.IPorts])
+        for comp in coupled.component_set:
+            if isinstance(comp, CoupledDEVS):
+                eic_count += Utils.count_eic(comp)
+            elif not isinstance(comp, AtomicDEVS):
+                raise RuntimeError("Unrecognized type of component")
+
+        return eic_count
+
+    @staticmethod
+    def count_eoc(coupled):
+        """
+        :return: Number of eoc couplings in a coupled model
+        """
+        eoc_count = sum([len(port.inline) for port in coupled.OPorts])
+        for comp in coupled.component_set:
+            if isinstance(comp, CoupledDEVS):
+                eoc_count += Utils.count_eoc(comp)
+            elif not isinstance(comp, AtomicDEVS):
+                raise RuntimeError("Unrecognized type of component")
+
+        return eoc_count
+
+    @staticmethod
+    def count_transitions(coupled):
+        """
+        :return: Number of atomic components in a coupled model
+        """
+        int_count = 0
+        ext_count = 0
+        for comp in coupled.component_set:
+            if isinstance(comp, DelayedAtomicStats):
+                int_count += comp.int_count
+                ext_count += comp.ext_count
+            elif isinstance(comp, CoupledDEVS):
+                pic, pec = Utils.count_transitions(comp)
+                int_count += pic
+                ext_count += pec
+            elif not isinstance(comp, Generator):
+                raise RuntimeError("Unexpected type of component: %s" % comp.__class__.__name__)
+
+        return int_count, ext_count
+
+
+class DevstoneUtilsTestCase(TestCase):
+
+    def __init__(self, name, num_valid_params_sets: int = 10):
+        super().__init__(name)
+        self.valid_high_params = []
+        self.valid_low_params = []
+
+        for _ in range(int(num_valid_params_sets)):
+            self.valid_high_params.append([random.randint(1, 100), random.randint(1, 200),
+                                           random.randint(1, 1000), random.randint(1, 1000)])
+
+        for _ in range(int(num_valid_params_sets)):
+            self.valid_low_params.append([random.randint(1, 20), random.randint(1, 30),
+                                      random.randint(0, 10), random.randint(0, 10)])
+
+    def check_invalid_inputs(self, base_class):
+        self.assertRaises(ValueError, base_class, "root", 0, 1, 1, 1)
+        self.assertRaises(ValueError, base_class, "root", 1, 0, 1, 1)
+        self.assertRaises(ValueError, base_class, "root", 1, 1, -1, 0)
+        self.assertRaises(ValueError, base_class, "root", 1, 1, 0, -1)
+        self.assertRaises(ValueError, base_class, "root", 0, 1, -1, -1)
+        self.assertRaises(ValueError, base_class, "root", 0, 0, -1, -1)
+
+    def test_behavior_sequential(self):
+        self._test_behavior(Simulator)
+
+    #def test_behavior_parallel(self):
+    #    self._test_behavior(ParallelThreadCoordinator)
+
+    @abstractmethod
+    def _test_behavior(self, Coordinator):
+        pass
+
+
+class TestLI(DevstoneUtilsTestCase):
+
+    def test_structure(self):
+        """
+        Check structure params: atomic modules, ic's, eic's and eoc's.
+        """
+        for params_tuple in self.valid_high_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                self._check_structure(**params)
+
+    def test_structure_corner_cases(self):
+        params = {"depth": 10, "width": 1, "int_delay": 1, "ext_delay": 1}
+        self._check_structure(**params)
+        params["depth"] = 1
+        self._check_structure(**params)
+
+    def _check_structure(self, **params):
+        li_root = LI("LI_root", **params)
+        self.assertEqual(Utils.count_atomics(li_root), (params["width"] - 1) * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eic(li_root), params["width"] * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eoc(li_root), params["depth"])
+        self.assertEqual(Utils.count_ic(li_root), 0)
+
+    def _test_behavior(self, coord_type):
+        """
+        Check behaviour params: number of int and ext transitions.
+        """
+        for params_tuple in self.valid_low_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                li_root = LI("LI_root", stats=True, **params)
+                li_env = DEVStoneEnvironment("LI_env", li_root)
+                sim = Simulator(li_env)
+                sim.setVerbose(None)
+                # sim.setTerminationTime(10.0)
+                # sim.setStateSaving("custom")
+                sim.simulate()
+
+                int_count, ext_count = Utils.count_transitions(li_root)
+                self.assertEqual(int_count, (params["width"] - 1) * (params["depth"] - 1) + 1)
+                self.assertEqual(ext_count, (params["width"] - 1) * (params["depth"] - 1) + 1)
+
+    def test_invalid_inputs(self):
+        super().check_invalid_inputs(LI)
+
+
+class TestHI(DevstoneUtilsTestCase):
+
+    def test_structure(self):
+        """
+        Check structure params: atomic modules, ic's, eic's and eoc's.
+        """
+        for params_tuple in self.valid_high_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                self._check_structure(**params)
+
+    def test_structure_corner_cases(self):
+        params = {"depth": 10, "width": 1, "int_delay": 1, "ext_delay": 1}
+        self._check_structure(**params)
+        params["depth"] = 1
+        self._check_structure(**params)
+
+    def _check_structure(self, **params):
+        hi_root = HI("HI_root", **params)
+        self.assertEqual(Utils.count_atomics(hi_root), (params["width"] - 1) * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eic(hi_root), params["width"] * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eoc(hi_root), params["depth"])
+        self.assertEqual(Utils.count_ic(hi_root),
+                         (params["width"] - 2) * (params["depth"] - 1) if params["width"] > 2 else 0)
+
+    def _test_behavior(self, coord_type):
+        """
+        Check behaviour params: number of int and ext transitions.
+        """
+        for params_tuple in self.valid_low_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                hi_root = HI("HI_root", stats=True, **params)
+                hi_env = DEVStoneEnvironment("HI_env", hi_root)
+                sim = Simulator(hi_env)
+                sim.setVerbose(None)
+                # sim.setTerminationTime(10.0)
+                # sim.setStateSaving("custom")
+                sim.simulate()
+
+                int_count, ext_count = Utils.count_transitions(hi_root)
+                self.assertEqual(int_count, (((params["width"] - 1) * params["width"]) / 2) * (params["depth"] - 1) + 1)
+                self.assertEqual(ext_count, (((params["width"] - 1) * params["width"]) / 2) * (params["depth"] - 1) + 1)
+
+    def test_invalid_inputs(self):
+        super().check_invalid_inputs(HI)
+
+
+class TestHO(DevstoneUtilsTestCase):
+
+    def test_structure(self):
+        """
+        Check structure params: atomic modules, ic's, eic's and eoc's.
+        """
+        for params_tuple in self.valid_high_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                self._check_structure(**params)
+
+    def test_structure_corner_cases(self):
+        params = {"depth": 10, "width": 1, "int_delay": 1, "ext_delay": 1}
+        self._check_structure(**params)
+        params["depth"] = 1
+        self._check_structure(**params)
+
+    def _check_structure(self, **params):
+        ho_root = HO("HO_root", **params)
+        self.assertEqual(Utils.count_atomics(ho_root), (params["width"] - 1) * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eic(ho_root), (params["width"] + 1) * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_eoc(ho_root), params["width"] * (params["depth"] - 1) + 1)
+        self.assertEqual(Utils.count_ic(ho_root),
+                         (params["width"] - 2) * (params["depth"] - 1) if params["width"] > 2 else 0)
+
+    def _test_behavior(self, coord_type):
+        """
+        Check behaviour params: number of int and ext transitions.
+        """
+        for params_tuple in self.valid_low_params:
+            params = dict(zip(("depth", "width", "int_delay", "ext_delay"), params_tuple))
+
+            with self.subTest(**params):
+                ho_root = HO("HO_root", stats=True, **params)
+                ho_env = DEVStoneEnvironment("HO_env", ho_root)
+                sim = Simulator(ho_env)
+                sim.setVerbose(None)
+                # sim.setTerminationTime(10.0)
+                # sim.setStateSaving("custom")
+                sim.simulate()
+
+                int_count, ext_count = Utils.count_transitions(ho_env)
+                self.assertEqual(int_count, (((params["width"] - 1) * params["width"]) / 2) * (params["depth"] - 1) + 1)
+                self.assertEqual(ext_count, (((params["width"] - 1) * params["width"]) / 2) * (params["depth"] - 1) + 1)
+
+    def test_invalid_inputs(self):
+        super().check_invalid_inputs(HO)