convert tipog procedure to itpog object

tests/itpog.py

@@ -0,0 +1,199 @@
+'''
+Tipog tests
+
+Usage:
+  tipog [-v] <filename> [-o <output>]
+
+Options:
+  -v --verbose       Verbose print
+  -o --output        Register the results in <output> file (.json)
+  -h --help          Show this screen.
+  --version          Show version.
+'''
+
+import cProfile
+from importlib import import_module
+import json
+import os
+import timeit
+
+from docopt import docopt
+import yaml
+
+__version__ = "0.2"
+
+class ItpogDataError(IOError):
+    pass
+
+class ItPog(object):
+    def __init__(self, ipfile, verbose=False, output=None, outfilename=None):
+        try:
+            with open(ipfile, "r") as f:
+                data = yaml.load(f)
+        except FileNotFoundError:
+            raise FileNotFoundError("no file named '{}'".format(ipfile))
+        except yaml.scanner.ScannerError:
+            raise ItpogDataError("unreadable yaml file '{}'".format(ipfile))
+
+        if not data:
+            raise ItpogDataError("file '{}' is empty".format(ipfile))
+
+        self.ipfile = ipfile
+        self.ipdata = data
+
+        self.verbose = verbose
+        self.output = output
+        self.outfilename = outfilename
+        self.out_result = []
+
+    @classmethod
+    def run_file(cls, *args, **kwargs):
+        """ convenient method to load a file and run the tests
+        returns an ItPog object"""
+        try:
+            itpog = cls(*args, **kwargs)
+            itpog.run()
+        except Exception as e:
+            print("{}: {}".format(e.__class__.__name__, str(e)))
+        return itpog
+
+    def run(self):
+        """ run the jobs as programmed in the given file """
+        for name, imp in self.ipdata["imports"].items():
+            try:
+                globals()[name] = import_module(imp)
+            except (ModuleNotFoundError, ImportError):
+                raise ItpogDataError("unable to import '{}'".format(imp))
+        try:
+            jobs = self.ipdata["jobs"]
+        except KeyError:
+            raise ItpogDataError("missing 'jobs' entry")
+
+        for function_name, job in jobs.items():
+            try:
+                self._run_job(function_name, job)
+            except Exception as e:
+                print("{}: {}".format(e.__class__.__name__, str(e)))
+
+        if self.output:
+            if not self.outfilename:
+                self.outfilename = "{}_result.json".format(os.path.splitext(self.ipfile)[0])
+
+            outfilepath = os.path.join(".", self.outfilename)
+            try:
+                os.remove(outfilepath)
+            except FileNotFoundError:
+                pass
+            with open(outfilepath, "w+") as f:
+                json.dump(self.out_result, f)
+
+        print("** End of the tests")
+
+    def _run_job(self, function_name, job):
+            try:
+                function = eval(function_name)
+            except NameError:
+                raise NameError("unknown function ('{}')".format(function_name))
+
+            if self.verbose:
+                print("** Test function '{}'".format(function_name))
+
+            try:
+                args_lst = job["args"]
+            except KeyError:
+                args_lst = [[]]
+
+            try:
+                validator_str = job["validator"]
+                try:
+                    validator = eval(validator_str)
+                    if self.verbose:
+                        print("> validator: '{}'".format(validator_str))
+                except NameError:
+                    raise ItpogDataError("unknown function as validator ('{}')".format(validator_str))
+
+            except (TypeError, KeyError, SyntaxError):
+                validator = None
+
+            for args in args_lst:
+                call_str = "{}(*{})".format(function_name, args)
+
+                print(">> {}".format(call_str))
+
+                exectime = self.ittime(call_str)
+                print("\t> Run in {} ms.".format(exectime))
+
+                if self.verbose:
+                    self.profile(call_str)
+
+                if validator:
+                    valid = self.validate(function, validator, args)
+                    print("\t> Validated: {}".format(valid))
+
+                if self.output:
+                    last_result = {"call": call_str, "result": function(*args), "exectime": exectime}
+                    try:
+                        last_result.update(job["infos"])
+                    except KeyError:
+                        pass
+                    self.out_result.append(last_result)
+
+            if self.verbose:
+                print("------------------------------")
+
+
+    @staticmethod
+    def profile(_call):
+        """ Use the cProfile module to measure the execution time of each call
+            _call has to be a string
+        """
+        cProfile.run(_call, sort='nfl')
+
+    @staticmethod
+    def ittime(_call):
+        """ returns the execution time in milli-seconds
+            _call has to be a string
+            (ex: 'time.sleep(1)', which will return 1000)
+        """
+        number, t = 1, 0
+        while t < 10 ** 8:
+            t = timeit.timeit(lambda: eval(_call), number=number)
+            if t >= 0.1:
+                return 1000 * t / number
+            number *= 10
+        else:
+            return -1
+
+    @staticmethod
+    def validate(fct, validator, args=[], kwargs={}):
+        """ compare the results of 'fct' and 'validator' methods,
+        with the same 'args' and 'kwargs' arguments """
+        result = fct(*args, **kwargs)
+        attended = validator(*args, **kwargs)
+        return result == attended
+
+    @staticmethod
+    def generate_sample(filename="itpog_sample_file.yml"):
+        """ generate a sample yaml configuration file """
+        data = {"imports": {"time": "time", "math": "math"},
+                "jobs": {
+                          "time.sleep": {"infos": {"note": "x second attended"},
+                                    "args": [[0.1], [1]],
+                                    "validator": ""},
+                          "math.pow":  {"infos": {},
+                                    "args": [[1, 2], [3, 2], [3, 3], [4, 2]],
+                                    "validator": "lambda x, y: x**y"}
+                         }
+                }
+        with open(os.path.join(".", filename), "w+") as f:
+            yaml.dump(data, f)
+
+
+if __name__ == "__main__":
+
+    sysargs = docopt(__doc__, version=__version__)
+
+    itpog = ItPog.run_file(sysargs["<filename>"], sysargs["--verbose"], sysargs["--output"], sysargs["<output>"])
+
+
+

