OpenGoPro/demos/python/sdk_wireless_camera_control/open_gopro/util/logger.py

# logger.py/Open GoPro, Version 2.0 (C) Copyright 2021 GoPro, Inc. (http://gopro.com/OpenGoPro).
# This copyright was auto-generated on Thu Aug 24 17:08:14 UTC 2023

"""Logger abstraction above default python logging"""

from __future__ import annotations

import http.client as http_client
import logging
from pathlib import Path
from typing import Any, Final

from rich import traceback
from rich.logging import RichHandler


class Logger:
    """A singleton class to manage logging for the Open GoPro internal modules

    Attributes:
        ARROW_HEAD_COUNT (Final[int]): Length of leading arrow
        ARROW_TAIL_COUNT (Final[int]): Length of trailing arrow

    Args:
        logger (logging.Logger): input logger that will be modified and then returned
        output (Path | None): Path of log file for file stream handler. If not set, will not log to file.
        modules (dict[str, int] | None): Optional override of modules / levels. Will be merged into default modules.
    """

    _instances: dict[type[Logger], Logger] = {}
    ARROW_HEAD_COUNT: Final[int] = 8
    ARROW_TAIL_COUNT: Final[int] = 14

    def __new__(cls, *_: Any) -> Any:  # noqa https://github.com/PyCQA/pydocstyle/issues/515
        if cls not in cls._instances:
            c = object.__new__(cls)
            cls._instances[cls] = c
            return c
        raise RuntimeError("The logger can only be setup once and this should be done at the top level.")

    def __init__(
        self,
        logger: logging.Logger,
        output: Path | None = None,
        modules: dict[str, int] | None = None,
    ) -> None:
        self.modules: dict[str, int] = {
            "open_gopro.gopro_base": logging.DEBUG,
            "open_gopro.gopro_wired": logging.DEBUG,
            "open_gopro.gopro_wireless": logging.DEBUG,  # TRACE for concurrency debugging
            "open_gopro.api.builders": logging.DEBUG,
            "open_gopro.api.http_commands": logging.DEBUG,
            "open_gopro.api.ble_commands": logging.DEBUG,
            "open_gopro.domain.communicator_interface": logging.DEBUG,
            "open_gopro.network.ble.adapters.bleak_wrapper": logging.DEBUG,
            "open_gopro.network.ble.client": logging.DEBUG,
            "open_gopro.parsers.bytes": logging.DEBUG,
            "open_gopro.parsers.json": logging.DEBUG,
            "open_gopro.parsers.response": logging.DEBUG,
            "open_gopro.parsers.general": logging.DEBUG,
            "open_gopro.network.wifi.adapters.wireless": logging.DEBUG,
            "open_gopro.network.wifi.mdns_scanner": logging.DEBUG,
            "open_gopro.domain.observable": logging.DEBUG,  # TRACE for concurrency debugging
            "open_gopro.domain.gopro_observable": logging.DEBUG,  # TRACE for observable debugging
            "open_gopro.models.response": logging.DEBUG,
            "open_gopro.models.network_scan_response": logging.DEBUG,
            "open_gopro.features.cohn_feature": logging.DEBUG,
            "open_gopro.features.access_point_feature": logging.DEBUG,
            "open_gopro.features.streaming.stream_feature": logging.DEBUG,
            "open_gopro.features.streaming.webcam_stream": logging.DEBUG,
            "open_gopro.features.streaming.livestream": logging.DEBUG,
            "open_gopro.features.streaming.preview_stream": logging.DEBUG,
            "open_gopro.util.util": logging.DEBUG,
            "open_gopro.demos.gui.video_display": logging.DEBUG,
            "open_gopro.database.db": logging.DEBUG,
            "bleak": logging.DEBUG,
            "urllib3": logging.DEBUG,
            "http.client": logging.DEBUG,
        }

        self.logger = logger
        self.modules = modules or self.modules
        self.handlers: list[logging.Handler] = []

        # monkey-patch a `print` global into the http.client module; all calls to
        # print() in that module will then use our logger's debug method
        http_client.HTTPConnection.debuglevel = 1
        http_client.print = lambda *args: logging.getLogger("http.client").debug(" ".join(args))  # type: ignore

        self.file_handler: logging.Handler | None
        if output:
            # Logging to file with millisecond timing
            self.file_handler = logging.FileHandler(output, mode="w")
            file_formatter = logging.Formatter(
                fmt="%(threadName)13s:%(asctime)s.%(msecs)03d %(filename)-40s %(lineno)4s %(levelname)-8s | %(message)s",
                datefmt="%H:%M:%S",
            )
            self.file_handler.setFormatter(file_formatter)
            self.file_handler.setLevel(logging.TRACE)  # type: ignore
            logger.addHandler(self.file_handler)
            self.addLoggingHandler(self.file_handler)
        else:
            self.file_handler = None

        # Use Rich for colorful console logging
        self.stream_handler = RichHandler(rich_tracebacks=True, enable_link_path=True, show_time=False)
        stream_formatter = logging.Formatter("%(asctime)s.%(msecs)03d %(message)s", datefmt="%H:%M:%S")
        self.stream_handler.setFormatter(stream_formatter)
        self.stream_handler.setLevel(logging.INFO)
        logger.addHandler(self.stream_handler)
        self.addLoggingHandler(self.stream_handler)

        self.addLoggingLevel("TRACE", logging.DEBUG - 5)
        logger.setLevel(logging.TRACE)  # type: ignore # pylint: disable=no-member

        traceback.install()  # Enable exception tracebacks in rich logger

    @classmethod
    def get_instance(cls) -> Logger:
        """Get the singleton instance

        Raises:
            RuntimeError: Has not yet been instantiated

        Returns:
            Logger: singleton instance
        """
        if not (logger := cls._instances.get(Logger, None)):
            raise RuntimeError("Logging must first be setup")
        return logger

    def addLoggingHandler(self, handler: logging.Handler) -> None:
        """Add a handler for all of the internal GoPro modules

        Args:
            handler (logging.Handler): handler to add
        """
        self.logger.addHandler(handler)
        self.handlers.append(handler)

        # Enable / disable logging in modules
        for module, level in self.modules.items():
            l = logging.getLogger(module)
            l.setLevel(level)
            l.addHandler(handler)

    # From https://stackoverflow.com/questions/2183233/how-to-add-a-custom-loglevel-to-pythons-logging-facility/35804945#35804945
    @staticmethod
    def addLoggingLevel(levelName: str, levelNum: int) -> None:
        """Comprehensively adds a new logging level to the `logging` module and the currently configured logging class.

        `levelName` becomes an attribute of the `logging` module with the value
        `levelNum`. `methodName` becomes a convenience method for both `logging`
        itself and the class returned by `logging.getLoggerClass()` (usually just
        `logging.Logger`). If `methodName` is not specified, `levelName.lower()` is
        used.

        To avoid accidental clobberings of existing attributes, this method will
        raise an `AttributeError` if the level name is already an attribute of the
        `logging` module or if the method name is already present

        Example:
        --------
        >>> addLoggingLevel('TRACE', logging.DEBUG - 5)
        >>> logging.getLogger(__name__).setLevel("TRACE")
        >>> logging.getLogger(__name__).trace('that worked')
        >>> logging.trace('so did this')
        >>> logging.TRACE
        5

        Args:
            levelName (str): name of level (i.e. TRACE)
            levelNum (int): integer level of new logging level
        """
        methodName = levelName.lower()

        def logForLevel(self: Any, message: str, *args: Any, **kwargs: Any) -> None:
            if self.isEnabledFor(levelNum):
                self._log(levelNum, message, args, **kwargs)

        def logToRoot(message: str, *args: Any, **kwargs: Any) -> None:
            logging.log(levelNum, message, *args, **kwargs)

        logging.addLevelName(levelNum, levelName)
        setattr(logging, levelName, levelNum)
        setattr(logging.getLoggerClass(), methodName, logForLevel)
        setattr(logging, methodName, logToRoot)

    @staticmethod
    def build_log_tx_str(stringable: Any) -> str:
        """Build a string with Tx arrows

        Args:
            stringable (Any): stringable object to surround with arrows

        Returns:
            str: string surrounded by Tx arrows
        """
        s = str(stringable).strip(r"{}")
        arrow = f"{'<'*Logger.ARROW_HEAD_COUNT}{'-'*Logger.ARROW_TAIL_COUNT}"
        return f"\n{arrow}{s}{arrow}\n"

    @staticmethod
    def build_log_rx_str(stringable: Any, asynchronous: bool = False) -> str:
        """Build a string with Rx arrows

        Args:
            stringable (Any): stringable object to surround with arrows
            asynchronous (bool): Should the arrows contain ASYNC?. Defaults to False.

        Returns:
            str: string surrounded by Rx arrows
        """
        s = str(stringable).strip(r"{}")
        assert Logger.ARROW_TAIL_COUNT > 5
        if asynchronous:
            arrow = f"{'-'*(Logger.ARROW_TAIL_COUNT//2-3)}ASYNC{'-'*(Logger.ARROW_TAIL_COUNT//2-2)}{'>'*Logger.ARROW_HEAD_COUNT}"
        else:
            arrow = f"{'-'*Logger.ARROW_TAIL_COUNT}{'>'*Logger.ARROW_HEAD_COUNT}"
        return f"\n{arrow}{s}{arrow}\n"


