Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
Size: Mime:
"""A progress reporter inspired from the logging modules"""
from __future__ import absolute_import, unicode_literals

import os
import time
from contextlib import contextmanager
from datetime import datetime

import py


class Verbosity(object):
    DEBUG = 2
    INFO = 1
    DEFAULT = 0
    QUIET = -1
    EXTRA_QUIET = -2


REPORTER_TIMESTAMP_ON_ENV = str("TOX_REPORTER_TIMESTAMP")
REPORTER_TIMESTAMP_ON = os.environ.get(REPORTER_TIMESTAMP_ON_ENV, False) == "1"
START = datetime.now()


class Reporter(object):
    def __init__(self, verbose_level=None, quiet_level=None):
        kwargs = {}
        if verbose_level is not None:
            kwargs["verbose_level"] = verbose_level
        if quiet_level is not None:
            kwargs["quiet_level"] = quiet_level
        self._reset(**kwargs)

    def _reset(self, verbose_level=0, quiet_level=0):
        self.verbose_level = verbose_level
        self.quiet_level = quiet_level
        self.reported_lines = []
        self.tw = py.io.TerminalWriter()

    @property
    def verbosity(self):
        return self.verbose_level - self.quiet_level

    def log_popen(self, cwd, outpath, cmd_args_shell, pid):
        """log information about the action.popen() created process."""
        msg = "[{}] {}$ {}".format(pid, cwd, cmd_args_shell)
        if outpath:
            if outpath.common(cwd) is not None:
                outpath = cwd.bestrelpath(outpath)
            msg = "{} >{}".format(msg, outpath)
        self.verbosity1(msg, of="logpopen")

    @property
    def messages(self):
        return [i for _, i in self.reported_lines]

    @contextmanager
    def timed_operation(self, name, msg):
        self.verbosity2("{} start: {}".format(name, msg), bold=True)
        start = time.time()
        yield
        duration = time.time() - start
        self.verbosity2(
            "{} finish: {} after {:.2f} seconds".format(name, msg, duration),
            bold=True,
        )

    def separator(self, of, msg, level):
        if self.verbosity >= level:
            self.reported_lines.append(("separator", "- summary -"))
            self.tw.sep(of, msg)

    def logline_if(self, level, of, msg, key=None, **kwargs):
        if self.verbosity >= level:
            message = str(msg) if key is None else "{}{}".format(key, msg)
            self.logline(of, message, **kwargs)

    def logline(self, of, msg, **opts):
        self.reported_lines.append((of, msg))
        timestamp = ""
        if REPORTER_TIMESTAMP_ON:
            timestamp = "{} ".format(datetime.now() - START)
        line_msg = "{}{}\n".format(timestamp, msg)
        self.tw.write(line_msg, **opts)

    def keyvalue(self, name, value):
        if name.endswith(":"):
            name += " "
        self.tw.write(name, bold=True)
        self.tw.write(value)
        self.tw.line()

    def line(self, msg, **opts):
        self.logline("line", msg, **opts)

    def info(self, msg):
        self.logline_if(Verbosity.DEBUG, "info", msg)

    def using(self, msg):
        self.logline_if(Verbosity.INFO, "using", msg, "using ", bold=True)

    def good(self, msg):
        self.logline_if(Verbosity.QUIET, "good", msg, green=True)

    def warning(self, msg):
        self.logline_if(Verbosity.QUIET, "warning", msg, "WARNING: ", red=True)

    def error(self, msg):
        self.logline_if(Verbosity.QUIET, "error", msg, "ERROR: ", red=True)

    def skip(self, msg):
        self.logline_if(Verbosity.QUIET, "skip", msg, "SKIPPED: ", yellow=True)

    def verbosity0(self, msg, **opts):
        self.logline_if(Verbosity.DEFAULT, "verbosity0", msg, **opts)

    def verbosity1(self, msg, of="verbosity1", **opts):
        self.logline_if(Verbosity.INFO, of, msg, **opts)

    def verbosity2(self, msg, **opts):
        self.logline_if(Verbosity.DEBUG, "verbosity2", msg, **opts)

    def quiet(self, msg):
        self.logline_if(Verbosity.QUIET, "quiet", msg)


_INSTANCE = Reporter()


def update_default_reporter(quiet_level, verbose_level):
    _INSTANCE.quiet_level = quiet_level
    _INSTANCE.verbose_level = verbose_level


def has_level(of):
    return _INSTANCE.verbosity > of


def verbosity():
    return _INSTANCE.verbosity


verbosity0 = _INSTANCE.verbosity0
verbosity1 = _INSTANCE.verbosity1
verbosity2 = _INSTANCE.verbosity2
error = _INSTANCE.error
warning = _INSTANCE.warning
good = _INSTANCE.good
using = _INSTANCE.using
skip = _INSTANCE.skip
info = _INSTANCE.info
line = _INSTANCE.line
separator = _INSTANCE.separator
keyvalue = _INSTANCE.keyvalue
quiet = _INSTANCE.quiet
timed_operation = _INSTANCE.timed_operation
log_popen = _INSTANCE.log_popen