tests/tipog_model.yml → tests/itpog_model.yml

@@ -1,5 +1,5 @@
 imports:
-- from pypog import geometry
+  geometry: pypog.geometry
   
 jobs:
   geometry.fhex2_line:

tests/profiler.py

@@ -1,42 +0,0 @@
-'''
-    Profiling performance of the geometry functions
-'''
-
-import cProfile
-import timeit
-
-from pypog import geometry
-
-_profile = ['geometry.squ2_line(0,0,1000,1000)',
-            'geometry.fhex2_line(0,0,1000,1000)',
-            'geometry.rectangle(0,0,1000,1000)',
-            'geometry.hollow_rectangle(0,0,1000,1000)',
-            'geometry.squ2_triangle(0,0,100,100,2)',
-            'geometry.fhex2_triangle(0,0,100,100,2)',
-#             'geometry.squ3_triangle(0,0,0,100,100,100,2)',
-#             'geometry.fhex3_triangle(0,0,0,100,100,100,2)',
-            'geometry.fhex2_pivot((0,0),[(i, j) for i in range(1, 100) for j in range(1,100)], 1)',
-            'geometry.squ2_pivot((0,0),[(i, j) for i in range(1, 100) for j in range(1,100)], 1)',
-            ]
-
-for _call in _profile:
-    print(">> {}".format(_call))
-    cProfile.run(_call, sort='nfl')
-
-    number = 1
-    t = 0
-    while 1:
-        t = timeit.timeit(lambda: eval(_call), number=number)
-        if t >= 1:
-            break
-        elif number > 1000000:
-            print("error: number superior to a million")
-            number = "err"
-            t = "?"
-            break
-        number *= 10
-
-    print("Timeit (x{}): {} s.\n".format(number, t))
-    print("".join(["-" for _ in range(30)]))
-
-

tests/tipog.py

