Source code for ska_sdp_piper.piper.utils.log_util

import logging
from copy import deepcopy

import dask
from distributed import WorkerPlugin
from ska_ser_logging import configure_logging

from ska_sdp_piper.piper.utils.log_config import LOGGING_CONFIG




class LogUtil:
    """
    A class to setup logging configuration
    in the context of piper based pipelines.

    This also stores some global settings related to the logging,
    which might be useful for the pipelines or runners.
    """

    level: int = logging.INFO
    "Stores the logging level to be used globally"
    verbose: bool = False
    "Stores whether user enabled verbose logs via the CLI options"
    log_file: str | None = None
    """
    Stores the file path to which the logs will be written. Piper
    will set this up on behalf of the user.
    """



    @classmethod
    def configure(
        cls,
        log_file: str | None = None,
        verbose: bool = False,
        disable_console: bool = False,
    ):
        """
        Setup logging config as per the SKA Logging Format.
        Additionally, this can setup logging to a file, and
        can disable logging to stdout.

        Every call to this function will override the existing
        global logging config. Ideally in the appplication's lifecycle,
        this function should be called once, as early as possible.

        Parameters
        ----------
        log_file
            Path to log file. If None or if value is "falsy",
            logs are not written to a file.
        verbose
            If True, set the log level to DEBUG, else set to INFO.
        disable_console
            By default, the ska_ser_logging logger logs to the
            sys.stdout stream. This can be disabled by setting
            disable_console value to True.
        """
        cls.verbose = verbose
        cls.log_file = log_file

        if cls.verbose:
            cls.level = logging.DEBUG

        overrides_config = deepcopy(LOGGING_CONFIG)

        if cls.log_file:
            overrides_config["handlers"]["file"] = {
                "()": logging.FileHandler,
                "formatter": "default",
                "filename": cls.log_file,
                "filters": deepcopy(
                    overrides_config["handlers"]["console"]["filters"]
                ),
            }
            overrides_config["root"]["handlers"].append("file")

        if disable_console:
            # Set console handler to None, so that
            # configure_logging will delete it when merging with
            # SKA standard config dictionary
            overrides_config["handlers"]["console"] = None
            overrides_config["root"]["handlers"].remove("console")

        configure_logging(
            level=cls.level,
            overrides=overrides_config,
        )






class LogPlugin(WorkerPlugin):
    """
    Dask worker plugin that configures logging on worker startup.
    """

    idempotent = True

    def __init__(
        self,
        log_file: str | None = None,
        verbose: bool = False,
        disable_console: bool = False,
    ):
        """
        This accepts same parameters
        as the :py:func:`LogUtil.configure()` function.
        Refer to that function for details.

        Example
        -------
        >>> plugin = LogPlugin(verbose=LogUtil.verbose)
        >>> client.register_worker_plugin(plugin)
        """
        self.log_file = log_file
        self.verbose = verbose
        self.disable_console = disable_console



    def setup(self, worker):
        LogUtil.configure(
            log_file=self.log_file,
            verbose=self.verbose,
            disable_console=self.disable_console,
        )




@dask.delayed
def delayed_log(logger: logging.Logger, formated_log_msg: str, **kwargs):
    """
    Log a formatted message in a Dask-delayed task.
    This ensures that dask delayed objects inside kwargs dictionary,
    whose results are needed for logging a message,
    are not computed eagerly.

    Parameters
    ----------
    logger
        Logging callable that accepts a single formatted message.
    formated_log_msg
        Format string used to build the log message.
    **kwargs
        Keyword arguments used to format ``formated_log_msg``.

    Example
    -------
    >>> # Assuming root logger can stream to stdout
    >>> logger = logging.getLogger()
    >>> task = delayed_log(
    ...     logger.info,
    ...     "{var_1} and {var_2}",
    ...     var_1=10,
    ...     var_2=dask.delayed(list)([1, 2, 3])
    ... )
    >>> task.compute()
    "10 and [1, 2, 3]"
    """
    logger(formated_log_msg.format(**kwargs))