"""
Logging system for the thesis framework.
This module provides a centralized logging configuration using loguru.
It supports both console and file output with rotation, colored output,
and structured logging for medical imaging processing pipelines.
Example:
>>> from thesis.core.logging import get_logger
>>> logger = get_logger(__name__)
>>> logger.info("Processing started")
>>> logger.debug(f"Image shape: {shape}")
"""
import logging as _stdlib_logging
import sys
import threading
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator, Optional, Union
from loguru import logger
from .formatters import get_console_format, get_file_format
from .handlers import InterceptHandler
__all__ = [
"get_logger",
"setup_logging",
"reset_logging",
"set_console_level",
"logger",
"InterceptHandler",
"suppress_nipype_native_logging",
"suspend_console_logging",
]
[docs]
def suppress_nipype_native_logging() -> None:
"""Remove nipype's own StreamHandlers to prevent duplicate log output.
Nipype attaches StreamHandlers directly to named loggers (nipype.workflow,
nipype.pipeline.*, etc.) that output in nipype's native format. Since the
root InterceptHandler already routes those records to loguru, the native
handlers produce duplicate lines. Call this once after nipype is imported.
"""
nipype_root = _stdlib_logging.getLogger("nipype")
nipype_root.handlers = []
nipype_root.propagate = True
for name, log in _stdlib_logging.root.manager.loggerDict.items():
if not isinstance(log, _stdlib_logging.Logger):
continue
if name.startswith("nipype"):
log.handlers = []
log.propagate = True
# Global flag to track if logging has been initialized
_logging_initialized = False
_logging_lock = threading.Lock()
[docs]
def setup_logging(
log_dir: Optional[Union[str, Path]] = None,
log_level: str = "INFO",
console_output: bool = True,
file_output: bool = True,
rotation: str = "10 MB",
retention: str = "7 days",
compression: str = "zip",
) -> None:
"""
Configure the logging system for the application.
Args:
log_dir: Directory for log files. If None, uses './logs'
log_level: Minimum log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
console_output: Whether to output logs to console
file_output: Whether to output logs to file
rotation: When to rotate log files (size or time-based)
retention: How long to keep old log files
compression: Compression format for rotated logs
Example:
>>> setup_logging(log_dir="./logs", log_level="DEBUG")
>>> logger.info("Logging configured")
"""
global _logging_initialized
with _logging_lock:
if _logging_initialized:
logger.warning("Logging already initialized, skipping setup")
return
# Remove default handler
logger.remove()
# Redirect all stdlib logging (used by Nipype, nipype.interfaces.fsl, etc.) to loguru
_stdlib_logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
# Add console handler if requested.
# Route through _ConsoleCoordinator so log lines don't
# collide with active progress bars (tqdm / spinner).
global _console_handler_id
if console_output:
_console_handler_id = logger.add(
_coordinated_stderr_sink,
format=get_console_format(),
level=log_level,
colorize=True,
backtrace=True,
diagnose=True,
)
# Add file handler if requested
if file_output:
if log_dir is None:
log_dir = Path("./logs")
else:
log_dir = Path(log_dir)
log_dir.mkdir(parents=True, exist_ok=True)
# Main application log
logger.add(
log_dir / "thesis_{time:YYYY-MM-DD}.log",
format=get_file_format(),
level=log_level,
rotation=rotation,
retention=retention,
compression=compression,
backtrace=True,
diagnose=True,
enqueue=True, # Thread-safe
)
# Error-only log
logger.add(
log_dir / "errors_{time:YYYY-MM-DD}.log",
format=get_file_format(),
level="ERROR",
rotation=rotation,
retention=retention,
compression=compression,
backtrace=True,
diagnose=True,
enqueue=True,
)
_logging_initialized = True
logger.info(f"Logging system initialized (level={log_level})")
[docs]
def get_logger(name: Optional[str] = None) -> Any:
"""
Get a logger instance for a module.
This function returns the global loguru logger with context about
the calling module. The logger is automatically configured on first use.
Args:
name: Name of the module/component (typically __name__)
Returns:
Logger instance with module context
Example:
>>> logger = get_logger(__name__)
>>> logger.info("Processing image")
"""
# Return logger with module context
if name:
return logger.bind(module=name)
return logger
_console_handler_id: Optional[int] = None
_console_level_name: str = "INFO"
def _coordinated_stderr_sink(message: str) -> None:
"""Loguru sink that writes directly to stderr.
Click's progressbar handles output coordination internally,
so we can write log messages directly without additional
synchronization.
"""
sys.stderr.write(str(message))
sys.stderr.flush()
[docs]
@contextmanager
def suspend_console_logging() -> Iterator[None]:
"""Temporarily disable console logging while progress UI is active."""
global _console_handler_id
if _console_handler_id is None:
yield
return
handler_id = _console_handler_id
logger.remove(handler_id)
_console_handler_id = None
try:
yield
finally:
_console_handler_id = logger.add(
_coordinated_stderr_sink,
format=get_console_format(),
level=_console_level_name,
colorize=True,
backtrace=True,
diagnose=True,
)
[docs]
def set_console_level(level: str) -> None:
"""Change the console handler's minimum log level at runtime.
This is used by the CLI output system to suppress verbose loguru
messages when the user has not requested ``-v`` / ``--verbose``.
File handlers are unaffected so the full log is always available.
Args:
level: New minimum level (e.g. ``"WARNING"``, ``"DEBUG"``).
"""
global _console_handler_id, _console_level_name
_console_level_name = level.upper()
if _console_handler_id is not None:
logger.remove(_console_handler_id)
_console_handler_id = logger.add(
_coordinated_stderr_sink,
format=get_console_format(),
level=_console_level_name,
colorize=True,
backtrace=True,
diagnose=True,
)
[docs]
def reset_logging() -> None:
"""
Reset the logging system (primarily for testing).
Removes all handlers and resets the initialization flag.
"""
global _logging_initialized, _console_handler_id, _console_level_name
logger.remove()
_logging_initialized = False
_console_handler_id = None
_console_level_name = "INFO"
