acts/framework/acts/logger.py

#!/usr/bin/env python3
#
#   Copyright 2016 - The Android Open Source Project
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

from __future__ import print_function

import datetime
import logging
import os
import re

from acts import tracelogger
from acts.libs.logging import log_stream
from acts.libs.logging.log_stream import LogStyles
from acts.utils import create_dir


log_line_format = "%(asctime)s.%(msecs).03d %(levelname)s %(message)s"
# The micro seconds are added by the format string above,
# so the time format does not include ms.
log_line_time_format = "%Y-%m-%d %H:%M:%S"
log_line_timestamp_len = 23

logline_timestamp_re = re.compile("\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d.\d\d\d")


def _parse_logline_timestamp(t):
    """Parses a logline timestamp into a tuple.

    Args:
        t: Timestamp in logline format.

    Returns:
        An iterable of date and time elements in the order of month, day, hour,
        minute, second, microsecond.
    """
    date, time = t.split(' ')
    year, month, day = date.split('-')
    h, m, s = time.split(':')
    s, ms = s.split('.')
    return year, month, day, h, m, s, ms


def is_valid_logline_timestamp(timestamp):
    if len(timestamp) == log_line_timestamp_len:
        if logline_timestamp_re.match(timestamp):
            return True
    return False


def logline_timestamp_comparator(t1, t2):
    """Comparator for timestamps in logline format.

    Args:
        t1: Timestamp in logline format.
        t2: Timestamp in logline format.

    Returns:
        -1 if t1 < t2; 1 if t1 > t2; 0 if t1 == t2.
    """
    dt1 = _parse_logline_timestamp(t1)
    dt2 = _parse_logline_timestamp(t2)
    for u1, u2 in zip(dt1, dt2):
        if u1 < u2:
            return -1
        elif u1 > u2:
            return 1
    return 0


def _get_timestamp(time_format, delta=None):
    t = datetime.datetime.now()
    if delta:
        t = t + datetime.timedelta(seconds=delta)
    return t.strftime(time_format)[:-3]


def epoch_to_log_line_timestamp(epoch_time):
    """Converts an epoch timestamp in ms to log line timestamp format, which
    is readable for humans.

    Args:
        epoch_time: integer, an epoch timestamp in ms.

    Returns:
        A string that is the corresponding timestamp in log line timestamp
        format.
    """
    s, ms = divmod(epoch_time, 1000)
    d = datetime.datetime.fromtimestamp(s)
    return d.strftime("%Y-%m-%d %H:%M:%S.") + str(ms)


def get_log_line_timestamp(delta=None):
    """Returns a timestamp in the format used by log lines.

    Default is current time. If a delta is set, the return value will be
    the current time offset by delta seconds.

    Args:
        delta: Number of seconds to offset from current time; can be negative.

    Returns:
        A timestamp in log line format with an offset.
    """
    return _get_timestamp("%Y-%m-%d %H:%M:%S.%f", delta)


def get_log_file_timestamp(delta=None):
    """Returns a timestamp in the format used for log file names.

    Default is current time. If a delta is set, the return value will be
    the current time offset by delta seconds.

    Args:
        delta: Number of seconds to offset from current time; can be negative.

    Returns:
        A timestamp in log file name format with an offset.
    """
    return _get_timestamp("%Y-%m-%d_%H-%M-%S-%f", delta)


def _setup_test_logger(log_path, prefix=None):
    """Customizes the root logger for a test run.

    The logger object has a stream handler and a file handler. The stream
    handler logs INFO level to the terminal, the file handler logs DEBUG
    level to files.

    Args:
        log_path: Location of the log file.
        prefix: A prefix for each log line in terminal.
    """
    logging.log_path = log_path
    log_styles = [LogStyles.LOG_INFO + LogStyles.TO_STDOUT,
                  LogStyles.DEFAULT_LEVELS + LogStyles.MONOLITH_LOG]
    terminal_format = log_line_format
    if prefix:
        terminal_format = "[{}] {}".format(prefix, log_line_format)
    stream_formatter = logging.Formatter(terminal_format, log_line_time_format)
    file_formatter = logging.Formatter(log_line_format, log_line_time_format)
    log = log_stream.create_logger('test_run', '', log_styles=log_styles,
                                   stream_format=stream_formatter,
                                   file_format=file_formatter)
    log.setLevel(logging.DEBUG)


def kill_test_logger(logger):
    """Cleans up a test logger object by removing all of its handlers.

    Args:
        logger: The logging object to clean up.
    """
    for h in list(logger.handlers):
        logger.removeHandler(h)
        if isinstance(h, logging.FileHandler):
            h.close()


def create_latest_log_alias(actual_path):
    """Creates a symlink to the latest test run logs.

    Args:
        actual_path: The source directory where the latest test run's logs are.
    """
    link_path = os.path.join(os.path.dirname(actual_path), "latest")
    if os.path.islink(link_path):
        os.remove(link_path)
    os.symlink(actual_path, link_path)


def setup_test_logger(log_path, prefix=None):
    """Customizes the root logger for a test run.

    Args:
        log_path: Location of the report file.
        prefix: A prefix for each log line in terminal.
        filename: Name of the files. The default is the time the objects
            are requested.
    """
    create_dir(log_path)
    _setup_test_logger(log_path, prefix)
    create_latest_log_alias(log_path)


def normalize_log_line_timestamp(log_line_timestamp):
    """Replace special characters in log line timestamp with normal characters.

    Args:
        log_line_timestamp: A string in the log line timestamp format. Obtained
            with get_log_line_timestamp.

    Returns:
        A string representing the same time as input timestamp, but without
        special characters.
    """
    norm_tp = log_line_timestamp.replace(' ', '_')
    norm_tp = norm_tp.replace(':', '-')
    return norm_tp


class LoggerAdapter(logging.LoggerAdapter):
    """A LoggerAdapter class that takes in a lambda for transforming logs."""

    def __init__(self, logging_lambda):
        self.logging_lambda = logging_lambda
        super(LoggerAdapter, self).__init__(logging.getLogger(), {})

    def process(self, msg, kwargs):
        return self.logging_lambda(msg), kwargs


def create_logger(logging_lambda=lambda message: message):
    """Returns a logger with logging defined by a given lambda.

    Args:
        logging_lambda: A lambda of the form:
            >>> lambda log_message: return 'string'
    """
    return tracelogger.TraceLogger(LoggerAdapter(logging_lambda))


def create_tagged_trace_logger(tag=''):
    """Returns a logger that logs each line with the given prefix.

    Args:
        tag: The tag of the log line, E.g. if tag == tag123, the output
            line would be:

            <TESTBED> <TIME> <LOG_LEVEL> [tag123] logged message
    """
    def logging_lambda(msg):
        return '[%s] %s' % (tag, msg)
    return create_logger(logging_lambda)