fix(ble): Redirect Python logging to RNS format for consistent output
Adds logging handler to redirect driver logs from Python's logging module (INFO:root:) to Reticulum's logging format ([Info] BLEInterface[...]). Changes: - Add RNSLoggingHandler to intercept root logger messages from linux_bluetooth_driver - Filter out verbose D-Bus debug logs from underlying libraries (bleak, dbus_fast) - Only redirect INFO level and above from root logger (driver messages) - Remove duplicate StreamHandlers to prevent double output - Map Python log levels to RNS log levels (DEBUG->LOG_DEBUG, INFO->LOG_INFO, etc.) Result: Clean, consistently formatted startup logs without verbose library noise. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
b503718bf8
commit
ae7c0288e3
1 changed files with 58 additions and 0 deletions
|
|
@ -40,6 +40,7 @@ import os
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import logging
|
||||||
from collections import deque
|
from collections import deque
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
|
|
@ -375,6 +376,9 @@ class BLEInterface(Interface):
|
||||||
self.driver.on_device_disconnected = self._device_disconnected_callback
|
self.driver.on_device_disconnected = self._device_disconnected_callback
|
||||||
self.driver.on_error = self._error_callback
|
self.driver.on_error = self._error_callback
|
||||||
|
|
||||||
|
# Redirect Python logging to RNS logging for proper formatting
|
||||||
|
self._setup_logging_redirect()
|
||||||
|
|
||||||
# Set driver power mode
|
# Set driver power mode
|
||||||
self.driver.set_power_mode(self.power_mode)
|
self.driver.set_power_mode(self.power_mode)
|
||||||
|
|
||||||
|
|
@ -464,6 +468,60 @@ class BLEInterface(Interface):
|
||||||
startup_thread = threading.Thread(target=self._start_advertising_when_identity_ready, daemon=True, name="BLE-Advertising-Startup")
|
startup_thread = threading.Thread(target=self._start_advertising_when_identity_ready, daemon=True, name="BLE-Advertising-Startup")
|
||||||
startup_thread.start()
|
startup_thread.start()
|
||||||
|
|
||||||
|
def _setup_logging_redirect(self):
|
||||||
|
"""
|
||||||
|
Redirect Python logging from the BLE driver to RNS logging for consistent formatting.
|
||||||
|
Only redirects logs from 'root' logger (used by linux_bluetooth_driver), not from
|
||||||
|
underlying libraries like bleak, dbus_fast, etc.
|
||||||
|
"""
|
||||||
|
class RNSLoggingHandler(logging.Handler):
|
||||||
|
def __init__(self, interface_name):
|
||||||
|
super().__init__()
|
||||||
|
self.interface_name = interface_name
|
||||||
|
|
||||||
|
def emit(self, record):
|
||||||
|
try:
|
||||||
|
# Only process logs from root logger (linux_bluetooth_driver)
|
||||||
|
# Ignore verbose logs from underlying libraries (bleak, dbus_fast, etc.)
|
||||||
|
if record.name != 'root':
|
||||||
|
return
|
||||||
|
|
||||||
|
# Map Python logging levels to RNS log levels
|
||||||
|
level_map = {
|
||||||
|
logging.DEBUG: RNS.LOG_DEBUG,
|
||||||
|
logging.INFO: RNS.LOG_INFO,
|
||||||
|
logging.WARNING: RNS.LOG_WARNING,
|
||||||
|
logging.ERROR: RNS.LOG_ERROR,
|
||||||
|
logging.CRITICAL: RNS.LOG_CRITICAL
|
||||||
|
}
|
||||||
|
rns_level = level_map.get(record.levelno, RNS.LOG_INFO)
|
||||||
|
|
||||||
|
# Format message
|
||||||
|
message = self.format(record)
|
||||||
|
|
||||||
|
# Log to RNS
|
||||||
|
RNS.log(f"{self.interface_name} {message}", rns_level)
|
||||||
|
except Exception:
|
||||||
|
# Silently fail if RNS logging fails (don't want to break the driver)
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Get root logger (used by linux_bluetooth_driver)
|
||||||
|
root_logger = logging.getLogger()
|
||||||
|
|
||||||
|
# Remove any existing stream handlers from root logger to prevent duplicate console output
|
||||||
|
for handler in root_logger.handlers[:]:
|
||||||
|
if isinstance(handler, logging.StreamHandler):
|
||||||
|
root_logger.removeHandler(handler)
|
||||||
|
|
||||||
|
# Only add handler if not already added (avoid duplicates)
|
||||||
|
handler_exists = any(isinstance(h, RNSLoggingHandler) for h in root_logger.handlers)
|
||||||
|
if not handler_exists:
|
||||||
|
handler = RNSLoggingHandler(str(self))
|
||||||
|
handler.setLevel(logging.INFO) # Only INFO and above from driver
|
||||||
|
handler.setFormatter(logging.Formatter('%(message)s'))
|
||||||
|
root_logger.addHandler(handler)
|
||||||
|
root_logger.setLevel(logging.INFO) # Don't capture DEBUG from libraries
|
||||||
|
|
||||||
def _start_advertising_when_identity_ready(self):
|
def _start_advertising_when_identity_ready(self):
|
||||||
"""
|
"""
|
||||||
Background thread that waits for Transport.identity, sets it on driver,
|
Background thread that waits for Transport.identity, sets it on driver,
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue