tessl/pypi-wurlitzer

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

Overview

Eval results

Files

System Integration

Name: tessl/pypi-wurlitzer
Author: tessl

Forward C-level output to Python's sys.stdout/stderr streams, enabling integration with existing Python output handling systems like Jupyter notebooks, IPython, and logging frameworks.

Capabilities

System Stream Forwarding

Forward C-level stdout/stderr to Python sys streams without creating intermediate buffers.

def sys_pipes(encoding='utf-8', bufsize=None):
    """Redirect C-level stdout/stderr to sys.stdout/stderr
    
    Args:
        encoding: Text encoding for output (default: 'utf-8')
        bufsize: Pipe buffer size in bytes (default: auto-detected)
        
    Returns:
        Context manager that forwards C output to sys streams
        
    Note:
        This is useful when sys.stdout/stderr are already being forwarded
        somewhere, e.g., in a Jupyter kernel. DO NOT USE if sys.stdout and 
        sys.stderr are not already being forwarded.
    """

Example usage:

import sys
from wurlitzer import sys_pipes

# Ensure sys.stdout is being handled (e.g., in Jupyter)
with sys_pipes():
    # C-level output will appear in the same place as Python print()
    c_function_with_output()
    print("This Python output and C output appear together")

Jupyter/IPython Integration

Seamlessly integrate C-level output with Jupyter notebook cell outputs.

from wurlitzer import sys_pipes

# In a Jupyter notebook cell
with sys_pipes():
    # C library calls now appear in cell output
    scientific_c_library.compute_results()
    numpy_function_with_c_warnings()

Combined with Python Output

Mix C-level and Python output in the same stream for unified logging.

import sys
from wurlitzer import sys_pipes

def debug_function():
    print("Python: Starting computation")
    
    with sys_pipes():
        c_computation_library.process()  # C output mixed with Python
        
    print("Python: Computation complete")

Custom Stream Redirection

Use sys_pipes when sys.stdout/stderr have been redirected to custom handlers.

import sys
from io import StringIO
from wurlitzer import sys_pipes

# Redirect Python sys streams
captured_output = StringIO()
original_stdout = sys.stdout
sys.stdout = captured_output

try:
    with sys_pipes():
        # Both Python and C output go to captured_output
        print("Python output")
        c_function_call()  # C output
finally:
    sys.stdout = original_stdout

all_output = captured_output.getvalue()

Integration Patterns

Testing Framework Integration

Integrate with testing frameworks that capture sys.stdout/stderr:

import pytest
from wurlitzer import sys_pipes

def test_c_function_output(capsys):
    with sys_pipes():
        my_c_extension.function_under_test()
    
    captured = capsys.readouterr()
    assert "expected C output" in captured.out

Logging Framework Integration

Forward C output through Python's logging system:

import logging
import sys
from io import StringIO
from wurlitzer import sys_pipes

class LoggingHandler(logging.StreamHandler):
    def __init__(self):
        super().__init__(StringIO())
        
    def emit(self, record):
        # Process log records that may include C output
        super().emit(record)

# Setup logging to capture sys output
logger = logging.getLogger()
handler = LoggingHandler()
logger.addHandler(handler)

# Redirect sys.stdout to logger
sys.stdout = handler.stream

with sys_pipes():
    c_function_call()  # Output goes through logging system

Web Application Integration

Integrate with web frameworks where sys.stdout is redirected:

from wurlitzer import sys_pipes
import flask

app = flask.Flask(__name__)

@app.route('/process')
def process_data():
    output_buffer = []
    
    with sys_pipes():
        # C processing output captured in web context
        result = c_data_processor.analyze(request.data)
        
    return {"result": result, "processing_complete": True}

Error Handling and Validation

Stream Validation

sys_pipes includes validation to prevent infinite recursion:

import sys
from wurlitzer import sys_pipes

# This will raise ValueError to prevent infinite recursion
try:
    # When sys.stdout is the same as sys.__stdout__
    with sys_pipes():
        pass
except ValueError as e:
    print(f"Cannot forward streams: {e}")
    # Use regular pipes() instead
    from wurlitzer import pipes
    with pipes():
        c_function_call()

Fallback Patterns

Graceful degradation when sys_pipes cannot be used:

from wurlitzer import sys_pipes, pipes

def safe_c_call():
    try:
        with sys_pipes():
            return c_function_call()
    except ValueError:
        # Fallback to regular pipes
        with pipes() as (out, err):
            result = c_function_call()
            
        # Handle captured output manually
        stdout_content = out.read()
        if stdout_content:
            print(stdout_content, end='')
            
        return result

Thread Safety

sys_pipes is thread-safe and works correctly in multi-threaded environments:

import threading
from wurlitzer import sys_pipes

def worker_thread():
    with sys_pipes():
        thread_specific_c_function()

# Multiple threads can safely use sys_pipes
threads = [threading.Thread(target=worker_thread) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()

Install with Tessl CLI