log

Attention : This should be part of the main() function, that is : a global logging mechanism and configuration.

Functions:

Name	Description
initialize_logger	Initialise logging to either stderr or a file ?
log_data_fingerprint
log_path_for_pid	Return a log file path unique to the process (e.g. logs/access.log -> logs/access.12345.log).

initialize_logger

initialize_logger(
    ctx: Context, log_level: None | int = None
)

Initialise logging to either stderr or a file ?

Notes

Attention : Used in typer_option_log !

Source code in pvgis/log.py

def initialize_logger(
    ctx: Context,
    log_level: None | int = None,
):
    """
    Initialise logging to either stderr or a file ?

    Notes
    -----
    Attention : Used in typer_option_log !

    """
    LOGURU_LEVELS = {
        0: "WARNING",  # Only show warnings and errors
        1: "INFO",  # Show info, warnings, and errors
        # Define more levels ?
        7: "DEBUG",  # Show debug messages
    }
    minimum_log_level = LOGURU_LEVELS.get(log_level, "WARNING")
    # print(f'Minimum log level : {minimum_log_level}')

    rich_handler = ctx.params.get("log_rich_handler")
    if rich_handler:
        logger.remove()
        import richuru

        richuru.install(level=0, rich_traceback=False)

    if log_level and not rich_handler:
        import sys

        # print(f"Logging to sys.stderr : {sys.stderr}")
        # logger.add(sys.stderr, enqueue=True, backtrace=True, diagnose=True)
        fmt = "{time} | {level: <8} | {name: ^15} | {function: ^15} | {line: >3} | {message}"
        logger.add(sys.stderr, format=fmt, level=minimum_log_level)
        logger.debug(f"Logging to sys.stderr : {sys.stderr}")

    log_file = ctx.params.get("log_file")
    # NOTE If logging is enabled but no explicit file path is provided,
    # fall back to an automatic timestamped file name (previous behavior).
    if log_level and not log_file:
        log_file = "pvgis_{time}.log"

    if log_file:
        if not rich_handler:
            logger.debug(f"Logging to file : {log_file}")

        else:
            logger.debug(
                f"Logging to file : {log_file}",
                alt=f"Logging to file : [reverse]{log_file}[/reverse] ?",
            )

        log_file_path = Path(log_file)
        log_file_path.parent.mkdir(parents=True, exist_ok=True)
        logger.add(
            str(log_file_path), level=minimum_log_level
        )  # , compression="tar.gz")
        # logger.debug(f'Logging to file : {log_file}', alt=f'Logging to file : [reverse]{log_file}[/reverse] ?')

    return log_level

log_data_fingerprint

log_data_fingerprint(
    data,
    log_level,
    hash_after_this_verbosity_level=2,
    output=None,
)

Source code in pvgis/log.py

def log_data_fingerprint(
    data,
    log_level,
    hash_after_this_verbosity_level=2,
    output=None,
):
    """ """
    if output:
        print(type(output))
    if log_level > hash_after_this_verbosity_level:
        import inspect

        caller_name = inspect.stack()[1].function
        data_hash = generate_hash(data)
        logger.debug(
            f"< Output {caller_name}() : {type(data)}, {data.dtype}, Hash {data_hash}",
            alt=f"< [bold]Output[/bold] of {caller_name}() : {type(data)}, [reverse]{data.dtype}[/reverse], Hash [code]{data_hash}[/code]",
        )
    if log_level > DEBUG_AFTER_THIS_VERBOSITY_LEVEL:
        from devtools import debug

        debug(locals())

log_path_for_pid

log_path_for_pid(path: str | None, pid: int) -> str | None

Return a log file path unique to the process (e.g. logs/access.log -> logs/access.12345.log).

Use in multi-worker setups (e.g. Gunicorn) so each worker rotates only its own file, avoiding FileNotFoundError when one worker rotates a file shared by others.

Source code in pvgis/log.py

def log_path_for_pid(path: str | None, pid: int) -> str | None:
    """
    Return a log file path unique to the process (e.g. logs/access.log -> logs/access.12345.log).

    Use in multi-worker setups (e.g. Gunicorn) so each worker rotates only its own file,
    avoiding FileNotFoundError when one worker rotates a file shared by others.
    """
    if not path:
        return None
    p = Path(path)
    return str(p.parent / f"{p.stem}.{pid}{p.suffix}")