PythonPDEVS DEVStone (minimal version)

devstone/pythonpdevs-minimal/devstone.py

@@ -0,0 +1,267 @@
+from abc import ABC, abstractmethod
+from collections import defaultdict
+
+from pystone import pystones
+from pypdevs.minimal import AtomicDEVS, CoupledDEVS
+from pypdevs.infinity import INFINITY
+from pypdevs.simulator import Simulator
+
+
+class DelayedAtomic(AtomicDEVS):
+    def __init__(self, name: str, int_delay: float, ext_delay: float, add_out_port: bool = False, prep_time=0):
+        super().__init__(name)
+
+        self.int_delay = int_delay
+        self.ext_delay = ext_delay
+        self.prep_time = prep_time
+
+        self.i_in = self.addInPort("i_in")
+
+        if add_out_port:
+            self.o_out = self.addOutPort("o_out")
+
+    def intTransition(self):
+        if self.int_delay:
+            pystones(self.int_delay)
+
+        return "passive"
+
+    def timeAdvance(self):
+        if self.state == "active":
+            return self.prep_time
+        else:
+            return INFINITY
+
+    def outputFnc(self):
+        if hasattr(self, "o_out"):
+            return {self.o_out: [0]}
+        return {}
+
+    def extTransition(self, inputs):
+        if self.ext_delay:
+            pystones(self.ext_delay)
+
+        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, 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")
+
+        if depth < 1:
+            raise ValueError("Invalid depth")
+        if width < 1:
+            raise ValueError("Invalid width")
+        if int_delay < 0:
+            raise ValueError("Invalid int_delay")
+        if ext_delay < 0:
+            raise ValueError("Invalid ext_delay")
+
+        if depth == 1:
+            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)
+            self.connectPorts(atomic.o_out, self.o_out)
+        else:
+            coupled = self.gen_coupled()
+            self.addSubModel(coupled)
+            self.connectPorts(self.i_in, coupled.i_in)
+            self.connectPorts(coupled.o_out, self.o_out)
+
+            for idx in range(width - 1):
+                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
+    def gen_coupled(self):
+        """ :return a coupled method with i_in and o_out ports"""
+        pass
+
+
+class LI(DEVStoneWrapper):
+
+    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,
+                  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, 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)
+            self.connectPorts(self.i_in, self.component_set[-1].i_in)
+
+        for idx in range(1, len(self.component_set) - 1):
+            assert isinstance(self.component_set[idx], AtomicDEVS)
+            self.connectPorts(self.component_set[idx].o_out, self.component_set[idx + 1].i_in)
+            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,
+                  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, 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")
+
+        assert len(self.component_set) > 0
+        if isinstance(self.component_set[0], CoupledDEVS):
+            self.connectPorts(self.i_in, self.component_set[0].i_in2)
+
+        if len(self.component_set) > 1:
+            assert isinstance(self.component_set[-1], AtomicDEVS)
+            self.connectPorts(self.i_in2, self.component_set[-1].i_in)
+            self.connectPorts(self.component_set[-1].o_out, self.o_out2)
+
+        for idx in range(1, len(self.component_set) - 1):
+            assert isinstance(self.component_set[idx], AtomicDEVS)
+            self.connectPorts(self.component_set[idx].o_out, self.component_set[idx + 1].i_in)
+            self.connectPorts(self.i_in2, self.component_set[idx].i_in)
+            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,
+                  prep_time=self.prep_time, stats=self.stats)
+
+
+class HOmod(CoupledDEVS):
+
+    def __init__(self, name: str, depth: int, width: int, int_delay: float, ext_delay: float, 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.i_in = self.addInPort("i_in")
+        self.i_in2 = self.addInPort("i_in2")
+        self.o_out = self.addOutPort("o_out")
+
+        if depth < 1:
+            raise ValueError("Invalid depth")
+        if width < 1:
+            raise ValueError("Invalid width")
+        if int_delay < 0:
+            raise ValueError("Invalid int_delay")
+        if ext_delay < 0:
+            raise ValueError("Invalid ext_delay")
+
+        if depth == 1:
+            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)
+            self.connectPorts(atomic.o_out, self.o_out)
+        else:
+            coupled = HOmod("Coupled_%d" % (self.depth - 1), self.depth - 1, self.width, self.int_delay, self.ext_delay, prep_time=prep_time, stats=stats)
+            self.addSubModel(coupled)
+            self.connectPorts(self.i_in, coupled.i_in)
+            self.connectPorts(coupled.o_out, self.o_out)
+
+            if width >= 2:
+                atomics = defaultdict(list)
+
+                # Generate atomic components
+                for i in range(width):
+                    min_row_idx = 0 if i < 2 else i - 1
+                    for j in range(min_row_idx, width - 1):
+                        if self.stats:
+                            atomic = DelayedAtomicStats("Atomic_%d_%d_%d" % (depth - 1, i, j), int_delay, ext_delay,
+                                                   add_out_port=True, prep_time=prep_time)
+                        else:
+                            atomic = DelayedAtomic("Atomic_%d_%d_%d" % (depth - 1, i, j), int_delay, ext_delay,
+                                                   add_out_port=True, prep_time=prep_time)
+                        self.addSubModel(atomic)
+                        atomics[i].append(atomic)
+
+                # Connect EIC
+                for atomic in atomics[0]:
+                    self.connectPorts(self.i_in2, atomic.i_in)
+                for i in range(1, width):
+                    atomic_set = atomics[i]
+                    self.connectPorts(self.i_in2, atomic_set[0].i_in)
+
+                # Connect IC
+                for atomic in atomics[0]:  # First row to coupled component
+                    self.connectPorts(atomic.o_out, coupled.i_in2)
+                for i in range(len(atomics[1])):  # Second to first rows
+                    for j in range(len(atomics[0])):
+                        self.connectPorts(atomics[1][i].o_out, atomics[0][j].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)
+
+
+if __name__ == '__main__':
+    import sys
+
+    sys.setrecursionlimit(10000)
+    root = HOmod("Root", 4, 3, 0, 0)
+    sim = Simulator(root)
+    sim.setVerbose(None)
+    # sim.setTerminationTime(10.0)
+    sim.setStateSaving("custom")
+    sim.simulate()

