new: working logger

pyproject.toml

@@ -23,6 +23,8 @@ dependencies = [
     "tomli_w",
     "numba",
     "tqdm",
+    "pydantic",
+    "pydantic-settings",
 ]
 
 [tool.ruff]

src/scgenerator/__init__.py

@@ -2,6 +2,7 @@
 from scgenerator import io, math, noise, operators, plotting
 from scgenerator.helpers import *
 from scgenerator.io import MemoryIOHandler, ZipFileIOHandler
+from scgenerator.logger import get_logger
 from scgenerator.math import abs2, argclosest, normalized, span, tspace, wspace
 from scgenerator.parameter import Parameters
 from scgenerator.physics import fiber, materials, plasma, pulse, units

src/scgenerator/env.py

@@ -1,91 +1,43 @@
 import os
+from enum import Enum
 from pathlib import Path
 from typing import Any, Dict, Literal, Optional, Set
 
-ENVIRON_KEY_BASE = "SCGENERATOR_"
-TMP_FOLDER_KEY_BASE = ENVIRON_KEY_BASE + "SC_TMP_"
-PREFIX_KEY_BASE = ENVIRON_KEY_BASE + "PREFIX_"
-
-PBAR_POLICY = ENVIRON_KEY_BASE + "PBAR_POLICY"
-LOG_FILE_LEVEL = ENVIRON_KEY_BASE + "LOG_FILE_LEVEL"
-LOG_PRINT_LEVEL = ENVIRON_KEY_BASE + "LOG_PRINT_LEVEL"
-START_RAY = ENVIRON_KEY_BASE + "START_RAY"
-NO_RAY = ENVIRON_KEY_BASE + "NO_RAY"
-OUTPUT_PATH = ENVIRON_KEY_BASE + "OUTPUT_PATH"
+from pydantic import ImportString, NonNegativeFloat, ValidationInfo, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
-global_config: dict[str, dict[str, Any]] = {
-    LOG_FILE_LEVEL: dict(
-        help="minimum lvl of message to be saved in the log file",
-        choices=["critical", "error", "warning", "info", "debug"],
-        default=None,
-        type=str,
-    ),
-    LOG_PRINT_LEVEL: dict(
-        help="minimum lvl of message to be printed to the standard output",
-        choices=["critical", "error", "warning", "info", "debug"],
-        default="error",
-        type=str,
-    ),
-    PBAR_POLICY: dict(
-        help="what to do with progress pars (print them, make them a txt file or nothing), default is print",
-        choices=["print", "file", "both", "none"],
-        default=None,
-        type=str,
-    ),
-    START_RAY: dict(action="store_true", help="initialize ray (ray must be installed)", type=bool),
-    NO_RAY: dict(action="store_true", help="force not to use ray", type=bool),
-    OUTPUT_PATH: dict(
-        short_name="-o", help="path to the final output folder", default=None, type=str
-    ),
-}
+class LogLevel(Enum):
+    DEBUG = "debug"
+    INFO = "info"
+    WARNING = "warning"
+    ERROR = "error"
+    CRITICAL = "critical"
 
 
-def data_folder(task_id: int) -> Optional[str]:
-    idstr = str(int(task_id))
-    tmp = os.getenv(TMP_FOLDER_KEY_BASE + idstr)
-    return tmp
+class Config(BaseSettings):
+    model_config = SettingsConfigDict(validate_default=True, env_prefix="SCG_")
+
+    @model_validator(mode="before")
+    @classmethod
+    def sort_log_stuff(
+        self, data: dict[str, Any], validation_info: ValidationInfo
+    ) -> dict[str, Any]:
+        if "log_file" in data and "log_output" not in data:
+            data["log_output"] = None
+
+        if "log_file" in data and "log_level" not in data:
+            data["log_level"] = LogLevel.INFO
+
+        return data
+
+    progress_interval: NonNegativeFloat = 2.0
+    log_level: LogLevel = LogLevel.ERROR
+    log_output: ImportString | None = "sys.stderr"
+    log_file: Path | None = None
 
 
-def get(key: str, default=None) -> Any:
-    str_value = os.environ.get(key)
-    if isinstance(str_value, str):
-        try:
-            t = global_config[key]["type"]
-            if t == bool:
-                return str_value.lower() == "true"
-            return t(str_value)
-        except (ValueError, KeyError):
-            pass
-    return default
-
-
-def all_environ() -> Dict[str, str]:
-    """returns a dictionary of all environment variables set by any instance of scgenerator"""
-    d = dict(filter(lambda el: el[0].startswith(ENVIRON_KEY_BASE), os.environ.items()))
-    return d
-
-
-def output_path() -> Path:
-    p = get(OUTPUT_PATH)
-    if p is not None:
-        return Path(p).resolve()
-    return None
-
-
-def pbar_policy() -> Set[Literal["print", "file"]]:
-    policy = get(PBAR_POLICY)
-    if policy == "print" or policy is None:
-        return {"print"}
-    elif policy == "file":
-        return {"file"}
-    elif policy == "both":
-        return {"file", "print"}
-    else:
-        return set()
-
-
-def log_file_level() -> Set[Literal["critical", "error", "warning", "info", "debug"]]:
+def log_level() -> Set[Literal["critical", "error", "warning", "info", "debug"]]:
     policy = get(LOG_FILE_LEVEL)
     try:
         policy = policy.lower()