def setup_logging(
    base: logging.Logger | str,
    output: Path | None = None,
    modules: dict[str, int] | None = None,
) -> logging.Logger:
    """Configure the GoPro modules for logging and get a logger that can be used by the application

    This can only be called once and should be done at the top level of the application.

    Args:
        base (logging.Logger | str): Name of application (i.e. __name__) or preconfigured logger to use as base
        output (Path | None): Path of log file for file stream handler. If not set, will not log to file.
        modules (dict[str, int] | None): Optional override of modules / levels. Will be merged into default modules.

    Raises:
        TypeError: Base logger is not of correct type

    Returns:
        logging.Logger: updated logger that the application can use for logging
    """
    if isinstance(base, str):
        base = logging.getLogger(base)
    elif not isinstance(base, logging.Logger):
        raise TypeError("Base must be of type logging.Logger or str")
    l = Logger(base, output, modules)
    return l.logger


def set_file_logging_level(level: int) -> None:
    """Change the global logging level for the default file output handler

    Args:
        level (int): level to set
    """
    if fh := Logger.get_instance().file_handler:
        fh.setLevel(level)


def set_stream_logging_level(level: int) -> None:
    """Change the global logging level for the default stream output handler

    Args:
        level (int): level to set
    """
    Logger.get_instance().stream_handler.setLevel(level)


def set_logging_level(level: int) -> None:
    """Change the global logging level for the default file and stream output handlers

    Args:
        level (int): level to set
    """
    set_file_logging_level(level)
    set_stream_logging_level(level)


def add_logging_handler(handler: logging.Handler) -> None:
    """Add a handler to all of the GoPro internal modules

    Args:
        handler (logging.Handler): handler to add
    """
    Logger.get_instance().addLoggingHandler(handler)