coloredlogcat.py

#!/usr/bin/python3
'''
    Copyright 2015, Jag Saund
    Copyright 2023, TBM13

    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.
'''

import argparse
import os
import re
import sys
import textwrap
import threading
import time

TERMINAL_WIDTH = os.get_terminal_size().columns

class LogLine:
    def __init__(self, timestamp: str = '', pid: str = '',
                 priority: str = '', process_tag: str = '',
                 message: str = '', hide_timestamp: bool = False,
                 hide_pid: bool = False, hide_priority: bool = False,
                 hide_process_tag: bool = False,
                 is_complete: bool = True) -> None:
        self.timestamp = timestamp
        self.pid = pid
        self.priority = priority
        self.process_tag = process_tag
        self.message = message
        self.hide_timestamp = hide_timestamp
        self.hide_pid = hide_pid
        self.hide_priority = hide_priority
        self.hide_process_tag = hide_process_tag
        self.is_complete = is_complete
class RawLogLine(LogLine):
    def __init__(self, message: str = '') -> None:
        self.message = message
        self.is_complete = True

class LogLineFilter:
    def __init__(self, filter_on_match: bool):
        self.pids: list[str]|None = None
        self.priorities: str|None = None
        self.tag_pattern: re.Pattern|None = None
        self.message_pattern: re.Pattern|None = None
        self.filter_on_match = filter_on_match

    def is_line_filtered(self, line: LogLine) -> bool:
        if self.pids is not None and not line.pid in self.pids:
            return not self.filter_on_match
        if self.priorities is not None and not line.priority in self.priorities:
            return not self.filter_on_match
        if (self.tag_pattern is not None and
            self.tag_pattern.match(line.process_tag) is None):
            return not self.filter_on_match
        if (self.message_pattern is not None and
            self.message_pattern.match(line.message) is None):
            return not self.filter_on_match
        
        return self.filter_on_match

class LogFormat:
    PATTERN = re.compile('')

    WIDTH_TIMESTAMP = 0
    WIDTH_PID = 0
    WIDTH_PRIORITY = 0
    # Process tag length will be dynamic, but this is
    # the minimum it will take up including chars like whitespaces
    WIDTH_PROCESS_TAG = 0

    STYLE_TIMESTAMP = '\033[0;38;5;134m'
    STYLE_PID = '\033[0;38;5;36;48;5;236m'
    STYLE_PRIORITY = {
        'V': '\033[0;38;5;255;48;5;36m',
        'I': '\033[0;38;5;255;48;5;40m',
        'D': '\033[0;38;5;255;48;5;33m',
        'W': '\033[0;38;5;255;48;5;208m',
        'E': '\033[0;38;5;255;48;5;124m',
        'F': '\033[0;38;5;255;48;5;196m',
        'S': '\033[0;38;5;255;48;5;248m',
    }
    STYLE_PROCESS_TAG = '\033[0;38;5;255;48;5;236m'
    STYLE_MESSAGE = {
        'V':  '\033[0;38;5;36m',
        'I': '\033[0;38;5;40m',
        'D': '\033[0;38;5;33m',
        'W': '\033[0;38;5;208m',
        'E': '\033[0;38;5;124m',
        'F': '\033[0;38;5;255;48;5;196m',
        'S': '\033[0;38;5;248m',
    }

    @staticmethod
    def from_line(line: str):
        """Detects the log format used in `line` and returns the respective `LogFormat`.
        
        If no format is detected, returns `None`.
        """
        for format in LogFormat.__subclasses__():
            if format.PATTERN.match(line) is not None: 
                return format()

        return None

    def __init__(self) -> None:
        # Calculate width of the header + whitespaces
        self.WIDTH_HEADER = (
            self.WIDTH_TIMESTAMP + min(self.WIDTH_TIMESTAMP, 1)
            + self.WIDTH_PID + min(self.WIDTH_PID, 1)
            + self.WIDTH_PRIORITY + min(self.WIDTH_PRIORITY, 1)
            + self.WIDTH_PROCESS_TAG
        )

    def construct_line(self, match: re.Match) -> LogLine:
        raise NotImplementedError()
    
    def _format_message(self, line: LogLine) -> str:
        indent = self.WIDTH_HEADER + len(line.process_tag)
        return textwrap.fill(
            line.message, width=TERMINAL_WIDTH - indent
        ).replace('\n', '\n' + ' ' * indent)

    def format_line(self, line: LogLine) -> str:
        out = ''

        # Timestamp
        if self.WIDTH_TIMESTAMP > 0:
            if line.hide_timestamp:
                out += ' ' * (self.WIDTH_TIMESTAMP + 1)
            else:
                out += (
                    self.STYLE_TIMESTAMP + 
                    line.timestamp +
                    '\033[0m '
                )
        # PID
        if self.WIDTH_PID > 0:
            if line.hide_pid:
                out += ' ' * (self.WIDTH_PID + 1)
            else:
                out += (
                    self.STYLE_PID +
                    line.pid.center(self.WIDTH_PID) +
                    '\033[0m '
                )
        # Priority
        if self.WIDTH_PRIORITY > 0:
            if line.hide_priority:
                out += ' ' * (self.WIDTH_PRIORITY + 1)
            else:
                out += (
                    self.STYLE_PRIORITY[line.priority] +
                    line.priority.center(self.WIDTH_PRIORITY) +
                    '\033[0m '
                )
        # Process tag
        if len(line.process_tag) > 0:
            if line.hide_process_tag:
                out += (
                    ' ' * (self.WIDTH_PROCESS_TAG + len(line.process_tag))
                )
            else:
                out += (
                    self.STYLE_PROCESS_TAG +
                    line.process_tag +
                    ':\033[0m '
                )

        msg = self._format_message(line)
        # Message
        out += (
            self.STYLE_MESSAGE[line.priority] +
            msg +
            '\033[0m'
        )

        return out

