Source code for smif.cli.log

import datetime
import logging
import logging.config
from collections import OrderedDict


# Make profiling methods available through the logger


def profiling_start(self, operation, key):
    time_placeholder = datetime.time(0, 0)
    level = sum(
        log[1]["stop"] == time_placeholder for log in logging.Logger._profile.items()
    )
    logging.Logger._profile[(operation, key)] = {
        "start": datetime.datetime.now(),
        "stop": time_placeholder,
        "level": level,
    }





def profiling_stop(self, operation, key):
    logging.Logger._profile[(operation, key)]["stop"] = datetime.datetime.now()





def summary(self, *args, **kws):

    if self.isEnabledFor(logging.INFO):
        summary = []
        columns = [30, 80, 12]
        column = (
            "{:"
            + str(columns[0])
            + "s}"
            + "{:"
            + str(columns[1])
            + "s}"
            + "{:"
            + str(columns[2])
            + "s}"
        )

        level_width = (
            max([profile["level"] for profile in logging.Logger._profile.values()]) * 2
        )
        total_width = sum(columns) + level_width

        # header
        summary.append(
            ("{:*^" + str(total_width) + "s}").format(" Modelrun time profile ")
        )
        summary.append(column.format("Function", "Operation", "Duration [hh:mm:ss]"))
        summary.append("*" * total_width)

        # body
        for profile in logging.Logger._profile.keys():

            # calculate time diff
            profile_data = logging.Logger._profile[profile]
            diff = profile_data["stop"] - profile_data["start"]
            s = diff.total_seconds()
            time_spent = "{:02d}:{:02d}:{:05.2f}".format(
                int(s // 3600), int(s % 3600 // 60), s % 60
            )

            # trunctuate long lines
            if len(profile[0]) > columns[0] - 2:
                func = profile[0][: columns[0] - 3] + ".."
            else:
                func = profile[0]
            if len(profile[1]) > columns[1] - 2:
                op = profile[1][: columns[1] - 3] + ".."
            else:
                op = profile[1]

            summary.append(
                profile_data["level"] * "| " + column.format(func, op, time_spent)
            )

        # footer
        summary.append("*" * total_width)

        for entry in summary:
            self._log(logging.INFO, entry, args)





def setup_logging(loglevel):
    config = {
        "version": 1,
        "formatters": {
            "default": {
                "format": "%(asctime)s %(name)-12s: %(levelname)-8s %(message)s"
            },
            "message": {"format": "\033[1;34m%(levelname)-8s\033[0m %(message)s"},
        },
        "handlers": {
            "file": {
                "class": "logging.FileHandler",
                "level": "DEBUG",
                "formatter": "default",
                "filename": "smif.log",
                "mode": "a",
                "encoding": "utf-8",
            },
            "stream": {
                "class": "logging.StreamHandler",
                "formatter": "message",
                "level": "DEBUG",
            },
        },
        "root": {"handlers": ["file", "stream"], "level": "DEBUG"},
        # disable_existing_loggers defaults to True, which causes problems with class/module
        # -specific loggers, especially in unit tests when this method might be called multiple
        # times
        "disable_existing_loggers": False,
    }

    if loglevel is None:
        config["root"]["level"] = logging.WARNING
    elif loglevel == 1:
        config["root"]["level"] = logging.INFO
    else:
        config["root"]["level"] = logging.DEBUG

    logging.config.dictConfig(config)
    logging.debug("Debug logging enabled.")



logging.Logger.profiling_start = profiling_start
logging.Logger.profiling_stop = profiling_stop
logging.Logger.summary = summary
logging.Logger._profile = OrderedDict()