Skip to content

logger

Logger #

A Python class that integrates with C++ logging functionality. This class manages a Python logger and connects it to C++ logging streams.

Source code in cogip/utils/logger.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class Logger:
    """
    A Python class that integrates with C++ logging functionality.
    This class manages a Python logger and connects it to C++ logging streams.
    """

    def __init__(self, name: str, *, level: int = logging.INFO, enable_cpp: bool = True):
        """
        Initialize the logger with a specific name and level.

        Args:
            name: Name of the logger (appears in log output)
            level: Minimum logging level
            enable_cpp: If True, enables C++ logging integration
        """
        self.name = name
        self.is_destroyed = False  # Flag to track destruction

        # Create the Python logger
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)

        # Prevent the log messages from being handled by parent loggers
        self.logger.propagate = False

        # Remove existing handlers if any
        for handler in self.logger.handlers[:]:
            self.logger.removeHandler(handler)

        formatter = logging.Formatter("[%(asctime)s][%(name)s][%(threadName)s] %(levelname)s: %(message)s")

        # Add console handler
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

        # Add file handler
        # Check if user has root permissions
        if os.geteuid() == 0:
            # If user has root permissions, like on Raspberry Pi,
            # use /var/log/cogip to allow log persistence
            log_dir = Path("/var/log/cogip")
        else:
            # If user does not have root permissions, like in Docker stack,
            # use /tmp since no persistent storage is required
            log_dir = Path("/tmp/cogip-logs")
        log_dir.mkdir(parents=True, exist_ok=True)
        robot_id = os.getenv("ROBOT_ID", "X")
        file_handler = logging.handlers.RotatingFileHandler(
            log_dir / f"robot{robot_id}-{name}.log",
            maxBytes=10 * 1024 * 1024,
            backupCount=5,
        )
        file_handler.setLevel(level)
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

        if enable_cpp:
            self.enable_cpp_logging()

        atexit.register(self.cleanup)  # Register cleanup function

    def __del__(self):
        self.cleanup()

    def enable_cpp_logging(self):
        """Enable C++ logging integration."""
        if not self.is_destroyed:
            cpp_logger.set_logger_callback(self.log_callback)

    def cleanup(self):
        """Cleanup function to unregister the callback."""
        if not self.is_destroyed:
            self.is_destroyed = True
            cpp_logger.unset_logger_callback()  # Unregister the callback

    def log_callback(self, message: str, level: cpp_logger.LogLevel):
        """
        Callback function for C++ logging.
        Routes C++ log messages to the appropriate Python logger method.

        Args:
            message: The log message from C++
            level: Logging level from C++
        """
        # Avoid processing if the logger is destroyed
        if self.is_destroyed:
            return
        if not message:
            return
        logger_func = getattr(self.logger, level.name.lower(), self.logger.info)
        logger_func(f"[C++] {message}")

    def debug(self, message):
        """Log a debug message from Python"""
        self.logger.debug(message)

    def info(self, message):
        """Log an info message from Python"""
        self.logger.info(message)

    def warning(self, message):
        """Log a warning message from Python"""
        self.logger.warning(message)

    def error(self, message):
        """Log an error message from Python"""
        self.logger.error(message)

    def setLevel(self, level: int):
        """
        Set the logging level for the logger.

        Args:
            level: The logging level to set
        """
        self.logger.setLevel(level)
        for handler in self.logger.handlers:
            handler.setLevel(level)

__init__(name, *, level=logging.INFO, enable_cpp=True) #

Initialize the logger with a specific name and level.

Parameters:

Name Type Description Default
name str

Name of the logger (appears in log output)

 required
level int

Minimum logging level

 INFO
enable_cpp bool

If True, enables C++ logging integration

 True
