utils.datalog_bridge

  1"""Bridge Python logging -> wpilib DataLogManager.
  2
  3Call ``setup_logging()`` once at robot init to:
  4  1. Forward all Python ``logging`` output into .wpilog files
  5  2. Create a persistent NT entry at ``/robot/logLevel`` that allows
  6     the dashboard to change the log level at runtime
  7
  8Accepted level strings (case-insensitive): debug, info, warn, warning,
  9error, critical, fatal, off.  WPILib-style k-prefixed names (kDebug,
 10kInfo, etc.) are also accepted.
 11"""
 12
 13import logging
 14
 15import ntcore
 16import wpilib
 17
 18
 19class _WPILogHandler(logging.Handler):
 20    """Forwards Python log records into the wpilog 'messages' entry."""
 21
 22    def emit(self, record: logging.LogRecord) -> None:
 23        try:
 24            msg = self.format(record)
 25            wpilib.DataLogManager.log(msg)
 26        except Exception:
 27            self.handleError(record)
 28
 29
 30# Normalized string -> Python logging level
 31_LOG_LEVEL_MAP: dict[str, int] = {
 32    "debug": logging.DEBUG,
 33    "kdebug": logging.DEBUG,
 34    "info": logging.INFO,
 35    "kinfo": logging.INFO,
 36    "warn": logging.WARNING,
 37    "warning": logging.WARNING,
 38    "kwarning": logging.WARNING,
 39    "error": logging.ERROR,
 40    "kerror": logging.ERROR,
 41    "critical": logging.CRITICAL,
 42    "kcritical": logging.CRITICAL,
 43    "fatal": logging.CRITICAL,
 44    "off": logging.CRITICAL + 10,
 45    "koff": logging.CRITICAL + 10,
 46}
 47
 48# References kept alive to prevent GC of NT subscribers/publishers
 49_nt_refs: list = []
 50_initialized: bool = False
 51
 52
 53def _on_log_level_changed(event: ntcore.Event) -> None:
 54    """NT listener callback — updates root logger level."""
 55    value = event.data.value.getString()
 56    level = _LOG_LEVEL_MAP.get(value.strip().lower())
 57    if level is not None:
 58        logging.getLogger().setLevel(level)
 59        # Use print to avoid feeding back through the WPILog handler
 60        print(f"Log level changed to {value.strip()} ({level})")
 61
 62
 63def setup_logging(default_level: int = logging.INFO) -> None:
 64    """Initialize the Python logging -> wpilog bridge.
 65
 66    - Attaches a handler to the root logger that forwards all records
 67      to ``DataLogManager.log()``.
 68    - Sets the root logger to *default_level*.
 69    - Creates a persistent NT string entry at ``/robot/logLevel``
 70      with a listener that updates the root level on change.
 71
 72    Safe to call multiple times — subsequent calls are no-ops.
 73    """
 74    global _initialized
 75    if _initialized:
 76        return
 77    _initialized = True
 78
 79    # Attach wpilog handler to root logger
 80    handler = _WPILogHandler()
 81    handler.setFormatter(logging.Formatter(
 82        "%(name)s [%(levelname)s] %(message)s"))
 83    root = logging.getLogger()
 84    root.addHandler(handler)
 85    root.setLevel(default_level)
 86
 87    # NT-controlled log level (persistent, callback-driven)
 88    inst = ntcore.NetworkTableInstance.getDefault()
 89    topic = inst.getStringTopic("/robot/logLevel")
 90
 91    publisher = topic.publish(
 92        ntcore.PubSubOptions(keepDuplicates=False))
 93    publisher.setDefault("INFO")
 94    inst.getTable("/robot").getEntry("logLevel").setPersistent()
 95
 96    subscriber = topic.subscribe("INFO")
 97    inst.addListener(
 98        subscriber,
 99        ntcore.EventFlags.kValueRemote | ntcore.EventFlags.kImmediate,
100        _on_log_level_changed,
101    )
102
103    # Prevent GC from collecting the publisher/subscriber
104    _nt_refs.extend([publisher, subscriber])