Skip to content

logging

Worker-side logging/printing integration (internal).

The parent process owns the main logging configuration. The worker routes its own logging records and print() output into the parent's log queue, so users see a single unified stream.

Functions

get_logger()

Get or create the brmspy logger instance.

Returns a configured logger with a custom formatter that outputs messages in the format: [brmspy][method_name] msg here

Returns:

Type Description
Logger

Configured brmspy logger instance

Examples:

>>> from brmspy.helpers.log import get_logger
>>> logger = get_logger()
>>> logger.info("Starting process")  # Prints: [brmspy][<module>] Starting process
Source code in brmspy/helpers/log.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def get_logger() -> logging.Logger:
    """
    Get or create the brmspy logger instance.

    Returns a configured logger with a custom formatter that outputs
    messages in the format: `[brmspy][method_name] msg here`

    Returns
    -------
    logging.Logger
        Configured brmspy logger instance

    Examples
    --------
    >>> from brmspy.helpers.log import get_logger
    >>> logger = get_logger()
    >>> logger.info("Starting process")  # Prints: [brmspy][<module>] Starting process
    """
    global _logger

    if _logger is None:
        _logger = logging.getLogger("brmspy")
        _logger.setLevel(logging.INFO)

        if not _logger.handlers:
            # Handler for "normal" logs
            normal_handler = logging.StreamHandler()
            normal_handler.setFormatter(BrmspyFormatter())
            normal_handler.addFilter(NonPrintFilter())
            _logger.addHandler(normal_handler)

            # print logs: preserve control chars and explicit \n/\r
            print_handler = logging.StreamHandler()
            print_handler.setFormatter(logging.Formatter("%(message)s"))
            print_handler.addFilter(PrintOnlyFilter())
            print_handler.terminator = ""
            _logger.addHandler(print_handler)

        if _running_under_pytest():
            _logger.propagate = True
        else:
            _logger.propagate = False

    return _logger

setup_worker_logging(log_queue, level=None)

Configure worker logging to forward into the parent's log queue.

Parameters:

Name Type Description Default
log_queue Queue

Queue owned by the parent; the worker will emit logging records into it.

 required
level int | None

Root log level for the worker process. Defaults to logging.INFO.

 None
Notes

When BRMSPY_WORKER=1, the worker replaces builtins.print to preserve raw control characters and line endings produced by R/cmdstan.

Source code in brmspy/_session/worker/logging.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def setup_worker_logging(log_queue: Queue, level: int | None = None) -> None:
    """
    Configure worker logging to forward into the parent's log queue.

    Parameters
    ----------
    log_queue : multiprocessing.queues.Queue
        Queue owned by the parent; the worker will emit `logging` records into it.
    level : int | None, optional
        Root log level for the worker process. Defaults to `logging.INFO`.

    Notes
    -----
    When `BRMSPY_WORKER=1`, the worker replaces [`builtins.print`](https://docs.python.org/3/library/functions.html#print)
    to preserve raw control characters and line endings produced by R/cmdstan.
    """
    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(level or logging.INFO)
    root.addHandler(QueueHandler(log_queue))

    logger = get_logger()

    def _print(*values: object, **kwargs):
        sep = kwargs.get("sep", " ")
        end = kwargs.get("end", "\n")

        # Preserve raw control chars and end exactly as R/cmdstan intended
        msg = sep.join(str(v) for v in values) + end

        if msg == "":
            return

        logger.info(
            msg,
            extra={
                "method_name": "_print",
                "no_prefix": True,
                "from_print": True,  # important for filters
            },
        )

    if os.environ.get("BRMSPY_WORKER") == "1":
        builtins.print = _print