Source code in cogip/utils/logger.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(self, name: str, *, level: int = logging.INFO, enable_cpp: bool = True):
    """
    Initialize the logger with a specific name and level.

    Args:
        name: Name of the logger (appears in log output)
        level: Minimum logging level
        enable_cpp: If True, enables C++ logging integration
    """
    self.name = name
    self.is_destroyed = False  # Flag to track destruction

    # Create the Python logger
    self.logger = logging.getLogger(name)
    self.logger.setLevel(level)

    # Prevent the log messages from being handled by parent loggers
    self.logger.propagate = False

    # Remove existing handlers if any
    for handler in self.logger.handlers[:]:
        self.logger.removeHandler(handler)

    formatter = logging.Formatter("[%(asctime)s][%(name)s][%(threadName)s] %(levelname)s: %(message)s")

    # Add console handler
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    self.logger.addHandler(console_handler)

    # Add file handler
    # Check if user has root permissions
    if os.geteuid() == 0:
        # If user has root permissions, like on Raspberry Pi,
        # use /var/log/cogip to allow log persistence
        log_dir = Path("/var/log/cogip")
    else:
        # If user does not have root permissions, like in Docker stack,
        # use /tmp since no persistent storage is required
        log_dir = Path("/tmp/cogip-logs")
    log_dir.mkdir(parents=True, exist_ok=True)
    robot_id = os.getenv("ROBOT_ID", "X")
    file_handler = logging.handlers.RotatingFileHandler(
        log_dir / f"robot{robot_id}-{name}.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=5,
    )
    file_handler.setLevel(level)
    file_handler.setFormatter(formatter)
    self.logger.addHandler(file_handler)

    if enable_cpp:
        self.enable_cpp_logging()

    atexit.register(self.cleanup)  # Register cleanup function

cleanup() #

Cleanup function to unregister the callback.

Source code in cogip/utils/logger.py 
80
81
82
83
84
def cleanup(self):
    """Cleanup function to unregister the callback."""
    if not self.is_destroyed:
        self.is_destroyed = True
        cpp_logger.unset_logger_callback()  # Unregister the callback

debug(message) #

Log a debug message from Python

Source code in cogip/utils/logger.py 
103
104
105
def debug(self, message):
    """Log a debug message from Python"""
    self.logger.debug(message)

enable_cpp_logging() #

Enable C++ logging integration.

Source code in cogip/utils/logger.py 
75
76
77
78
def enable_cpp_logging(self):
    """Enable C++ logging integration."""
    if not self.is_destroyed:
        cpp_logger.set_logger_callback(self.log_callback)

error(message) #

Log an error message from Python

Source code in cogip/utils/logger.py 
115
116
117
def error(self, message):
    """Log an error message from Python"""
    self.logger.error(message)

info(message) #

Log an info message from Python

Source code in cogip/utils/logger.py 
107
108
109
def info(self, message):
    """Log an info message from Python"""
    self.logger.info(message)

log_callback(message, level) #

Callback function for C++ logging. Routes C++ log messages to the appropriate Python logger method.

Parameters:

Name Type Description Default
message str

The log message from C++

 required
level LogLevel

Logging level from C++

 required
Source code in cogip/utils/logger.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def log_callback(self, message: str, level: cpp_logger.LogLevel):
    """
    Callback function for C++ logging.
    Routes C++ log messages to the appropriate Python logger method.

    Args:
        message: The log message from C++
        level: Logging level from C++
    """
    # Avoid processing if the logger is destroyed
    if self.is_destroyed:
        return
    if not message:
        return
    logger_func = getattr(self.logger, level.name.lower(), self.logger.info)
    logger_func(f"[C++] {message}")

setLevel(level) #

Set the logging level for the logger.

Parameters:

Name Type Description Default
level int

The logging level to set

 required
Source code in cogip/utils/logger.py 
119
120
121
122
123
124
125
126
127
128
def setLevel(self, level: int):
    """
    Set the logging level for the logger.

    Args:
        level: The logging level to set
    """
    self.logger.setLevel(level)
    for handler in self.logger.handlers:
        handler.setLevel(level)

warning(message) #

Log a warning message from Python

Source code in cogip/utils/logger.py 
111
112
113
def warning(self, message):
    """Log a warning message from Python"""
    self.logger.warning(message)