@@ -105,3 +57,6 @@ def log_print_level() -> Set[Literal["critical", "error", "warning", "info", "de
     except AttributeError:
         pass
     return None
+
+
+CONFIG = Config()

src/scgenerator/logger.py

@@ -1,6 +1,6 @@
 import logging
 
-from scgenerator.env import log_file_level, log_print_level
+from scgenerator.env import CONFIG
 
 lvl_map: dict[str, int] = dict(
     debug=logging.DEBUG,
@@ -27,7 +27,7 @@ def get_logger(name=None):
     logging.Logger obj
         logger
     """
-    name = __name__ if name is None else name
+    name = name or __name__
     logger = logging.getLogger(name)
     return configure_logger(logger)
 
@@ -40,29 +40,26 @@ def configure_logger(logger: logging.Logger):
     ----------
     logger : logging.Logger
         logger to configure
-    logfile : str or None, optional
-        path to log file
 
     Returns
     -------
     logging.Logger obj
         updated logger
     """
-    if not hasattr(logger, "already_configured"):
-        print_lvl = lvl_map.get(log_print_level(), logging.NOTSET)
-        file_lvl = lvl_map.get(log_file_level(), logging.NOTSET)
-
-        if file_lvl > logging.NOTSET:
+    if hasattr(logger, "already_configured"):
+        return logger
+    log_level = lvl_map.get(CONFIG.log_level.value, logging.NOTSET)
+    if CONFIG.log_file is not None:
         formatter = logging.Formatter("{levelname}: {name}: {message}", style="{")
-            file_handler1 = logging.FileHandler("scgenerator.log", "a+")
+        file_handler1 = logging.FileHandler(CONFIG.log_file, "a+")
         file_handler1.setFormatter(formatter)
-            file_handler1.setLevel(file_lvl)
+        file_handler1.setLevel(log_level)
         logger.addHandler(file_handler1)
-        if print_lvl > logging.NOTSET:
-            stream_handler = logging.StreamHandler()
-            stream_handler.setLevel(print_lvl)
+    if CONFIG.log_output is not None:
+        stream_handler = logging.StreamHandler(CONFIG.log_output)
+        stream_handler.setLevel(log_level)
         logger.addHandler(stream_handler)
 
-        logger.setLevel(logging.DEBUG)
+    logger.setLevel(log_level)
     logger.already_configured = True
     return logger

tests/test_logger.py

@@ -0,0 +1,7 @@
+import scgenerator as sc
+
+
+def test_get_logger():
+    assert sc.get_logger() is sc.get_logger()
+    assert sc.get_logger("this_test") is sc.get_logger("this_test")
+    assert sc.get_logger("this_test") is not sc.get_logger()