@@ -1,153 +0,0 @@
-'''
-Tipog tests
-
-Usage:
-  tipog [-v] <filename> [-o <output>]
-
-Options:
-  -v --verbose       Verbose result
-  -o --output        Register the results in <output> file (json)
-  -h --help          Show this screen.
-  --version          Show version.
-'''
-
-import cProfile
-import json
-import os
-import sys
-import timeit
-
-from docopt import docopt
-import yaml
-
-
-__version__ = "0.1"
-
-sysargs = docopt(__doc__, version=__version__)
-
-yaml_file = sysargs["<filename>"]
-
-try:
-    with open(yaml_file, "r") as f:
-        data = yaml.load(f)
-except FileNotFoundError:
-    print("ERR: no file named '{}'".format(yaml_file))
-    sys.exit(1)
-except yaml.scanner.ScannerError:
-    print("ERR: unreadable yaml file '{}'".format(yaml_file))
-    sys.exit(1)
-
-if not data:
-    print("ERR: empty yaml file '{}'".format(yaml_file))
-    sys.exit(1)
-
-for to_import in data["imports"]:
-    try:
-        # TODO: avoid the 'exec'
-        exec(to_import)
-    except (ModuleNotFoundError, ImportError):
-        print("ERR: unable to import '{}'".format(to_import))
-        sys.exit(1)
-
-try:
-    jobs = data["jobs"]
-except KeyError:
-    print("ERR: missing 'jobs' entry")
-    sys.exit(1)
-
-
-
-
-def profile(_call, verbose=False):
-
-    print(">> {}".format(_call))
-
-    if verbose:
-        cProfile.run(_call, sort='nfl')
-
-    number = 1
-    t = 0
-    while 1:
-        t = timeit.timeit(lambda: eval(_call), number=number)
-        if t >= 0.1:
-            break
-        elif number > 10000000:
-            print("ERR: unable to compute the execution time")
-            number = "err"
-            t = "?"
-            break
-        number *= 10
-
-    return t / 1000
-
-
-
-def validate(fct, validator, args, verbose=False):
-    result = fct(*args)
-    attended = validator(*args)
-    return result == attended
-
-to_register = []
-
-for function_name, job in jobs.items():
-
-    try:
-        function = eval(function_name)
-    except NameError:
-        print("ERR: unknown function ('{}')".format(function_name))
-        continue
-
-    if sysargs["--verbose"]:
-        print("** Test function '{}'".format(function_name))
-
-    try:
-        args_lst = job["args"]
-    except KeyError:
-        args_lst = [[]]
-
-    try:
-        validator_str = job["validator"]
-        try:
-            validator = eval(validator_str)
-            if sysargs["--verbose"]:
-                print("> validator: '{}'".format(validator_str))
-        except NameError:
-            print("ERR: unknown function as validator ('{}')".format(validator_str))
-    except (TypeError, KeyError):
-        validator = None
-
-    for args in args_lst:
-        call_str = "{}(*{})".format(function_name, args)
-
-        exectime = profile(call_str, sysargs["--verbose"])
-        print("\t> Run in {} ms.".format(exectime))
-
-        if validator:
-            valid = validate(function, validator, args, sysargs["--verbose"])
-            print("\t> Validated: {}".format(valid))
-
-        if sysargs["--output"]:
-            last_result = {"call": call_str, "result": function(*args), "exectime": exectime}
-            try:
-                last_result.update(job["infos"])
-            except KeyError:
-                pass
-            to_register.append(last_result)
-
-    if sysargs["--verbose"]:
-        print("------------------------------")
-
-if sysargs["--output"]:
-    output_name = sysargs["<output>"]
-    if not output_name:
-        output_name = "{}_result".format(os.path.splitext(sysargs["<filename>"])[0])
-
-    reg_file_path = os.path.join(".", output_name)
-    try:
-        os.remove(reg_file_path)
-    except FileNotFoundError:
-        pass
-    with open(reg_file_path, "w+") as f:
-        json.dump(to_register, f)
-
-print("** End of the tests")