tessl/pypi-wurlitzer

Capture C-level stdout/stderr pipes in Python via os.dup2

Overview

Eval results

Files

Output Capture

Name: tessl/pypi-wurlitzer
Author: tessl

Context manager-based C-level output capture for temporary redirection of stdout and stderr streams.

Capabilities

Basic Output Capture

Capture both stdout and stderr in separate streams using the default PIPE mode.

def pipes(stdout=PIPE, stderr=PIPE, encoding='utf-8', bufsize=None):
    """Capture C-level stdout/stderr in a context manager.
    
    Args:
        stdout: None, PIPE, Writable, or Logger object (default: PIPE)
        stderr: None, PIPE, STDOUT, Writable, or Logger object (default: PIPE)
        encoding: Text encoding for captured output (default: 'utf-8')
        bufsize: Pipe buffer size in bytes (default: auto-detected)
        
    Returns:
        Tuple of (stdout, stderr) streams when used as context manager
    """

Example usage:

from wurlitzer import pipes

with pipes() as (out, err):
    # C-level function calls here
    libc.printf(b"Hello from C stdout\n")
    libc.fprintf(stderr_fd, b"Error from C stderr\n")

stdout_content = out.read()  # "Hello from C stdout\n"
stderr_content = err.read()  # "Error from C stderr\n"

Combined Stream Capture

Redirect both stdout and stderr to a single destination using the STDOUT constant.

from wurlitzer import pipes, STDOUT

with pipes(stderr=STDOUT) as (out, err):
    # Both stdout and stderr go to 'out'
    call_c_function_with_mixed_output()
    # err will be None

combined_output = out.read()

Custom Stream Destinations

Direct output to specific file-like objects, files, or StringIO instances.

from io import StringIO
from wurlitzer import pipes, STDOUT

# Redirect to custom StringIO
output_buffer = StringIO()
with pipes(stdout=output_buffer, stderr=STDOUT):
    c_function_call()

captured_text = output_buffer.getvalue()

File Output

Write captured output directly to files for persistent storage.

from wurlitzer import pipes, STDOUT

with open("output.log", "w") as logfile:
    with pipes(stdout=logfile, stderr=STDOUT):
        long_running_c_process()
# Output is written directly to output.log

Logger Integration

Forward captured output to Python logging objects with configurable log levels.

def pipes(stdout=Logger, stderr=Logger, encoding='utf-8', bufsize=None):
    """When Logger objects are passed, each line becomes a log message.
    
    stdout Logger messages: INFO level
    stderr Logger messages: ERROR level
    """

Example with logging:

import logging
from wurlitzer import pipes, STDOUT

logger = logging.getLogger("c_output")
logger.setLevel(logging.INFO)

# Add handlers as needed
handler = logging.FileHandler("c_output.log")
logger.addHandler(handler)

with pipes(stdout=logger, stderr=STDOUT):
    c_function_with_debug_output()
    # Each line of C output becomes a log entry

Binary Output Mode

Capture raw bytes without text encoding for binary data handling.

from wurlitzer import pipes

with pipes(encoding=None) as (out, err):
    c_function_producing_binary_data()

binary_stdout = out.read()  # Returns bytes object
binary_stderr = err.read()  # Returns bytes object

Buffer Size Configuration

Control pipe buffer sizes for performance optimization, especially on Linux systems.

from wurlitzer import pipes

# Large buffer for high-volume output
with pipes(bufsize=1024*1024) as (out, err):  # 1MB buffer
    c_function_with_lots_of_output()

# Disable buffer size setting
with pipes(bufsize=0) as (out, err):
    c_function_call()

Usage Patterns

Scientific Computing

Capture output from NumPy, SciPy, or other libraries with C extensions:

import numpy as np
from wurlitzer import pipes

with pipes() as (out, err):
    # NumPy operations that may produce C-level warnings
    result = np.linalg.solve(matrix_a, vector_b)

warnings = err.read()
if warnings:
    print(f"NumPy warnings: {warnings}")

Testing and Debugging

Verify that C-level functions produce expected output:

from wurlitzer import pipes

def test_c_function_output():
    with pipes() as (out, err):
        my_c_extension.debug_function()
    
    output = out.read()
    assert "Expected debug message" in output
    assert err.read() == ""  # No errors

Jupyter Notebook Usage

Clean output capture in notebook environments:

from wurlitzer import pipes

with pipes() as (out, err):
    # C library calls won't pollute notebook output
    some_noisy_c_library.process_data()

# Examine output programmatically
if out.read().strip():
    print("C function produced output")

Install with Tessl CLI