tessl/pypi-pyvips

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

—

Pending

Overview

Eval results

Files

I/O Connections

Name: tessl/pypi-pyvips
Author: tessl

Advanced I/O system supporting custom sources and targets for streaming operations, enabling integration with various data sources and destinations beyond the filesystem. This system provides flexibility for network streams, databases, cloud storage, and custom protocols.

Capabilities

Source Objects

Input connections for reading image data from various sources with streaming support.

class Source:
    """Input connection for streaming image data."""
    
    @classmethod
    def new_from_file(cls, filename: str) -> 'Source':
        """
        Create source from file.
        
        Parameters:
        - filename: str, path to input file
        
        Returns:
        Source object for the file
        """
    
    @classmethod  
    def new_from_memory(cls, data: bytes) -> 'Source':
        """
        Create source from memory buffer.
        
        Parameters:
        - data: bytes, image data in memory
        
        Returns:
        Source object for the memory buffer
        """
    
    @classmethod
    def new_from_descriptor(cls, descriptor: int) -> 'Source':
        """
        Create source from file descriptor.
        
        Parameters:
        - descriptor: int, file descriptor
        
        Returns:
        Source object for the descriptor
        """
    
    def filename(self) -> str:
        """Get associated filename if available."""
    
    def nick(self) -> str:
        """Get human-readable name for the source."""

Example usage:

# File source
source = pyvips.Source.new_from_file('input.jpg')
image = pyvips.Image.new_from_source(source, '')

# Memory source
with open('image.png', 'rb') as f:
    data = f.read()
source = pyvips.Source.new_from_memory(data)
image = pyvips.Image.new_from_source(source, '.png')

# File descriptor source (Unix systems)
import os
fd = os.open('image.tiff', os.O_RDONLY)
source = pyvips.Source.new_from_descriptor(fd)
image = pyvips.Image.new_from_source(source, '.tiff')
os.close(fd)

# Source introspection
print(f"Source filename: {source.filename()}")
print(f"Source name: {source.nick()}")

Target Objects

Output connections for writing image data to various destinations with streaming support.

class Target:
    """Output connection for streaming image data."""
    
    @classmethod
    def new_to_file(cls, filename: str) -> 'Target':
        """
        Create target for file output.
        
        Parameters:
        - filename: str, path to output file
        
        Returns:
        Target object for the file
        """
    
    @classmethod
    def new_to_memory(cls) -> 'Target':
        """
        Create target for memory output.
        
        Returns:
        Target object for memory buffer
        """
    
    @classmethod
    def new_to_descriptor(cls, descriptor: int) -> 'Target':
        """
        Create target for file descriptor.
        
        Parameters:
        - descriptor: int, file descriptor
        
        Returns:
        Target object for the descriptor
        """
    
    def filename(self) -> str:
        """Get associated filename if available."""
    
    def nick(self) -> str:
        """Get human-readable name for the target."""

Example usage:

# File target
target = pyvips.Target.new_to_file('output.jpg')
image.write_to_target(target, '.jpg', Q=90)

# Memory target
target = pyvips.Target.new_to_memory()
image.write_to_target(target, '.png', compression=6)
# Note: Getting data from memory target requires libvips >= 8.13

# File descriptor target (Unix systems)
import os
fd = os.open('output.webp', os.O_WRONLY | os.O_CREAT, 0o644)
target = pyvips.Target.new_to_descriptor(fd)
image.write_to_target(target, '.webp', Q=80)
os.close(fd)

# Target introspection
print(f"Target filename: {target.filename()}")
print(f"Target name: {target.nick()}")

Custom Sources

Create custom input sources for specialized data sources and protocols.

class SourceCustom(Source):
    """Custom input source for specialized data sources."""
    
    def __init__(self):
        """Create new custom source."""
    
    def on_read(self, handler: callable) -> None:
        """
        Attach read handler.
        
        Parameters:
        - handler: callable, function(size) -> bytes
          Should return up to 'size' bytes, empty bytes for EOF
        """
    
    def on_seek(self, handler: callable) -> None:
        """
        Attach seek handler (optional).
        
        Parameters:
        - handler: callable, function(offset, whence) -> int
          Should seek to position and return new position
          whence: 0=SEEK_SET, 1=SEEK_CUR, 2=SEEK_END
        """

Example usage:

# HTTP source example
import requests

class HTTPSource:
    def __init__(self, url):
        self.response = requests.get(url, stream=True)
        self.iter = iter(self.response.iter_content(chunk_size=8192))
        self.position = 0
        
    def read_handler(self, size):
        try:
            chunk = next(self.iter)
            self.position += len(chunk)
            return chunk
        except StopIteration:
            return b''  # EOF

# Use HTTP source
http_source = HTTPSource('https://example.com/image.jpg')
source = pyvips.SourceCustom()
source.on_read(http_source.read_handler)
image = pyvips.Image.new_from_source(source, '')

# Database source example
import sqlite3

class DatabaseSource:
    def __init__(self, db_path, image_id):
        self.conn = sqlite3.connect(db_path)
        cursor = self.conn.cursor()
        cursor.execute("SELECT image_data FROM images WHERE id = ?", (image_id,))
        self.data = cursor.fetchone()[0]
        self.position = 0
        
    def read_handler(self, size):
        if self.position >= len(self.data):
            return b''
        chunk = self.data[self.position:self.position + size]
        self.position += len(chunk)
        return chunk
        
    def seek_handler(self, offset, whence):
        if whence == 0:  # SEEK_SET
            self.position = offset
        elif whence == 1:  # SEEK_CUR
            self.position += offset
        elif whence == 2:  # SEEK_END
            self.position = len(self.data) + offset
        return self.position

