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
129
130
131
132
133
134
135 | 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)
# Add syslog handler
if Path("/dev/log").exists():
syslog_handler = logging.handlers.SysLogHandler(address="/dev/log")
syslog_handler.setLevel(level)
syslog_handler.setFormatter(formatter)
self.logger.addHandler(syslog_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
|
|
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
71
72
73
74
75
76
77 | 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)
# Add syslog handler
if Path("/dev/log").exists():
syslog_handler = logging.handlers.SysLogHandler(address="/dev/log")
syslog_handler.setLevel(level)
syslog_handler.setFormatter(formatter)
self.logger.addHandler(syslog_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
| 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
| 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
| 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
| 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
| 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
|
|
required
|
level |
LogLevel
|
|
required
|
Source code in cogip/utils/logger.py
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108 | 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
|
|
required
|
Source code in cogip/utils/logger.py
126
127
128
129
130
131
132
133
134
135 | 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
| def warning(self, message):
"""Log a warning message from Python"""
self.logger.warning(message)
|