refactor(packet_capture_service): enhance logging configuration and per-packet summary

- Removed global logger level setting and introduced a method to apply log levels based on service-specific verbose/debug settings.
- Added a new method for logging per-packet summaries, allowing for more granular logging control based on verbosity and debug flags.
- Updated packet logging to utilize the new summary method, ensuring appropriate log levels are used for packet capture actions.
This commit is contained in:
agessaman
2026-05-17 11:28:41 -07:00
parent 4205780492
commit 41fe46babb
2 changed files with 94 additions and 3 deletions
@@ -66,7 +66,6 @@ class PacketCaptureService(BaseServicePlugin):
# Setup logging (use bot's formatter and configuration)
self.logger = logging.getLogger("PacketCaptureService")
self.logger.setLevel(bot.logger.level)
# Only setup handlers if none exist to prevent duplicates
if not self.logger.handlers:
@@ -185,6 +184,7 @@ class PacketCaptureService(BaseServicePlugin):
# Verbose/debug
self.verbose = config.getboolean("PacketCapture", "verbose", fallback=False)
self.debug = config.getboolean("PacketCapture", "debug", fallback=False)
self._apply_log_level()
# MQTT configuration
self.mqtt_enabled = config.getboolean("PacketCapture", "mqtt_enabled", fallback=True)
@@ -369,6 +369,22 @@ class PacketCaptureService(BaseServicePlugin):
"""
return self.bot.config.get("PacketCapture", key, fallback=fallback)
def _apply_log_level(self) -> None:
"""Set service logger level from PacketCapture verbose/debug, not global bot log_level."""
if self.debug:
self.logger.setLevel(logging.DEBUG)
else:
self.logger.setLevel(logging.INFO)
def _log_packet_summary(self, message: str) -> None:
"""Log per-packet summary when verbose or debug is enabled."""
if not (self.verbose or self.debug):
return
if self.debug:
self.logger.debug(message)
else:
self.logger.info(message)
def _auth_token_iat_exp(self, broker_config: dict[str, Any]) -> tuple[int, int]:
"""Unix iat/exp for JWT payload (exp = iat + ttl). Non-positive TTL uses 86400s."""
iat = int(time.time())
@@ -894,7 +910,7 @@ class PacketCaptureService(BaseServicePlugin):
publish_metrics["skipped_unparseable"] = skip_mqtt_unparseable
publish_metrics["skipped_invalid_advert_signature"] = skip_mqtt_invalid_advert_signature
# Log DEBUG level for each packet (verbose; use INFO only for service lifecycle)
# Per-packet summary: INFO when verbose, DEBUG when debug (lifecycle stays INFO)
if publish_metrics.get("skipped_unparseable"):
action = "Captured (MQTT skipped: zero hash / unparseable)"
elif publish_metrics.get("skipped_invalid_advert_signature"):
@@ -903,7 +919,7 @@ class PacketCaptureService(BaseServicePlugin):
action = "Skipping"
else:
action = "Captured"
self.logger.debug(
self._log_packet_summary(
f"📦 {action} packet #{self.packet_count}: {formatted_packet['route']} type {formatted_packet['packet_type']}, {formatted_packet['len']} bytes, SNR: {formatted_packet['SNR']}, RSSI: {formatted_packet['RSSI']}, hash: {formatted_packet['hash']} (MQTT: {publish_metrics['succeeded']}/{publish_metrics['attempted']})"
)
@@ -0,0 +1,75 @@
"""PacketCapture service log level respects verbose/debug, not global bot log_level."""
from __future__ import annotations
import configparser
import logging
from unittest.mock import MagicMock
from modules.service_plugins.packet_capture_service import PacketCaptureService
def _svc_from_ini(ini: str) -> PacketCaptureService:
cp = configparser.ConfigParser()
cp.read_string(ini.strip())
bot = MagicMock()
bot.config = cp
bot.logger = logging.getLogger("test_bot")
bot.logger.setLevel(logging.DEBUG)
svc = object.__new__(PacketCaptureService)
svc.bot = bot
svc.logger = logging.getLogger("PacketCaptureService.test")
svc._load_config()
return svc
def test_debug_false_uses_info_even_when_bot_is_debug():
svc = _svc_from_ini(
"""
[PacketCapture]
enabled = true
verbose = false
debug = false
"""
)
assert svc.debug is False
assert svc.verbose is False
assert svc.logger.level == logging.INFO
def test_debug_true_uses_debug_level():
svc = _svc_from_ini(
"""
[PacketCapture]
enabled = true
verbose = false
debug = true
"""
)
assert svc.logger.level == logging.DEBUG
def test_log_packet_summary_only_when_verbose_or_debug():
svc = _svc_from_ini(
"""
[PacketCapture]
enabled = true
verbose = false
debug = false
"""
)
svc.logger.handlers.clear()
records: list[logging.LogRecord] = []
class _Capture(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
records.append(record)
svc.logger.addHandler(_Capture())
svc._log_packet_summary("packet line")
assert records == []
svc.verbose = True
svc._log_packet_summary("packet line")
assert len(records) == 1
assert records[0].levelno == logging.INFO