From 2eaa672035c3dbfa66a2910fbc7c1177f2667d7a Mon Sep 17 00:00:00 2001 From: iback Date: Mon, 29 Jun 2026 11:51:32 +0000 Subject: [PATCH 1/5] feat(logger): route terminal output through Loguru (native colored levels) Add loguru dep and a new TPTBox/logger/_loguru_backend.py: each Log_Type maps to a custom TPTBOX_* Loguru level whose color markup reproduces the exact ANSI of type2bcolors (light-cyan->96, bg blue->44, ...). print_to_terminal now emits via a colorized Loguru function-sink (writing to the live sys.stdout, honoring end so end='\r' progress lines are intact). WARNING_THROW still routes to warnings.warn. Public API/classes/verbose semantics unchanged. Verified: 28/28 golden cases render identically (only invisible reset-code differences) and all 15 non-default Log_Types emit the exact expected color code. Co-Authored-By: Claude Opus 4.8 --- TPTBox/logger/_loguru_backend.py | 174 +++++++++++++++++++++++++++++++ TPTBox/logger/log_file.py | 5 +- pyproject.toml | 1 + 3 files changed, 179 insertions(+), 1 deletion(-) create mode 100644 TPTBox/logger/_loguru_backend.py diff --git a/TPTBox/logger/_loguru_backend.py b/TPTBox/logger/_loguru_backend.py new file mode 100644 index 0000000..5621101 --- /dev/null +++ b/TPTBox/logger/_loguru_backend.py @@ -0,0 +1,174 @@ +"""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", + "configure", + "emit_file", + "emit_terminal", + "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 emit_file(text: str, key, ltype: Log_Type = Log_Type.TEXT) -> 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).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 diff --git a/TPTBox/logger/log_file.py b/TPTBox/logger/log_file.py index b6db6d6..3c8cce6 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, @@ -561,7 +562,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 = [ From c51ebc81a37ec6dc4b29184b5996322cc13cddc2 Mon Sep 17 00:00:00 2001 From: iback Date: Mon, 29 Jun 2026 11:57:27 +0000 Subject: [PATCH 2/5] feat(logger): route Logger file output through Loguru sinks (+ rotation/retention) Logger now writes via a Loguru file sink instead of raw file.write. Default keeps its own handle behind a Loguru function-sink (flush() works, 'end' honored, ANSI-free, atexit-safe duration line). New rotation=/retention=/enqueue kwargs switch to a Loguru-owned path sink for rotation & retention. Per-instance record filtering keeps multiple Loggers' files isolated. Verified: exact filename scheme + header/duration lines, no ANSI in files, rotation produces multiple files, two-logger isolation, and the duration line is still written at interpreter exit for unclosed loggers. Co-Authored-By: Claude Opus 4.8 --- TPTBox/logger/_loguru_backend.py | 30 ++++++++++++++++++-- TPTBox/logger/log_file.py | 47 +++++++++++++++++++++++--------- 2 files changed, 62 insertions(+), 15 deletions(-) diff --git a/TPTBox/logger/_loguru_backend.py b/TPTBox/logger/_loguru_backend.py index 5621101..a6136a3 100644 --- a/TPTBox/logger/_loguru_backend.py +++ b/TPTBox/logger/_loguru_backend.py @@ -159,11 +159,37 @@ def add_file_sink(filepath, key, *, rotation=None, retention=None, enqueue: bool ) -def emit_file(text: str, key, ltype: Log_Type = Log_Type.TEXT) -> None: +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).log(level_name(ltype), text) + logger.bind(tptbox_channel="file", tptbox_file_id=key, tptbox_end=end).log(level_name(ltype), text) def remove_sink(sink_id: int) -> None: diff --git a/TPTBox/logger/log_file.py b/TPTBox/logger/log_file.py index 3c8cce6..0bcdc67 100755 --- a/TPTBox/logger/log_file.py +++ b/TPTBox/logger/log_file.py @@ -278,6 +278,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. @@ -287,6 +290,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 @@ -308,10 +315,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) @@ -349,14 +366,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. @@ -378,7 +395,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) @@ -399,8 +417,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: From 4b23d72c52f7f712bf14b2e08ff0d7f8e54d0dfd Mon Sep 17 00:00:00 2001 From: iback Date: Mon, 29 Jun 2026 12:02:56 +0000 Subject: [PATCH 3/5] feat(logger): exception capture + expose Loguru config/logger for user sinks print_error now also emits a structured Loguru record (opt(exception=True), bound to an 'exception' channel the default sinks ignore) so user sinks capture the traceback; the human-readable text output is unchanged. Add install_excepthook() to route uncaught exceptions through Loguru (opt-in). Re-export configure, install_excepthook, and the configured loguru_logger from TPTBox.logger so callers can attach their own sinks (JSON via serialize=True, level filtering on TPTBOX_*). Configure now runs eagerly at import so the take-over logger.remove() only drops Loguru's own default handler and never wipes user sinks added afterwards. Co-Authored-By: Claude Opus 4.8 --- TPTBox/logger/__init__.py | 14 ++++++++++++ TPTBox/logger/_loguru_backend.py | 37 ++++++++++++++++++++++++++++++++ TPTBox/logger/log_file.py | 8 ++++++- 3 files changed, 58 insertions(+), 1 deletion(-) 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 index a6136a3..56ef6a8 100644 --- a/TPTBox/logger/_loguru_backend.py +++ b/TPTBox/logger/_loguru_backend.py @@ -26,9 +26,12 @@ __all__ = [ "add_file_sink", + "add_file_stream_sink", "configure", + "emit_exception", "emit_file", "emit_terminal", + "install_excepthook", "level_name", "logger", "remove_sink", @@ -198,3 +201,37 @@ def remove_sink(sink_id: int) -> None: 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 0bcdc67..d43939a 100755 --- a/TPTBox/logger/log_file.py +++ b/TPTBox/logger/log_file.py @@ -161,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 From 7baaf68e931eeb83888a053eed72108142787647 Mon Sep 17 00:00:00 2001 From: iback Date: Mon, 29 Jun 2026 12:16:46 +0000 Subject: [PATCH 4/5] test(logger): lock color/format contract + document Loguru backend Add unit_tests/test_logger.py (10 tests) asserting the backwards-compat contract: per-Log_Type terminal coloring matches type2bcolors, carriage-return end preserved, WARNING_THROW->warnings.warn, verbose gating, ANSI-free + isolated file output, and structured exception capture into user sinks. Document the Loguru backend, TPTBOX_* levels, take-over opt-out, rotation/retention, user sinks, and exception capture in the logger README. Full unit suite: 405 passed / 4 skipped; thread safety verified (3200 concurrent lines, no interleaving). Co-Authored-By: Claude Opus 4.8 --- TPTBox/logger/README.md | 30 ++++++++ unit_tests/test_logger.py | 142 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+) create mode 100644 unit_tests/test_logger.py 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/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() From 497c6beebf005d804b591a81aecd5253705e7c57 Mon Sep 17 00:00:00 2001 From: iback Date: Wed, 8 Jul 2026 13:44:26 +0000 Subject: [PATCH 5/5] consistent one import logger across the whole toolbox --- TPTBox/__init__.py | 21 +- TPTBox/core/bids_files.py | 95 ++++---- TPTBox/core/dicom/dicom_extract.py | 6 +- TPTBox/core/internal/elastic_deform.py | 5 +- TPTBox/core/nii_poi_abstract.py | 4 +- TPTBox/core/np_utils.py | 9 +- TPTBox/core/poi.py | 10 +- TPTBox/core/poi_fun/_help.py | 3 +- TPTBox/core/poi_fun/ray_casting.py | 10 +- TPTBox/core/poi_fun/save_load.py | 2 +- TPTBox/core/vert_constants.py | 2 +- TPTBox/logger/README.md | 240 +++++++++++++++---- TPTBox/logger/__init__.py | 23 +- TPTBox/logger/_loguru_backend.py | 240 ++++++++++++++----- TPTBox/logger/log_file.py | 177 +++++++++++++- TPTBox/mesh3D/mesh.py | 4 +- TPTBox/segmentation/VibeSeg/auto_download.py | 8 +- TPTBox/segmentation/spineps.py | 6 +- TPTBox/spine/snapshot2D/snapshot_modular.py | 19 +- TPTBox/stitching/stitching.py | 66 +++-- unit_tests/test_logger.py | 207 +++++++++++++++- 21 files changed, 915 insertions(+), 242 deletions(-) diff --git a/TPTBox/__init__.py b/TPTBox/__init__.py index 70c66dc..66139d2 100755 --- a/TPTBox/__init__.py +++ b/TPTBox/__init__.py @@ -29,8 +29,20 @@ from TPTBox.core.poi_fun.poi_global import POI_Global from TPTBox.core.vert_constants import ZOOMS, Location, Vertebra_Instance, v_idx2name, v_idx_order, v_name2idx -# Logger -from TPTBox.logger import Log_Type, Logger, Logger_Interface, Print_Logger, String_Logger +# Logger — everything goes through `configure_logging()` and the `logger` proxy. +from TPTBox.logger import ( + Log_Type, + Logger, + Logger_Interface, + Print_Logger, + Reflection_Logger, + String_Logger, + configure_logging, + get_default_logger, + log, + logger, + loguru_logger, +) from TPTBox.logger.log_file import No_Logger Centroids = POI @@ -57,8 +69,13 @@ "calc_centroids", "calc_poi_from_subreg_vert", "calc_poi_from_two_segs", + "configure_logging", "core", + "get_default_logger", "load_poi", + "log", + "logger", + "loguru_logger", "np_utils", "to_nii", "v_idx2name", diff --git a/TPTBox/core/bids_files.py b/TPTBox/core/bids_files.py index f459b95..a850d48 100755 --- a/TPTBox/core/bids_files.py +++ b/TPTBox/core/bids_files.py @@ -14,6 +14,7 @@ import numpy as np import TPTBox +from TPTBox.logger import logger if TYPE_CHECKING: from TPTBox.core.nii_poi_abstract import Grid @@ -66,28 +67,29 @@ def validate_entities(key: str, value: str, name: str, verbose: bool) -> bool: return True try: key = key.lower() + log = logger if key not in entities_keys: - print( - f"[!] {key} is not in list of legal keys. This name '{name}' is invalid. Legal keys are: {list(entities_keys.keys())}. \nFor use see https://bids-specification.readthedocs.io/en/stable/99-appendices/09-entities.html" + log.on_fail( + f"{key} is not in list of legal keys. This name '{name}' is invalid. Legal keys are: {list(entities_keys.keys())}. \nFor use see https://bids-specification.readthedocs.io/en/stable/99-appendices/09-entities.html" ) entities_keys[key] = key return False if key in entity_alphanumeric and not value.isalnum(): - print(f"[!] value for {key} must be alphanumeric. This name '{name}' is invalid, with value {value}") + log.on_fail(f"value for {key} must be alphanumeric. This name '{name}' is invalid, with value {value}") return False if key in entity_decimal and not value.isdecimal(): - print(f"[!] value for {key} must be decimal. This name '{name}' is invalid, with value {value}") + log.on_fail(f"value for {key} must be decimal. This name '{name}' is invalid, with value {value}") return False # if int(value) == 0: - # print(f"[!] value for {key} must be not 0. This name '{name}' is invalid, with value {value}") + # log.on_fail(f"value for {key} must be not 0. This name '{name}' is invalid, with value {value}") if key in entity_format and value not in formats_relaxed: - print(f"[!] value for {key} must be a format. This name '{name}' is invalid, with value {value}") + log.on_fail(f"value for {key} must be a format. This name '{name}' is invalid, with value {value}") return False if key in entity_on_off and value not in ["on", "off"]: - print(f"[!] value for {key} must be in {['on', 'off']}. This name '{name}' is invalid, with value {value}") + log.on_fail(f"value for {key} must be in {['on', 'off']}. This name '{name}' is invalid, with value {value}") return False if key in entity_left_right and value not in ["L", "R"]: - print(f"[!] value for {key} must be in {['L', 'R']}. This name '{name}' is invalid, with value {value}") + log.on_fail(f"value for {key} must be in {['L', 'R']}. This name '{name}' is invalid, with value {value}") return False # parts = [ # "mag", @@ -117,7 +119,7 @@ def validate_entities(key: str, value: str, name: str, verbose: bool) -> bool: else: return True except Exception as e: - print(e) + logger.on_fail(e) return False @@ -148,29 +150,30 @@ def get_values_from_name(path: Path | str, verbose: bool) -> tuple[str, dict[str keys = bids_key.split("_") bids_format = keys[-1] + log = logger if bids_format not in formats_relaxed and verbose: - print(f"[!] Unknown format {bids_format} in file {name}", formats) + log.on_fail(f"Unknown format {bids_format} in file {name}", formats) formats_relaxed.append(bids_format) if file_type not in file_types and verbose: - print(f"[!] Unknown file_type {file_type} in file {name}") + log.on_fail(f"Unknown file_type {file_type} in file {name}") dic = {} for idx, s in enumerate(keys[:-1]): try: key, value = s.split("-", maxsplit=1) if idx == 0 and key != "sub" and verbose: - print(f"[!] First key must be sub not {key}. This name '{name}' is invalid") + log.on_fail(f"First key must be sub not {key}. This name '{name}' is invalid") if idx != 1 and key == "ses" and verbose: - print(f"[!] Session must be second key. This name '{name}' is invalid") + log.on_fail(f"Session must be second key. This name '{name}' is invalid") if key in dic and verbose: - print(f"[!] {bids_key} contains copies of the same key twice. This name '{name}' is invalid") + log.on_fail(f"{bids_key} contains copies of the same key twice. This name '{name}' is invalid") validate_entities(key, value, name, verbose) dic[key] = value except Exception: if verbose: - print(f'[!] "{s}" is not a valid key/value pair. Expected "KEY-VALUE" in {name}') + log.on_fail(f'"{s}" is not a valid key/value pair. Expected "KEY-VALUE" in {name}') return bids_format, dic, bids_key, file_type @@ -231,9 +234,10 @@ def save_buffer(f: Path, buffer_name: str) -> list[Path]: try: with open(str(f / buffer_name), "wb") as b: pickle.dump(new_buffer, b) - print("\n[ ] Save Buffer:", f) if verbose else None + if verbose: + logger.on_neutral("Save Buffer:", f) except OSError: - print("Saving not allowed") + logger.on_fail("Saving not allowed") _cont = 0 return new_buffer @@ -243,7 +247,8 @@ def save_buffer(f: Path, buffer_name: str) -> list[Path]: assert "/" not in parent, "only top parent folder allowed" folder = Path(dataset, parent) if not folder.exists(): - print("[ ] Dose not exist:", (folder), f"{' ':20}") if verbose else None + if verbose: + logger.on_neutral("Dose not exist:", (folder), f"{' ':20}") continue if (folder / buffer_name).exists(): import datetime @@ -253,40 +258,31 @@ def save_buffer(f: Path, buffer_name: str) -> list[Path]: age = today - file_mod_time if age.days >= int(max_age_days): - ( - print( - "[ ] Delete Buffer - to old:", + if verbose: + logger.on_neutral( + "Delete Buffer - to old:", (folder / buffer_name), f"{' ':20}", ) - if verbose - else None - ) (folder / buffer_name).unlink() if (folder / buffer_name).exists() and parent not in recompute_parents: with open((folder / buffer_name), "rb") as b: l = pickle.load(b) - ( - print( - f"[{len(l):8}] Read Buffer:", + if verbose: + logger.on_neutral( + f"Read Buffer [{len(l):8}]:", (folder / buffer_name), f"{' ':20}", ) - if verbose - else None - ) files[dataset] += l else: - ( - print( - f"[{_cont:8}] Create new Buffer:", + if verbose: + logger.on_neutral( + f"Create new Buffer [{_cont:8}]:", (folder / buffer_name), f"{' ':20}", end="\r", ) - if verbose - else None - ) files[dataset] += save_buffer((folder), buffer_name) if filter_file is not None: files: dict[Path | str, list[Path]] = {d: [g for g in f if filter_file(g)] for d, f in files.items()} @@ -315,7 +311,7 @@ def _scan_tree(path, lvl=1, filter_folder=lambda _x, _y: True, verbose=False): yield from _scan_tree(entry.path, lvl=lvl + 1, verbose=verbose) elif entry.name[0] != ".": if verbose: - print(f"[{_cont:8}]", end="\r") + logger.on_neutral(f"[{_cont:8}]", end="\r", ignore_prefix=True) _cont += 1 yield entry @@ -365,10 +361,10 @@ def __init__( for ds in datasets: ds_path = Path(ds) if isinstance(ds, str) else ds if not ds_path.name.startswith("dataset-"): - print(f"[!] Dataset {ds_path.name} does not start with 'dataset-'") + logger.on_fail(f"Dataset {ds_path.name} does not start with 'dataset-'") for ps in parents: if not any(ps.startswith(lp) for lp in parents): - print(f"[!] Parentfolder {ps} is not a legal name") + logger.on_fail(f"Parentfolder {ps} is not a legal name") self.datasets = datasets self.parents = parents @@ -437,7 +433,7 @@ def add_file_2_subject(self, bids: BIDS_FILE | Path, ds: Path | str | None = Non bids_key, file_type = str(bids).rsplit("/", maxsplit=1)[-1].split(".", maxsplit=1) # print(bids_key) except Exception: - print("[!] skip file with out a type declaration:", bids.name) + logger.on_fail("skip file with out a type declaration:", bids.name) # raise e return @@ -455,14 +451,12 @@ def add_file_2_subject(self, bids: BIDS_FILE | Path, ds: Path | str | None = Non if subject not in self.subjects: self.subjects[subject] = Subject_Container(subject, self.sequence_splitting_keys) self.count_file += 1 - ( - print( + if self.verbose: + logger.on_neutral( f"Found: {subject}, total file keys {(self.count_file)}, total subjects = {len(self.subjects)} ", end="\r", + ignore_prefix=True, ) - if self.verbose - else None - ) self.subjects[subject].add(bids) def enumerate_subjects(self, sort: bool = False, shuffle: bool = False) -> list[tuple[str, Subject_Container]]: @@ -1093,13 +1087,10 @@ def get_changed_path( # noqa: C901 for key, value in info.items(): # New Keys are getting checked! if non_strict_mode: - ( - print( - f"[!] {key} is not in list of legal keys. This name '{key}' is invalid. Legal keys are: {list(entities_keys.keys())}. \nFor use see https://bids-specification.readthedocs.io/en/stable/99-appendices/09-entities.html" + if key not in entities_keys: + logger.on_fail( + f"{key} is not in list of legal keys. This name '{key}' is invalid. Legal keys are: {list(entities_keys.keys())}. \nFor use see https://bids-specification.readthedocs.io/en/stable/99-appendices/09-entities.html" ) - if key not in entities_keys - else None - ) else: assert key in entities_keys, ( f"[!] {key} is not in list of legal keys. This name '{key}' is invalid. Legal keys are: {list(entities_keys.keys())}. \nFor use see https://bids-specification.readthedocs.io/en/stable/99-appendices/09-entities.html" @@ -1601,7 +1592,7 @@ def get_identifier(self, sequence_splitting_keys: list[str]) -> str: sequence_splitting_keys (list[str]): list of keys to use for splitting """ if "sub" not in self.info: - print(f"family_id, no sub-key, got {self.info}") + logger.on_fail(f"family_id, no sub-key, got {self.info}") identifier = "sub-404" else: identifier = "sub-" + self.info["sub"] diff --git a/TPTBox/core/dicom/dicom_extract.py b/TPTBox/core/dicom/dicom_extract.py index 93519da..4c3f8ec 100644 --- a/TPTBox/core/dicom/dicom_extract.py +++ b/TPTBox/core/dicom/dicom_extract.py @@ -201,7 +201,7 @@ def dicom_to_nifti_multiframe(ds: pydicom.FileDataset, nii_path: str | Path) -> # } # ], # --- No geometry info (e.g. RGB screen captures or video frames) --- - print("⚠️ No spatial metadata found — assuming pixel size = 1mm and identity orientation.") + logger.on_warning("No spatial metadata found — assuming pixel size = 1mm and identity orientation.") affine = np.eye(4) affine[0, 0] = 1.0 affine[1, 1] = 1.0 @@ -332,7 +332,7 @@ def _extract_nii_from_dicom(dicom_out_path, nii_path): return False except Exception: - print(nii_path) + logger.on_fail(nii_path) return True @@ -525,7 +525,7 @@ def _add_grid_info_to_json(nii_path: Path | str, simp_json: Path | str, force_up json_dict = load_json(simp_json) if Path(simp_json).exists() else {} if "grid" in json_dict and not force_update: return json_dict - print("Read Grid info", Path(simp_json).exists(), "grid" in json_dict) + logger.on_neutral("Read Grid info", Path(simp_json).exists(), "grid" in json_dict) nii = NII.load(nii_path, False) gird = { "shape": nii.shape, diff --git a/TPTBox/core/internal/elastic_deform.py b/TPTBox/core/internal/elastic_deform.py index 29aecbf..3d48e98 100644 --- a/TPTBox/core/internal/elastic_deform.py +++ b/TPTBox/core/internal/elastic_deform.py @@ -5,6 +5,7 @@ from numpy.typing import NDArray from TPTBox import NII +from TPTBox.logger import logger def deformed_nii( @@ -51,7 +52,7 @@ def deformed_nii( if sigma is None or points is None: sigma, points = get_random_deform_parameter(deform_factor=deform_factor) - print("deformation parameter sigma = ", round(sigma, 4), "; n_points = ", points) + logger.on_neutral("deformation parameter sigma = ", round(sigma, 4), "; n_points = ", points) t = time.time() values = list(nii_dic.values()) # Deform @@ -73,7 +74,7 @@ def deformed_nii( out2: dict[str, NII] = {} for (k, nii), arr in zip(nii_dic.items(), out, strict=True): out2[k] = nii.set_array(arr[p:-p, p:-p, p:-p]) - print("Deformation took", round(time.time() - t, 1), "Seconds") + logger.on_neutral("Deformation took", round(time.time() - t, 1), "Seconds") return out2 diff --git a/TPTBox/core/nii_poi_abstract.py b/TPTBox/core/nii_poi_abstract.py index 0205dcd..f107dde 100755 --- a/TPTBox/core/nii_poi_abstract.py +++ b/TPTBox/core/nii_poi_abstract.py @@ -12,7 +12,7 @@ from TPTBox.core.np_utils import np_count_nonzero from TPTBox.core.vert_constants import COORDINATE -from TPTBox.logger import Log_Type +from TPTBox.logger import Log_Type, logger from .vert_constants import ( AFFINE, @@ -95,7 +95,7 @@ def __str__(self) -> str: try: zoom = "(" + ",".join([f"{a:.2f}" for a in self.zoom]) + ")" except Exception as e: - print(e) + logger.on_warning(e) zoom = self.zoom return f"shape={self.shape_int},spacing={zoom}, origin={origin}, ori={self.orientation}" # type: ignore diff --git a/TPTBox/core/np_utils.py b/TPTBox/core/np_utils.py index 2b3ebbd..59c54bd 100755 --- a/TPTBox/core/np_utils.py +++ b/TPTBox/core/np_utils.py @@ -29,6 +29,7 @@ from TPTBox.core.compat import zip_strict from TPTBox.core.vert_constants import COORDINATE, LABEL_MAP, LABEL_REFERENCE +from TPTBox.logger import logger UINT = TypeVar("UINT", bound=np.unsignedinteger[Any]) INT = TypeVar("INT", bound=np.signedinteger[Any]) @@ -110,7 +111,7 @@ def cc3dstatistics(arr: UINTARRAY, use_crop: bool = True) -> dict: arrc = arr[crop] return _cc3dstats(arrc) except ValueError as e: - print(e) + logger.on_warning(e) return _cc3dstats(arr) @@ -1317,7 +1318,8 @@ def _convex_hull( """ points = np.transpose(np.where(arr)) if len(points) <= 3: - print("To few points") if verbose else None + if verbose: + logger.on_warning("To few points") return np.zeros_like(arr, dtype=arr.dtype) hull = scipy.spatial.ConvexHull(points) deln = scipy.spatial.Delaunay(points[hull.vertices]) @@ -1401,7 +1403,6 @@ def infect(x, y, z): infect(*infect_list.pop()) boundary[boundary == 0] = 2 boundary -= 1 - print(boundary.sum()) return boundary @@ -1449,7 +1450,7 @@ def np_betti_numbers(img: np.ndarray, verbose=False) -> tuple[int, int, int]: b2 -= 1 b1 = b0 + b2 - euler_char_num # Euler number = Betti:0 - Betti:1 + Betti:2 if verbose: - print(f"Betti number: b0 = {b0}, b1 = {b1}, b2 = {b2}") + logger.on_neutral(f"Betti number: b0 = {b0}, b1 = {b1}, b2 = {b2}") return b0, b1, b2 diff --git a/TPTBox/core/poi.py b/TPTBox/core/poi.py index 6a3d53a..e5f513c 100755 --- a/TPTBox/core/poi.py +++ b/TPTBox/core/poi.py @@ -962,7 +962,7 @@ def wrap(*args, **kwargs): len_pref = 0 if buffer_file is not None and Path(buffer_file).exists(): assert extend_to is None - print("load") + log.on_neutral("load") extend_to = POI.load(buffer_file) len_pref = len(extend_to) kwargs["extend_to"] = extend_to @@ -1053,7 +1053,7 @@ def calc_poi_from_subreg_vert( # Step 1 get all required locations, crop vert/subreg # Step 2 calc centroids - print("step 2", subreg_id_int) if _print_phases else None + log.on_debug("step 2", subreg_id_int) if _print_phases else None if len(subreg_id_int_phase_1) != 0: arr = vert_msk.get_array() arr[arr >= 100] = 0 @@ -1072,7 +1072,7 @@ def calc_poi_from_subreg_vert( ) [subreg_id_int.remove(i) for i in subreg_id_int_phase_1] # Step 3 Vertebra_Full - print("step 3", subreg_id_int) if _print_phases else None + log.on_debug("step 3", subreg_id_int) if _print_phases else None if Location.Vertebra_Full.value in subreg_id_int: log.print("Calc centroid from subregion id", "Vertebra_Full", verbose=verbose) full = Location.Vertebra_Full.value @@ -1085,7 +1085,7 @@ def calc_poi_from_subreg_vert( extend_to = calc_centroids(vert_msk.set_array(arr), decimals=decimals, second_stage=full, extend_to=extend_to, inplace=True) subreg_id_int.remove(full) # Step 4 IVD / Endplates Superior / Endplate Inferior - print("step 4", subreg_id_int) if _print_phases else None + log.on_debug("step 4", subreg_id_int) if _print_phases else None mapping_vert = { Location.Vertebra_Disc.value: 100, Location.Vertebral_Body_Endplate_Superior.value: 200, @@ -1112,7 +1112,7 @@ def calc_poi_from_subreg_vert( subreg_id_int.remove(loc) # Step 5 call non_centroid_pois # Prepare mask to binary mask - print("step 5", subreg_id_int) if _print_phases else None + log.on_debug("step 5", subreg_id_int) if _print_phases else None vert_arr = vert_msk.get_seg_array() subreg_arr = subreg_msk.get_seg_array() assert subreg_msk.shape == vert_arr.shape, "Shape miss-match" + str(subreg_msk.shape) + str(vert_arr.shape) diff --git a/TPTBox/core/poi_fun/_help.py b/TPTBox/core/poi_fun/_help.py index 334d0f5..75bc038 100644 --- a/TPTBox/core/poi_fun/_help.py +++ b/TPTBox/core/poi_fun/_help.py @@ -9,6 +9,7 @@ from TPTBox.core.compat import zip_strict from TPTBox.core.nii_wrapper import NII from TPTBox.core.vert_constants import Vertebra_Instance +from TPTBox.logger import logger _log = Print_Logger() sacrum_w_o_arcus = (Vertebra_Instance.COCC.value, Vertebra_Instance.S6.value, Vertebra_Instance.S5.value, Vertebra_Instance.S4.value) @@ -106,7 +107,7 @@ def paint_into_NII( assert a is not None direction = np.array(poi[idx, goal]) - np.array(poi[idx, start]) if abs(direction.sum().item()) < 0.000000000001: - print("skip", idx, goal, "-", start) + logger.on_debug("skip", idx, goal, "-", start) continue a = add_ray_to_img(poi[idx, start], direction, a, True, value=199, dilate=2) # type: ignore except KeyError: diff --git a/TPTBox/core/poi_fun/ray_casting.py b/TPTBox/core/poi_fun/ray_casting.py index 08c6140..62df7b4 100644 --- a/TPTBox/core/poi_fun/ray_casting.py +++ b/TPTBox/core/poi_fun/ray_casting.py @@ -11,6 +11,7 @@ from TPTBox.core.poi_fun._help import sacrum_w_o_arcus, to_local_np from TPTBox.core.poi_fun.pixel_based_point_finder import get_direction from TPTBox.core.vert_constants import COORDINATE, DIRECTIONS, Location +from TPTBox.logger import logger from TPTBox.logger.log_file import Logger_Interface _log = Print_Logger() @@ -102,7 +103,7 @@ def max_distance_ray_cast_convex_npfast( y = start_coord[1] + norm_vec[1] * mid z = start_coord[2] + norm_vec[2] * mid val = trilinear_interpolate(region_array, x, y, z) - print(f"Raycast check at distance {mid:.2f}: value={val:.4f}") + logger.on_debug(f"Raycast check at distance {mid:.2f}: value={val:.4f}") if val > 0.5: min_v = mid else: @@ -585,9 +586,10 @@ def calculate_pca_normal_np( pca.fit(points) # First, second, and third principal components if verbose: - print(f"Main Axis (PC1): {pca.components_[0]}") - print(f"Secondary Axis (PC2): {pca.components_[1]}") - print(f"Third Axis (PC3): {pca.components_[2]}") + _log = logger + _log.on_neutral(f"Main Axis (PC1): {pca.components_[0]}") + _log.on_neutral(f"Secondary Axis (PC2): {pca.components_[1]}") + _log.on_neutral(f"Third Axis (PC3): {pca.components_[2]}") normal_vector = pca.components_[pca_component] if zoom is not None: normal_vector = normal_vector / np.array(zoom) diff --git a/TPTBox/core/poi_fun/save_load.py b/TPTBox/core/poi_fun/save_load.py index 966d991..cb2e1e1 100644 --- a/TPTBox/core/poi_fun/save_load.py +++ b/TPTBox/core/poi_fun/save_load.py @@ -416,7 +416,7 @@ def _load_docker_centroids(dict_list: list, centroids: POI_Descriptor, format_: subreg_id = conversion_poi[subreg] centroids[vert_id, subreg_id] = (d["X"], d["Y"], d["Z"]) except Exception: - print(f"Label {d['label']} is not an integer and cannot be converted to an int") + log.on_warning(f"Label {d['label']} is not an integer and cannot be converted to an int") centroids[0, d["label"]] = (d["X"], d["Y"], d["Z"]) else: raise ValueError(d) diff --git a/TPTBox/core/vert_constants.py b/TPTBox/core/vert_constants.py index d7ef0e8..1534b49 100755 --- a/TPTBox/core/vert_constants.py +++ b/TPTBox/core/vert_constants.py @@ -33,10 +33,10 @@ import numpy as np from TPTBox.logger import log_file +from TPTBox.logger.log_file import logger as log # noqa: F401 # backwards-compat alias ROUNDING_LVL = 7 ##################### -log = log_file.Reflection_Logger() logging = Union[bool, log_file.Logger_Interface] # R: Right, L: Left; S: Superior (up), I: Inferior (down); A: Anterior (front), P: Posterior (back) DIRECTIONS = Literal["R", "L", "S", "I", "A", "P"] diff --git a/TPTBox/logger/README.md b/TPTBox/logger/README.md index b6d0b46..51625a0 100644 --- a/TPTBox/logger/README.md +++ b/TPTBox/logger/README.md @@ -1,78 +1,216 @@ # Logger (`TPTBox.logger`) -Structured, consistent logging for long-running medical image processing pipelines. -Provides a simple interface with configurable verbosity, message categories, and output targets. +Consistent, colored, structured logging for TPTBox pipelines. One import for +library code, one function to configure everything. -## Public API +## Quick start ```python -from TPTBox import Logger, Print_Logger, No_Logger, String_Logger, Log_Type +from TPTBox import configure_logging, logger + +configure_logging(file="run.log", verbose=True) +logger.on_ok("segmentation finished") +logger.on_warning("no cord found in sub-007") +logger.on_fail("could not open file") ``` -## Key classes +Terminal output: -| Class | Description | -|---|---| -| `Logger` | Base logger; prints to stdout with optional file output and timestamps | -| `Print_Logger` | Always-verbose logger — prints every message regardless of `verbose` flag | -| `No_Logger` | Silent logger — discards all messages; useful in batch/library code | -| `String_Logger` | Accumulates messages into an in-memory string; useful for testing | -| `Reflection_Logger` | Wraps another logger and mirrors its messages to a second logger | -| `Logger_Interface` | Abstract base class for custom logger implementations | +``` +[+] 2026-07-07 08:11:53 segmentation finished +[?] 2026-07-07 08:11:53 no cord found in sub-007 +[!] 2026-07-07 08:11:53 could not open file +``` -## Log_Type enum +The same lines (ANSI-free) also land in the log file next to `run.log`. -| Member | Meaning | -|---|---| -| `Log_Type.BOLD` | Highlighted/important message | -| `Log_Type.OK` | Success confirmation | -| `Log_Type.WARNING` | Non-fatal warning | -| `Log_Type.FAIL` | Error or failure | -| `Log_Type.TEXT` | Plain informational text | +## Everything in one function + +```python +configure_logging( + file=None, # str | Path — enable file logging + filename=None, # str — stem when `file` is a directory + level=None, # int | str | Log_Type — terminal severity threshold + verbose=None, # bool — default_verbose on the default logger + silent=False, # bool — nothing prints (kill-switch) + capture_warnings=None, # bool — warnings.warn → logger + capture_exceptions=None, # bool — uncaught exceptions → logger + loguru_sink=None, # dict — extra Loguru sink (e.g. structured JSON) +) +``` -## Example +Call with no arguments to inspect the current config: ```python -from TPTBox import Logger, Log_Type +>>> configure_logging() +{'file': PosixPath('.../logs/....log'), 'level': 30, 'verbose': True, + 'capture_warnings': False, 'capture_exceptions': False} +``` -log = Logger(path="run.log", log_filename="pipeline", default_verbose=True) +### Recipes -log.print("Starting segmentation", Log_Type.BOLD) -log.print("Loaded 42 subjects", Log_Type.OK) -log.print("Missing T2w for sub-007", Log_Type.WARNING) +```python +# Terminal + file for everything +configure_logging(file="run.log", verbose=True) -# Suppress all output (e.g. in a library function) -from TPTBox import No_Logger -log = No_Logger() -log.print("This is silently discarded") -``` +# Only warnings/errors on terminal; file still gets everything +configure_logging(file="run.log", verbose=True, level="WARNING") -## Loguru backend +# File-only, no terminal +configure_logging(file="run.log", verbose=False) -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`. +# Silence everything +configure_logging(silent=True) -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. +# Capture warnings and uncaught exceptions +configure_logging(capture_warnings=True, capture_exceptions=True) -### What this enables +# Attach a structured JSON sink alongside the terminal/file output +configure_logging(loguru_sink={"sink": "run.jsonl", "serialize": True, + "level": "TPTBOX_WARNING"}) +``` + +Environment variables (auto-applied on import): + +| Variable | Effect | +|---------------------------------|--------| +| `TPTBOX_LOG_LEVEL=WARNING` | Global terminal severity threshold. | +| `TPTBOX_SILENT=1` | Silence every printout. | +| `TPTBOX_CAPTURE_WARNINGS=1` | `warnings.warn(...)` → logger. | +| `TPTBOX_CAPTURE_EXCEPTIONS=1` | Uncaught exceptions → logger. | +| `TPTBOX_LOGGER_TAKEOVER=0` | Keep Loguru's built-in default stderr handler. | + +## The `verbose` kwargs on TPTBox functions + +Many TPTBox functions accept a `verbose: bool` argument +(`nii.reorient_(..., verbose=True)`, `BIDS_Global_info(..., verbose=True)`, ...). +They control **only** whether *that specific line* reaches the **terminal**. +The **file** sink always records everything (as long as a `Logger` is the +default). + +Combined with `configure_logging`, that gives you two dials: + +| You want … | Set … | +|---------------------------------------------------|------------------------------------------------------| +| Every message to terminal AND file | `configure_logging(file=..., verbose=True)` | +| Only WARNING+ on terminal; file gets everything | `configure_logging(file=..., verbose=True, level="WARNING")` | +| Only *specific* noisy calls on terminal | leave `verbose=False` globally, pass `verbose=True` at the call sites you want to see | +| No terminal at all; file gets everything | `configure_logging(file=..., verbose=False)` | +| No terminal, no file — nothing anywhere | `configure_logging(silent=True)` | + +## Log types + +`Log_Type` — the type marker controls prefix, color, and severity. + +| Log_Type | Prefix | Color | Convenience helper | +|-------------------|--------------|------------------|--------------------| +| `TEXT` | `[*]` | default | `logger.on_text` / `logger.info` | +| `NEUTRAL` | `[ ]` | default | `logger.on_neutral` | +| `SAVE` | `[*]` | cyan | `logger.on_save` | +| `LOG` | `[#]` | blue | `logger.on_log` | +| `OK` | `[+]` | green | `logger.on_ok` | +| `WARNING` | `[?]` | yellow | `logger.on_warning` / `logger.warning` | +| `WARNING_THROW` | `[?]` | yellow (via `warnings.warn`) | `logger.print(..., ltype=Log_Type.WARNING_THROW)` | +| `FAIL` | `[!]` | red | `logger.on_fail` / `logger.error` | +| `STRANGE` | `[-]` | magenta | `logger.on_debug` | +| `BOLD` | `[*]` | bold | `logger.on_bold` | +| `UNDERLINE` | `[_]` | underline | `logger.print(..., ltype=Log_Type.UNDERLINE)` | +| `ITALICS` | `[ ]` | italics | `logger.print(..., ltype=Log_Type.ITALICS)` | +| `Yellow` | `[*]` | yellow | `logger.print(..., ltype=Log_Type.Yellow)` | +| `DOCKER` | `[Docker]` | italics | `logger.print(..., ltype=Log_Type.DOCKER)` | +| `TOTALSEG` | `[TOTALSEG]` | italics | `logger.print(..., ltype=Log_Type.TOTALSEG)` | +| `STAGE` | `[*]` | blue background | `logger.print(..., ltype=Log_Type.STAGE)` | + +`level=` in `configure_logging` accepts: +- integer Loguru severities (`10`, `20`, `25`, `30`, `40`, ...), +- TPTBox names — `"TEXT"`, `"OK"`, `"WARNING"`, `"FAIL"`, `"TPTBOX_OK"`, ..., or a `Log_Type` value, +- standard Loguru names — `"TRACE"`, `"DEBUG"`, `"INFO"`, `"SUCCESS"`, `"WARNING"`, `"ERROR"`, `"CRITICAL"`. + +## For developers adding new code + +Import the module-level proxy and call it. Nothing else. ```python -from TPTBox.logger import loguru_logger, configure, install_excepthook +# my_module.py +from TPTBox.logger import logger + +def compute_something(x, verbose: bool = False): + logger.print("starting compute for", x, verbose=verbose) + if x < 0: + logger.on_warning("negative input, using absolute value") + x = abs(x) + ... + logger.on_ok("compute done") +``` + +Rules of thumb: -# 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 +- Do **not** manually prepend `[!]` / `[?]` / `[*]` to your message. The helpers do it. +- Do **not** import a concrete `Logger` class in library code. Use the proxy. +- Pass `verbose=verbose` on `logger.print(...)` when the caller has explicit + control over terminal noise. Skip the `verbose=` for warnings and failures — + they should always reach the terminal if the level threshold permits. -# 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) +If a user has not called `configure_logging(...)`, the default is a silent +`No_Logger` — so library code stays quiet unless the caller opts in. -# 3) Exception capture: print_error() emits the structured exception to your sinks; -# optionally route *uncaught* exceptions through Loguru too: -install_excepthook() +## Escape hatches + +For fine-grained control beyond `configure_logging`: + +- **Custom Loguru sink**: `from TPTBox import loguru_logger` then + `loguru_logger.add(sink, ...)`. Structured records for exceptions and file + output flow through Loguru; the terminal is written directly to `sys.stdout` + and does not go through Loguru sinks. +- **Direct access to the current logger**: `from TPTBox import get_default_logger`. +- **Private helpers** (rarely needed) live in `TPTBox.logger.log_file` and + `TPTBox.logger._loguru_backend` under underscore-prefixed names + (`_set_default_logger`, `_set_verbosity_level`, `_set_logger_silence`, + `_install_warnings_hook`, `_install_excepthook`, `_capture_exceptions`, ...). + +## Loguru backend details + +- Terminal output is written **directly to `sys.stdout`** with our own ANSI + coloring — it does NOT go through Loguru's global handler routing. That + guarantees TPTBox's terminal output is unaffected by whatever handlers a host + application has active on the Loguru logger. +- File output (from `Logger`) and structured exception records DO go through + Loguru so your custom sinks receive them. Each `Logger` instance's file has + its own Loguru sink with a per-instance filter. +- `configure()` runs eagerly at import. It removes Loguru's built-in default + stderr handler (id 0) so file/exception records don't get printed with the + `time | LEVEL | module:func:line - msg` default format. Sinks a host + application registered *before* importing TPTBox are preserved. Set + `TPTBOX_LOGGER_TAKEOVER=0` before importing TPTBox to keep the default. + +Line format: + +``` +[prefix] YYYY-MM-DD HH:MM:SS body ``` -Thread-safety: Loguru serializes sink writes, so concurrent `print()` calls no longer interleave -mid-line (`enqueue=True` additionally makes a sink thread/process-safe). +The prefix (`[*]`, `[!]`, `[?]`, `[+]`, `[#]`, `[Docker]`, ...) comes first, +then the timestamp, then the message body. No Loguru level name is emitted. +`logger.print()` with no arguments prints exactly one blank line. + +Thread-safety: Loguru serializes file-sink writes; terminal writes acquire a +small lock in `emit_terminal`. `enqueue=True` when constructing `Logger` +additionally makes the sink thread/process-safe. + +## Public API + +```python +from TPTBox import ( + # the proxy — use everywhere in library code + logger, log, + # one-call setup + configure_logging, + # the current default (for advanced inspection) + get_default_logger, + # concrete logger classes + Logger, Print_Logger, No_Logger, String_Logger, Reflection_Logger, + Logger_Interface, Log_Type, + # Loguru escape hatch (add your own sinks) + loguru_logger, +) +``` diff --git a/TPTBox/logger/__init__.py b/TPTBox/logger/__init__.py index 0021100..a51cef1 100755 --- a/TPTBox/logger/__init__.py +++ b/TPTBox/logger/__init__.py @@ -1,9 +1,18 @@ from __future__ import annotations -from ._loguru_backend import configure, install_excepthook +from ._loguru_backend import configure, current_level 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 ( + Logger, + Logger_Interface, + Reflection_Logger, + String_Logger, + configure_logging, + get_default_logger, + log, + logger, +) from .log_file import No_Logger as Print_Logger __all__ = [ @@ -13,7 +22,11 @@ "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 + "configure", + "configure_logging", + "current_level", + "get_default_logger", + "log", + "logger", + "loguru_logger", ] diff --git a/TPTBox/logger/_loguru_backend.py b/TPTBox/logger/_loguru_backend.py index 56ef6a8..321f9e9 100644 --- a/TPTBox/logger/_loguru_backend.py +++ b/TPTBox/logger/_loguru_backend.py @@ -1,68 +1,79 @@ -"""Loguru-backed emission layer for the TPTBox logger. +"""Loguru-backed emission layer. -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. +Terminal output is written directly to ``sys.stdout`` with our own ANSI coloring; +file sinks and structured exception records go through Loguru so users can attach +their own sinks to :data:`loguru_logger`. """ from __future__ import annotations +import datetime import os import sys +import threading from loguru import logger -from TPTBox.logger.log_constants import Log_Type +from TPTBox.logger.log_constants import Log_Type, bcolors, type2bcolors __all__ = [ + "SILENT_LEVEL", + "_install_excepthook", + "_install_warnings_hook", + "_restore_warnings_hook", + "_set_verbosity_level", "add_file_sink", "add_file_stream_sink", "configure", + "current_level", "emit_exception", "emit_file", "emit_terminal", - "install_excepthook", "level_name", "logger", "remove_sink", ] +# Threshold higher than any real level; used by silence(). +SILENT_LEVEL = 1000 + # 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.STRANGE: ("TPTBOX_STRANGE", 10, ""), 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, ""), + Log_Type.SAVE: ("TPTBOX_SAVE", 22, ""), + Log_Type.OK: ("TPTBOX_OK", 25, ""), + Log_Type.WARNING: ("TPTBOX_WARNING", 30, ""), + Log_Type.WARNING_THROW: ("TPTBOX_WARNING_THROW", 30, ""), + Log_Type.FAIL: ("TPTBOX_FAIL", 40, ""), +} + +_LEVEL_ALIASES: dict[str, int] = { + "TRACE": 5, + "DEBUG": 10, + "INFO": 20, + "SUCCESS": 25, + "WARNING": 30, + "ERROR": 40, + "CRITICAL": 50, } -_TERMINAL_FORMAT = "{message}" _FILE_FORMAT = "{message}" _configured = False +_current_level: int = 0 +_stdout_lock = threading.Lock() def level_name(ltype: Log_Type) -> str: @@ -79,58 +90,127 @@ def _ensure_levels() -> None: 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``. +def _resolve_level(level: int | str) -> int: + """Translate an integer/string/Log_Type level to a numeric severity threshold.""" + if isinstance(level, Log_Type): + return _LEVELS[level][1] + if isinstance(level, int): + return level + key = str(level).strip().upper() + if key in _LEVEL_ALIASES: + return _LEVEL_ALIASES[key] + for name, no, _ in _LEVELS.values(): + if name == key or name.removeprefix("TPTBOX_") == key: + return no + raise ValueError(f"Unknown log level: {level!r}") + + +def _set_verbosity_level(level: int | str) -> None: + """Set the global severity threshold for terminal output. - Loguru always appends a ``\n``; we strip it and append the original ``end`` so - ``end="\r"`` progress lines survive unchanged. + Records with severity below this threshold are dropped by :func:`emit_terminal`. + File sinks are not affected — use Loguru's per-sink ``level=`` when adding files. + + Args: + level: Either a numeric severity (e.g. ``20``), a standard Loguru level name + (``"WARNING"``, ``"ERROR"``, ...), or a TPTBox level name (``"TPTBOX_OK"``, + ``"OK"``). """ - 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) + global _current_level # noqa: PLW0603 + _current_level = _resolve_level(level) + + +def current_level() -> int: + """Return the current global terminal severity threshold.""" + return _current_level def configure(take_over: bool | None = None) -> None: - """Configure the global Loguru logger for TPTBox (idempotent). + """Configure the Loguru side of the logger (idempotent). + + Terminal output does NOT go through Loguru; this function only sets up levels and + optionally strips Loguru's default stderr handler so :func:`emit_file` and + :func:`emit_exception` records (which DO go through Loguru) don't get echoed with + the ``time | LEVEL | module:func:line - msg`` default format. 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). + take_over: If True (default), remove Loguru's built-in stderr handler (id 0) + so the default file:function:line format never appears alongside TPTBox + output. Any sinks a host application registered *before* importing TPTBox + are preserved. If False, leave every existing Loguru handler alone. If + None, read ``TPTBOX_LOGGER_TAKEOVER`` (default ``1``). """ global _configured # noqa: PLW0603 _ensure_levels() + env_level = os.environ.get("TPTBOX_LOG_LEVEL") + if env_level: + try: + _set_verbosity_level(env_level) + except ValueError: + pass 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, - ) + # Remove only Loguru's built-in default stderr handler; leave anything a host + # application already attached in place. `remove(0)` raises ValueError if id 0 + # was already removed by the user — treat that as a no-op. + try: + logger.remove(0) + except ValueError: + pass _configured = True + # Env-var opt-ins, applied on first configure() only. + from TPTBox.logger.log_file import _set_logger_silence # lazy: avoids circular import + + for env_name, hook in ( + ("TPTBOX_SILENT", _set_logger_silence), + ("TPTBOX_CAPTURE_WARNINGS", _install_warnings_hook), + ("TPTBOX_CAPTURE_EXCEPTIONS", _install_excepthook), + ): + if _env_truthy(env_name): + hook() + + +def _env_truthy(name: str) -> bool: + """Return True if env var ``name`` is set to a truthy value.""" + return os.environ.get(name, "0") not in ("0", "", "false", "False", "no", "No") + + +def _timestamp() -> str: + """Return the current wall-clock time as ``YYYY-MM-DD HH:MM:SS``.""" + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def format_terminal_line(text: str, ltype: Log_Type = Log_Type.TEXT) -> str: + """Return ``text`` wrapped with the ANSI color of ``ltype`` (no trailing newline). + + Empty ``text`` is returned unchanged so ``logger.print()`` prints a clean blank line. + """ + if text == "": + return "" + color = type2bcolors.get(ltype, (bcolors.ENDC, ""))[0] + if not color or color == bcolors.ENDC: + return text + return f"{color}{text}{bcolors.ENDC}" + 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. + r"""Write one already-prefixed line directly to :data:`sys.stdout`. - ``text`` is passed as the single ``{message}`` value with NO format args, so literal - ``{}``/``<>`` in the message are never interpreted. + Bypasses Loguru's global routing entirely so the terminal output is not affected by + whatever handlers a host application has active. Uses our own ANSI color for + ``ltype`` (matching the legacy ``type2bcolors`` mapping) and honors ``end`` so + ``end="\r"`` progress lines stay on one line. """ - if not _configured: - configure() - logger.bind(tptbox_channel="terminal", tptbox_end=end).log(level_name(ltype), text) + if _LEVELS.get(ltype, _LEVELS[Log_Type.TEXT])[1] < _current_level: + return + line = format_terminal_line(text, ltype) + with _stdout_lock: + # Look up sys.stdout at write-time (not at add-time) so redirect_stdout / capsys work. + sys.stdout.write(line + end) def add_file_sink(filepath, key, *, rotation=None, retention=None, enqueue: bool = False, mode: str = "w") -> int: @@ -216,7 +296,7 @@ def emit_exception(message: str, ltype: Log_Type = Log_Type.FAIL) -> None: logger.opt(exception=True).bind(tptbox_channel="exception").log(level_name(ltype), message) -def install_excepthook() -> None: +def _install_excepthook() -> None: """Route uncaught exceptions through Loguru (opt-in; replaces ``sys.excepthook``).""" def _hook(exc_type, exc_value, exc_tb): @@ -230,8 +310,50 @@ def _hook(exc_type, exc_value, exc_tb): 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. +_original_showwarning = None + + +def _install_warnings_hook() -> None: + """Route ``warnings.warn(...)`` through the TPTBox logger (idempotent). + + Overrides :func:`warnings.showwarning` so every warning is emitted as a WARNING-level + line via the module-wide default logger AND as a structured Loguru record on the + ``"warning"`` channel (for user sinks that ``.add(..., serialize=True)``). + + Call :func:`_restore_warnings_hook` to undo. + """ + global _original_showwarning # noqa: PLW0603 + import warnings + + from TPTBox.logger.log_file import get_default_logger + + if _original_showwarning is not None: + return + _original_showwarning = warnings.showwarning + + def _show(message, category, filename, lineno, file=None, line=None): # noqa: ARG001 + text = f"{category.__name__}: {message} ({filename}:{lineno})" + get_default_logger().on_warning(text) + if _configured: + logger.bind(tptbox_channel="warning").log(level_name(Log_Type.WARNING), text) + + warnings.showwarning = _show + + +def _restore_warnings_hook() -> None: + """Undo :func:`_install_warnings_hook`. No-op if the hook is not installed.""" + global _original_showwarning # noqa: PLW0603 + import warnings + + if _original_showwarning is None: + return + warnings.showwarning = _original_showwarning + _original_showwarning = None + + +# Configure eagerly at import: strips Loguru's built-in default handler (id 0) so +# structured records (`emit_file` / `emit_exception`) don't double-print through +# Loguru's `time | LEVEL | file:func:line - msg` format. Sinks a host application +# added *before* importing TPTBox survive. Set TPTBOX_LOGGER_TAKEOVER=0 before +# importing TPTBox to keep Loguru's default handler in place. configure() diff --git a/TPTBox/logger/log_file.py b/TPTBox/logger/log_file.py index d43939a..c93191c 100755 --- a/TPTBox/logger/log_file.py +++ b/TPTBox/logger/log_file.py @@ -73,6 +73,9 @@ def print( def _preprocess_text(self, text: tuple[str, ...], ltype=Log_Type.TEXT, ignore_prefix: bool = False) -> str: """Processes given text parts, converting manually specified datatypes, and adds the prefix and ltype corresponding color. + Layout: ``[prefix] YYYY-MM-DD HH:MM:SS body`` — the ``[]`` prefix always comes + first, then the timestamp, then the message body. No level name is emitted. + Args: text (tuple[str, ...]): _description_ type (_type_, optional): _description_. Defaults to Log_Type.TEXT. @@ -86,7 +89,7 @@ def _preprocess_text(self, text: tuple[str, ...], ltype=Log_Type.TEXT, ignore_pr if "[" not in string[:3] and not ignore_prefix: prefix = self._get_logger_prefix(ltype) - string = prefix + " " + string + string = prefix + " " + _backend._timestamp() + " " + string return string @@ -632,4 +635,174 @@ def _set_indent(indent_change: int | bool): return indentation_level -log = No_Logger() +_default_logger: Logger_Interface = No_Logger() + + +def _set_default_logger(logger: Logger_Interface) -> None: + """Install a module-wide default logger. Prefer :func:`configure_logging`.""" + global _default_logger # noqa: PLW0603 + _default_logger = logger + + +def get_default_logger() -> Logger_Interface: + """Return the module-wide default logger.""" + return _default_logger + + +def _set_logger_silence() -> None: + """Silence every TPTBox printout. + + default logger becomes a non-verbose ``No_Logger`` and the global terminal severity threshold is raised above every real level. + """ + from TPTBox.logger import _loguru_backend as _backend_mod + + _set_default_logger(No_Logger()) + _default_logger.default_verbose = False # type: ignore[attr-defined] + _backend_mod._set_verbosity_level(_backend_mod.SILENT_LEVEL) + + +class _LoggerProxy: + """Attribute-forwarding proxy: every access resolves to the current default logger. + + Import once at module top-level: + + from TPTBox.logger import logger + + Then use it exactly like a ``Logger``: + + logger.on_ok("done") + logger.on_fail("bad") + logger.print("plain", ltype=Log_Type.LOG) + + ``logger.`` always looks up ```` on whatever :func:`get_default_logger` + currently returns — so :func:`configure_logging` reconfigures every callsite at + once, without library code having to be re-imported or refactored. + """ + + __slots__ = () + + def __getattr__(self, name: str): + return getattr(_default_logger, name) + + def __repr__(self) -> str: + return f"" + + +logger = _LoggerProxy() +# `log` is kept as a backwards-compatible alias for the older `log.print(...)` idiom. +log = logger + + +def _capture_exceptions(fn): + """Decorator: log any exception raised by ``fn`` through the default logger, then re-raise. + + The uncaught exception is captured via :func:`Logger_Interface.print_error`, which + emits a colored traceback to the terminal AND a structured Loguru record so + user-added sinks receive the exception too. + """ + import functools + + @functools.wraps(fn) + def _wrapped(*args, **kwargs): + try: + return fn(*args, **kwargs) + except Exception: + get_default_logger().print_error() + raise + + return _wrapped + + +def configure_logging( + file=None, + filename=None, + level=None, + verbose=None, + silent: bool = False, + capture_warnings: bool | None = None, + capture_exceptions: bool | None = None, + loguru_sink=None, +) -> dict: + """One-call TPTBox logging setup — the single entry point for configuring output. + + Every argument is optional; unset args leave the corresponding aspect untouched. + Call with no arguments to inspect the current effective configuration. + + Args: + file: Path to a log file, or a directory / dataset root (in which case a + timestamped ``__log.log`` is created inside a ``logs/`` + subfolder). ``None`` leaves the current default logger in place. + filename: The log file's stem when ``file`` is a directory / dataset root. + Ignored when ``file`` is a full path or ``None``. + level: Global terminal severity threshold. Accepts an int (loguru severity), + a string (``"WARNING"``, ``"TPTBOX_OK"``, ``"INFO"``, ...), or a + :class:`Log_Type`. + verbose: Sets ``default_verbose`` on the default logger. When True, every + ``logger.print(..., verbose=None)`` call inside TPTBox reaches the terminal; + when False, only file sinks receive them. Independent of ``level``. + silent: Shortcut for ``verbose=False`` plus terminal threshold above every + real level. Nothing prints. + capture_warnings: If True, install a hook so ``warnings.warn(...)`` is routed + through the logger. If False, restore the previous behavior. + capture_exceptions: If True, replace ``sys.excepthook`` so uncaught exceptions + are routed through Loguru. If False, restore ``sys.__excepthook__``. + loguru_sink: Optional dict of kwargs forwarded to ``loguru_logger.add(...)`` — + e.g. ``{"sink": "run.jsonl", "serialize": True, "level": "TPTBOX_WARNING"}``. + + Returns: + A dict with the resulting configuration: ``file``, ``level``, ``verbose``, + ``capture_warnings``, ``capture_exceptions``. + """ + import sys + + from TPTBox.logger import _loguru_backend as _backend_mod + + if file is not None: + _set_default_logger(_build_logger(file, filename)) + + if silent: + _set_logger_silence() + else: + if level is not None: + _backend_mod._set_verbosity_level(level) + if verbose is not None: + _default_logger.default_verbose = verbose # type: ignore[attr-defined] + + if capture_warnings is True: + _backend_mod._install_warnings_hook() + elif capture_warnings is False: + _backend_mod._restore_warnings_hook() + + if capture_exceptions is True: + _backend_mod._install_excepthook() + elif capture_exceptions is False: + sys.excepthook = sys.__excepthook__ + + if loguru_sink is not None: + from TPTBox.logger._loguru_backend import logger as _loguru + + _loguru.add(**loguru_sink) + + return { + "file": getattr(_default_logger, "_filepath", None), + "level": _backend_mod.current_level(), + "verbose": getattr(_default_logger, "default_verbose", None), + "capture_warnings": _backend_mod._original_showwarning is not None, + "capture_exceptions": sys.excepthook is not sys.__excepthook__, + } + + +def _build_logger(file, filename) -> Logger: + """Instantiate a :class:`Logger` from ``file`` (path-or-directory) and ``filename``. + + - ``file`` is a directory: use it as-is as the log dataset root, with ``filename`` as + the log stem (defaults to ``"run"``). + - ``file`` is a file path: use its parent as the root and its stem as the log stem + (``filename`` overrides the stem if provided). + """ + from pathlib import Path + + file = Path(file) + if file.exists() and file.is_dir(): + return Logger(file, filename or "run") + return Logger(file.parent or Path("."), filename or file.stem or "run") diff --git a/TPTBox/mesh3D/mesh.py b/TPTBox/mesh3D/mesh.py index 9bd3e2d..0be6257 100644 --- a/TPTBox/mesh3D/mesh.py +++ b/TPTBox/mesh3D/mesh.py @@ -11,7 +11,7 @@ import pyvista as pv from skimage.measure import marching_cubes -from TPTBox import NII, POI, Image_Reference, Log_Type, to_nii_seg +from TPTBox import NII, POI, Image_Reference, Log_Type, logger, to_nii_seg from TPTBox.core import vert_constants as vc from TPTBox.core.np_utils import np_bbox_binary from TPTBox.core.poi import COORDINATE @@ -122,7 +122,7 @@ def __init__(self, int_arr: np.ndarray | Image_Reference) -> None: # Force dtype to uint if np.issubdtype(int_arr.dtype, np.floating): - print("input is of type float, converting to int") + logger.on_warning("input is of type float, converting to int") int_arr.astype(np.uint16) # calculate bounding box cutout bbox_crop = np_bbox_binary(int_arr, px_dist=2) diff --git a/TPTBox/segmentation/VibeSeg/auto_download.py b/TPTBox/segmentation/VibeSeg/auto_download.py index 2a9bad8..6b6a1f3 100644 --- a/TPTBox/segmentation/VibeSeg/auto_download.py +++ b/TPTBox/segmentation/VibeSeg/auto_download.py @@ -24,6 +24,8 @@ from tqdm import tqdm +from TPTBox.logger import logger + logger = logging.getLogger(__name__) WEIGHTS_URL_ = "https://github.com/robert-graf/VibeSegmentator/releases/download/v1.0.0/" env_name = "VIBESEG_WEIGHTS_PATH" @@ -96,9 +98,9 @@ def _download(weights_url: str, weights_dir: Path, text: str = "", is_zip: bool with urllib.request.urlopen(str(weights_url)) as response: file_size = int(response.info().get("Content-Length", -1)) except Exception: - print("Download attempt failed:", weights_url) + logger.on_fail("Download attempt failed:", weights_url) return - print(f"Downloading {text}...") + logger.on_neutral(f"Downloading {text}...") with tqdm(total=file_size, unit="B", unit_scale=True, unit_divisor=1024, desc=Path(weights_url).name) as pbar: @@ -111,7 +113,7 @@ def update_progress(block_num: int, block_size: int, total_size: int) -> None: # Download the file urllib.request.urlretrieve(str(weights_url), zip_path, reporthook=update_progress) if is_zip: - print(f"Extracting {text}...") + logger.on_neutral(f"Extracting {text}...") with zipfile.ZipFile(zip_path, "r") as zip_ref: zip_ref.extractall(weights_dir) os.remove(zip_path) # noqa: PTH107 diff --git a/TPTBox/segmentation/spineps.py b/TPTBox/segmentation/spineps.py index 026b763..f949321 100644 --- a/TPTBox/segmentation/spineps.py +++ b/TPTBox/segmentation/spineps.py @@ -5,6 +5,7 @@ from typing import Literal from TPTBox import BIDS_FILE, NII, Print_Logger +from TPTBox.logger import logger logger = Print_Logger() @@ -169,9 +170,10 @@ def _run_spineps_all(nii_dataset: Path | str) -> None: try: # Execute the command and capture output result = subprocess.run(command, check=True, capture_output=True, text=True) - print(f"Model {model_semantic} processing complete:\n", result.stdout) + _logger = logger + _logger.on_ok(f"Model {model_semantic} processing complete:\n", result.stdout) except subprocess.CalledProcessError as e: - print(f"Error during processing {model_semantic}:\n", e.stderr) + logger.on_fail(f"Error during processing {model_semantic}:\n", e.stderr) def _run_spineps_internal( diff --git a/TPTBox/spine/snapshot2D/snapshot_modular.py b/TPTBox/spine/snapshot2D/snapshot_modular.py index 7c31f06..ff3c3ac 100755 --- a/TPTBox/spine/snapshot2D/snapshot_modular.py +++ b/TPTBox/spine/snapshot2D/snapshot_modular.py @@ -25,6 +25,7 @@ Location, POI_Reference, calc_centroids, + logger, to_nii, to_nii_optional, v_idx2name, @@ -395,7 +396,7 @@ def curve_projected_mip( try: thicke = [int(i // y_zoom) + int(i % y_zoom > 0) for i in thick] except Exception: - print("thick infinity bug", y_zoom, thick_t, thick) + logger.on_fail("thick infinity bug", y_zoom, thick_t, thick) thicke = (*thick_t,) thick = thicke y_post_rel_to_border = y_ref + int(0.4 * (shp[1] - 1 - y_ref)) # one-third distance to border @@ -493,7 +494,7 @@ def curve_projection_axial_fallback( axis=0, ) except Exception as e: - print(e) + logger.on_warning(e) axl_plane = np.zeros((1, 1)) return axl_plane @@ -689,7 +690,7 @@ def plot_sag_centroids( zorder=2, ) except Exception as e: - print(e) + logger.on_warning(e) if "line_segments_sag" in ctd.info: for color, x, (c, d) in ctd.info["line_segments_sag"]: if len(x) == 2: @@ -1050,7 +1051,7 @@ def to_cdt(ctd_bids: POI_Reference | None) -> POI | None: ctd = POI.load(ctd_bids, allow_global=True) if len(ctd) > 0: # handle case if empty centroid file given return ctd - print("[!][snp] To few centroids", ctd) + logger.on_fail("[snp] To few centroids", ctd) return None @@ -1169,7 +1170,7 @@ def create_snapshot( # noqa: C901 else: ctd_tmp = ctd except Exception: - print("did not manage to calc ctd_tmp\n", frame) + logger.on_fail("did not manage to calc ctd_tmp\n", frame) raise if frame.curve_location is None: if frame.show_these_subreg_poi is not None: @@ -1208,7 +1209,7 @@ def create_snapshot( # noqa: C901 else: sag_seg, cor_seg, axl_seg = (None, None, None) except Exception: - print(frame) + logger.on_fail(frame) raise # Conversion to 2D image done, now normalization try: @@ -1219,7 +1220,8 @@ def create_snapshot( # noqa: C901 max_cor = np.percentile(cor_img[cor_img != 0], 99) except Exception: max_cor = 1 - print("max sag/cor", max_sag, max_cor) if verbose else None + if verbose: + logger.on_neutral("max sag/cor", max_sag, max_cor) ##MRT## if frame.mode == "MRI": max_intens = max(max_sag, max_cor) # type: ignore @@ -1368,7 +1370,8 @@ def create_snapshot( # noqa: C901 snp_path = [str(snp_path)] for path in snp_path: fig.savefig(str(path)) - print("[*] Snapshot saved:", path) if verbose else None + if verbose: + logger.on_save("Snapshot saved:", path) plt.close() return snp_path diff --git a/TPTBox/stitching/stitching.py b/TPTBox/stitching/stitching.py index affb85d..806bfd8 100755 --- a/TPTBox/stitching/stitching.py +++ b/TPTBox/stitching/stitching.py @@ -12,6 +12,8 @@ from scipy.spatial import ConvexHull from skimage.exposure import match_histograms +from TPTBox.logger import logger + def get_rotation_and_spacing_from_affine(affine: np.ndarray) -> tuple[np.ndarray, np.ndarray]: """Decompose a NIfTI affine into its rotation matrix and voxel spacing. @@ -190,11 +192,15 @@ def get_max_affine_and_shape( new_spacing = np.maximum(min_spacing, new_spacing) shape: np.ndarray = np.ceil(min_shape / new_spacing) - print("Choose the following spacing:", new_spacing) if verbose else None - print(f"Output shape is {shape}, which utilizes {min_possible_volume / min_volume * 100:.1f} % of all voxels.") if verbose else None + if verbose: + _log = logger + _log.on_neutral("Choose the following spacing:", new_spacing) + _log.on_neutral(f"Output shape is {shape}, which utilizes {min_possible_volume / min_volume * 100:.1f} % of all voxels.") affine = get_ras_affine(min_rotation, new_spacing, origen[0]) - print("The new origin is ", np.round(affine[:3, 3], 2)) if verbose else None - print("The optimal rotation came from file number ", opt_id, " ", np.round(min_rotation.reshape(-1), 2)) if verbose else None + if verbose: + _log = logger + _log.on_neutral("The new origin is ", np.round(affine[:3, 3], 2)) + _log.on_neutral("The optimal rotation came from file number ", opt_id, " ", np.round(min_rotation.reshape(-1), 2)) return nib.Nifti1Image(np.zeros(shape.astype(int), dtype=dtype), affine) # type: ignore @@ -332,7 +338,7 @@ def from_nibabel(nib_image): ndim = nib_image.ndim if ndim < 3: - print("Dimensionality is less than 3.") + logger.on_fail("Dimensionality is less than 3.") return None q_form = nib_image.get_qform() @@ -506,16 +512,19 @@ def main( # noqa: C901 min_value = 0 match_histogram = False histogram = None + _log = logger if len(images) == 0 or len(images) == 1: - print("!!! Need at least two images (-i ...nii.gz ...nii.gz) to stitch!!!\n Got " + str(images)) + _log.on_fail("Need at least two images (-i ...nii.gz ...nii.gz) to stitch! Got " + str(images)) return None, None corners = [] affines = [] niis: list[nib.nifti1.Nifti1Image] = [] - print("### loading ###") if verbose else None + if verbose: + _log.on_neutral("### loading ###") for f_name in images: if isinstance(f_name, (Path, str)): - print("Load ", f_name, Path(f_name)) if verbose else None + if verbose: + _log.on_neutral("Load ", f_name, Path(f_name)) # Load Nii nii: nib.nifti1.Nifti1Image = nib.load(f_name) # type: ignore @@ -529,13 +538,16 @@ def main( # noqa: C901 if len(niis) == 0: reference = None else: - print("Histogram equalization with previous file") if verbose else None + if verbose: + _log.on_neutral("Histogram equalization with previous file") reference = get_array(niis[-1]) elif histogram.isdigit(): - print("Histogram equalization", images[int(histogram)]) if verbose else None + if verbose: + _log.on_neutral("Histogram equalization", images[int(histogram)]) reference = buffer_reference(images[int(histogram)], bias_field=bias_field, crop=crop_to_bias_field) # type: ignore else: - print("Histogram equalization with file", histogram) if verbose else None + if verbose: + _log.on_neutral("Histogram equalization with file", histogram) reference = buffer_reference(histogram, bias_field=bias_field, crop=crop_to_bias_field) # type: ignore if reference is not None: image = get_array(nii) @@ -555,7 +567,8 @@ def main( # noqa: C901 corners_current = np.concatenate(corners, axis=0) # compute output shape and affine - print("### compute output shape and affine ###") if verbose else None + if verbose: + _log.on_neutral("### compute output shape and affine ###") if is_segmentation: max_value = max([x.get_fdata().max() for x in niis]) if max_value < 256: @@ -573,9 +586,11 @@ def main( # noqa: C901 target_list = [] occupancy_list = [] # get resampled arrays and occupancy - print("### resample to new space ###") if verbose else None + if verbose: + _log.on_neutral("### resample to new space ###") for i, nii in enumerate(niis, 1): - print(f"{i:2}/{len(niis):2} resampled", end="\r") if verbose else None + if verbose: + _log.on_neutral(f"{i:2}/{len(niis):2} resampled", end="\r", ignore_prefix=True) nii_new = nip.resample_from_to(nii, nii_out, 0 if is_segmentation else 3, mode="constant", cval=min_value) arr_new = get_array(nii_new) target_list.append(arr_new) @@ -588,11 +603,13 @@ def main( # noqa: C901 else: occupancy_list.append(get_array(b).astype(np.float32)) - print("\n### ramp stitching ###") if verbose else None + if verbose: + _log.on_neutral("\n### ramp stitching ###") # ramp stitching combinations = list(itertools.combinations(range(len(target_list)), 2)) for idx, item in enumerate(combinations, 1): - print(f"{idx:2}/{len(combinations):2} ramp stitching", end="\r") if verbose else None + if verbose: + _log.on_neutral(f"{idx:2}/{len(combinations):2} ramp stitching", end="\r", ignore_prefix=True) # TODO fix intersection with more than two occupancies arr_1_full = occupancy_list[item[0]] arr_2_full = occupancy_list[item[1]] @@ -639,9 +656,8 @@ def main( # noqa: C901 if kick_out_fully_integrated_images: images.pop(item[1]) if (arr_1_full.max() != 1 or arr_2_full.max() != 1) and kick_out_fully_integrated_images: - print("kick_out_fully_integrated_images") - - print(images) + _log.on_warning("kick_out_fully_integrated_images") + _log.on_neutral(images) return main( images, output, @@ -672,7 +688,8 @@ def main( # noqa: C901 target_arr = target_arr.astype(dtype2) target_arr = target_arr.sum(0) target_arr[target_arr <= min_value] = min_value - print("\n### Save ###") if verbose else None + if verbose: + _log.on_neutral("\n### Save ###") if output is not None: output = str(output) if not output.endswith(".nii.gz"): @@ -695,7 +712,8 @@ def main( # noqa: C901 if save: nib.save(nii_out, output) # type: ignore - print("Saved ", output) if verbose else None + if verbose: + _log.on_save("Saved ", output) if store_ramp: occupancy_arr = np.stack(occupancy_list, -1) @@ -707,9 +725,11 @@ def main( # noqa: C901 output = output.replace(".nii.gz", "_ramps.nii.gz").replace("_msk_", "_") if ramp_path is None else ramp_path if save: nib.save(nii_occ, output) # type: ignore - print("Saved ", output) if verbose else None + if verbose: + _log.on_save("Saved ", output) return nii_out, nii_occ - print("\n### Finished ###") if verbose else None + if verbose: + _log.on_neutral("\n### Finished ###") return nii_out, None diff --git a/unit_tests/test_logger.py b/unit_tests/test_logger.py index 4a86f6c..4cf5a64 100644 --- a/unit_tests/test_logger.py +++ b/unit_tests/test_logger.py @@ -20,7 +20,7 @@ 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_constants import type2bcolors from TPTBox.logger.log_file import Logger, No_Logger @@ -37,15 +37,26 @@ def _norm(s: str) -> str: return s.removeprefix("\x1b[0m") +_TS_RE = r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}" + + class TestLoggerColorContract(unittest.TestCase): def test_each_log_type_renders_like_type2bcolors(self): + # Format: `[prefix] YYYY-MM-DD HH:MM:SS body\n`. `[]` comes first, + # then the timestamp, no level name. Compare against a regex per Log_Type. 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}") + color = re.escape(type2bcolors[lt][0]) + prefix = re.escape(type2bcolors[lt][1]) + if color and type2bcolors[lt][0] != "\x1b[0m": + pattern = rf"^{color}{prefix} {_TS_RE} Hello World\x1b\[0m\n$" + else: + # TEXT/NEUTRAL currently map to the reset code; the emitter skips + # coloring so no ANSI is present. + pattern = rf"^{prefix} {_TS_RE} Hello World\n$" + self.assertRegex(got, pattern, f"color/format changed for {lt.name}") def test_exact_color_code_present(self): for lt in Log_Type: @@ -62,6 +73,11 @@ def test_end_carriage_return_preserved(self): def test_verbose_false_suppresses_terminal(self): self.assertEqual(_cap(lambda: No_Logger().print("hidden", ltype=Log_Type.SAVE, verbose=False)), "") + def test_empty_print_emits_only_newline(self): + # `logger.print()` with no args must produce exactly one blank line — no prefix, + # no timestamp, no color codes. + self.assertEqual(_cap(lambda: No_Logger().print()), "\n") + def test_warning_throw_emits_warning_not_stdout(self): with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") @@ -70,19 +86,19 @@ def test_warning_throw_emits_warning_not_stdout(self): 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"))) + self.assertRegex(_cap(lambda: No_Logger().on_ok("ok")), rf"\x1b\[92m\[\+\] {_TS_RE} ok") + self.assertRegex(_cap(lambda: No_Logger().on_fail("bad")), rf"\x1b\[91m\[!\] {_TS_RE} bad") def with_prefix(): lg = No_Logger() lg.prefix = "API" lg.print("hi", ltype=Log_Type.SAVE) - self.assertIn("[API] hi", _cap(with_prefix)) + self.assertRegex(_cap(with_prefix), rf"\[API\] {_TS_RE} hi") 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.assertRegex(got, rf"\[\*\] {_TS_RE} Saved: /p/f 42") self.assertTrue(got.startswith(type2bcolors[Log_Type.SAVE][0])) @@ -98,8 +114,10 @@ def test_file_is_ansi_free_and_well_formed(self): 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) + # Each line begins with `[prefix] YYYY-MM-DD HH:MM:SS ` — assert the pattern + # per needle rather than the exact literal. + for prefix, body in [("#", "Log started at:"), (r"\*", "Saved /p/f.nii.gz"), ("!", "oops"), ("#", "Program duration:")]: + self.assertRegex(content, rf"\[{prefix}\] {_TS_RE} {re.escape(body)}") def test_two_loggers_isolated(self): d = tempfile.mkdtemp() @@ -113,6 +131,175 @@ def test_two_loggers_isolated(self): self.assertNotIn("only-B", ca) +class TestLoggerProxyAndSilence(unittest.TestCase): + def test_logger_proxy_forwards_to_current_default(self): + from TPTBox.logger import get_default_logger, logger + from TPTBox.logger.log_file import String_Logger, _set_default_logger + + previous = get_default_logger() + buf = String_Logger() + _set_default_logger(buf) + try: + logger.on_ok("hello") + finally: + _set_default_logger(previous) + self.assertIn("hello", buf.log_content) + + def test_log_is_alias_of_logger(self): + from TPTBox.logger import log, logger + + self.assertIs(log, logger) + + def test_vert_constants_log_is_proxy(self): + from TPTBox.core.vert_constants import log as vc_log + from TPTBox.logger import logger + + self.assertIs(vc_log, logger) + + def test_configure_logging_silent_kill_switch(self): + from TPTBox.logger import configure_logging, get_default_logger + from TPTBox.logger._loguru_backend import _set_verbosity_level + from TPTBox.logger.log_file import No_Logger, _set_default_logger + + previous = get_default_logger() + configure_logging(silent=True) + try: + for method in ("on_text", "on_ok", "on_fail", "on_warning", "on_neutral"): + self.assertEqual(_cap(lambda m=method: getattr(No_Logger(), m)("x")), "", f"{method} not silenced") + finally: + _set_default_logger(previous) + _set_verbosity_level(0) + + def test_configure_logging_warnings_capture_round_trip(self): + import warnings + + from TPTBox.logger import configure_logging, get_default_logger + from TPTBox.logger.log_file import String_Logger, _set_default_logger + + previous = get_default_logger() + buf = String_Logger() + _set_default_logger(buf) + configure_logging(capture_warnings=True) + try: + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn("captured!", UserWarning, stacklevel=1) + finally: + configure_logging(capture_warnings=False) + _set_default_logger(previous) + self.assertIn("UserWarning: captured!", buf.log_content) + + def test_capture_exceptions_logs_and_reraises(self): + from TPTBox.logger import get_default_logger + from TPTBox.logger.log_file import String_Logger, _capture_exceptions, _set_default_logger + + previous = get_default_logger() + buf = String_Logger() + _set_default_logger(buf) + + @_capture_exceptions + def boom(): + raise ValueError("kaboom") + + try: + with self.assertRaises(ValueError): + boom() + finally: + _set_default_logger(previous) + self.assertIn("ValueError", buf.log_content) + self.assertIn("kaboom", buf.log_content) + + +class TestConfigureLogging(unittest.TestCase): + def test_configure_logging_end_to_end(self): + import tempfile + from pathlib import Path + + from TPTBox.logger import configure_logging, get_default_logger, logger + from TPTBox.logger._loguru_backend import _set_verbosity_level + from TPTBox.logger.log_file import _set_default_logger + + previous = get_default_logger() + d = tempfile.mkdtemp() + cfg = configure_logging(file=d, filename="pipeline", level="WARNING", verbose=True) + try: + self.assertEqual(cfg["level"], 30) + self.assertTrue(cfg["verbose"]) + self.assertIsNotNone(cfg["file"]) + + # Level threshold gates terminal + self.assertEqual(_cap(lambda: logger.on_ok("hidden")), "") + self.assertNotEqual(_cap(lambda: logger.on_warning("visible")), "") + + # File captures both regardless + get_default_logger().close() + files = list(Path(d).glob("logs/*.log")) + self.assertEqual(len(files), 1) + content = files[0].read_text() + self.assertIn("hidden", content) + self.assertIn("visible", content) + finally: + _set_default_logger(previous) + _set_verbosity_level(0) + + def test_configure_logging_no_args_returns_current_config(self): + from TPTBox.logger import configure_logging + + cfg = configure_logging() + self.assertIn("level", cfg) + self.assertIn("verbose", cfg) + self.assertIn("capture_warnings", cfg) + self.assertIn("capture_exceptions", cfg) + + def test_readme_symbols_all_importable(self): + """Guard against fabricated names in the README code fences.""" + import importlib + + tp = importlib.import_module("TPTBox") + for name in ( + "logger", + "log", + "configure_logging", + "get_default_logger", + "loguru_logger", + "Logger", + "Print_Logger", + "No_Logger", + "String_Logger", + "Reflection_Logger", + "Logger_Interface", + "Log_Type", + ): + self.assertTrue(hasattr(tp, name), f"README cites `TPTBox.{name}` but it isn't exported") + + +class TestGlobalLevelAndDefaultLogger(unittest.TestCase): + def test_configure_logging_level_filters_below_threshold(self): + from TPTBox.logger import configure_logging + from TPTBox.logger._loguru_backend import _set_verbosity_level + + configure_logging(level="WARNING") # 30 — drops TEXT (20), OK (25) + try: + self.assertEqual(_cap(lambda: No_Logger().on_text("hidden")), "") + self.assertEqual(_cap(lambda: No_Logger().on_ok("hidden")), "") + self.assertNotEqual(_cap(lambda: No_Logger().on_warning("visible")), "") + self.assertNotEqual(_cap(lambda: No_Logger().on_fail("visible")), "") + finally: + _set_verbosity_level(0) + + def test_default_logger_roundtrip(self): + from TPTBox.logger import get_default_logger + from TPTBox.logger.log_file import String_Logger, _set_default_logger + + previous = get_default_logger() + buf = String_Logger() + _set_default_logger(buf) + try: + self.assertIs(get_default_logger(), buf) + finally: + _set_default_logger(previous) + + class TestExceptionCapture(unittest.TestCase): def test_print_error_text_and_structured_record(self): import json