🔇 ctx#

This module provides context managers for various utilities such as logging to a file, measuring execution time, silencing output, and setting multiprocessing start methods.

LogToFile - Context manager to log to a file every print call (with LINE_UP handling)
MeasureTime - Context manager to measure execution time of a code block
Muffle - Silences output and can replay it on error (like @silent)
DoNothing - Context manager that does nothing (no-op)
SetMPStartMethod - Context manager to temporarily set multiprocessing start method

import time
import stouputils as stp

# Everything printed inside this block also goes to run.log
with stp.LogToFile("run.log"):
    stp.info("Training started")

    # Print how long the indented block took
    with stp.MeasureTime(stp.progress, "Loading the dataset"):
        time.sleep(0.6)

    # Hide everything printed inside this block
    with stp.Muffle():
        print("A chatty library prints this, nobody sees it")

    stp.warning("Validation loss went up")

stp.info("Content of run.log:")
print(stp.read_file("run.log"))

Terminal output of the example, the log file holding the same lines

common#

class AbstractBothContextManager[source]#

Bases: AbstractContextManager, AbstractAsyncContextManager, Generic

Abstract base class for context managers that support both synchronous and asynchronous usage.

do_nothing#

class DoNothing(*args: Any, **kwargs: Any)[source]#

Bases: AbstractBothContextManager[DoNothing]

Context manager that does nothing.

This is a no-op context manager that can be used as a placeholder or for conditional context management.

Different from contextlib.nullcontext because it handles args and kwargs, along with async context management.

>>> with DoNothing():
...     print("This will be printed normally")
This will be printed normally

>>> # Conditional context management
>>> some_condition = True
>>> ctx = DoNothing() if some_condition else Muffle()
>>> with ctx:
...     print("May or may not be printed depending on condition")
May or may not be printed depending on condition

NullContextManager[source]#: Alias for DoNothing context manager

log_to_file#

ROUTING_LOCK: allocate_lock = <unlocked _thread.lock object>[source]#: Keeps two threads opening or closing a LogToFile at once from each putting their own TeeMultiOutput in place.

class LogToFile( path: str, mode: str = 'w', encoding: str = 'utf-8', tee_stdout: bool = True, tee_stderr: bool = True, strip_colors: bool = False, ignore_lineup: bool = True, restore_on_exit: bool = True, capture_fd: bool = False, )[source]#

Bases: AbstractBothContextManager[LogToFile]

Context manager to log to a file.

This context manager allows you to temporarily log output to a file while still printing normally.

Every LogToFile writes through the same TeeMultiOutput, put in place by the first one and taken away with the last one, so they nest and close in any order. One opened in the main thread receives the output of every thread, one opened in another thread only the output of that thread.

Parameters:

path – Path to the log file
mode – Mode to open the file in (default: “w”)
encoding – Encoding to use for the file (default: “utf-8”)
tee_stdout – Whether to redirect stdout to the file (default: True)
tee_stderr – Whether to redirect stderr to the file (default: True)
ignore_lineup – Whether the file receives what a terminal ends up showing (default: True): a progress bar once in its final state, and no line the cursor moves back to
restore_on_exit – Whether the last LogToFile to close gives stdout/stderr back (default: True)
capture_fd – Whether the file also receives what bypasses sys.stdout (default: False): C extensions, os.system, uncaptured child processes. Skipped with a warning where unsupported, see FdCapture.

> import stouputils as stp
> with stp.LogToFile("output.log"):
>     stp.info("This will be logged to output.log and printed normally")
>     print("This will also be logged")

> with stp.LogToFile("output.log") as log_ctx:
>     stp.warning("This will be logged to output.log and printed normally")
>     log_ctx.change_file("new_file.log")
>     print("This will be logged to new_file.log")

path: str[source]#: Attribute remembering path to the log file

mode: str[source]#: Attribute remembering mode to open the file in

encoding: str[source]#: Attribute remembering encoding to use for the file

tee_stdout: bool[source]#: Whether to redirect stdout to the file

tee_stderr: bool[source]#: Whether to redirect stderr to the file

strip_colors: bool[source]#: Whether to strip ANSI color codes from output sent to non-stdout/stderr files

ignore_lineup: bool[source]#: Whether the file receives what a terminal ends up showing

restore_on_exit: bool[source]#: Whether the last LogToFile to close gives stdout/stderr back

file: IO[Any][source]#: Attribute remembering opened file

tees: dict[str, TeeMultiOutput][source]#: The stream objects holding the file, by the name of the sys attribute they stand in for

capture_fd: bool[source]#: Whether the file also receives what bypasses sys.stdout

fd_capture: FdCapture | None[source]#: Capture of the descriptors while it runs, through which the file then receives everything

change_file(new_path: str) → None[source]#

Change the log file to a new path.

Parameters:: new_path – New path to the log file

static common(

logs_folder: str,

filepath: str,

func: CallableAny,

init_kwargs: dict[str, Any] | None = None,

*args: Any,

**kwargs: Any,

) → Any[source]#

Common code used at the beginning of a program to launch main function

Parameters:

