diff --git a/TPTBox/logger/README.md b/TPTBox/logger/README.md index f423b78..b6d0b46 100644 --- a/TPTBox/logger/README.md +++ b/TPTBox/logger/README.md @@ -46,3 +46,33 @@ from TPTBox import No_Logger log = No_Logger() log.print("This is silently discarded") ``` + +## Loguru backend + +Emission is backed by [Loguru](https://github.com/Delgan/loguru) (`TPTBox/logger/_loguru_backend.py`). +The public API, classes, `verbose` semantics, and the terminal **color coding** are unchanged — +each `Log_Type` maps to a custom `TPTBOX_*` Loguru level whose color reproduces the original ANSI, +so terminal output renders identically. `WARNING_THROW` still raises a Python `warnings.warn`. + +On import, TPTBox configures the global Loguru logger (removing Loguru's own default handler). +Set `TPTBOX_LOGGER_TAKEOVER=0` *before importing TPTBox* to opt out. + +### What this enables + +```python +from TPTBox.logger import loguru_logger, configure, install_excepthook + +# 1) Attach your own sink — e.g. structured JSON logs, with level filtering: +loguru_logger.add("run.jsonl", serialize=True, level="TPTBOX_WARNING") # WARNING and worse + +# 2) File rotation / retention on the file-backed Logger: +from TPTBox import Logger +log = Logger("dataset/", "pipeline", rotation="20 MB", retention="10 days", enqueue=True) + +# 3) Exception capture: print_error() emits the structured exception to your sinks; +# optionally route *uncaught* exceptions through Loguru too: +install_excepthook() +``` + +Thread-safety: Loguru serializes sink writes, so concurrent `print()` calls no longer interleave +mid-line (`enqueue=True` additionally makes a sink thread/process-safe). diff --git a/TPTBox/logger/__init__.py b/TPTBox/logger/__init__.py index 4551510..0021100 100755 --- a/TPTBox/logger/__init__.py +++ b/TPTBox/logger/__init__.py @@ -1,5 +1,19 @@ from __future__ import annotations +from ._loguru_backend import configure, install_excepthook +from ._loguru_backend import logger as loguru_logger from .log_constants import Log_Type from .log_file import Logger, Logger_Interface, Reflection_Logger, String_Logger from .log_file import No_Logger as Print_Logger + +__all__ = [ + "Log_Type", + "Logger", + "Logger_Interface", + "Print_Logger", + "Reflection_Logger", + "String_Logger", + "configure", # opt out of the global-Loguru take-over / customize sinks + "install_excepthook", # route uncaught exceptions through Loguru (opt-in) + "loguru_logger", # the configured Loguru logger; add your own sinks (JSON, files, ...) to it +] diff --git a/TPTBox/logger/_loguru_backend.py b/TPTBox/logger/_loguru_backend.py new file mode 100644 index 0000000..56ef6a8 --- /dev/null +++ b/TPTBox/logger/_loguru_backend.py @@ -0,0 +1,237 @@ +"""Loguru-backed emission layer for the TPTBox logger. + +This is the only module that imports :mod:`loguru`. The public logger classes in +:mod:`TPTBox.logger.log_file` keep owning everything that decides the output *bytes* +(prefix building, ``ltype``-in-``*text`` detection, ``verbose`` gating, the +``datatype_to_string`` conversion). This module only takes the final, fully-built +message string and routes it to Loguru sinks: + +* a colorized terminal sink writing to the live ``sys.stdout`` (so output stays + capturable/redirectable just like the old ``print``), and +* per-instance file sinks for :class:`~TPTBox.logger.log_file.Logger`. + +Each :class:`~TPTBox.logger.log_constants.Log_Type` maps to a custom Loguru level +whose ``color`` markup reproduces the exact ANSI code of the old ``type2bcolors`` +table, so the terminal coloring renders identically while being level-driven. +""" + +from __future__ import annotations + +import os +import sys + +from loguru import logger + +from TPTBox.logger.log_constants import Log_Type + +__all__ = [ + "add_file_sink", + "add_file_stream_sink", + "configure", + "emit_exception", + "emit_file", + "emit_terminal", + "install_excepthook", + "level_name", + "logger", + "remove_sink", +] + +# Log_Type -> (loguru level name, severity, color markup). +# The markup is chosen so Loguru emits the SAME ANSI escape as the old `type2bcolors` +# (validated: -> \033[96m, -> \033[44m, etc.). Empty color == default. +_LEVELS: dict[Log_Type, tuple[str, int, str]] = { + Log_Type.TEXT: ("TPTBOX_TEXT", 20, ""), + Log_Type.NEUTRAL: ("TPTBOX_NEUTRAL", 20, ""), + Log_Type.SAVE: ("TPTBOX_SAVE", 22, ""), + Log_Type.WARNING: ("TPTBOX_WARNING", 30, ""), + Log_Type.WARNING_THROW: ("TPTBOX_WARNING_THROW", 30, ""), + Log_Type.LOG: ("TPTBOX_LOG", 20, ""), + Log_Type.OK: ("TPTBOX_OK", 25, ""), + Log_Type.FAIL: ("TPTBOX_FAIL", 40, ""), + Log_Type.Yellow: ("TPTBOX_YELLOW", 20, ""), + Log_Type.STRANGE: ("TPTBOX_STRANGE", 10, ""), + Log_Type.UNDERLINE: ("TPTBOX_UNDERLINE", 20, ""), + Log_Type.ITALICS: ("TPTBOX_ITALICS", 20, ""), + Log_Type.BOLD: ("TPTBOX_BOLD", 20, ""), + Log_Type.DOCKER: ("TPTBOX_DOCKER", 20, ""), + Log_Type.TOTALSEG: ("TPTBOX_TOTALSEG", 20, ""), + Log_Type.STAGE: ("TPTBOX_STAGE", 20, ""), +} + +_TERMINAL_FORMAT = "{message}" +_FILE_FORMAT = "{message}" + +_configured = False + + +def level_name(ltype: Log_Type) -> str: + """Return the Loguru level name registered for a given ``Log_Type``.""" + return _LEVELS.get(ltype, _LEVELS[Log_Type.TEXT])[0] + + +def _ensure_levels() -> None: + """Register the custom ``TPTBOX_*`` levels (idempotent).""" + for name, no, color in _LEVELS.values(): + try: + logger.level(name) + except ValueError: + logger.level(name, no=no, color=color) + + +def _terminal_sink(message) -> None: + r"""Write a colorized record to the live ``sys.stdout``, honoring the call's ``end``. + + Loguru always appends a ``\n``; we strip it and append the original ``end`` so + ``end="\r"`` progress lines survive unchanged. + """ + end = message.record["extra"].get("tptbox_end", "\n") + text = str(message) + text = text.removesuffix("\n") + # Look up sys.stdout at write-time (not at add-time) so redirect_stdout / capsys work. + sys.stdout.write(text + end) + + +def configure(take_over: bool | None = None) -> None: + """Configure the global Loguru logger for TPTBox (idempotent). + + Args: + take_over: If True, remove Loguru's pre-existing handlers so only the + TPTBox terminal sink is active (the default — matches the old logger + which was the sole stdout writer). If False, leave any handlers a host + application registered and only add TPTBox's filtered stdout sink. + If None, read the ``TPTBOX_LOGGER_TAKEOVER`` env var (default True). + """ + global _configured # noqa: PLW0603 + _ensure_levels() + if _configured: + return + if take_over is None: + take_over = os.environ.get("TPTBOX_LOGGER_TAKEOVER", "1") not in ("0", "false", "False") + if take_over: + logger.remove() + logger.add( + _terminal_sink, + format=_TERMINAL_FORMAT, + colorize=True, + level=0, + filter=lambda r: r["extra"].get("tptbox_channel") == "terminal", + enqueue=False, + catch=False, + ) + _configured = True + + +def emit_terminal(text: str, ltype: Log_Type = Log_Type.TEXT, end: str = "\n") -> None: + """Emit one already-built, already-prefixed message to the terminal sink. + + ``text`` is passed as the single ``{message}`` value with NO format args, so literal + ``{}``/``<>`` in the message are never interpreted. + """ + if not _configured: + configure() + logger.bind(tptbox_channel="terminal", tptbox_end=end).log(level_name(ltype), text) + + +def add_file_sink(filepath, key, *, rotation=None, retention=None, enqueue: bool = False, mode: str = "w") -> int: + """Register a Loguru file sink dedicated to one ``Logger`` instance. + + Args: + filepath: Destination log file (Loguru owns/creates it). + key: Unique id bound on each record so only this instance's lines land here. + rotation/retention: Optional Loguru file rotation/retention policies. + enqueue: If True, writes go through a background thread (thread/process-safe). + mode: File open mode (``"w"`` truncates, matching the old behavior). + + Returns: + The Loguru sink id (pass to :func:`remove_sink`). + """ + if not _configured: + configure() + return logger.add( + str(filepath), + format=_FILE_FORMAT, + colorize=False, + level=0, + filter=lambda r, _k=key: r["extra"].get("tptbox_file_id") == _k, + rotation=rotation, + retention=retention, + enqueue=enqueue, + mode=mode, + catch=False, + ) + + +def add_file_stream_sink(stream, key, *, enqueue: bool = False) -> int: + """Register a Loguru function sink writing ANSI-free lines to ``stream`` (a file handle). + + Unlike :func:`add_file_sink` (Loguru owns the file) this keeps the caller's handle, so + ``flush()`` works and the call's ``end`` is honored. No rotation/retention. + """ + if not _configured: + configure() + + def _sink(message, _s=stream) -> None: + end = message.record["extra"].get("tptbox_end", "\n") + text = str(message) + text = text.removesuffix("\n") + _s.write(text + end) + + return logger.add( + _sink, + format=_FILE_FORMAT, + colorize=False, + level=0, + filter=lambda r, _k=key: r["extra"].get("tptbox_file_id") == _k, + enqueue=enqueue, + catch=False, + ) + + +def emit_file(text: str, key, ltype: Log_Type = Log_Type.TEXT, end: str = "\n") -> None: + """Emit one ANSI-free line to the file sink identified by ``key``.""" + if not _configured: + configure() + logger.bind(tptbox_channel="file", tptbox_file_id=key, tptbox_end=end).log(level_name(ltype), text) + + +def remove_sink(sink_id: int) -> None: + """Remove a Loguru sink, tolerating an already-removed id (atexit double-remove).""" + try: + logger.remove(sink_id) + except (ValueError, KeyError): + pass + + +def emit_exception(message: str, ltype: Log_Type = Log_Type.FAIL) -> None: + """Emit a structured record carrying the *active* exception, for user sinks. + + The human-readable traceback text is still emitted separately by the facade + (``print_error``). This extra record is bound to the ``"exception"`` channel so the + default terminal/file sinks ignore it (no double traceback there); a user sink added + with ``serialize=True`` (or any permissive filter) receives the structured exception. + """ + if not _configured: + configure() + logger.opt(exception=True).bind(tptbox_channel="exception").log(level_name(ltype), message) + + +def install_excepthook() -> None: + """Route uncaught exceptions through Loguru (opt-in; replaces ``sys.excepthook``).""" + + def _hook(exc_type, exc_value, exc_tb): + if issubclass(exc_type, KeyboardInterrupt): + sys.__excepthook__(exc_type, exc_value, exc_tb) + return + logger.opt(exception=(exc_type, exc_value, exc_tb)).bind(tptbox_channel="exception").log( + level_name(Log_Type.FAIL), "Uncaught exception" + ) + + sys.excepthook = _hook + + +# Configure eagerly at import: this runs before any user code can add Loguru sinks, so the +# take-over `logger.remove()` only drops Loguru's own default handler (nothing user-owned yet). +# Sinks a caller adds afterwards survive. Set TPTBOX_LOGGER_TAKEOVER=0 before importing TPTBox +# to opt out of the take-over. +configure() diff --git a/TPTBox/logger/log_file.py b/TPTBox/logger/log_file.py index b6db6d6..d43939a 100755 --- a/TPTBox/logger/log_file.py +++ b/TPTBox/logger/log_file.py @@ -9,6 +9,7 @@ import numpy as np +from TPTBox.logger import _loguru_backend as _backend from TPTBox.logger.log_constants import ( Log_Type, _clean_all_color_from_text, @@ -160,8 +161,14 @@ def add_sub_logger(self, name: str, default_verbose: bool = False) -> String_Log return self # type: ignore def print_error(self, **args) -> None: - """Log the current exception traceback with ``Log_Type.FAIL`` severity.""" + """Log the current exception traceback with ``Log_Type.FAIL`` severity. + + Also emits a structured Loguru record carrying the active exception so user sinks + (e.g. ``logger.add(..., serialize=True)``) can capture it; the human-readable + traceback text below is unchanged. + """ self.print(traceback.format_exc(), ltype=Log_Type.FAIL, **args) + _backend.emit_exception("Logged exception") logging_state = None @@ -277,6 +284,9 @@ def __init__( default_verbose: bool = False, log_arguments=None, prefix: str | None = None, + rotation=None, + retention=None, + enqueue: bool = False, ): """Initialise a file-backed logger, creating the log directory and file automatically. @@ -286,6 +296,10 @@ def __init__( default_verbose: Default verbose behavior when not specified in calls. log_arguments: If set, will print the contents in a "run with arguments" section. prefix: If set, will use this string as prefix instead of the automatically chosen one. + rotation: Optional Loguru rotation policy (e.g. ``"10 MB"``, ``"00:00"``). When set + (or ``retention``), Loguru owns the file and rotates it. + retention: Optional Loguru retention policy (e.g. ``"10 days"``, ``5``). + enqueue: If True, file writes go through a background thread (thread/process-safe). """ path = Path(path) # ensure pathlib object # Get Start time @@ -307,10 +321,20 @@ def __init__( log_path = Path(path).joinpath("logs") if not Path.exists(log_path): Path.mkdir(log_path) - # Open log file - self.f = open(log_path.joinpath(log_filename_full), "w") # noqa: SIM115 - # calls close() if program terminates - self._finalizer = weakref.finalize(self.f, self.close) + # Open log file -> route writes through a Loguru sink. + self._filepath = log_path.joinpath(log_filename_full) + self._file_key = id(self) + self._closed = False + if rotation is None and retention is None: + # Default: own the handle (flush() works, `end` honored, atexit-safe duration line). + self.f = open(self._filepath, "w") # noqa: SIM115 + self._sink_id = _backend.add_file_stream_sink(self.f, self._file_key, enqueue=enqueue) + else: + # Loguru owns the file -> rotation/retention available (lines are \n-terminated). + self.f = None + self._sink_id = _backend.add_file_sink(self._filepath, self._file_key, rotation=rotation, retention=retention, enqueue=enqueue) + # calls close() at interpreter exit (matches the previous atexit cleanup) + self._finalizer = weakref.finalize(self, self.close) self.default_verbose = default_verbose # Log file always start with their name and start log time self.print(log_filename_processed[:-1], verbose=False, ltype=Log_Type.LOG) @@ -348,14 +372,14 @@ def create_from_bids( path = bids_file.dataset return Logger(path, log_filename, default_verbose=default_verbose, prefix=override_prefix) - def _log(self, text: str, end: str = "\n", ltype=Log_Type.TEXT) -> None: # noqa: ARG002 - """Write a plain-text line to the log file.""" - self.f.write(str(text)) - self.f.write(end) + def _log(self, text: str, end: str = "\n", ltype=Log_Type.TEXT) -> None: + """Write a plain-text line to the log file (via the Loguru file sink).""" + _backend.emit_file(str(text), self._file_key, ltype, end) def flush(self) -> None: - """Flush the underlying file buffer to disk.""" - self.f.flush() + """Flush the underlying file buffer to disk (Loguru file sinks flush per write).""" + if self.f is not None: + self.f.flush() def flush_sub_logger(self, sublogger: String_Logger, closed: bool = False) -> None: """Append a sub-logger's accumulated content into this log file. @@ -377,7 +401,8 @@ def remove(self) -> None: def close(self) -> None: """Flush all sub-loggers, write timing information, and close the log file.""" - if not self.f.closed: + if not self._closed: + self._closed = True self.sub_loggers = [s for s in self.sub_loggers if s.log_content != ""] if len(self.sub_loggers) > 0: self.print(ignore_prefix=True) @@ -398,8 +423,11 @@ def close(self) -> None: verbose=False, ltype=Log_Type.LOG, ) - self.f.flush() - self.f.close() + # remove the sink AFTER the closing lines were emitted, then release our handle + _backend.remove_sink(self._sink_id) + if self.f is not None: + self.f.flush() + self.f.close() @property def removed(self) -> bool: @@ -561,7 +589,9 @@ def print_to_terminal(text: str, end: str, ltype: Log_Type = Log_Type.TEXT) -> N if ltype == Log_Type.WARNING_THROW: warnings.warn(color_log_text(ltype=ltype, text=text), Warning, stacklevel=3) else: - print(color_log_text(ltype=ltype, text=text), end=end) + # Route through Loguru (native level-driven coloring). The facade already built the + # exact prefixed text; the backend applies the matching color and honors `end`. + _backend.emit_terminal(text, ltype, end) def sub_log_call_func(name: str, logger: Logger, function, default_verbose: bool | None = None, **kwargs) -> object: diff --git a/pyproject.toml b/pyproject.toml index c5356aa..3a603aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ joblib = "*" scikit-learn = "*" pynrrd = "*" requests = "*" +loguru = "^0.7.2" # --- OLD STACK (Python < 3.11) numpy = [ diff --git a/unit_tests/test_logger.py b/unit_tests/test_logger.py new file mode 100644 index 0000000..4a86f6c --- /dev/null +++ b/unit_tests/test_logger.py @@ -0,0 +1,142 @@ +"""Backwards-compatibility regression tests for the Loguru-backed logger. + +These lock the public contract that must survive the Loguru migration: the terminal +color coding (the `type2bcolors` ANSI + `[*]/[!]/...` prefixes), carriage-return (`end`) +progress lines, `WARNING_THROW` -> warnings.warn, verbose gating, ANSI-free file output, +and exception capture into user sinks. Coloring is now Loguru-native, so we compare after +normalizing the (invisible) ANSI reset codes rather than byte-for-byte. +""" + +from __future__ import annotations + +import glob +import io +import os +import re +import tempfile +import unittest +import warnings +from contextlib import redirect_stdout +from pathlib import Path + +from TPTBox.logger import Log_Type, Print_Logger +from TPTBox.logger.log_constants import color_log_text, type2bcolors +from TPTBox.logger.log_file import Logger, No_Logger + + +def _cap(fn) -> str: + buf = io.StringIO() + with redirect_stdout(buf): + fn() + return buf.getvalue() + + +def _norm(s: str) -> str: + """Collapse runs of resets and drop a leading reset (all invisible) for render-equality.""" + s = re.sub(r"(\x1b\[0m)+", "\x1b[0m", s) + return s.removeprefix("\x1b[0m") + + +class TestLoggerColorContract(unittest.TestCase): + def test_each_log_type_renders_like_type2bcolors(self): + for lt in Log_Type: + if lt == Log_Type.WARNING_THROW: + continue # routed to warnings.warn, asserted separately + got = _cap(lambda lt=lt: No_Logger().print("Hello World", ltype=lt)) + prefix = type2bcolors[lt][1] + expected = color_log_text(lt, f"{prefix} Hello World") + "\n" # the old reference rendering + self.assertEqual(_norm(got), _norm(expected), f"color/format changed for {lt.name}") + + def test_exact_color_code_present(self): + for lt in Log_Type: + if lt in (Log_Type.WARNING_THROW, Log_Type.TEXT, Log_Type.NEUTRAL): + continue # throw-type and default (reset) colors + got = _cap(lambda lt=lt: No_Logger().print("x", ltype=lt)) + self.assertTrue(got.startswith(type2bcolors[lt][0]), f"{lt.name} missing its ANSI color") + + def test_end_carriage_return_preserved(self): + got = _cap(lambda: No_Logger().print("progress", ltype=Log_Type.SAVE, end="\r")) + self.assertTrue(got.endswith("\r")) + self.assertNotIn("\n", got) + + def test_verbose_false_suppresses_terminal(self): + self.assertEqual(_cap(lambda: No_Logger().print("hidden", ltype=Log_Type.SAVE, verbose=False)), "") + + def test_warning_throw_emits_warning_not_stdout(self): + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + out = _cap(lambda: No_Logger().print("danger", ltype=Log_Type.WARNING_THROW)) + self.assertEqual(out, "") + self.assertEqual(len(w), 1) + + def test_on_helpers_and_prefix_attr(self): + self.assertIn("\x1b[92m[+] ok", _cap(lambda: No_Logger().on_ok("ok"))) + self.assertIn("\x1b[91m[!] bad", _cap(lambda: No_Logger().on_fail("bad"))) + + def with_prefix(): + lg = No_Logger() + lg.prefix = "API" + lg.print("hi", ltype=Log_Type.SAVE) + + self.assertIn("[API] hi", _cap(with_prefix)) + + def test_multi_arg_and_positional_ltype(self): + got = _cap(lambda: No_Logger().print("Saved:", "/p/f", 42, Log_Type.SAVE)) + self.assertIn("[*] Saved: /p/f 42", got) + self.assertTrue(got.startswith(type2bcolors[Log_Type.SAVE][0])) + + +class TestLoggerFileBackend(unittest.TestCase): + def test_file_is_ansi_free_and_well_formed(self): + d = tempfile.mkdtemp() + lg = Logger(d, "unit", default_verbose=False) + lg.print("Saved", "/p/f.nii.gz", ltype=Log_Type.SAVE) + lg.print("oops", ltype=Log_Type.FAIL) + lg.close() + files = glob.glob(os.path.join(d, "logs", "*.log")) + self.assertEqual(len(files), 1) + self.assertTrue(os.path.basename(files[0]).endswith("_unit_log.log")) + content = Path(files[0]).read_text() + self.assertNotIn("\x1b", content) # no ANSI in the file + for needle in ["[#] Log started at:", "[*] Saved /p/f.nii.gz", "[!] oops", "[#] Program duration:"]: + self.assertIn(needle, content) + + def test_two_loggers_isolated(self): + d = tempfile.mkdtemp() + a, b = Logger(d, "AAA"), Logger(d, "BBB") + a.print("only-A", verbose=False) + b.print("only-B", verbose=False) + a.close() + b.close() + ca = Path(next(f for f in glob.glob(os.path.join(d, "logs", "*.log")) if "AAA" in f)).read_text() + self.assertIn("only-A", ca) + self.assertNotIn("only-B", ca) + + +class TestExceptionCapture(unittest.TestCase): + def test_print_error_text_and_structured_record(self): + import json + + from TPTBox.logger import loguru_logger + + records = [] + sid = loguru_logger.add(lambda m: records.append(str(m)), serialize=True, level=0) + try: + out = _cap(self._raise_and_log) + finally: + loguru_logger.remove(sid) + self.assertIn("ZeroDivisionError", out) + self.assertTrue(out.startswith(type2bcolors[Log_Type.FAIL][0])) # FAIL-colored text + recs = [json.loads(r)["record"] for r in records] + self.assertTrue(any(r.get("exception") and r["exception"]["type"] == "ZeroDivisionError" for r in recs)) + + @staticmethod + def _raise_and_log(): + try: + _ = 1 / 0 + except ZeroDivisionError: + Print_Logger().print_error() + + +if __name__ == "__main__": + unittest.main()