room_control/console.py

import logging
import logging.config
import re
from abc import ABC
from dataclasses import asdict

from rich.console import Console
from rich.highlighter import RegexHighlighter
from rich.theme import Theme

console = Console(
    width=100,
    theme=Theme(
        {
            'log.time': 'none',
            # 'logging.level.info': 'none',
            'room': 'italic bright_cyan',
            'component': 'dark_violet',
            'friendly_name': 'yellow',
            'light': 'light_slate_blue',
            'sensor': 'green',
            'time': 'yellow',
            'z2m': 'bright_black',
            'topic': 'chartreuse2',
            'true': 'green',
            'false': 'red',
        }
    ),
    log_time_format='%Y-%m-%d %I:%M:%S %p',
    # highlighter=RCHighlighter(),
)


class RCHighlighter(RegexHighlighter):
    highlights = [
        r'(?P<light>(light|switch)\.\w+)',
        r'(?P<time>\d+:\d+:\d+)',
        r'(?P<z2m>zigbee2mqtt/)',
        r'(?P<sensor>binary_sensor\.\w+)',
        # r"'state': '(?P<on>on)|(?P<off>off)'"
        r'(?P<true>True)|(?P<false>False)',
    ]


RICH_HANDLER_CFG = {
    '()': 'rich.logging.RichHandler',
    'markup': True,
    'show_path': False,
    # 'show_time': False,
    'omit_repeated_times': False,
    'console': console,
    'highlighter': RCHighlighter(),
}


class ContextSettingFilter(logging.Filter, ABC):
    def filter(self, record: logging.LogRecord) -> logging.LogRecord:
        for name, val in asdict(self).items():
            if val is not None:
                setattr(record, name, val)
        return record


class RoomFilter(logging.Filter):
    """Used to filter out messages that have a component field because they will have already been printed by their respective logger."""

    def filter(self, record: logging.LogRecord) -> bool:
        return getattr(record, 'component', None) is None


# @dataclass
# class RoomControllerFilter(ContextSettingFilter):
#     room: str
#     component: Optional[str] = None


# class RoomControllerFormatter(logging.Formatter):
#     def format(self, record: logging.LogRecord):
#         return super().format(record)


class UnMarkupFilter(logging.Filter):
    md_regex = re.compile(r'(?P<open>\[.*?\])(?P<text>.*?)(?P<close>\[\/\])')

    def filter(self, record: logging.LogRecord) -> logging.LogRecord:
        record.msg = self.md_regex.sub(r'\g<text>', record.msg)
        return record


def room_logging_config(name: str):
    return {
        'version': 1,
        'disable_existing_loggers': False,
        'filters': {
            'room': {'()': 'console.RoomFilter'},
            'unmarkup': {'()': 'console.UnMarkupFilter'},
        },
        'formatters': {
            'rich_room': {
                'style': '{',
                'format': '[room]{room}[/] {message}',
                'datefmt': '%H:%M:%S.%f',
            },
            'file': {
                'style': '{',
                'format': '{asctime}.{msecs:03.0f} {levelname:8} {name}: {message}',
                'datefmt': '%Y-%m-%d %H:%M:%S',
            },
        },
        'handlers': {
            'rich_room': {'formatter': 'rich_room', 'filters': ['room'], **RICH_HANDLER_CFG},
            'file': {
                'filters': ['unmarkup'],
                'formatter': 'file',
                'class': 'logging.handlers.RotatingFileHandler',
                # 'class': 'logging.FileHandler',
                'filename': f'/logs/{name}.log',
                'mode': 'w',
                'maxBytes': 1000000,
                'backupCount': 3,
            },
        },
        'loggers': {
            f'AppDaemon.{name}': {
                'level': 'INFO',
                'propagate': False,
                'handlers': ['rich_room', 'file'],
            },
        },
    }


def component_logging_config(parent_room: str, component: str):
    logger_name = f'AppDaemon.{parent_room}.{component}'

    LOG_CFG = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'rich_component': {
                'style': '{',
                'format': '[room]{room}[/] [component]{component}[/] {message}',
                'datefmt': '%H:%M:%S.%f',
            },
        },
        'handlers': {
            'rich_component': {
                'formatter': 'rich_component',
                **RICH_HANDLER_CFG,
            },
        },
        'loggers': {
            logger_name: {
                # 'level': 'INFO',
                'propagate': True,
                'handlers': ['rich_component'],
            }
        },
    }
    return LOG_CFG


def setup_component_logging(self) -> logging.Logger:
    """Creates a logger for a subcomponent with a RichHandler"""
    component = type(self).__name__
    parent = self.args['app']
    cfg_dict = component_logging_config(parent_room=parent, component=component)
    logger_name = next(iter(cfg_dict['loggers']))

    try:
        logging.config.dictConfig(cfg_dict)
    except Exception:
        console.print_exception()
    else:
        logger = logging.getLogger(logger_name)
        logger = logging.LoggerAdapter(logger, {'room': parent, 'component': component})
        return logger