Mod: Update log utils for cmd line customization.

src/aac_datasets/check.py

@@ -9,13 +9,12 @@
 
 import yaml
 
-import aac_datasets
 from aac_datasets.datasets.audiocaps import AudioCaps, AudioCapsCard
 from aac_datasets.datasets.clotho import Clotho, ClothoCard
 from aac_datasets.datasets.macs import MACS, MACSCard
 from aac_datasets.datasets.wavcaps import WavCaps, WavCapsCard
-from aac_datasets.download import setup_logging
 from aac_datasets.utils.globals import get_default_root
+from aac_datasets.utils.log_utils import setup_logging_verbose
 
 DATASETS_NAMES = (AudioCapsCard.NAME, ClothoCard.NAME, MACSCard.NAME, WavCapsCard.NAME)
 
@@ -124,7 +123,7 @@ def _get_main_check_args() -> Namespace:
 
 def _main_check() -> None:
     args = _get_main_check_args()
-    setup_logging(aac_datasets.__package__, args.verbose)
+    setup_logging_verbose("aac_datasets", args.verbose)
 
     if args.verbose >= 2:
         pylog.debug(yaml.dump({"Arguments": args.__dict__}, sort_keys=False))

src/aac_datasets/download.py

@@ -6,7 +6,6 @@
 
 import yaml
 
-import aac_datasets
 from aac_datasets.datasets.functional.audiocaps import (
     AudioCapsCard,
     download_audiocaps_datasets,
@@ -17,18 +16,14 @@
     WavCapsCard,
     download_wavcaps_datasets,
 )
-from aac_datasets.utils.cmdline import (
-    _str_to_bool,
-    _str_to_opt_int,
-    _str_to_opt_str,
-    setup_logging,
-)
+from aac_datasets.utils.cmdline import _str_to_bool, _str_to_opt_int, _str_to_opt_str
 from aac_datasets.utils.globals import (
     get_default_ffmpeg_path,
     get_default_root,
     get_default_ytdlp_path,
     get_default_zip_path,
 )
+from aac_datasets.utils.log_utils import setup_logging_verbose
 
 pylog = logging.getLogger(__name__)
 
@@ -175,7 +170,7 @@ def _get_main_download_args() -> Namespace:
 
 def _main_download() -> None:
     args = _get_main_download_args()
-    setup_logging(aac_datasets.__package__, args.verbose)
+    setup_logging_verbose("aac_datasets", args.verbose)
 
     if args.verbose >= 2:
         pylog.debug(yaml.dump({"Arguments": args.__dict__}, sort_keys=False))

src/aac_datasets/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", "t", "yes", "y", "1")
 _FALSE_VALUES = ("false", "f", "no", "n", "0")
@@ -38,63 +34,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_datasets/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