Source code for tracmass.verbose

"""Verbose debug output: timestamped progress messages for each run stage.

tracmass logs what it is doing (adapter calls, flux reads with their size,
seeding, kernel compilation, each window's phases) to the standard
:mod:`logging` logger ``"tracmass"`` at ``DEBUG`` level.  Nothing is shown
unless verbose output is turned on, in any of these ways:

- ``tracmass.run(..., verbose=True)`` or ``verbose = true`` in the active
  ``tracmass.toml`` environment (also honoured by the integrators);
- :func:`set_verbose` for the rest of the session;
- standard logging config, e.g.
  ``logging.getLogger("tracmass").setLevel("DEBUG")`` plus a handler.

Messages go to ``stderr``, one line each, with the wall-clock time, so a run
that seems stuck shows which step it is in and how long the previous steps
took.
"""

from __future__ import annotations

import contextlib
import logging
import sys
import time

logger = logging.getLogger("tracmass")

_handler: logging.Handler | None = None
_saved_level: int = logging.NOTSET




def set_verbose(verbose: bool = True) -> None:
    """Turn verbose debug output on or off for the session.

    Parameters
    ----------
    verbose : bool, optional
        True attaches a ``stderr`` handler to the ``"tracmass"`` logger and
        sets it to ``DEBUG``; False removes the handler again and restores
        the previous level.  Default True.

    Examples
    --------
    >>> tracmass.set_verbose()
    >>> result = tracmass.run(env="gog")
    """
    global _handler, _saved_level
    if verbose and _handler is None:
        _handler = logging.StreamHandler(sys.stderr)
        _handler.setFormatter(
            logging.Formatter("%(asctime)s %(name)s: %(message)s", "%H:%M:%S")
        )
        _saved_level = logger.level
        logger.addHandler(_handler)
        logger.setLevel(logging.DEBUG)
    elif not verbose and _handler is not None:
        logger.removeHandler(_handler)
        logger.setLevel(_saved_level)
        _handler = None





def is_verbose() -> bool:
    """``True`` while :func:`set_verbose` output is on."""
    return _handler is not None





def resolve_verbose(verbose: bool | None = None) -> bool:
    """*verbose* if given, else the ``verbose`` setting (default False)."""
    if verbose is not None:
        return bool(verbose)
    try:
        from tracmass.config import settings

        return bool(settings.get("verbose", False))
    except Exception:  # pragma: no cover - config is optional
        return False





@contextlib.contextmanager
def verbose_output(verbose: bool | None = None):
    """Turn verbose output on for a ``with`` block, if *verbose* resolves True.

    Output that was already on (from :func:`set_verbose` or an outer block)
    stays on afterwards; output turned on here is turned off on exit.
    """
    turn_on = resolve_verbose(verbose) and not is_verbose()
    if turn_on:
        set_verbose(True)
    try:
        yield
    finally:
        if turn_on:
            set_verbose(False)





@contextlib.contextmanager
def timed(log: logging.Logger, msg: str, *args):
    """Log ``msg ...`` at entry and ``msg: done in N s`` at exit (DEBUG)."""
    if not log.isEnabledFor(logging.DEBUG):
        yield
        return
    log.debug(msg + " ...", *args)
    t0 = time.perf_counter()
    yield
    log.debug(msg + ": done in %.1f s", *args, time.perf_counter() - t0)





def gb(nbytes: float) -> str:
    """*nbytes* formatted as ``"1.23 GB"``."""
    return f"{nbytes / 1e9:.2f} GB"