# Use database source
db_source = DatabaseSource('images.db', 123)
source = pyvips.SourceCustom()
source.on_read(db_source.read_handler)
source.on_seek(db_source.seek_handler)
image = pyvips.Image.new_from_source(source, '')

Custom Targets

Create custom output targets for specialized data destinations and protocols.

class TargetCustom(Target):
    """Custom output target for specialized destinations."""
    
    def __init__(self):
        """Create new custom target."""
    
    def on_write(self, handler: callable) -> None:
        """
        Attach write handler.
        
        Parameters:
        - handler: callable, function(data) -> int
          Should write bytes and return number of bytes written
        """
    
    def on_read(self, handler: callable) -> None:
        """
        Attach read handler (optional, for read-back capability).
        
        Parameters:
        - handler: callable, function(size) -> bytes
        """
    
    def on_seek(self, handler: callable) -> None:
        """
        Attach seek handler (optional).
        
        Parameters:
        - handler: callable, function(offset, whence) -> int
        """
    
    def on_end(self, handler: callable) -> None:
        """
        Attach end handler (libvips >= 8.13).
        
        Parameters:
        - handler: callable, function() -> None
          Called when write operation completes
        """
    
    def on_finish(self, handler: callable) -> None:
        """
        Attach finish handler (deprecated, use on_end).
        
        Parameters:
        - handler: callable, function() -> None
        """

Example usage:

# HTTP upload target example
import requests

class HTTPUploadTarget:
    def __init__(self, upload_url):
        self.upload_url = upload_url
        self.buffer = bytearray()
        
    def write_handler(self, data):
        self.buffer.extend(data)
        return len(data)
        
    def end_handler(self):
        # Upload complete buffer
        files = {'image': ('image.jpg', bytes(self.buffer), 'image/jpeg')}
        response = requests.post(self.upload_url, files=files)
        print(f"Upload status: {response.status_code}")

# Use HTTP upload target
upload_target = HTTPUploadTarget('https://api.example.com/upload')
target = pyvips.TargetCustom()
target.on_write(upload_target.write_handler)
target.on_end(upload_target.end_handler)
image.write_to_target(target, '.jpg', Q=90)

# Cloud storage target example
import boto3

class S3Target:
    def __init__(self, bucket, key):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.key = key
        self.parts = []
        self.buffer = bytearray()
        
    def write_handler(self, data):
        self.buffer.extend(data)
        return len(data)
        
    def end_handler(self):
        # Upload to S3
        self.s3.put_object(
            Bucket=self.bucket,
            Key=self.key,
            Body=bytes(self.buffer)
        )
        print(f"Uploaded to s3://{self.bucket}/{self.key}")

# Use S3 target
s3_target = S3Target('my-bucket', 'processed/image.jpg')
target = pyvips.TargetCustom()
target.on_write(s3_target.write_handler)
target.on_end(s3_target.end_handler)
image.write_to_target(target, '.jpg', Q=85)

# Database target example
class DatabaseTarget:
    def __init__(self, db_path, image_id):
        self.conn = sqlite3.connect(db_path)
        self.image_id = image_id
        self.buffer = bytearray()
        
    def write_handler(self, data):
        self.buffer.extend(data)
        return len(data)
        
    def end_handler(self):
        cursor = self.conn.cursor()
        cursor.execute(
            "UPDATE images SET image_data = ? WHERE id = ?",
            (bytes(self.buffer), self.image_id)
        )
        self.conn.commit()
        self.conn.close()

# Use database target
db_target = DatabaseTarget('images.db', 456)
target = pyvips.TargetCustom()
target.on_write(db_target.write_handler)
target.on_end(db_target.end_handler)
image.write_to_target(target, '.png', compression=6)

Connection Base Class

Common functionality for all connection types.

class Connection:
    """Abstract base connection class."""
    
    def filename(self) -> str:
        """
        Get associated filename.
        
        Returns:
        str, filename if available, empty string otherwise
        """
    
    def nick(self) -> str:
        """
        Get human-readable name.
        
        Returns:
        str, descriptive name for the connection
        """

Streaming Benefits

The I/O connection system provides several advantages:

Memory Efficiency

Process large images without loading entirely into memory
Constant memory usage regardless of image size
Efficient for gigapixel+ images

Network Integration

Stream images directly from HTTP/HTTPS sources
Upload processed results without temporary files
Support for various protocols and APIs

Database Integration

Load and save images directly to/from databases
Support for BLOB fields and binary data
No intermediate file storage required

Custom Protocols

Implement specialized data sources and sinks
Support for encrypted or compressed streams
Integration with existing data pipelines

Error Handling

try:
    source = pyvips.Source.new_from_file('nonexistent.jpg')
    image = pyvips.Image.new_from_source(source, '')
except pyvips.Error as e:
    print(f"Source creation failed: {e.message}")

# Custom source error handling
class SafeHTTPSource:
    def __init__(self, url):
        try:
            self.response = requests.get(url, stream=True, timeout=30)
            self.response.raise_for_status()
            self.iter = iter(self.response.iter_content(chunk_size=8192))
        except requests.RequestException as e:
            raise pyvips.Error(f"HTTP request failed: {e}")
    
    def read_handler(self, size):
        try:
            chunk = next(self.iter)
            return chunk
        except StopIteration:
            return b''
        except Exception as e:
            raise pyvips.Error(f"Read failed: {e}")

# Robust custom target
class SafeFileTarget:
    def __init__(self, filename):
        self.filename = filename
        self.file = None
        
    def write_handler(self, data):
        try:
            if self.file is None:
                self.file = open(self.filename, 'wb')
            written = self.file.write(data)
            return written
        except IOError as e:
            raise pyvips.Error(f"Write failed: {e}")
    
    def end_handler(self):
        if self.file:
            self.file.close()