utils.datalog_bridge

Bridge Python logging -> wpilib DataLogManager.

Call setup_logging() once at robot init to:

Forward all Python logging output into .wpilog files
Create a persistent NT entry at /robot/logLevel that allows the dashboard to change the log level at runtime

Accepted level strings (case-insensitive): debug, info, warn, warning, error, critical, fatal, off. WPILib-style k-prefixed names (kDebug, kInfo, etc.) are also accepted.

View Source

 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
51
52def _on_log_level_changed(event: ntcore.Event) -> None:
53    """NT listener callback — updates root logger level."""
54    value = event.data.value.getString()
55    level = _LOG_LEVEL_MAP.get(value.strip().lower())
56    if level is not None:
57        logging.getLogger().setLevel(level)
58        logging.info("Log level changed to %s (%d)", value.strip(), level)
59
60
61def setup_logging(default_level: int = logging.INFO) -> None:
62    """Initialize the Python logging -> wpilog bridge.
63
64    - Attaches a handler to the root logger that forwards all records
65      to ``DataLogManager.log()``.
66    - Sets the root logger to *default_level*.
67    - Creates a persistent NT string entry at ``/robot/logLevel``
68      with a listener that updates the root level on change.
69
70    Safe to call once at robot init.  Calling multiple times will add
71    duplicate handlers — avoid that.
72    """
73    # Attach wpilog handler to root logger
74    handler = _WPILogHandler()
75    handler.setFormatter(logging.Formatter(
76        "%(name)s [%(levelname)s] %(message)s"))
77    root = logging.getLogger()
78    root.addHandler(handler)
79    root.setLevel(default_level)
80
81    # NT-controlled log level (persistent, callback-driven)
82    inst = ntcore.NetworkTableInstance.getDefault()
83    topic = inst.getStringTopic("/robot/logLevel")
84
85    publisher = topic.publish(
86        ntcore.PubSubOptions(keepDuplicates=False))
87    publisher.setDefault("INFO")
88    inst.getTable("/robot").getEntry("logLevel").setPersistent()
89
90    subscriber = topic.subscribe("INFO")
91    inst.addListener(
92        subscriber,
93        ntcore.EventFlags.kValueRemote | ntcore.EventFlags.kImmediate,
94        _on_log_level_changed,
95    )
96
97    # Prevent GC from collecting the publisher/subscriber
98    _nt_refs.extend([publisher, subscriber])

def setup_logging(default_level: int = 20) -> None: View Source

62def setup_logging(default_level: int = logging.INFO) -> None:
63    """Initialize the Python logging -> wpilog bridge.
64
65    - Attaches a handler to the root logger that forwards all records
66      to ``DataLogManager.log()``.
67    - Sets the root logger to *default_level*.
68    - Creates a persistent NT string entry at ``/robot/logLevel``
69      with a listener that updates the root level on change.
70
71    Safe to call once at robot init.  Calling multiple times will add
72    duplicate handlers — avoid that.
73    """
74    # Attach wpilog handler to root logger
75    handler = _WPILogHandler()
76    handler.setFormatter(logging.Formatter(
77        "%(name)s [%(levelname)s] %(message)s"))
78    root = logging.getLogger()
79    root.addHandler(handler)
80    root.setLevel(default_level)
81
82    # NT-controlled log level (persistent, callback-driven)
83    inst = ntcore.NetworkTableInstance.getDefault()
84    topic = inst.getStringTopic("/robot/logLevel")
85
86    publisher = topic.publish(
87        ntcore.PubSubOptions(keepDuplicates=False))
88    publisher.setDefault("INFO")
89    inst.getTable("/robot").getEntry("logLevel").setPersistent()
90
91    subscriber = topic.subscribe("INFO")
92    inst.addListener(
93        subscriber,
94        ntcore.EventFlags.kValueRemote | ntcore.EventFlags.kImmediate,
95        _on_log_level_changed,
96    )
97
98    # Prevent GC from collecting the publisher/subscriber
99    _nt_refs.extend([publisher, subscriber])

Initialize the Python logging -> wpilog bridge.

Attaches a handler to the root logger that forwards all records to DataLogManager.log().
Sets the root logger to default_level.
Creates a persistent NT string entry at /robot/logLevel with a listener that updates the root level on change.

Safe to call once at robot init. Calling multiple times will add duplicate handlers — avoid that.