room_control/console.py

import logging
import logging.config
import re
from abc import ABC
from dataclasses import asdict, dataclass
from typing import Optional

from appdaemon.logging import AppNameFormatter
from rich.console import Console
from rich.highlighter import RegexHighlighter
from rich.logging import RichHandler
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