#!/usr/bin/env python3 """ Centralized logging configuration for water monitoring system """ import logging import logging.handlers import os from datetime import datetime from typing import Optional class ColoredFormatter(logging.Formatter): """Colored console formatter""" 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 } def format(self, record): if hasattr(record, "levelname"): color = self.COLORS.get(record.levelname, self.COLORS["RESET"]) record.levelname = f"{color}{record.levelname}{self.COLORS['RESET']}" return super().format(record) def setup_logging( log_level: str = "INFO", log_file: Optional[str] = None, max_file_size: int = 10 * 1024 * 1024, # 10MB backup_count: int = 5, enable_console: bool = True, enable_colors: bool = True, ) -> logging.Logger: """ Setup comprehensive logging configuration Args: log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_file: Path to log file (optional) max_file_size: Maximum size of log file before rotation backup_count: Number of backup files to keep enable_console: Whether to enable console logging enable_colors: Whether to enable colored console output Returns: Configured logger instance """ # Create logs directory if it doesn't exist if log_file: log_dir = os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir) # Configure root logger logger = logging.getLogger() logger.setLevel(getattr(logging, log_level.upper())) # Clear existing handlers logger.handlers.clear() # Create formatters detailed_formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) simple_formatter = logging.Formatter( "%(asctime)s - %(levelname)s - %(message)s", datefmt="%H:%M:%S" ) # Console handler if enable_console: console_handler = logging.StreamHandler() if enable_colors and os.name != "nt": # Don't use colors on Windows console_formatter = ColoredFormatter( "%(asctime)s - %(levelname)s - %(message)s", datefmt="%H:%M:%S" ) else: console_formatter = simple_formatter console_handler.setFormatter(console_formatter) console_handler.setLevel(getattr(logging, log_level.upper())) logger.addHandler(console_handler) # File handler with rotation if log_file: file_handler = logging.handlers.RotatingFileHandler( log_file, maxBytes=max_file_size, backupCount=backup_count, encoding="utf-8" ) file_handler.setFormatter(detailed_formatter) file_handler.setLevel(logging.DEBUG) # Always log everything to file logger.addHandler(file_handler) # Add performance logger for metrics perf_logger = logging.getLogger("performance") if log_file: perf_file = log_file.replace(".log", "_performance.log") perf_handler = logging.handlers.RotatingFileHandler( perf_file, maxBytes=max_file_size, backupCount=backup_count, encoding="utf-8", ) perf_formatter = logging.Formatter( "%(asctime)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) perf_handler.setFormatter(perf_formatter) perf_logger.addHandler(perf_handler) perf_logger.setLevel(logging.INFO) perf_logger.propagate = False return logger def log_performance_metric( operation: str, duration: float, additional_info: Optional[str] = None ): """Log performance metrics""" perf_logger = logging.getLogger("performance") message = f"PERF: {operation} took {duration:.3f}s" if additional_info: message += f" - {additional_info}" perf_logger.info(message) def get_logger(name: str) -> logging.Logger: """Get a logger with the specified name""" return logging.getLogger(name)