tessl/pypi-pyvips

Python binding for the libvips image processing library with high-performance streaming architecture

—

Pending

Overview

Eval results

Files

System Control

Name: tessl/pypi-pyvips
Author: tessl

Advanced functionality for operation introspection, cache management, and low-level system control for fine-tuning performance and behavior. These utilities provide control over libvips internals and enable advanced optimization strategies.

Capabilities

Operation System

Direct access to libvips operations and introspection capabilities for dynamic operation calling and documentation generation.

class Operation:
    """Call libvips operations and provide introspection."""
    
    @classmethod
    def call(cls, operation_name: str, *args, **kwargs):
        """
        Call any libvips operation by name.
        
        Parameters:
        - operation_name: str, name of the libvips operation
        - *args: positional arguments for the operation
        - **kwargs: keyword arguments for the operation
        
        Returns:
        Operation result (varies by operation)
        """
    
    @classmethod
    def generate_docstring(cls, operation_name: str) -> str:
        """
        Generate Google-style docstring for operation.
        
        Parameters:
        - operation_name: str, operation name
        
        Returns:
        str, formatted docstring with parameters and description
        """
    
    @classmethod
    def generate_sphinx(cls, operation_name: str) -> str:
        """
        Generate Sphinx-style docstring for operation.
        
        Parameters:
        - operation_name: str, operation name
        
        Returns:
        str, Sphinx-formatted docstring
        """
    
    @classmethod
    def generate_sphinx_all(cls) -> str:
        """
        Generate Sphinx documentation for all operations.
        
        Returns:
        str, complete Sphinx documentation
        """

class Introspect:
    """Build and cache introspection data for operations."""
    
    @classmethod
    def get(cls, operation_name: str) -> 'Introspect':
        """
        Get cached introspection data for operation.
        
        Parameters:
        - operation_name: str, operation name
        
        Returns:
        Introspect object with operation details
        """
    
    # Properties
    description: str           # Operation description
    flags: int                # Operation flags
    required_input: list      # Required input parameters
    optional_input: list      # Optional input parameters  
    required_output: list     # Required output parameters
    optional_output: list     # Optional output parameters

Example usage:

# Direct operation calling
result = pyvips.Operation.call('add', image1, image2)
blurred = pyvips.Operation.call('gaussblur', image, 2.0)

# Advanced operation with all parameters
resized = pyvips.Operation.call('resize', image, 
    scale=0.5, 
    vscale=0.8,
    kernel='lanczos3')

# Get operation documentation
doc = pyvips.Operation.generate_docstring('gaussblur')
print(doc)

# Introspect operation details
introspect = pyvips.Introspect.get('resize')
print(f"Description: {introspect.description}")
print(f"Required inputs: {introspect.required_input}")
print(f"Optional inputs: {introspect.optional_input}")

# List all available operations
import pyvips
operations = []
def collect_operations(name, operation):
    operations.append(name)
pyvips.type_map(pyvips.type_from_name('VipsOperation'), collect_operations)
print(f"Available operations: {len(operations)}")

# Dynamic operation discovery
def call_operation_safely(op_name, *args, **kwargs):
    """Safely call operation with error handling."""
    try:
        introspect = pyvips.Introspect.get(op_name)
        print(f"Calling {op_name}: {introspect.description}")
        return pyvips.Operation.call(op_name, *args, **kwargs)
    except pyvips.Error as e:
        print(f"Operation {op_name} failed: {e}")
        return None

# Use dynamic calling
result = call_operation_safely('thumbnail', image, 200)

Cache Management

Control libvips operation cache for memory management and performance optimization.

def cache_set_max(mx: int) -> None:
    """
    Set maximum number of operations in cache.
    
    Parameters:
    - mx: int, maximum cache entries (0 = unlimited)
    """

def cache_set_max_mem(mx: int) -> None:
    """
    Set maximum cache memory usage.
    
    Parameters:
    - mx: int, maximum memory in bytes (0 = unlimited)
    """

def cache_set_max_files(mx: int) -> None:
    """
    Set maximum open files in cache.
    
    Parameters:
    - mx: int, maximum open files (0 = unlimited)
    """

def cache_set_trace(trace: bool) -> None:
    """
    Enable/disable cache operation tracing.
    
    Parameters:
    - trace: bool, enable tracing for debugging
    """

def cache_get_max() -> int:
    """Get maximum cache entries setting."""

def cache_get_size() -> int:
    """Get current cache size (number of entries)."""

def cache_get_max_mem() -> int:
    """Get maximum cache memory setting."""

def cache_get_max_files() -> int:
    """Get maximum open files setting."""

Example usage:

# Cache configuration for memory-constrained environments
pyvips.cache_set_max_mem(100 * 1024 * 1024)  # 100MB limit
pyvips.cache_set_max_files(50)                # Limit open files
pyvips.cache_set_max(200)                     # Limit cache entries

# Cache configuration for high-performance environments
pyvips.cache_set_max_mem(1 * 1024 * 1024 * 1024)  # 1GB limit
pyvips.cache_set_max_files(1000)                    # More open files
pyvips.cache_set_max(0)                             # Unlimited entries

