""" Centralized Logging Configuration for Yuga Planner This module provides a unified logging configuration that: 1. Respects the YUGA_DEBUG environment variable for debug logging 2. Uses consistent formatting across the entire codebase 3. Eliminates the need for individual logging.basicConfig() calls Usage: from utils.logging_config import setup_logging, get_logger # Initialize logging (typically done once per module) setup_logging() logger = get_logger(__name__) # Use logging methods logger.debug("Debug message - only shown when YUGA_DEBUG=true") logger.info("Info message - always shown") logger.warning("Warning message") logger.error("Error message") Environment Variables: YUGA_DEBUG: Set to "true" to enable debug logging Migration from old logging: Replace: import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) With: from utils.logging_config import setup_logging, get_logger setup_logging() logger = get_logger(__name__) """ import os, sys, logging, threading, time from typing import Optional from collections import deque class LogCapture: """Capture logs for real-time streaming to UI""" def __init__(self, max_lines: int = 1000): self.max_lines = max_lines self.log_buffer = deque(maxlen=max_lines) self.session_buffer = deque(maxlen=max_lines) # Current session logs self.lock = threading.Lock() self.session_start_time = None def add_log(self, record: logging.LogRecord): """Add a log record to the UI streaming buffer (filtered for essential logs only)""" # This only affects UI streaming - console logs are handled separately logger_name = record.name message = record.getMessage() # Skip all UI, gradio, httpx, and other system logs for UI streaming skip_loggers = [ "gradio", "httpx", "uvicorn", "fastapi", "urllib3", "ui.pages.chat", "ui.", "asyncio", "websockets", "handlers.tool_call_handler", "services.mcp_client", ] # Skip if it's a system logger if any(skip in logger_name for skip in skip_loggers): return # Only include essential task splitting and constraint solver logs for UI essential_patterns = [ "=== Step 1: Task Breakdown ===", "=== Step 2: Time Estimation ===", "=== Step 3: Skill Matching ===", "Processing", "tasks for time estimation", "Completed time estimation", "Completed skill matching", "Generated", "tasks with skills", "Starting solve process", "Preparing schedule for solving", "Starting schedule solver", "solving", "constraint", "optimization", ] # Check if this log message contains essential information is_essential = any( pattern.lower() in message.lower() for pattern in essential_patterns ) # Only include essential logs from factory and handler modules for UI allowed_modules = ["factory.", "handlers.mcp_backend", "services.schedule"] module_allowed = any( logger_name.startswith(module) for module in allowed_modules ) if not (module_allowed and is_essential): return # Format for clean streaming display in UI timestamp = time.strftime("%H:%M:%S", time.localtime(record.created)) # Clean up the message for better display match message: case msg if "===" in msg: # Task breakdown steps formatted_log = f"⏳ {msg.replace('===', '').strip()}" case msg if "Processing" in msg and "time estimation" in msg: formatted_log = f"⏱️ {msg}" case msg if "Completed" in msg: formatted_log = f"✅ {msg}" case msg if "Generated" in msg and "tasks" in msg: formatted_log = f"🎯 {msg}" case msg if "Starting solve process" in msg or "Starting schedule solver" in msg: formatted_log = f"⚡ {msg}" case msg if "Preparing schedule" in msg: formatted_log = f"📋 {msg}" case _: formatted_log = f"🔧 {message}" with self.lock: self.log_buffer.append(formatted_log) # Add to session buffer if session is active if self.session_start_time and record.created >= self.session_start_time: self.session_buffer.append(formatted_log) def start_session(self): """Start capturing logs for current session""" with self.lock: self.session_start_time = time.time() self.session_buffer.clear() def get_session_logs(self) -> list: """Get all logs from current session""" with self.lock: return list(self.session_buffer) def get_recent_logs(self, count: int = 50) -> list: """Get recent logs""" with self.lock: return list(self.log_buffer)[-count:] class StreamingLogHandler(logging.Handler): """Custom log handler that captures logs for streaming""" def __init__(self, log_capture: LogCapture): super().__init__() self.log_capture = log_capture def emit(self, record): try: self.log_capture.add_log(record) except Exception: self.handleError(record) # Global log capture instance _log_capture = LogCapture() _streaming_handler = None def setup_logging(level: Optional[str] = None) -> None: """ Set up centralized logging configuration for the application. Args: level: Override the logging level. If None, uses YUGA_DEBUG environment variable. """ global _streaming_handler # Determine logging level if level is not None: log_level = getattr(logging, level.upper(), logging.INFO) else: debug_enabled = os.getenv("YUGA_DEBUG", "false").lower() == "true" log_level = logging.DEBUG if debug_enabled else logging.INFO # Get root logger root_logger = logging.getLogger() # Only configure if not already configured if not root_logger.handlers or _streaming_handler is None: # Clear existing handlers to avoid duplicates root_logger.handlers.clear() # Create formatter formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # Console handler for terminal output (shows ALL logs) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(log_level) console_handler.setFormatter(formatter) # Streaming handler for UI capture (filtered to essential logs only) _streaming_handler = StreamingLogHandler(_log_capture) _streaming_handler.setLevel( logging.DEBUG ) # Capture all levels, but filter in handler # Configure root logger root_logger.setLevel(logging.DEBUG) # Add both handlers root_logger.addHandler(console_handler) root_logger.addHandler(_streaming_handler) # Log the configuration logger = logging.getLogger(__name__) logger.debug("Debug logging enabled via YUGA_DEBUG environment variable") def get_logger(name: str) -> logging.Logger: """ Get a logger instance with the specified name. Args: name: Name for the logger, typically __name__ Returns: Configured logger instance """ return logging.getLogger(name) def is_debug_enabled() -> bool: """Check if debug logging is enabled via environment variable.""" return os.getenv("YUGA_DEBUG", "false").lower() == "true" def get_log_capture() -> LogCapture: """Get the global log capture instance for UI streaming""" return _log_capture def start_session_logging(): """Start capturing logs for the current chat session""" _log_capture.start_session() def get_session_logs() -> list: """Get all logs from the current session for streaming to UI""" return _log_capture.get_session_logs()