Mod: Update log utils for cmd line customization.

src/aac_metrics/download.py

@@ -12,7 +12,6 @@
 
 from torch.hub import download_url_to_file
 
-import aac_metrics
 from aac_metrics.classes.bert_score_mrefs import BERTScoreMRefs
 from aac_metrics.classes.fense import FENSE
 from aac_metrics.functional.meteor import DNAME_METEOR_CACHE
@@ -22,13 +21,14 @@
     FNAME_SPICE_JAR,
     check_spice_install,
 )
-from aac_metrics.utils.cmdline import _str_to_bool, setup_logging
+from aac_metrics.utils.cmdline import _str_to_bool
 from aac_metrics.utils.globals import (
     _get_cache_path,
     _get_tmp_path,
     get_default_cache_path,
     get_default_tmp_path,
 )
+from aac_metrics.utils.log_utils import setup_logging_verbose
 from aac_metrics.utils.tokenization import FNAME_STANFORD_CORENLP_3_4_1_JAR
 
 pylog = logging.getLogger(__name__)
@@ -384,7 +384,7 @@ def _get_main_download_args() -> Namespace:
 
 def _main_download() -> None:
     args = _get_main_download_args()
-    setup_logging(aac_metrics.__package__, args.verbose)
+    setup_logging_verbose("aac_metrics", args.verbose)
 
     download_metrics(
         cache_path=args.cache_path,

src/aac_metrics/eval.py

@@ -9,19 +9,19 @@
 
 import yaml
 
-import aac_metrics
 from aac_metrics.functional.evaluate import (
     DEFAULT_METRICS_SET_NAME,
     METRICS_SETS,
     evaluate,
 )
 from aac_metrics.utils.checks import check_java_path, check_metric_inputs
-from aac_metrics.utils.cmdline import _str_to_bool, _str_to_opt_str, setup_logging
+from aac_metrics.utils.cmdline import _str_to_bool, _str_to_opt_str
 from aac_metrics.utils.globals import (
     get_default_cache_path,
     get_default_java_path,
     get_default_tmp_path,
 )
+from aac_metrics.utils.log_utils import setup_logging_verbose
 
 pylog = logging.getLogger(__name__)
 
@@ -227,7 +227,7 @@ def _get_main_evaluate_args() -> Namespace:
 
 def _main_eval() -> None:
     args = _get_main_evaluate_args()
-    setup_logging(aac_metrics.__package__, args.verbose)
+    setup_logging_verbose("aac_metrics", args.verbose)
 
     if not check_java_path(args.java_path):
         raise RuntimeError(f"Invalid Java executable. ({args.java_path})")

src/aac_metrics/utils/cmdline.py

@@ -1,11 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-import logging
-import sys
-from logging import Logger
-from types import ModuleType
-from typing import Optional, Sequence, Union
+from typing import Optional
 
 _TRUE_VALUES = ("true", "1", "t", "yes", "y")
 _FALSE_VALUES = ("false", "0", "f", "no", "n")
@@ -29,63 +25,3 @@ def _str_to_opt_str(s: str) -> Optional[str]:
         return None
     else:
         return s
-
-
-def setup_logging(
-    package_or_logger: Union[
-        str,
-        ModuleType,
-        None,
-        Logger,
-        Sequence[Union[str, ModuleType, None]],
-        Sequence[Logger],
-    ],
-    verbose: int,
-    format_: Optional[str] = "[%(asctime)s][%(name)s][%(levelname)s] - %(message)s",
-) -> None:
-    if package_or_logger is None or isinstance(
-        package_or_logger, (str, Logger, ModuleType)
-    ):
-        package_or_logger_lst = [package_or_logger]
-    else:
-        package_or_logger_lst = list(package_or_logger)
-
-    name_or_logger_lst = [
-        pkg.__name__ if isinstance(pkg, ModuleType) else pkg
-        for pkg in package_or_logger_lst
-    ]
-    logger_lst = [
-        logging.getLogger(pkg_i) if not isinstance(pkg_i, Logger) else pkg_i
-        for pkg_i in name_or_logger_lst
-    ]
-
-    handler = logging.StreamHandler(sys.stdout)
-    if format_ is not None:
-        handler.setFormatter(logging.Formatter(format_))
-
-    for logger in logger_lst:
-        found = False
-        for handler in logger.handlers:
-            if (
-                isinstance(handler, logging.StreamHandler)
-                and handler.stream is sys.stdout
-            ):
-                found = True
-                break
-        if not found:
-            logger.addHandler(handler)
-
-        level = _verbose_to_logging_level(verbose)
-        logger.setLevel(level)
-
-
-def _verbose_to_logging_level(verbose: int) -> int:
-    if verbose < 0:
-        level = logging.ERROR
-    elif verbose == 0:
-        level = logging.WARNING
-    elif verbose == 1:
-        level = logging.INFO
-    else:
-        level = logging.DEBUG
-    return level

src/aac_metrics/utils/log_utils.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import logging
+import sys
+from functools import cache
+from logging import Logger
+from types import ModuleType
+from typing import List, Optional, Sequence, Union
+
+pylog = logging.getLogger(__name__)
+
+DEFAULT_FMT = "[%(asctime)s][%(name)s][%(levelname)s] - %(message)s"
+
+
+@cache
+def warn_once(msg: str, logger: Union[Logger, ModuleType, None]) -> None:
+    if logger is None:
+        pylog = logging.root
+    elif isinstance(logger, ModuleType):
+        pylog: Logger = logger.root
+    else:
+        pylog = logger
+
+    pylog.warning(msg)
+
+
+def setup_logging_verbose(
+    package_or_logger: Union[
+        str,
+        ModuleType,
+        None,
+        Logger,
+        Sequence[Union[str, ModuleType, None]],
+        Sequence[Logger],
+    ],
+    verbose: int,
+    fmt: Optional[str] = DEFAULT_FMT,
+) -> None:
+    level = _verbose_to_logging_level(verbose)
+    return setup_logging_level(package_or_logger, level=level, fmt=fmt)
+
+
+def setup_logging_level(
+    package_or_logger: Union[
+        str,
+        ModuleType,
+        None,
+        Logger,
+        Sequence[Union[str, ModuleType, None]],
+        Sequence[Logger],
+    ],
+    level: int,
+    fmt: Optional[str] = DEFAULT_FMT,
+) -> None:
+    logger_lst = _get_loggers(package_or_logger)
+    handler = logging.StreamHandler(sys.stdout)
+    if fmt is not None:
+        handler.setFormatter(logging.Formatter(fmt))
+
+    for logger in logger_lst:
+        found = False
+        for handler in logger.handlers:
+            if (
+                isinstance(handler, logging.StreamHandler)
+                and handler.stream is sys.stdout
+            ):
+                found = True
+                break
+        if not found:
+            logger.addHandler(handler)
+
+        logger.setLevel(level)
+
+
+def _get_loggers(
+    package_or_logger: Union[
+        str,
+        ModuleType,
+        None,
+        Logger,
+        Sequence[Union[str, ModuleType, None]],
+        Sequence[Logger],
+    ],
+) -> List[Logger]:
+    if package_or_logger is None or isinstance(
+        package_or_logger, (str, Logger, ModuleType)
+    ):
+        package_or_logger_lst = [package_or_logger]
+    else:
+        package_or_logger_lst = list(package_or_logger)
+
+    name_or_logger_lst = [
+        pkg.__name__ if isinstance(pkg, ModuleType) else pkg
+        for pkg in package_or_logger_lst
+    ]
+    logger_lst = [
+        logging.getLogger(pkg_i) if not isinstance(pkg_i, Logger) else pkg_i
+        for pkg_i in name_or_logger_lst
+    ]
+    return logger_lst
+
+
+def _verbose_to_logging_level(verbose: int) -> int:
+    if verbose < 0:
+        level = logging.ERROR
+    elif verbose == 0:
+        level = logging.WARNING
+    elif verbose == 1:
+        level = logging.INFO
+    else:
+        level = logging.DEBUG
+    return level