quecpython.helios-service/code/usr/bin/log_service.py

from usr.bin.components.abstract_service import AbstractService
from usr.utils.service_utils import Singleton
from usr.utils.resolver import TimeResolver
from usr.bin.components.monitor import ServiceMonitor
from machine import UART
import uos

LOGGER = "LOG"


class LOG_LV:
    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    CRITICAL = "CRITICAL"


class AbstractLogOutputUtil(object):
    def open(self):
        pass

    def output(self, message, **kwargs):
        pass

    def close(self):
        pass

    def __call__(self, message, **kwargs):
        self.output(message, **kwargs)


@Singleton
class UartLogOutputUtil(AbstractLogOutputUtil):
    def __init__(self, UARTn=UART.UART2):
        self.uart = UART(UARTn, 115200, 8, 0, 1, 0)

    def output(self, message, **kwargs):
        self.uart.write(message)


@Singleton
class PrintLogOutputUtil(AbstractLogOutputUtil):
    def output(self, message, **kwargs):
        print(message)


@Singleton
class MqttLogOutputUtil(AbstractLogOutputUtil):

    def __init__(self):
        super().__init__()

    def output(self, message, **kwargs):
        pass


@Singleton
class ConsoleLogOutputUtil(AbstractLogOutputUtil):
    def __init__(self):
        super().__init__()

    def output(self, message, **kwargs):
        pass


@Singleton
class FileLogOutputUtil(AbstractLogOutputUtil):
    def __init__(self):
        super().__init__()
        self.file_name = "log.txt"
        self.abs_path = "/usr/log/"
        self.content = []
        self.f = None

    def open(self):
        if "log" not in uos.listdir("/usr"):
            uos.mkdir(self.abs_path)
        self.f = open(self.abs_path + self.file_name, "w+")

    def output(self, message, **kwargs):
        self.open()
        if len(self.content) > 10:
            self.close()

    def close(self):
        self.f.close()
        self.f = None


class AbstractFormatUtil(object):
    def format(self, *args, **kwargs):
        pass


class LogFormatUtil(AbstractFormatUtil):
    def format(self, *args, **kwargs):
        log_format = "{} {} [{}] - {}\n".format(args[0], args[1], args[2], args[3])
        return log_format


class LogServiceMonitor(ServiceMonitor):

    @staticmethod
    def create_monitor(config=None):
        log_service = LogService()
        if config is not None:
            level = config.get('level')
            log_service.set_level(level)
        lsm = LogServiceMonitor(log_service)
        if config is not None:
            lsm.set_exception_handlers(config.get('exceptionHandlers', None))
        return lsm


@Singleton
class LogService(AbstractService):
    """
        default log is async queue
        you can set
    """

    def __init__(self):
        super().__init__(LOGGER)
        self.__reporter = PrintLogOutputUtil()
        self.__tr = TimeResolver()
        self.format_util = LogFormatUtil()
        self.__level_map = {
            LOG_LV.DEBUG: 0,
            LOG_LV.INFO: 1,
            LOG_LV.WARNING: 2,
            LOG_LV.ERROR: 3,
            LOG_LV.CRITICAL: 4
        }
        self.low_level = 0

    def __set_report(self, report):
        if isinstance(report, AbstractLogOutputUtil):
            self.__reporter = report

    def set_output(self, out_obj):
        if isinstance(out_obj, AbstractLogOutputUtil):
            self.__reporter.close()
            self.__reporter = out_obj
            self.__reporter.open()
        else:
            self.log_send(self.name, LOG_LV.ERROR, '"{}" is not extend AbstractLogOutputUtil'.format(out_obj))
            raise Exception('"{}" is not extend AbstractLogOutputUtil'.format(out_obj))

    def set_level(self, level):
        if level in self.__level_map:
            self.low_level = self.__level_map[level]
        else:
            self.low_level = 0

    def log_send(self, sign, level, msg, mode=1):
        """send log deal"""
        if self.mode is not None:
            mode = self.mode
        if self.__level_map[level] >= self.low_level:
            if mode:
                self.send_msg_async(message=self.format_msg(sign, level, msg))
            else:
                self.send_msg_sync(message=self.format_msg(sign, level, msg))

    def format_msg(self, sign, level, msg):
        """
            format msg
            year-month-day hour-minute-second  weekday service [level] - message
        """
        return self.format_util.format(self.__tr.resolver(), sign, level, msg)

    def output_msg(self, *args, **kwargs):
        msg = "message"
        em = kwargs.get(msg, False)
        if em:
            self.__reporter.output(em[msg])

    def prepare_before_stop(self):
        self.__reporter.close()

    def prepare_before_start(self):
        self.signal.connect(self.output_msg, sender="anonymous")
        self.__reporter.open()


class LogAdapter(object):
    """
        log adapter mode
        mode used : adapter proxy single LogService

    """

    def __init__(self, name, enable=1):
        self.log_service = LogService()
        self.name = name
        self.enable = enable
        self.mode = 1
        self.tag = None

    def get_tag(self):
        if self.tag is None:
            return self.name
        else:
            return self.tag

    def critical(self, msg):
        if self.enable:
            self.log_service.log_send(self.name, LOG_LV.CRITICAL, msg, self.mode)

    def debug(self, msg):
        if self.enable:
            self.log_service.log_send(self.name, LOG_LV.DEBUG, msg, self.mode)

    def info(self, msg):
        if self.enable:
            self.log_service.log_send(self.name, LOG_LV.INFO, msg, self.mode)

    def warning(self, msg):
        if self.enable:
            self.log_service.log_send(self.name, LOG_LV.WARNING, msg, self.mode)

    def error(self, msg):
        if self.enable:
            self.log_service.log_send(self.name, LOG_LV.ERROR, msg, self.mode)


if __name__ == '__main__':
    net_ser = LogAdapter("Net")
    ls = LogService()
    ls.start()
    net_ser.debug("111")