devstone/pythonpdevs-minimal/generator.py

@@ -0,0 +1,36 @@
+from typing import Any
+
+from pypdevs.minimal import AtomicDEVS
+from pypdevs.infinity import INFINITY
+
+
+class Generator(AtomicDEVS):
+
+    def __init__(self, name, num_outputs=1, period=None):
+        super(Generator, self).__init__(name=name)
+        self.num_outputs = num_outputs
+        self.period = period
+        self.state = "active"
+        self.o_out = self.addOutPort("o_out")
+
+    def intTransition(self):
+        if self.state == "active" and self.period is not None:
+            self.state = "waiting"
+        elif self.state == "waiting":
+            self.state = "active"
+        else:
+            self.state = "passive"
+
+    def timeAdvance(self):
+        if self.state == "active":
+            return 0
+        elif self.state == "waiting":
+            return self.period
+        else:
+            return INFINITY
+
+    def outputFnc(self):
+        return {self.o_out: list(range(self.num_outputs))}
+
+    def extTransition(self, inputs):
+        pass

devstone/pythonpdevs-minimal/main.py

@@ -0,0 +1,80 @@
+import argparse
+import sys
+import time
+from pypdevs.minimal import CoupledDEVS
+from pypdevs.minimal import Simulator
+
+from devstone import LI, HI, HO, HOmod
+from generator import Generator
+
+sys.setrecursionlimit(10000)
+
+
+class DEVStoneEnvironment(CoupledDEVS):
+
+    def __init__(self, name, devstone_model, num_gen_outputs=1, gen_period=None):
+        super(DEVStoneEnvironment, self).__init__(name=name)
+        generator = Generator("generator", num_gen_outputs, gen_period)
+        self.addSubModel(generator)
+        self.addSubModel(devstone_model)
+
+        self.connectPorts(generator.o_out, devstone_model.i_in)
+
+        if isinstance(devstone_model, HO) or isinstance(devstone_model, HOmod):
+            self.connectPorts(generator.o_out, devstone_model.i_in2)
+
+
+MODEL_TYPES = ("LI", "HI", "HO", "HOmod")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description='Script to compare DEVStone implementations with different engines')
+
+    parser.add_argument('-m', '--model-type', required=True, help='DEVStone model type (LI, HI, HO, HOmod)')
+    parser.add_argument('-d', '--depth', type=int, required=True, help='Number of recursive levels of the model.')
+    parser.add_argument('-w', '--width', type=int, required=True, help='Width of each coupled model.')
+    parser.add_argument('-i', '--int-cycles', type=int, default=0, help='Dhrystone cycles executed in internal transitions')
+    parser.add_argument('-e', '--ext-cycles', type=int, default=0, help='Dhrystone cycles executed in external transitions')
+    parser.add_argument('-f', '--flatten', action="store_true", help='Activate flattening on model')
+    parser.add_argument('-c', '--chained', action="store_true", help='Use chained coordinator')
+
+    args = parser.parse_args()
+
+    if args.model_type not in MODEL_TYPES:
+        raise RuntimeError("Unrecognized model type.")
+
+    return args
+
+
+if __name__ == '__main__':
+
+    args = parse_args()
+
+    devstone_model = None
+
+    start_time = time.time()
+    if args.model_type == "LI":
+        devstone_model = LI("LI_root", args.depth, args.width, args.int_cycles, args.ext_cycles)
+    elif args.model_type == "HI":
+        devstone_model = HI("HI_root", args.depth, args.width, args.int_cycles, args.ext_cycles)
+    elif args.model_type == "HO":
+        devstone_model = HO("HO_root", args.depth, args.width, args.int_cycles, args.ext_cycles)
+    elif args.model_type == "HOmod":
+        devstone_model = HOmod("HOmod_root", args.depth, args.width, args.int_cycles, args.ext_cycles)
+
+    env = DEVStoneEnvironment("DEVStoneEnvironment", devstone_model)
+    model_created_time = time.time()
+
+    sim = Simulator(env)
+    sim.setVerbose(None)
+    # sim.setTerminationTime(10.0)
+    # sim.setStateSaving("custom")
+    engine_setup_time = time.time()
+
+    sim.simulate()
+    sim_time = time.time()
+
+    print("Model creation time: {}".format(model_created_time - start_time))
+    print("Engine setup time: {}".format(engine_setup_time - model_created_time))
+    print("Simulation time: {}".format(sim_time - engine_setup_time))
+

