Skip to content

Commit

Permalink
Merge dev into feature/365/convenience-click-types
Browse files Browse the repository at this point in the history
  • Loading branch information
TimothyWillard committed Jan 10, 2025
2 parents dcb899b + da7d816 commit 4071d37
Show file tree
Hide file tree
Showing 8 changed files with 442 additions and 91 deletions.
169 changes: 169 additions & 0 deletions flepimop/gempyor_pkg/src/gempyor/logging.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,169 @@
"""
Logging utilities for consistent script output.
This module provides functionality for creating consistent outputs from CLI tools
provided by this package. Currently exported are:
- `ClickHandler`: Custom logging handler specifically designed for CLI output using
click.
- `get_script_logger`: Factory for creating a logger instance with a consistent style
across CLI tools.
"""

__all__ = ["ClickHandler", "get_script_logger"]


import logging
import os
import sys
from typing import Any, IO

import click


DEFAULT_LOG_FORMAT = "%(asctime)s:%(levelname)s:%(name)s> %(message)s"


class ClickHandler(logging.Handler):
"""
Custom logging handler specifically for click based CLI tools.
"""

_punctuation = (".", ",", "?", "!", ":")

def __init__(
self,
level: int | str = 0,
file: IO[Any] | None = None,
nl: bool = True,
err: bool = False,
color: bool | None = None,
punctuate: bool = True,
) -> None:
"""
Initialize an instance of the click handler.
Args:
level: The logging level to use for this handler.
file: The file to write to. Defaults to stdout.
nl: Print a newline after the message. Enabled by default.
err: Write to stderr instead of stdout.
color: Force showing or hiding colors and other styles. By default click
will remove color if the output does not look like an interactive
terminal.
punctuate: A boolean indicating if punctuation should be added to the end
of a log message provided if missing.
Notes:
For more details on the `file`, `nl`, `err`, and `color` args please refer
to [`click.echo`](https://click.palletsprojects.com/en/8.1.x/api/#click.echo).
"""
super().__init__(level)
self._file = file
self._nl = nl
self._err = err
self._color = color
self._punctuate = punctuate

def emit(self, record: logging.LogRecord) -> None:
"""
Emit a given log record via `click.echo`
Args:
record: The log record to output.
See Also:
[`logging.Handler.emit`](https://docs.python.org/3/library/logging.html#logging.Handler.emit)
"""
msg = self.format(record)
msg = f"{msg}." if self._punctuate and not msg.endswith(self._punctuation) else msg
click.echo(
message=msg, file=self._file, nl=self._nl, err=self._err, color=self._color
)


def get_script_logger(
name: str,
verbosity: int,
handler: logging.Handler | None = None,
log_format: str = DEFAULT_LOG_FORMAT,
) -> logging.Logger:
"""
Create a logger for use in scripts.
Args:
name: The name to display in the log message, useful for locating the source
of logging messages. Almost always `__name__`.
verbosity: A non-negative integer for the verbosity level.
handler: An optional logging handler to use in creating the logger returned, or
`None` to just use the `ClickHandler`.
log_format: The format to use for logged messages. Passed directly to the `fmt`
argument of [logging.Formatter](https://docs.python.org/3/library/logging.html#logging.Formatter).
Returns:
An instance of `logging.Logger` that has the appropriate level set based on
`verbosity` and a custom handler for outputting for CLI tools.
Examples:
>>> from gempyor.logging import get_script_logger
>>> logger = get_script_logger(__name__, 3)
>>> logger.info("This is a log info message")
2024-10-29 16:07:20,272:INFO:__main__> This is a log info message.
"""
logger = logging.getLogger(name)
logger.setLevel(_get_logging_level(verbosity))
handler = ClickHandler() if handler is None else handler
log_formatter = logging.Formatter(log_format)
for old_handler in logger.handlers:
logger.removeHandler(old_handler)
handler.setFormatter(log_formatter)
logger.addHandler(handler)
# pytest-dev/pytest#3697
logger.propagate = os.path.basename(sys.argv[0]) == "pytest" if sys.argv else False
return logger