class FormatBrief(LogFormat):
    PATTERN = re.compile(
        r'^([VDIWEFS])\/(.*?)\( *(\d+)\): (.*)$'
    )
    WIDTH_TIMESTAMP = 0
    WIDTH_PID = 5
    WIDTH_PRIORITY = 3
    WIDTH_PROCESS_TAG = 2
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            priority=match.group(1),
            process_tag=match.group(2).rstrip(' '),
            pid=match.group(3),
            message=match.group(4)
        )

class FormatLong(LogFormat):
    PATTERN = re.compile(
        r'^\[ \d\d-\d\d (\d\d:\d\d:\d\d\.\d\d\d) +?(\d+?): ?.* ([VDIWEFS])\/(.*?) ]$'
    )
    WIDTH_TIMESTAMP = 12
    WIDTH_PID = 5
    WIDTH_PRIORITY = 3
    WIDTH_PROCESS_TAG = 2
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            timestamp=match.group(1),
            pid=match.group(2),
            priority=match.group(3),
            process_tag=match.group(4).rstrip(' '),
            is_complete=False
        )

    def _format_message(self, line: LogLine) -> str:
        line.message = line.message.rstrip('\n') + '\n'

        if line.message.count('\n') == 1:
            # If header and message fit in one line, return it as-is
            if (self.WIDTH_HEADER + len(line.process_tag) +
                len(line.message) <= TERMINAL_WIDTH):
                return line.message.strip('\n')

        return '\n' + line.message.rstrip('\n')

class FormatProcess(LogFormat):
    PATTERN = re.compile(
        r'^([VDIWEFS])\( *(\d+)\) (.*)$'
    )
    WIDTH_PRIORITY = 3
    WIDTH_PID=5
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            priority=match.group(1),
            pid=match.group(2),
            message=match.group(3)
        )

class FormatTag(LogFormat):
    PATTERN = re.compile(
        r'^([VDIWEFS])\/ *(.*?): (.*)$'
    )
    WIDTH_TIMESTAMP = 0
    WIDTH_PID = 0
    WIDTH_PRIORITY = 3
    WIDTH_PROCESS_TAG = 2
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            priority=match.group(1),
            process_tag=match.group(2).rstrip(' '),
            message=match.group(3)
        )

class FormatThread(LogFormat):
    PATTERN = re.compile(
        r'^([VDIWEFS])\( *(\d+): *\d+\) (.*)$'
    )
    WIDTH_TIMESTAMP = 0
    WIDTH_PID = 5
    WIDTH_PRIORITY = 3
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            priority=match.group(1),
            pid=match.group(2),
            message=match.group(3)
        )