devstone/pythonpdevs-minimal/pystone.py

@@ -0,0 +1,277 @@
+#! /usr/bin/python3.6
+
+"""
+"PYSTONE" Benchmark Program
+
+Version:        Python/1.2 (corresponds to C/1.1 plus 3 Pystone fixes)
+
+Author:         Reinhold P. Weicker,  CACM Vol 27, No 10, 10/84 pg. 1013.
+
+                Translated from ADA to C by Rick Richardson.
+                Every method to preserve ADA-likeness has been used,
+                at the expense of C-ness.
+
+                Translated from C to Python by Guido van Rossum.
+
+Version History:
+
+                Version 1.1 corrects two bugs in version 1.0:
+
+                First, it leaked memory: in Proc1(), NextRecord ends
+                up having a pointer to itself.  I have corrected this
+                by zapping NextRecord.PtrComp at the end of Proc1().
+
+                Second, Proc3() used the operator != to compare a
+                record to None.  This is rather inefficient and not
+                true to the intention of the original benchmark (where
+                a pointer comparison to None is intended; the !=
+                operator attempts to find a method __cmp__ to do value
+                comparison of the record).  Version 1.1 runs 5-10
+                percent faster than version 1.0, so benchmark figures
+                of different versions can't be compared directly.
+
+                Version 1.2 changes the division to floor division.
+
+                Under Python 3 version 1.1 would use the normal division
+                operator, resulting in some of the operations mistakenly
+                yielding floats. Version 1.2 instead uses floor division
+                making the benchmark an integer benchmark again.
+
+"""
+
+LOOPS = 50000
+
+from time import time
+
+__version__ = "1.2"
+
+[Ident1, Ident2, Ident3, Ident4, Ident5] = range(1, 6)
+
+class Record:
+
+    def __init__(self, PtrComp = None, Discr = 0, EnumComp = 0,
+                       IntComp = 0, StringComp = 0):
+        self.PtrComp = PtrComp
+        self.Discr = Discr
+        self.EnumComp = EnumComp
+        self.IntComp = IntComp
+        self.StringComp = StringComp
+
+    def copy(self):
+        return Record(self.PtrComp, self.Discr, self.EnumComp,
+                      self.IntComp, self.StringComp)
+
+TRUE = 1
+FALSE = 0
+
+def main(loops=LOOPS):
+    benchtime, stones = pystones(loops)
+    print("Pystone(%s) time for %d passes = %g" % \
+          (__version__, loops, benchtime))
+    print("This machine benchmarks at %g pystones/second" % stones)
+
+
+def pystones(loops=LOOPS):
+    return Proc0(loops)
+
+IntGlob = 0
+BoolGlob = FALSE
+Char1Glob = '\0'
+Char2Glob = '\0'
+Array1Glob = [0]*51
+Array2Glob = [x[:] for x in [Array1Glob]*51]
+PtrGlb = None
+PtrGlbNext = None
+
+def Proc0(loops=LOOPS):
+    global IntGlob
+    global BoolGlob
+    global Char1Glob
+    global Char2Glob
+    global Array1Glob
+    global Array2Glob
+    global PtrGlb
+    global PtrGlbNext
+
+    starttime = time()
+    for i in range(loops):
+        pass
+    nulltime = time() - starttime
+
+    PtrGlbNext = Record()
+    PtrGlb = Record()
+    PtrGlb.PtrComp = PtrGlbNext
+    PtrGlb.Discr = Ident1
+    PtrGlb.EnumComp = Ident3
+    PtrGlb.IntComp = 40
+    PtrGlb.StringComp = "DHRYSTONE PROGRAM, SOME STRING"
+    String1Loc = "DHRYSTONE PROGRAM, 1'ST STRING"
+    Array2Glob[8][7] = 10
+
+    starttime = time()
+
+    for i in range(loops):
+        Proc5()
+        Proc4()
+        IntLoc1 = 2
+        IntLoc2 = 3
+        String2Loc = "DHRYSTONE PROGRAM, 2'ND STRING"
+        EnumLoc = Ident2
+        BoolGlob = not Func2(String1Loc, String2Loc)
+        while IntLoc1 < IntLoc2:
+            IntLoc3 = 5 * IntLoc1 - IntLoc2
+            IntLoc3 = Proc7(IntLoc1, IntLoc2)
+            IntLoc1 = IntLoc1 + 1
+        Proc8(Array1Glob, Array2Glob, IntLoc1, IntLoc3)
+        PtrGlb = Proc1(PtrGlb)
+        CharIndex = 'A'
+        while CharIndex <= Char2Glob:
+            if EnumLoc == Func1(CharIndex, 'C'):
+                EnumLoc = Proc6(Ident1)
+            CharIndex = chr(ord(CharIndex)+1)
+        IntLoc3 = IntLoc2 * IntLoc1
+        IntLoc2 = IntLoc3 // IntLoc1
+        IntLoc2 = 7 * (IntLoc3 - IntLoc2) - IntLoc1
+        IntLoc1 = Proc2(IntLoc1)
+
+    benchtime = time() - starttime - nulltime
+    if benchtime == 0.0:
+        loopsPerBenchtime = 0.0
+    else:
+        loopsPerBenchtime = (loops / benchtime)
+    return benchtime, loopsPerBenchtime
+
+def Proc1(PtrParIn):
+    PtrParIn.PtrComp = NextRecord = PtrGlb.copy()
+    PtrParIn.IntComp = 5
+    NextRecord.IntComp = PtrParIn.IntComp
+    NextRecord.PtrComp = PtrParIn.PtrComp
+    NextRecord.PtrComp = Proc3(NextRecord.PtrComp)
+    if NextRecord.Discr == Ident1:
+        NextRecord.IntComp = 6
+        NextRecord.EnumComp = Proc6(PtrParIn.EnumComp)
+        NextRecord.PtrComp = PtrGlb.PtrComp
+        NextRecord.IntComp = Proc7(NextRecord.IntComp, 10)
+    else:
+        PtrParIn = NextRecord.copy()
+    NextRecord.PtrComp = None
+    return PtrParIn
+
+def Proc2(IntParIO):
+    IntLoc = IntParIO + 10
+    while 1:
+        if Char1Glob == 'A':
+            IntLoc = IntLoc - 1
+            IntParIO = IntLoc - IntGlob
+            EnumLoc = Ident1
+        if EnumLoc == Ident1:
+            break
+    return IntParIO
+
+def Proc3(PtrParOut):
+    global IntGlob
+
+    if PtrGlb is not None:
+        PtrParOut = PtrGlb.PtrComp
+    else:
+        IntGlob = 100
+    PtrGlb.IntComp = Proc7(10, IntGlob)
+    return PtrParOut
+
+def Proc4():
+    global Char2Glob
+
+    BoolLoc = Char1Glob == 'A'
+    BoolLoc = BoolLoc or BoolGlob
+    Char2Glob = 'B'
+
+def Proc5():
+    global Char1Glob
+    global BoolGlob
+
+    Char1Glob = 'A'
+    BoolGlob = FALSE
+
+def Proc6(EnumParIn):
+    EnumParOut = EnumParIn
+    if not Func3(EnumParIn):
+        EnumParOut = Ident4
+    if EnumParIn == Ident1:
+        EnumParOut = Ident1
+    elif EnumParIn == Ident2:
+        if IntGlob > 100:
+            EnumParOut = Ident1
+        else:
+            EnumParOut = Ident4
+    elif EnumParIn == Ident3:
+        EnumParOut = Ident2
+    elif EnumParIn == Ident4:
+        pass
+    elif EnumParIn == Ident5:
+        EnumParOut = Ident3
+    return EnumParOut
+
+def Proc7(IntParI1, IntParI2):
+    IntLoc = IntParI1 + 2
+    IntParOut = IntParI2 + IntLoc
+    return IntParOut
+
+def Proc8(Array1Par, Array2Par, IntParI1, IntParI2):
+    global IntGlob
+
+    IntLoc = IntParI1 + 5
+    Array1Par[IntLoc] = IntParI2
+    Array1Par[IntLoc+1] = Array1Par[IntLoc]
+    Array1Par[IntLoc+30] = IntLoc
+    for IntIndex in range(IntLoc, IntLoc+2):
+        Array2Par[IntLoc][IntIndex] = IntLoc
+    Array2Par[IntLoc][IntLoc-1] = Array2Par[IntLoc][IntLoc-1] + 1
+    Array2Par[IntLoc+20][IntLoc] = Array1Par[IntLoc]
+    IntGlob = 5
+
+def Func1(CharPar1, CharPar2):
+    CharLoc1 = CharPar1
+    CharLoc2 = CharLoc1
+    if CharLoc2 != CharPar2:
+        return Ident1
+    else:
+        return Ident2
+
+def Func2(StrParI1, StrParI2):
+    IntLoc = 1
+    while IntLoc <= 1:
+        if Func1(StrParI1[IntLoc], StrParI2[IntLoc+1]) == Ident1:
+            CharLoc = 'A'
+            IntLoc = IntLoc + 1
+    if CharLoc >= 'W' and CharLoc <= 'Z':
+        IntLoc = 7
+    if CharLoc == 'X':
+        return TRUE
+    else:
+        if StrParI1 > StrParI2:
+            IntLoc = IntLoc + 7
+            return TRUE
+        else:
+            return FALSE
+
+def Func3(EnumParIn):
+    EnumLoc = EnumParIn
+    if EnumLoc == Ident3: return TRUE
+    return FALSE
+
+if __name__ == '__main__':
+    import sys
+    def error(msg):
+        print(msg, end=' ', file=sys.stderr)
+        print("usage: %s [number_of_loops]" % sys.argv[0], file=sys.stderr)
+        sys.exit(100)
+    nargs = len(sys.argv) - 1
+    if nargs > 1:
+        error("%d arguments are too many;" % nargs)
+    elif nargs == 1:
+        try: loops = int(sys.argv[1])
+        except ValueError:
+            error("Invalid argument %r;" % sys.argv[1])
+    else:
+        loops = LOOPS
+    main(loops)

devstone/pythonpdevs-minimal/test_devstone.py

@@ -0,0 +1,347 @@
+from abc import abstractmethod
+from unittest import TestCase
+import random
+
+from pypdevs.minimal 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)