logs_folder – Folder to store logs in
filepath – Path to the main function
func – Main function to launch
init_kwargs – Keyword arguments to pass to LogToFile constructor
*args – Arguments to pass to the main function
**kwargs – Keyword arguments to pass to the main function

Returns:

Return value of the main function

>>> if __name__ == "__main__":
...     LogToFile.common(f"{ROOT}/logs", __file__, main, init_kwargs={"strip_colors": True})

measure_time#

class MeasureTime(print_func: ~collections.abc.Callable[[...], None] = <function debug>, message: str = 'Execution time', perf_counter: bool = True)[source]#

Bases: AbstractBothContextManager[MeasureTime]

Context manager to measure execution time.

This context manager measures the execution time of the code block it wraps and prints the result using a specified print function.

Parameters:

print_func – Function to use to print the execution time (e.g. debug, info, warning, error, etc.).
message – Message to display with the execution time. Defaults to “Execution time”.
perf_counter – Whether to use time.perf_counter_ns or time.time_ns. Defaults to True.

> import time
> import stouputils as stp
> with stp.MeasureTime(stp.info, message="My operation"):
...     time.sleep(0.5)
> # [INFO HH:MM:SS] My operation: 500.123ms (500123456ns)

> with stp.MeasureTime(): # Uses debug by default
...     time.sleep(0.1)
> # [DEBUG HH:MM:SS] Execution time: 100.456ms (100456789ns)

print_func: Callable[[...], None][source]#: Function to use for printing the execution time

message: str[source]#: Message to display with the execution time

perf_counter: bool[source]#: Whether to use time.perf_counter_ns or time.time_ns

ns: Callable[[], int][source]#: Time function to use

start_ns: int[source]#: Start time in nanoseconds

muffle#

class ErrorLevelDetector(level: int)[source]#

Bases: Handler

Logging handler that only remembers whether a record at/above its level was emitted.

Used by Muffle to decide whether captured output should be replayed, since some libraries log their failures instead of raising.

triggered: bool[source]#: Whether a record at/above the configured level has been seen

class Muffle( mute_stderr: bool = False, replay_on_error: bool = False, error_log_level: int | None = None, watch_loggers: Sequence[str] | None = None, )[source]#

Bases: AbstractBothContextManager[Muffle]

Context manager that temporarily silences output. (No thread-safety guaranteed)

Alternative to @silent

By default the output is sent to devnull and lost. When replay_on_error is enabled, the output is captured in memory instead and only written back to the original stream if an error occurs inside the block. An “error” is either an exception propagating out of the block, or - when error_log_level is set - any logging record at/above that level emitted by the watched loggers (handy for libraries that log their failures instead of raising).

Parameters:

mute_stderr – Whether to mute stderr as well as stdout
replay_on_error – Capture output in memory and replay it if an error occurs
error_log_level – Also treat log records at/above this level as an error (e.g. logging.ERROR)
watch_loggers – Names of the loggers to watch for error_log_level (default: root logger only)

>>> with Muffle():
...     print("This will not be printed")

>>> # Replays the captured output because an exception escapes the block
>>> try:
...     with Muffle(replay_on_error=True):
...         print("Context that explains the failure below")
...         raise ValueError("boom")
... except ValueError:
...     pass
Context that explains the failure below

>>> # Stays silent because nothing went wrong
>>> import logging
>>> with Muffle(replay_on_error=True, error_log_level=logging.ERROR):
...     print("This will not be printed")

mute_stderr: bool[source]#: Attribute remembering if stderr should be muted

replay_on_error: bool[source]#: Attribute remembering if captured output should be replayed on error

error_log_level: int | None[source]#: Attribute remembering the logging level that counts as an error (if any)

watch_loggers: Sequence[str] | None[source]#: Attribute remembering which loggers to watch for error records

original_stdout: IO[Any][source]#: Attribute remembering original stdout

original_stderr: IO[Any][source]#: Attribute remembering original stderr

attach_detector(level: int) → None[source]#: Watch the loggers for records at level or above, so a failure that is only logged also replays the output.

detach_detector() → bool[source]#

Stop watching the loggers.

Returns:: Whether a record at the watched level was logged meanwhile.

static write_safely( stream: IO[Any], text: str, ) → None[source]#

Write text to a stream, replacing characters the stream’s encoding cannot represent.

Captured output may contain unicode (e.g. Rich panels) that a narrow console encoding (such as cp1252 on Windows) cannot encode, which would otherwise raise UnicodeEncodeError.

set_mp_start_method#

class SetMPStartMethod(start_method: str | None)[source]#

Bases: AbstractBothContextManager[SetMPStartMethod]

Context manager to temporarily set multiprocessing start method.

This context manager allows you to temporarily change the multiprocessing start method and automatically restores the original method when exiting the context.

Parameters:: start_method – The start method to use: “spawn”, “fork”, or “forkserver”

> import multiprocessing as mp
> import stouputils as stp
> # Temporarily use spawn method
> with stp.SetMPStartMethod("spawn"):
> ...     # Your multiprocessing code here
> ...     pass

> # Original method is automatically restored

start_method: str | None[source]#: The start method to use

old_method: str | None[source]#: The original start method to restore