class FormatThreadTime(LogFormat):
    PATTERN = re.compile(
        r'^\d\d-\d\d (\d\d:\d\d:\d\d\.\d\d\d) +?(\d+) +?\d+ ([VDIWEFS]) +(.*?): (.*)$'
    )
    WIDTH_TIMESTAMP = 12
    WIDTH_PID = 5
    WIDTH_PRIORITY = 3
    WIDTH_PROCESS_TAG = 2
    
    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            timestamp=match.group(1),
            pid=match.group(2),
            priority=match.group(3),
            process_tag=match.group(4).rstrip(' '),
            message=match.group(5)
        )

class FormatTime(LogFormat):
    PATTERN = re.compile(
        r'^\d\d-\d\d (\d\d:\d\d:\d\d\.\d\d\d) ([VDIWEFS])\/ *(.*?)\( *(\d+)\): (.*)$'
    )
    WIDTH_TIMESTAMP = 12
    WIDTH_PID = 5
    WIDTH_PRIORITY = 3
    WIDTH_PROCESS_TAG = 2

    def construct_line(self, match: re.Match) -> LogLine:
        return LogLine(
            timestamp=match.group(1),
            priority=match.group(2),
            process_tag=match.group(3).rstrip(' '),
            pid=match.group(4),
            message=match.group(5)
        )

class PipeReader(threading.Thread):
    def __init__(self, pipe, out_file):
        self.pipe = pipe
        self.out_file = out_file
        self.log_format: LogFormat|None = None
        self.lines: list[LogLine] = [
            RawLogLine('ColoredLogcat: Start Read')
        ]

        self.abort = False
        self.doing_work = False
        super().__init__()

    def run(self):
        try:
            self.doing_work = True
            while not self.abort:
                self._read_line()

        finally:
            self.out_file.close()
            self.doing_work = False

    def _read_line(self):
        try:
            line = self.pipe.readline()
        except UnicodeError:
            return
        
        if not line:
            self.abort = True
            return
        self.out_file.write(line)

        if self.log_format is None:
            self.log_format = LogFormat.from_line(line)
        if self.log_format is None:
            self.lines.append(RawLogLine(line))
            return
        
        match = self.log_format.PATTERN.match(line)
        if match is None:
            # Long format's messages are split into multiple lines
            if isinstance(self.log_format, FormatLong):
                # Assume this line is part of the previous LogLine's message
                # This is usually (but not always) the case (see https://github.com/TBM13/ColoredLogcat/issues/1)
                self.lines[-1].message += line
                return

            # Not a LogLine (e. g. "--------- beginning of system")
            self.lines.append(RawLogLine(line))
            return
        
        # Since this line is a new LogLine, mark previous LogLine as completed
        self.lines[-1].is_complete = True

        line = self.log_format.construct_line(match)
        self.lines.append(line)

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '-f', '--file',
        help='Write raw and unfiltered log to file.'
    )
    parser.add_argument(
        '-F', '--filter-file',
        help='Load filters from file. ' \
             'See filter_file_example for more info.'
    )
    parser.add_argument(
        '-p', '--priority',
        help='Filter by priority. Example: -p WEF'
    )
    parser.add_argument(
        '-P', '--pid',
        help='Filter by PID(s). Example: -P 4713,6452'
    )
    parser.add_argument(
        '-t', '--tag',
        help='Filter by process tag (case insensitive). Regex supported. ' \
             'Example: -t "^(system|androidruntime)$"'
    )
    parser.add_argument(
        '-T', '--tag-s',
        help='Filter by process tag (case sensitive). Regex supported. ' \
             'Example: -t "^(System|AndroidRuntime)$"'
    )
    parser.add_argument(
        '-m', '--message',
        help='Filter by message (case insensitive). Regex supported. ' \
             'Example: -m "sending signal \d to.*"'
    )
    parser.add_argument(
        '-M', '--message-s',
        help='Filter by message (case sensitive). Regex supported. ' \
             'Example: -m "Sending signal \d to.*"'
    )

    adb_options = parser.add_argument_group('ADB options')
    adb_options.add_argument(
        '-v', '--format',
        default='time',
        help='Changes the format of ADB\'s output. Supported formats: ' \
             'brief, long, process, tag, thread, threadtime, time. ' \
             'Default and recommended format is "time".'
    )

    return parser.parse_args()