# Monitor cache usage
def print_cache_stats():
    print(f"Cache size: {pyvips.cache_get_size()}")
    print(f"Max entries: {pyvips.cache_get_max()}")
    print(f"Max memory: {pyvips.cache_get_max_mem()} bytes")
    print(f"Max files: {pyvips.cache_get_max_files()}")

print_cache_stats()

# Enable cache tracing for debugging
pyvips.cache_set_trace(True)
image = pyvips.Image.new_from_file('test.jpg')
resized = image.resize(0.5)  # Will show cache operations
pyvips.cache_set_trace(False)

# Adaptive cache management
def setup_cache_for_batch_processing(image_count, avg_image_size_mb):
    """Configure cache for batch processing workload."""
    
    # Estimate memory needs
    total_mem_mb = image_count * avg_image_size_mb * 2  # Factor for operations
    
    # Set conservative limits
    pyvips.cache_set_max_mem(min(total_mem_mb * 1024 * 1024, 512 * 1024 * 1024))
    pyvips.cache_set_max_files(min(image_count * 2, 200))
    pyvips.cache_set_max(min(image_count * 5, 500))
    
    print(f"Cache configured for {image_count} images")
    print_cache_stats()

# Dynamic cache adjustment
def monitor_and_adjust_cache():
    """Monitor cache performance and adjust settings."""
    import psutil
    
    # Get system memory info
    memory = psutil.virtual_memory()
    available_mb = memory.available // (1024 * 1024)
    
    # Adjust cache based on available memory
    if available_mb > 2048:  # > 2GB available
        pyvips.cache_set_max_mem(512 * 1024 * 1024)  # Use 512MB
    elif available_mb > 1024:  # > 1GB available
        pyvips.cache_set_max_mem(256 * 1024 * 1024)  # Use 256MB
    else:  # Low memory
        pyvips.cache_set_max_mem(64 * 1024 * 1024)   # Use 64MB
    
    print(f"Adjusted cache for {available_mb}MB available memory")

Version Information

Access libvips version information for compatibility checking and feature detection.

def version(flag: int) -> int:
    """
    Get libvips version information.
    
    Parameters:
    - flag: int, version component (0=major, 1=minor, 2=micro)
    
    Returns:
    int, version number component
    """

def at_least_libvips(x: int, y: int) -> bool:
    """
    Check if libvips version is at least x.y.
    
    Parameters:
    - x: int, major version
    - y: int, minor version
    
    Returns:
    bool, True if version >= x.y
    """

Example usage:

# Get version information
major = pyvips.version(0)
minor = pyvips.version(1) 
micro = pyvips.version(2)
print(f"libvips version: {major}.{minor}.{micro}")

# Check for specific features
if pyvips.at_least_libvips(8, 13):
    print("TargetCustom.on_end() is available")
    # Use newer features
else:
    print("Using legacy TargetCustom.on_finish()")
    # Use older API

# Feature detection
def has_feature(feature_name, min_major, min_minor):
    """Check if a feature is available in current libvips version."""
    return pyvips.at_least_libvips(min_major, min_minor)

# Check various features
features = {
    'custom_sources': (8, 9),
    'target_custom_end': (8, 13),
    'heif_support': (8, 8),
    'webp_animation': (8, 9)
}

for feature, (maj, min) in features.items():
    available = has_feature(feature, maj, min)
    print(f"{feature}: {'✓' if available else '✗'}")

# Version-dependent operation calling
def safe_operation_call(op_name, required_version, *args, **kwargs):
    """Call operation only if libvips version supports it."""
    maj, min = required_version
    if pyvips.at_least_libvips(maj, min):
        return pyvips.Operation.call(op_name, *args, **kwargs)
    else:
        raise pyvips.Error(f"Operation {op_name} requires libvips >= {maj}.{min}")

# Use version checking
try:
    result = safe_operation_call('heifload', (8, 8), 'image.heic')
    print("HEIF loading successful")
except pyvips.Error as e:
    print(f"HEIF not supported: {e}")

Security Controls (libvips >= 8.13)

Security features for controlling potentially dangerous operations in untrusted environments.

def block_untrusted_set(state: bool) -> None:
    """
    Set global block state for untrusted operations.
    
    Parameters:
    - state: bool, True to block untrusted operations
    """

def operation_block_set(name: str, state: bool) -> None:
    """
    Set block state for specific operation.
    
    Parameters:
    - name: str, operation name to block/unblock
    - state: bool, True to block operation
    """

Example usage:

# Security setup for web service
if pyvips.at_least_libvips(8, 13):
    # Block potentially dangerous operations
    pyvips.block_untrusted_set(True)
    
    # Block specific operations that might be security risks
    dangerous_ops = [
        'system',           # System command execution
        'pdfload',          # PDF might contain malicious content
        'svgload',          # SVG can contain scripts
        'magickload'        # ImageMagick has security history
    ]
    
    for op in dangerous_ops:
        try:
            pyvips.operation_block_set(op, True)
            print(f"Blocked operation: {op}")
        except:
            print(f"Could not block {op} (might not exist)")