def _get_logging_level(verbosity: int) -> int:
"""
An internal method to convert verbosity to a logging level.
Args:
verbosity: A non-negative integer for the verbosity level or level from
`logging` that will be returned as is.
Examples:
>>> _get_logging_level(0)
40
>>> _get_logging_level(1)
30
>>> _get_logging_level(2)
20
>>> _get_logging_level(3)
10
>>> _get_logging_level(4)
10
>>> import logging
>>> _get_logging_level(logging.ERROR) == logging.ERROR
True
Raises:
ValueError: If `verbosity` is less than zero.
Returns:
The log level from the `logging` module corresponding to the given `verbosity`.
"""
if verbosity < 0:
raise ValueError(f"`verbosity` must be non-negative, was given '{verbosity}'.")
if verbosity in (
logging.DEBUG,
logging.INFO,
logging.WARNING,
logging.ERROR,
logging.CRITICAL,
):
return verbosity
verbosity_to_logging_level = {
0: logging.ERROR,
1: logging.WARNING,
2: logging.INFO,
}
return verbosity_to_logging_level.get(verbosity, logging.DEBUG)
48 changes: 14 additions & 34 deletions flepimop/gempyor_pkg/src/gempyor/parameters.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,8 @@ def __init__(
self.npar = len(self.pnames)
if self.npar != len(set([name.lower() for name in self.pnames])):
raise ValueError(
"Parameters of the SEIR model have the same name (remember that case is not sufficient!)"
"Parameters of the SEIR model have the same "
"name (remember that case is not sufficient!)"
)

# Attributes of dictionary
Expand Down Expand Up @@ -128,16 +129,18 @@ def __init__(
print("loaded dates:", df.index)
raise ValueError(
f"Issue loading file '{fn_name}' for parameter '{pn}': "
f"Provided file dates span '{str(df.index[0])}' to '{str(df.index[-1])}', "
f"but the config dates span '{ti}' to '{tf}'."
f"Provided file dates span '{str(df.index[0])}' to "
f"'{str(df.index[-1])}', but the config dates "
f"span '{ti}' to '{tf}'."
)
if not (pd.date_range(ti, tf) == df.index).all():
print("config dates:", pd.date_range(ti, tf))
print("loaded dates:", df.index)
raise ValueError(
f"Issue loading file '{fn_name}' for parameter '{pn}': "
f"Provided file dates span '{str(df.index[0])}' to '{str(df.index[-1])}', "
f"but the config dates span '{ti}' to '{tf}'."
f"Provided file dates span '{str(df.index[0])}' to "
f"'{str(df.index[-1])}', but the config dates "
f"span '{ti}' to '{tf}'."
)

self.pdata[pn]["ts"] = df
Expand All @@ -148,7 +151,8 @@ def __init__(
else:
self.pdata[pn]["stacked_modifier_method"] = "product"
logging.debug(
f"No 'stacked_modifier_method' for parameter {pn}, assuming multiplicative NPIs"
f"No 'stacked_modifier_method' for parameter {pn}, "
"assuming multiplicative NPIs."
)

if self.pconfig[pn]["rolling_mean_windows"].exists():
Expand All @@ -165,32 +169,6 @@ def __init__(
logging.debug(f"Index in arrays are: {self.pnames2pindex}")
logging.debug(f"NPI overlap operation is {self.stacked_modifier_method} ")

def picklable_lamda_alpha(self):
"""
Read the `alpha_val` attribute.
This defunct method returns the `alpha_val` attribute of this class which is
never set by this class. If this method is called and the `alpha_val` attribute
is not set an AttributeError will be raised.
Returns:
The `alpha_val` attribute.
"""
return self.alpha_val

def picklable_lamda_sigma(self):
"""
Read the `sigma_val` attribute.
This defunct method returns the `sigma_val` attribute of this class which is
never set by this class. If this method is called and the `sigma_val` attribute
is not set an AttributeError will be raised.
Returns:
The `sigma_val` attribute.
"""
return self.sigma_val

def get_pnames2pindex(self) -> dict:
"""
Read the `pnames2pindex` attribute.
Expand Down Expand Up @@ -235,7 +213,8 @@ class via `subpop_names`.
else:
param_arr[idx] = self.pdata[pn]["ts"].values

return param_arr # we don't store it as a member because this object needs to be small to be pickable
# we don't store it as a member because this object needs to be small to be pickable
return param_arr

def parameters_load(
self, param_df: pd.DataFrame, n_days: int, nsubpops: int
Expand Down Expand Up @@ -275,7 +254,8 @@ def parameters_load(
param_arr[idx] = self.pdata[pn]["ts"].values
else:
print(
f"PARAM: parameter {pn} NOT found in loadID file. Drawing from config distribution"
f"PARAM: parameter {pn} NOT found in loadID file. "
"Drawing from config distribution"
)
pval = self.pdata[pn]["dist"]()
param_arr[idx] = np.full((n_days, nsubpops), pval)
Expand Down
53 changes: 51 additions & 2 deletions flepimop/gempyor_pkg/src/gempyor/shared_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,14 @@
import multiprocessing
import pathlib

from typing import Any, Callable, Literal
from typing import Any, Callable
import warnings

import click
import confuse

from .utils import as_list, config
from .logging import get_script_logger
from .utils import config, as_list


@click.group()
Expand Down Expand Up @@ -277,3 +278,51 @@ def _parse_option(param: click.Parameter, value: Any) -> Any:
cfg[option] = _parse_option(config_file_options[option], value)

return cfg


def log_cli_inputs(kwargs: dict[str, Any], verbosity: int | None = None) -> None:
"""
Log CLI inputs for user debugging.
This function only logs debug messages so the verbosity has to be set quite high
to see the output of this function.
Args:
kwargs: The CLI arguments given as a dictionary of key word arguments.
verbosity: The verbosity level of the CLI tool being used or `None` to infer
from the given `kwargs`.
Examples:
>>> from gempyor.shared_cli import log_cli_inputs
>>> log_cli_inputs({"abc": 123, "def": True}, 3)
2024-11-05 09:27:58,884:DEBUG:gempyor.shared_cli> CLI was given 2 arguments:
2024-11-05 09:27:58,885:DEBUG:gempyor.shared_cli> abc = 123.
2024-11-05 09:27:58,885:DEBUG:gempyor.shared_cli> def = True.
>>> log_cli_inputs({"abc": 123, "def": True}, 2)
>>> from pathlib import Path
>>> kwargs = {
... "input_file": Path("config.in"),
... "stochastic": True,
... "cluster": "longleaf",
... "verbosity": 3,
... }
>>> log_cli_inputs(kwargs)
2024-11-05 09:29:21,666:DEBUG:gempyor.shared_cli> CLI was given 4 arguments:
2024-11-05 09:29:21,667:DEBUG:gempyor.shared_cli> input_file = /Users/twillard/Desktop/GitHub/HopkinsIDD/flepiMoP/flepimop/gempyor_pkg/config.in.
2024-11-05 09:29:21,667:DEBUG:gempyor.shared_cli> stochastic = True.
2024-11-05 09:29:21,668:DEBUG:gempyor.shared_cli> cluster = longleaf.
2024-11-05 09:29:21,668:DEBUG:gempyor.shared_cli> verbosity = 3.
"""
verbosity = kwargs.get("verbosity") if verbosity is None else verbosity
if verbosity is None:
return
logger = get_script_logger(__name__, verbosity)
longest_key = -1
total_keys = 0
for k, _ in kwargs.items():
longest_key = len(k) if len(k) > longest_key else longest_key
total_keys += 1
logger.debug("CLI was given %u arguments:", total_keys)
for k, v in kwargs.items():
v = v.absolute() if isinstance(v, pathlib.Path) else v
logger.debug("%s = %s", k.ljust(longest_key, " "), v)
32 changes: 32 additions & 0 deletions flepimop/gempyor_pkg/tests/logging/test__get_logging_level.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
import logging

import pytest

from gempyor.logging import _get_logging_level


@pytest.mark.parametrize("verbosity", (-1, -100))
def test__get_logging_level_negative_verbosity_value_error(verbosity: int) -> None:
with pytest.raises(
ValueError, match=f"`verbosity` must be non-negative, was given '{verbosity}'."
):
_get_logging_level(verbosity)


@pytest.mark.parametrize(
("verbosity", "expected_level"),
(
(logging.ERROR, logging.ERROR),
(logging.WARNING, logging.WARNING),
(logging.INFO, logging.INFO),
(logging.DEBUG, logging.DEBUG),
(0, logging.ERROR),
(1, logging.WARNING),
(2, logging.INFO),
(3, logging.DEBUG),
(4, logging.DEBUG),
(5, logging.DEBUG),
),
)
def test__get_logging_level_output_validation(verbosity: int, expected_level: int) -> None:
assert _get_logging_level(verbosity) == expected_level
Loading

0 comments on commit 4071d37

Please sign in to comment.