rpx_benchmark.logging_utils

  1"""Structured logging for the RPX benchmark toolkit.
  2
  3Library modules do **not** configure logging themselves. They only call
  4:func:`get_logger` at module top::
  5
  6    from rpx_benchmark.logging_utils import get_logger
  7
  8    log = get_logger(__name__)
  9
 10    def do_the_thing():
 11        log.info("starting the thing")
 12        log.debug("parameter x=%s", x)
 13
 14The CLI entrypoint (and any application code driving the toolkit)
 15calls :func:`configure_logging` exactly once to attach a handler. If
 16``rich`` is installed, a pretty :class:`rich.logging.RichHandler` is
 17used so library log lines render with colour, level icons, and path
 18markers alongside the benchmark progress UI. Otherwise we fall back to
 19the standard library handler.
 20
 21The logger hierarchy mirrors the package structure
 22(``rpx_benchmark``, ``rpx_benchmark.hub``, ``rpx_benchmark.runner``...)
 23so power users can turn individual modules up or down with one call::
 24
 25    import logging
 26    logging.getLogger("rpx_benchmark.hub").setLevel(logging.DEBUG)
 27
 28Log level conventions
 29---------------------
 30
 31- ``DEBUG`` — internal state useful for bug reports. Off by default.
 32- ``INFO`` — user-facing progress ("downloading split", "loaded model").
 33- ``WARNING`` — degradations the user should know about but that do
 34  not abort the run ("CUDA unavailable; falling back to CPU",
 35  "profiler unavailable; FLOPs not reported").
 36- ``ERROR`` — recoverable failures that abort one operation.
 37- ``CRITICAL`` — reserved for unrecoverable state.
 38"""
 39
 40from __future__ import annotations
 41
 42import logging
 43import os
 44import sys
 45from typing import Optional
 46
 47_ROOT_LOGGER_NAME = "rpx_benchmark"
 48_CONFIGURED = False
 49
 50
 51# --------------------------------------------------------------------------- #
 52# Public API
 53# --------------------------------------------------------------------------- #
 54
 55
 56def get_logger(name: str) -> logging.Logger:
 57    """Return a module-scoped logger nested under the ``rpx_benchmark`` root.
 58
 59    Parameters
 60    ----------
 61    name : str
 62        Usually ``__name__``. If it starts with ``rpx_benchmark`` it is
 63        returned as-is; otherwise it is attached as a child of the root
 64        logger so external integrations can still funnel output.
 65
 66    Returns
 67    -------
 68    logging.Logger
 69        A logger ready to use. Call ``log.info(...)`` etc.
 70
 71    Examples
 72    --------
 73    >>> from rpx_benchmark.logging_utils import get_logger
 74    >>> log = get_logger("rpx_benchmark.hub")
 75    >>> log.name
 76    'rpx_benchmark.hub'
 77    """
 78    if not name:
 79        name = _ROOT_LOGGER_NAME
 80    if name == _ROOT_LOGGER_NAME or name.startswith(_ROOT_LOGGER_NAME + "."):
 81        return logging.getLogger(name)
 82    return logging.getLogger(f"{_ROOT_LOGGER_NAME}.{name}")
 83
 84
 85def configure_logging(
 86    level: str | int = "INFO",
 87    *,
 88    force: bool = False,
 89    use_rich: Optional[bool] = None,
 90) -> logging.Logger:
 91    """Install a handler on the root ``rpx_benchmark`` logger.
 92
 93    Safe to call multiple times: subsequent calls are no-ops unless
 94    ``force=True``. The CLI calls this once at the start of ``main``;
 95    library users can call it from their own entrypoint.
 96
 97    Parameters
 98    ----------
 99    level : str or int
100        Logging level name (``"DEBUG"``, ``"INFO"``, ``"WARNING"``,
101        ``"ERROR"``) or numeric level. Can be overridden at runtime
102        via the ``RPX_LOG_LEVEL`` environment variable.
103    force : bool
104        If True, re-install the handler even if one is already
105        present. Use with care — multiple handlers cause duplicate
106        output.
107    use_rich : bool, optional
108        Force rich or plain handler selection. When omitted, auto-
109        detects: rich if the ``rich`` package is importable and stderr
110        is a TTY, otherwise plain.
111
112    Returns
113    -------
114    logging.Logger
115        The configured root logger, for chaining.
116
117    Examples
118    --------
119    >>> from rpx_benchmark.logging_utils import configure_logging
120    >>> log = configure_logging("DEBUG")  # doctest: +SKIP
121    >>> log.info("benchmark starting")    # doctest: +SKIP
122    """
123    global _CONFIGURED
124
125    env_level = os.environ.get("RPX_LOG_LEVEL")
126    if env_level:
127        level = env_level
128
129    if isinstance(level, str):
130        level_value = logging.getLevelName(level.upper())
131        if not isinstance(level_value, int):
132            # Unknown name -> default to INFO with a warning emitted
133            # the first time.
134            level_value = logging.INFO
135    else:
136        level_value = int(level)
137
138    root = logging.getLogger(_ROOT_LOGGER_NAME)
139    root.setLevel(level_value)
140    root.propagate = False
141
142    if _CONFIGURED and not force:
143        return root
144
145    # Wipe any prior handlers so re-configuration is idempotent.
146    for h in list(root.handlers):
147        root.removeHandler(h)
148
149    handler = _build_handler(level_value, use_rich=use_rich)
150    root.addHandler(handler)
151    _CONFIGURED = True
152    return root
153
154
155def set_level(level: str | int) -> None:
156    """Shortcut for ``logging.getLogger('rpx_benchmark').setLevel(level)``.
157
158    Useful for quick toggles without re-running :func:`configure_logging`.
159    """
160    if isinstance(level, str):
161        level = logging.getLevelName(level.upper())
162    logging.getLogger(_ROOT_LOGGER_NAME).setLevel(level)
163
164
165# --------------------------------------------------------------------------- #
166# Internal handler factory
167# --------------------------------------------------------------------------- #
168
169
170def _rich_available() -> bool:
171    try:
172        import rich  # noqa: F401
173
174        return True
175    except ImportError:
176        return False
177
178
179def _build_handler(
180    level: int,
181    *,
182    use_rich: Optional[bool],
183) -> logging.Handler:
184    """Build a log handler, preferring rich if available and stderr is a tty."""
185    if use_rich is None:
186        use_rich = _rich_available() and sys.stderr.isatty()
187
188    if use_rich:
189        try:
190            from rich.logging import RichHandler
191
192            handler: logging.Handler = RichHandler(
193                level=level,
194                show_time=False,
195                show_path=False,
196                rich_tracebacks=True,
197                markup=False,
198            )
199            handler.setFormatter(logging.Formatter("%(message)s"))
200            return handler
201        except ImportError:
202            pass  # fall through to plain
203
204    handler = logging.StreamHandler(sys.stderr)
205    handler.setLevel(level)
206    handler.setFormatter(
207        logging.Formatter(
208            fmt="[%(levelname)s] %(name)s: %(message)s",
209        )
210    )
211    return handler
212
213
214__all__ = [
215    "get_logger",
216    "configure_logging",
217    "set_level",
218]