Source code for pyavs.utils.logging

"""
Centralized logging configuration for pyAVS package.

This module provides a unified logging system for all pyAVS components,
allowing for consistent log formatting, levels, and output handling.
"""

import logging
import sys
from pathlib import Path
from typing import Optional, Union, Dict, Any
import datetime


[docs] class ColoredFormatter(logging.Formatter): """Colored formatter for console output.""" # ANSI color codes COLORS = { 'DEBUG': '\033[36m', # Cyan 'INFO': '\033[32m', # Green 'WARNING': '\033[33m', # Yellow 'ERROR': '\033[31m', # Red 'CRITICAL': '\033[35m', # Magenta 'RESET': '\033[0m' # Reset }
[docs] def format(self, record): # Add color to levelname levelname = record.levelname if levelname in self.COLORS: record.levelname = f"{self.COLORS[levelname]}{levelname}{self.COLORS['RESET']}" return super().format(record)
[docs] class PyAVSLogger: """ Centralized logger for pyAVS package. Provides consistent logging across all modules with configurable output levels, formats, and destinations. """ _instance = None _loggers: Dict[str, logging.Logger] = {} _configured = False def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance
[docs] @classmethod def configure(cls, level: Union[str, int] = 'INFO', console: bool = True, file_path: Optional[Union[str, Path]] = None, format_string: Optional[str] = None, date_format: Optional[str] = None, use_colors: bool = True, max_file_size: int = 10 * 1024 * 1024, # 10MB backup_count: int = 5) -> None: """ Configure the global logging system for pyAVS. Parameters ---------- level : str or int, optional Logging level ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL') (default: 'INFO') console : bool, optional Whether to output to console (default: True) file_path : str or Path, optional Path to log file. If None, no file logging (default: None) format_string : str, optional Custom format string for log messages date_format : str, optional Custom date format for timestamps use_colors : bool, optional Whether to use colored output in console (default: True) max_file_size : int, optional Maximum log file size in bytes before rotation (default: 10MB) backup_count : int, optional Number of backup files to keep during rotation (default: 5) """ if cls._configured: cls.get_logger('pyavs').warning("Logging already configured. Reconfiguring...") # Convert string level to logging constant if isinstance(level, str): level = getattr(logging, level.upper()) # Default formats if format_string is None: format_string = '[%(asctime)s] %(name)s - %(levelname)s - %(message)s' if date_format is None: date_format = '%Y-%m-%d %H:%M:%S' # Configure root logger for pyavs root_logger = logging.getLogger('pyavs') root_logger.setLevel(level) # Remove existing handlers to avoid duplicates for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) # Console handler if console: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(level) if use_colors and sys.stdout.isatty(): # Only use colors if terminal supports it console_formatter = ColoredFormatter(format_string, date_format) else: console_formatter = logging.Formatter(format_string, date_format) console_handler.setFormatter(console_formatter) root_logger.addHandler(console_handler) # File handler with rotation if file_path is not None: from logging.handlers import RotatingFileHandler file_path = Path(file_path) file_path.parent.mkdir(parents=True, exist_ok=True) file_handler = RotatingFileHandler( file_path, maxBytes=max_file_size, backupCount=backup_count ) file_handler.setLevel(level) file_formatter = logging.Formatter(format_string, date_format) file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) # Prevent propagation to root logger to avoid duplicate messages root_logger.propagate = False cls._configured = True # Log configuration success logger = cls.get_logger('pyavs.logging') logger.info(f"Logging configured: level={logging.getLevelName(level)}, " f"console={console}, file={file_path is not None}")
[docs] @classmethod def get_logger(cls, name: str) -> logging.Logger: """ Get a logger instance for a specific module. Parameters ---------- name : str Logger name, typically the module name Returns ------- logging.Logger Logger instance """ if not cls._configured: # Configure with defaults if not already configured cls.configure() if name not in cls._loggers: # Ensure name starts with 'pyavs.' if not name.startswith('pyavs'): name = f'pyavs.{name}' logger = logging.getLogger(name) cls._loggers[name] = logger return cls._loggers[name]
[docs] @classmethod def set_level(cls, level: Union[str, int], logger_name: Optional[str] = None) -> None: """ Set logging level for specific logger or all loggers. Parameters ---------- level : str or int Logging level logger_name : str, optional Specific logger name. If None, applies to root pyavs logger """ if isinstance(level, str): level = getattr(logging, level.upper()) if logger_name is None: # Apply to root logger and all handlers root_logger = logging.getLogger('pyavs') root_logger.setLevel(level) for handler in root_logger.handlers: handler.setLevel(level) else: logger = cls.get_logger(logger_name) logger.setLevel(level)
[docs] @classmethod def add_file_handler(cls, file_path: Union[str, Path], level: Union[str, int] = 'INFO') -> None: """ Add an additional file handler to the logging system. Parameters ---------- file_path : str or Path Path to additional log file level : str or int, optional Logging level for this handler (default: 'INFO') """ if isinstance(level, str): level = getattr(logging, level.upper()) file_path = Path(file_path) file_path.parent.mkdir(parents=True, exist_ok=True) root_logger = logging.getLogger('pyavs') file_handler = logging.FileHandler(file_path) file_handler.setLevel(level) formatter = logging.Formatter( '[%(asctime)s] %(name)s - %(levelname)s - %(message)s', '%Y-%m-%d %H:%M:%S' ) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler)
# Convenience functions for common logging tasks
[docs] def get_logger(name: str) -> logging.Logger: """ Get a logger instance (convenience function). Parameters ---------- name : str Logger name Returns ------- logging.Logger Logger instance """ return PyAVSLogger.get_logger(name)
[docs] def configure_logging(**kwargs) -> None: """ Configure logging system (convenience function). Parameters ---------- **kwargs Arguments passed to PyAVSLogger.configure() """ PyAVSLogger.configure(**kwargs)
[docs] def set_log_level(level: Union[str, int], logger_name: Optional[str] = None) -> None: """ Set logging level (convenience function). Parameters ---------- level : str or int Logging level logger_name : str, optional Specific logger name """ PyAVSLogger.set_level(level, logger_name)
# Context manager for temporary log level changes
[docs] class temporary_log_level: """ Context manager to temporarily change log level. Usage: with temporary_log_level('DEBUG'): # Debug logging enabled logger.debug("This will be shown") """
[docs] def __init__(self, level: Union[str, int], logger_name: Optional[str] = None): self.new_level = level self.logger_name = logger_name self.original_level = None
def __enter__(self): if isinstance(self.new_level, str): self.new_level = getattr(logging, self.new_level.upper()) if self.logger_name is None: logger = logging.getLogger('pyavs') else: logger = PyAVSLogger.get_logger(self.logger_name) self.original_level = logger.level logger.setLevel(self.new_level) return self def __exit__(self, exc_type, exc_val, exc_tb): if self.logger_name is None: logger = logging.getLogger('pyavs') else: logger = PyAVSLogger.get_logger(self.logger_name) logger.setLevel(self.original_level)
# Progress logging utilities
[docs] def log_processing_start(logger: logging.Logger, operation: str, details: Optional[Dict[str, Any]] = None) -> None: """ Log the start of a processing operation. Parameters ---------- logger : logging.Logger Logger instance operation : str Description of the operation details : dict, optional Additional details to log """ if details: detail_str = ', '.join([f"{k}={v}" for k, v in details.items()]) logger.info(f"Starting {operation} ({detail_str})") else: logger.info(f"Starting {operation}")
[docs] def log_processing_end(logger: logging.Logger, operation: str, success: bool = True, duration: Optional[float] = None, details: Optional[Dict[str, Any]] = None) -> None: """ Log the end of a processing operation. Parameters ---------- logger : logging.Logger Logger instance operation : str Description of the operation success : bool, optional Whether operation was successful (default: True) duration : float, optional Duration in seconds details : dict, optional Additional details to log """ status = "completed" if success else "failed" message_parts = [f"{operation} {status}"] if duration is not None: message_parts.append(f"in {duration:.2f}s") if details: detail_str = ', '.join([f"{k}={v}" for k, v in details.items()]) message_parts.append(f"({detail_str})") message = ' '.join(message_parts) if success: logger.info(message) else: logger.error(message)
# Default configuration when module is imported if not PyAVSLogger._configured: # Configure with sensible defaults configure_logging( level='INFO', console=True, file_path=None, use_colors=True )