def main():
    args = parse_args()

    # Prepare out file
    if args.file is not None:
        if os.path.exists(args.file):
            answer = input(f'File "{args.file}" already exists. Overwrite? [y/N] ')
            if answer.lower() != 'y':
                return
            
            os.remove(args.file)
    out_file = open(args.file or os.devnull, 'w', errors='replace')

    filters = []
    # Build filter(s) from args
    if args.message is not None and args.message_s is not None:
        raise ValueError('Can\'t use both -m and -M')
    if args.tag is not None and args.tag_s is not None:
        raise ValueError('Can\'t use both -t and -T')

    filter = LogLineFilter(filter_on_match=False)
    if args.pid is not None:
        filter.pids = args.pid.split(',')
    if args.priority is not None:
        filter.priorities = args.priority.upper()

    tag_pattern = args.tag or args.tag_s
    tag_flags = re.IGNORECASE if args.tag is not None else 0
    if tag_pattern is not None:
        if not tag_pattern.startswith('^'):
            tag_pattern = '.*' + tag_pattern
        if not tag_pattern.endswith('$'):
            tag_pattern += '.*'

        filter.tag_pattern = re.compile(tag_pattern, tag_flags)

    msg_pattern = args.message or args.message_s
    msg_flags = re.IGNORECASE if args.message is not None else 0
    if msg_pattern is not None:
        if not msg_pattern.startswith('^'):
            msg_pattern = '.*' + msg_pattern
        if not msg_pattern.endswith('$'):
            msg_pattern += '.*'

        filter.message_pattern = re.compile(msg_pattern, msg_flags)
    
    filters.append(filter)

    # Build filters from filter file
    ff_pattern = re.compile(r'^(!)?([\d,]+|-)? ([VDIWEFS]+?|-) (.+?) (.+?)$',
                            re.IGNORECASE)
    if args.filter_file is not None:
        with open(args.filter_file, 'r') as f:
            for l in f.readlines():
                l = l.strip('\n')
                if len(l) == 0 or l.startswith('#'):
                    continue

                m = ff_pattern.match(l)
                if m is None:
                    raise ValueError(f'Invalid filter: "{l}"')
                
                filter = LogLineFilter(filter_on_match=m.group(1) is None)
                if m.group(2) != '-':
                    filter.pids = m.group(2).split(',')
                if m.group(3) != '-':
                    filter.priorities = m.group(3).upper()
                if m.group(4) != '-':
                    filter.tag_pattern = re.compile(m.group(4), re.IGNORECASE)
                if m.group(5) != '-':
                    filter.message_pattern = re.compile(m.group(5), re.IGNORECASE)
                
                filters.append(filter)

    # Prepare pipe
    if not os.isatty(sys.stdin.fileno()):
        pipe = sys.stdin
    else:
        pipe = os.popen(f'adb logcat -v {args.format.lower()}')

    # Start reading
    reader = PipeReader(pipe, out_file)
    reader.start()
    i = 0
    while reader.doing_work:
        # Print all available lines
        while i < len(reader.lines):
            if not reader.lines[i].is_complete:
                break

            print_line(
                line=reader.lines[i],
                log_format=reader.log_format,
                filters=filters
            )
            i += 1

        # While the reader is waiting for new lines, handle the command bar
        # command_bar()
        time.sleep(0.001)

last_printed_line: LogLine|None = None
def print_line(line: LogLine, log_format: LogFormat|None,
               filters: list[LogLineFilter]):
    if isinstance(line, RawLogLine):
        # This happens when messages end with more than one newlines
        if line.message == '\n':
            return

        print(line.message)
        return

    modify_line(line)
    for f in filters:
        if f.is_line_filtered(line):
            return
    
    global last_printed_line
    # Hide repeated information between this line and previous one
    if last_printed_line is not None:
        if line.timestamp == last_printed_line.timestamp:
            line.hide_timestamp = True

        if (line.pid == last_printed_line.pid and
            line.priority == last_printed_line.priority and
            line.process_tag == last_printed_line.process_tag):
            line.hide_pid = True
            line.hide_priority = True
            line.hide_process_tag = True

    print(log_format.format_line(line))
    last_printed_line=line

def modify_line(line: LogLine):
    if line.pid == '0':
        handle_kernel_log(line)

def handle_kernel_log(line: LogLine):
    if len(line.process_tag) != 0:
        return

    # On my device, some kernel logs contain the tag in the message, e.g:
    # I/        (    0): [3:android.hardwar: 4392] sec_bat_get_property cable type = 4 sleep_mode = 0
    if not line.message.startswith('['):
        return
    tag_end = line.message.find(']') + 1
    if tag_end == 0:
        return

    line.process_tag = line.message[:tag_end]
    line.message = line.message[tag_end + 1:]

def command_bar():
    # TODO
    pass

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass