Thesis/devstone/pythonpdevs/test_devstone.py

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, HOmod
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[0]:
            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)


class TestHOmod(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):
        homod_root = HOmod("HOmod_root", **params)
        self.assertEqual(Utils.count_atomics(homod_root),
                         ((params["width"] - 1) + ((params["width"] - 1) * params["width"]) / 2) * (
                                     params["depth"] - 1) + 1)
        self.assertEqual(Utils.count_eic(homod_root), (2 * (params["width"] - 1) + 1) * (params["depth"] - 1) + 1)
        self.assertEqual(Utils.count_eoc(homod_root), params["depth"])
        # ICs relative to the "triangular" section
        exp_ic = (((params["width"] - 2) * (params["width"] - 1)) / 2) if params["width"] > 1 else 0
        # Plus the ones relative to the connection from the 2nd to 1st row...
        exp_ic += (params["width"] - 1) ** 2
        # ...and from the 1st to the couple component
        exp_ic += params["width"] - 1
        # Multiplied by the number of layers (except the deepest one, that doesn't have ICs)
        exp_ic *= (params["depth"] - 1)
        self.assertEqual(Utils.count_ic(homod_root), exp_ic)

    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):
                homod_root = HOmod("HOmod_root", stats=True, **params)
                homod_env = DEVStoneEnvironment("HOmod_env", homod_root)
                sim = Simulator(homod_env)
                sim.setVerbose(None)
                # sim.setTerminationTime(10.0)
                # sim.setStateSaving("custom")
                sim.simulate()

                calc_in = lambda x, w: 1 + (x - 1) * (w - 1)
                exp_trans = 1
                tr_atomics = sum(range(1, params["width"]))
                for i in range(1, params["depth"]):
                    num_inputs = calc_in(i, params["width"])
                    trans_first_row = (params["width"] - 1) * (num_inputs + params["width"] - 1)
                    exp_trans += num_inputs * tr_atomics + trans_first_row

                int_count, ext_count = Utils.count_transitions(homod_env)
                self.assertEqual(int_count, exp_trans)
                self.assertEqual(ext_count, exp_trans)

    def test_invalid_inputs(self):
        super().check_invalid_inputs(HOmod)