# Safe image processing for user uploads
def process_user_upload(image_data, max_size_mb=10):
    """Safely process user-uploaded images."""
    
    # Size check
    if len(image_data) > max_size_mb * 1024 * 1024:
        raise ValueError(f"Image too large: {len(image_data)} bytes")
    
    # Enable security restrictions
    if pyvips.at_least_libvips(8, 13):
        pyvips.block_untrusted_set(True)
    
    try:
        # Load with restrictions
        image = pyvips.Image.new_from_buffer(image_data, '')
        
        # Limit output size
        if image.width > 4096 or image.height > 4096:
            image = image.thumbnail_image(4096)
        
        # Safe processing only
        result = (image
            .colourspace('srgb')
            .resize(0.8)
            .sharpen())
        
        return result.write_to_buffer('.jpg', Q=85, strip=True)
    
    except pyvips.Error as e:
        raise ValueError(f"Image processing failed: {e}")
    
    finally:
        # Reset security state if needed
        if pyvips.at_least_libvips(8, 13):
            pyvips.block_untrusted_set(False)

System Management

Low-level system control and resource management functions.

def leak_set(leak: bool) -> None:
    """
    Enable/disable libvips leak checking.
    
    Parameters:
    - leak: bool, enable leak detection for debugging
    """

def shutdown() -> None:
    """
    Shut down libvips system.
    Call before program exit to clean up resources.
    """

Example usage:

# Development/debugging setup
import atexit

def setup_debug_environment():
    """Configure libvips for debugging."""
    # Enable leak checking
    pyvips.leak_set(True)
    
    # Enable cache tracing
    pyvips.cache_set_trace(True)
    
    # Register cleanup
    atexit.register(pyvips.shutdown)
    
    print("Debug environment configured")

# Production setup
def setup_production_environment():
    """Configure libvips for production."""
    # Disable leak checking for performance
    pyvips.leak_set(False)
    
    # Optimize cache for server workload
    pyvips.cache_set_max_mem(256 * 1024 * 1024)  # 256MB
    pyvips.cache_set_max_files(100)
    pyvips.cache_set_max(1000)
    
    # Security setup
    if pyvips.at_least_libvips(8, 13):
        pyvips.block_untrusted_set(True)
    
    # Register cleanup
    atexit.register(pyvips.shutdown)
    
    print("Production environment configured")

# Application lifecycle management
class PyVipsManager:
    """Manage PyVips lifecycle and configuration."""
    
    def __init__(self, environment='production'):
        self.environment = environment
        self.setup()
    
    def setup(self):
        """Setup based on environment."""
        if self.environment == 'development':
            setup_debug_environment()
        else:
            setup_production_environment()
    
    def get_system_info(self):
        """Get system and version information."""
        return {
            'pyvips_version': pyvips.__version__,
            'libvips_version': f"{pyvips.version(0)}.{pyvips.version(1)}.{pyvips.version(2)}",
            'api_mode': pyvips.API_mode,
            'cache_size': pyvips.cache_get_size(),
            'cache_max': pyvips.cache_get_max(),
            'cache_max_mem': pyvips.cache_get_max_mem(),
            'cache_max_files': pyvips.cache_get_max_files()
        }
    
    def cleanup(self):
        """Clean shutdown."""
        print("Shutting down PyVips...")
        pyvips.shutdown()

# Use lifecycle manager
manager = PyVipsManager('production')
print("System info:", manager.get_system_info())

# Register cleanup
import atexit
atexit.register(manager.cleanup)

Performance Monitoring

import time
import psutil

class PerformanceMonitor:
    """Monitor PyVips performance and resource usage."""
    
    def __init__(self):
        self.start_time = time.time()
        self.initial_memory = psutil.Process().memory_info().rss
    
    def log_stats(self, operation):
        """Log performance statistics."""
        current_time = time.time()
        current_memory = psutil.Process().memory_info().rss
        
        cache_size = pyvips.cache_get_size()
        
        print(f"Operation: {operation}")
        print(f"  Time: {current_time - self.start_time:.2f}s")
        print(f"  Memory: {(current_memory - self.initial_memory) / 1024 / 1024:.1f}MB")
        print(f"  Cache size: {cache_size} entries")
        
        self.start_time = current_time
    
    def optimize_for_workload(self, workload_type):
        """Optimize cache settings for specific workload."""
        if workload_type == 'batch':
            pyvips.cache_set_max(2000)
            pyvips.cache_set_max_mem(512 * 1024 * 1024)
        elif workload_type == 'interactive':
            pyvips.cache_set_max(100)
            pyvips.cache_set_max_mem(128 * 1024 * 1024)
        elif workload_type == 'memory_constrained':
            pyvips.cache_set_max(50)
            pyvips.cache_set_max_mem(64 * 1024 * 1024)

# Use performance monitoring
monitor = PerformanceMonitor()
monitor.optimize_for_workload('batch')

image = pyvips.Image.new_from_file('large.tiff')
monitor.log_stats('load')

processed = image.resize(0.5).sharpen()
monitor.log_stats('process')

processed.write_to_file('output.jpg')
monitor.log_stats('save')

Install with Tessl CLI