openclaw-voice/utils/logging.py

"""Structured logging with per-module configuration and latency tracking."""

import logging
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Optional

from .config import LoggingConfig


# Global logger registry
_loggers: dict[str, logging.Logger] = {}
_latency_tracking_enabled: bool = True


def setup_logging(config: LoggingConfig) -> None:
    """
    Initialize logging system with configuration.

    Args:
        config: Logging configuration object
    """
    global _latency_tracking_enabled

    # Set latency tracking flag
    _latency_tracking_enabled = config.track_latency

    # Configure root logger
    root_logger = logging.getLogger()
    root_logger.setLevel(getattr(logging, config.level.upper()))

    # Clear existing handlers
    root_logger.handlers.clear()

    # Create formatter
    formatter = logging.Formatter(config.format)

    # Console handler
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    root_logger.addHandler(console_handler)

    # File handler (if configured)
    if config.file:
        file_path = Path(config.file)
        file_path.parent.mkdir(parents=True, exist_ok=True)

        if config.rotation.get("enabled", False):
            from logging.handlers import RotatingFileHandler

            file_handler = RotatingFileHandler(
                config.file,
                maxBytes=config.rotation.get("max_bytes", 10485760),
                backupCount=config.rotation.get("backup_count", 5),
            )
        else:
            file_handler = logging.FileHandler(config.file)

        file_handler.setFormatter(formatter)
        root_logger.addHandler(file_handler)

    # Configure per-module log levels
    for module_name, level in config.modules.items():
        module_logger = logging.getLogger(module_name)
        module_logger.setLevel(getattr(logging, level.upper()))

    root_logger.info("Logging system initialized")


def get_logger(name: str) -> logging.Logger:
    """
    Get or create a logger for a module.

    Args:
        name: Logger name (typically __name__ of calling module)

    Returns:
        Logger instance
    """
    if name not in _loggers:
        _loggers[name] = logging.getLogger(name)

    return _loggers[name]


@contextmanager
def log_latency(logger: logging.Logger, operation: str, level: int = logging.DEBUG):
    """
    Context manager to track and log operation latency.

    Usage:
        with log_latency(logger, "transcribe_audio"):
            result = transcribe(audio)

    Args:
        logger: Logger instance
        operation: Operation name for logging
        level: Log level for latency message
    """
    if not _latency_tracking_enabled:
        yield
        return

    start_time = time.perf_counter()
    exception_occurred = False

    try:
        yield
    except Exception:
        exception_occurred = True
        raise
    finally:
        elapsed_ms = (time.perf_counter() - start_time) * 1000

        if exception_occurred:
            logger.log(
                level,
                f"{operation} FAILED after {elapsed_ms:.2f}ms",
            )
        else:
            logger.log(
                level,
                f"{operation} completed in {elapsed_ms:.2f}ms",
            )


class LatencyTracker:
    """
    Track cumulative latency across multiple operations.

    Usage:
        tracker = LatencyTracker()

        with tracker.track("vad"):
            detect_speech(audio)

        with tracker.track("stt"):
            transcribe(audio)

        logger.info(tracker.summary())
    """

    def __init__(self):
        self._timings: dict[str, list[float]] = {}
        self._current_operation: Optional[str] = None
        self._operation_start: Optional[float] = None

    @contextmanager
    def track(self, operation: str):
        """Track latency for an operation."""
        if not _latency_tracking_enabled:
            yield
            return

        self._current_operation = operation
        self._operation_start = time.perf_counter()

        try:
            yield
        finally:
            if self._operation_start is not None:
                elapsed = time.perf_counter() - self._operation_start
                if operation not in self._timings:
                    self._timings[operation] = []
                self._timings[operation].append(elapsed)

            self._current_operation = None
            self._operation_start = None

    def get_timing(self, operation: str) -> Optional[float]:
        """
        Get total time for an operation in milliseconds.

        Args:
            operation: Operation name

        Returns:
            Total time in ms, or None if operation not tracked
        """
        if operation not in self._timings:
            return None

        return sum(self._timings[operation]) * 1000

    def get_average(self, operation: str) -> Optional[float]:
        """
        Get average time for an operation in milliseconds.

        Args:
            operation: Operation name

        Returns:
            Average time in ms, or None if operation not tracked
        """
        if operation not in self._timings:
            return None

        timings = self._timings[operation]
        return (sum(timings) / len(timings)) * 1000

    def total_time_ms(self) -> float:
        """Get total time across all operations in milliseconds."""
        total = 0.0
        for timings in self._timings.values():
            total += sum(timings)
        return total * 1000

    def summary(self) -> str:
        """
        Generate a summary of all tracked operations.

        Returns:
            Formatted summary string
        """
        if not self._timings:
            return "No operations tracked"

        lines = ["Latency Summary:"]
        for operation, timings in self._timings.items():
            total_ms = sum(timings) * 1000
            count = len(timings)
            avg_ms = total_ms / count
            lines.append(f"  {operation}: {total_ms:.2f}ms total ({count}x, avg {avg_ms:.2f}ms)")

        lines.append(f"  TOTAL: {self.total_time_ms():.2f}ms")

        return "\n".join(lines)

    def reset(self) -> None:
        """Clear all tracked timings."""
        self._timings.clear()


# Example usage function for testing
def _example_usage():
    """Example of how to use logging utilities."""
    from .config import LoggingConfig

    # Setup logging
    config = LoggingConfig(level="DEBUG", track_latency=True)
    setup_logging(config)

    # Get logger
    logger = get_logger(__name__)

    # Simple logging
    logger.info("Starting operation")
    logger.debug("Debug information")

    # Latency tracking - single operation
    with log_latency(logger, "expensive_operation"):
        time.sleep(0.1)  # Simulate work

    # Latency tracking - multiple operations
    tracker = LatencyTracker()

    with tracker.track("step_1"):
        time.sleep(0.05)

    with tracker.track("step_2"):
        time.sleep(0.03)

    with tracker.track("step_1"):  # Same operation again
        time.sleep(0.02)

    logger.info(tracker.summary())


if __name__ == "__main__":
    _example_usage()