kneron_model_converter/vendor/sys_flow/exceptions.py

#! /usr/bin/env python3

import logging
import traceback

from functools import wraps

import time
from blinker import signal


def print_err(e, do_print=True):
    """will print python error if do_print is true.

    Always call this function, but it will not run if turned off in config.
    """
    if do_print:
        print(e)
        traceback.print_tb(e.__traceback__)


def print_command(cmd, do_print=False):
    """will print generate command line if do_print is true.

    Always call this function, but it will not run if turned off in config.
    """
    if do_print:
        if type(cmd) is list:
            try:
                cmd = " ".join(cmd)
            except Exception as e:
                print_err(e, True)
                print("ERROR: cannot convert cmd to string: {}".format(cmd))
                return

        if type(cmd) is str:
            print("\n{}\n".format(cmd))


def run_module(module_name=None):
    def run_module_inner(func, module_name=module_name):
        """decorator to catch exceptions in test_case.run_xx call

        NOTE: ONLY for `test_case.run_xx(self, config, *args, **kargs)` calls.
            which has no return value.

        Not for dynasty calls, which is function calls with different parameters.
        """

        @wraps(func)
        def wrapper(self, *args, module_name=module_name, **kargs):

            if module_name is None:
                module_name = func.__name__
            if "hw_mode" in kargs:
                m_name = module_name.split("/")[-1]
                hw_mode = kargs["hw_mode"]
                module_name = f"kdp{hw_mode}/{m_name}"

            # prepare parameters
            config_ready = hasattr(self, "config")
            # maybe: no self.config when initial

            try:
                print_time_stamp = config_ready and self.config["regression"]["generate_time_stamps"]
                unit_to_denominator = {"second": 1, "minute": 60, "hour": 3600}
                curr_unit = self.config["regression"]["time_stamps_unit"]
                col_name = f"{module_name}:t"
            except:
                print_time_stamp = False

            try:
                if print_time_stamp:
                    start_time = time.perf_counter()
                    results = func(self, *args, **kargs)
                    # func may failed already
                    end_time = time.perf_counter()
                    total_time = (end_time - start_time) / unit_to_denominator[curr_unit]
                    signal("time_sender").send((self.model_id, col_name, int(total_time)))
                else:
                    results = func(self, *args, **kargs)

                signal("data_sender").send((self.model_id, module_name, "✓"))
                return results

            except Exception as e:
                # try to get time as well
                try:
                    if print_time_stamp:
                        end_time = time.perf_counter()
                        total_time = (end_time - start_time) / unit_to_denominator[curr_unit]
                        signal("time_sender").send((self.model_id, col_name, int(total_time)))
                except Exception as e2:
                    pass

                if type(e) is RegressionError:
                    raise
                elif type(e) is MultiRegressionError:
                    raise
                else:
                    print_err(e, config_ready and self.config["regression"]["print_error"])
                    raise RegressionError(module_name, self.model_id)

        return wrapper
    return run_module_inner


class MultiRegressionError(Exception):
    def __init__(self, error_list):
        # TODO: what if just general error?

        self.errors = [e for e in error_list if type(e) is RegressionError]


class GeneralError(Exception):
    """use this GeneralError to send error test_case.py for better hanle"""
    def __init__(self, msg, details=None):
        self.msg = msg
        self.details = details


class RegressionError(Exception):
    """Customized Exception for flow error recording.

    If a moduel failed, it will raise this exception with info.
    Exceptions will be recorded in report.

    * Dynasty related module and "Success" are defined manually.
    * other modules are using the corresponding function names.

    """


    def __init__(self, module_name, model_name, msg=None, print_err=False):
        """Initial RegressionError by give certain infomation for future record.

        - module_name: knerex updater 520, dynasty 520wq, etc
        - model_name: is the model name running regression
        - msg: is the brief message to print in summary report
        """
        if print_err:
            traceback.print_exc()

        self.module_name = module_name
        self.model_name = model_name
        self.msg = msg

        if module_name == "general/Success":
            flag = "Success"
        else:
            flag = "Failure"

        err_msg = '{} for model "{}" when running "{}"'.format(flag, model_name, module_name)

        if flag == "Failure":
            red = "\033[91m {}\033[00m"
            err_msg = red.format(err_msg)
        else:
            green = "\033[92m {}\033[00m"
            err_msg = green.format(err_msg)

        logging.getLogger().error(err_msg, exc_info=False)

        # usually we will print command. below info are redundant.
        # if command is not None:
        #     logging.getLogger().debug(command, exc_info=False)


def excepthook_print_last(etype, value, tb):
    """Print last exception, instead of all.

    To use, run:
    > sys.excepthook = excepthook_print_last
    """
    traceback_list = traceback.format_exception(etype, value, tb)
    print(traceback_list[-1])

    stack_summary = traceback.extract_tb(tb)
    last_frame = stack_summary[-1]
    filename, lineno, name, line = last_frame

    print(f"File: {filename}, Line: {lineno}")
    print(f"Code: {line}")

    # print all variables. not needed now
    # frame = tb.tb_frame
    # for key, value in frame.f_locals.items():
    #     print(f"{key} = {value}")