Source code for bhf_smarthealth.log.log_config

"""This module configures Logging objects used within the pipeline logging system"""

import logging
import os

from datetime import datetime, date, timezone
from pathlib import Path

from bhf_smarthealth.models import DataStreamNames
from bhf_smarthealth.utils import workflow_utils




def generate_log_timestamp_utc(timespec: str = "seconds"):
    """Generate a timestamp for 'now' in UTC zero offset

    Parameters
    ----------
    timespec : str, optional
       The optional argument timespec specifies the number of additional terms of the time to include. Valid options are 'auto', 'hours', 'minutes', 'seconds', 'milliseconds' and 'microseconds'., by default "seconds"

    Returns
    -------
    str
       String of UTC zero offset timestamp for 'now
    """
    return datetime.now(timezone.utc).isoformat(timespec=timespec).replace(":", "-")





def build_log_paths(
    date_to_process: date,
    outputs: dict[str, str],
    run_id: str,
    data_type: DataStreamNames | None = None,
) -> dict[str, str]:
    """Build a dictionary of logging output paths for each output location specified in outputs.keys() and the corresponding part-path in outputs.values()

    Parameters
    ----------
    date_to_process : date
        Date object of date being processed
    outputs : dict[str, str]
         Dictionary containing the log OUTPUT name as keys and the corresponding path value
    run_id : str
        String identifer of a unique pipeline run
    data_type : DataStreamNames | None, optional
        StrEnum of data stream being processed. If specified, a "data_type" subdir will be appended to each log path. By default None

    Returns
    -------
    dict[str, str]
        Dictionary of log names in the form {OUTPATH}_{LOGTYPE}_LOG as keys with path to log as the value.
    """

    # check the date_to_process is a valid date object
    date_to_process = workflow_utils.validate_date_to_process(date_to_process)

    # convert date object to string in specified format
    date_to_process_str = date_to_process.strftime("%Y%m%d")

    subdirs = ["metadata", "runs", "ops"]

    log_paths = {}

    if not data_type:
        data_type = ""  # type: ignore - if not specified, don't add a data_type stem

    for output_name, output_path in outputs.items():
        for subdir in subdirs:
            key = f"{output_name.upper()}_{subdir.upper()}_LOG"
            log_paths[key] = (
                f"{output_path.rstrip('/')}/logs/{subdir}/{date_to_process_str}/{run_id}/{data_type}"
            ).rstrip(
                "/"
            )  # remove trailing '/' in case of no data type to prevent path errors

    return log_paths





def default_log_format() -> str:
    return "%(asctime)s [%(levelname)s] | %(name)s: %(message)s"





def setup_global_logger(
    format: str = default_log_format(),
    log_path: Path | str = os.getcwd(),
    level: int | str = logging.INFO,
) -> None:
    """Initialises a global logger with File Handler to capture Snakemake workflow outputs.

    Parameters
    ----------
    format : str, optional
        Format options for the logged outputs, by default default_log_format()
    log_path : str, optional
        Path to output the log file to, by default the current working directory
    level : int, optional
        Logging level, by default logging.INFO
    """

    # Prevent re-running config if it has already been called
    if logging.getLogger().hasHandlers():
        return

    # Convert to human-readable logging level
    if isinstance(level, int):
        level = logging.getLevelName(level)

    # Get root logger
    logger = logging.getLogger()
    logger.setLevel(level)

    log_path = Path(log_path)

    # Check log path exists. If not, use the current working directory
    if not Path(log_path).exists():
        log_path = Path(os.getcwd())
        init_message = f"Log output path does not exist. Defaulting to {log_path} with level {level}"
    else:
        init_message = f"Log files will be output to {log_path} with level {level}"

    # Define log format
    if not format:
        format = default_log_format()
    formatter = logging.Formatter(format)

    # Prevent overwriting logs on subsequent runs by attaching timestamp to filename
    log_file_name = str(f"{generate_log_timestamp_utc()}_snakemake.log")
    log_file_path = Path(log_path / log_file_name)

    # Add FileHandler which outputs logs to specified file path
    file_handler = logging.FileHandler(log_file_path)
    file_handler.setLevel(level)
    file_handler.setFormatter(formatter)

    logger.addHandler(file_handler)

    logger.info("Log initialised.")
    logger.info(init_message)





def setup_rule_log(
    log_file: Path, level: int | str = logging.INFO, format: str = default_log_format()
) -> logging.Logger:
    """Initialises a logger with File Handler to capture outputs from Snakemake rules.

    Parameters
    ----------
    log_file : Path
        Path of the file to log to
    level : int | str, optional
        Logging level, by default logging.INFO
    format : str, optional
        Format options for the logged outputs, by default default_log_format()

    Returns
    -------
    logger : logging.Logger
        A configured Logger object with a File Handler
    """
    # Do basic configuration of a logging system with a FileHandler
    logging.basicConfig(
        filename=log_file,
        level=level,
        format=format,
    )

    logger = logging.getLogger(__name__)

    # Convert to human-readable logging level
    if isinstance(level, int):
        level = logging.getLevelName(level)

    logger.info(f"Log initialised. Logging to {log_file} with level {level}")

    return logger