tessl/pypi-asdf

Python implementation of the Advanced Scientific Data Format (ASDF) Standard

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Extension System

Name: tessl/pypi-asdf
Author: tessl

Plugin architecture for extending ASDF with custom types, validators, compressors, and tags. Enables seamless integration with domain-specific libraries and custom data formats while maintaining compatibility with the ASDF standard.

Capabilities

Base Extension Class

Abstract base class for creating ASDF extensions that define custom types, validation rules, and serialization behavior.

class Extension:
    """
    Abstract base class for ASDF extensions.
    """
    
    @property
    def extension_uri(self) -> str:
        """
        Unique URI identifying this extension.
        Must be implemented by subclasses.
        """
    
    @property
    def tags(self) -> list:
        """
        List of YAML tag URIs supported by this extension.
        """
    
    @property
    def converters(self) -> list:
        """
        List of Converter objects for handling custom types.
        """
    
    @property
    def compressors(self) -> list:
        """
        List of Compressor objects for custom compression schemes.
        """
    
    @property  
    def validators(self) -> list:
        """
        List of Validator objects for additional validation.
        """

Extension Proxy

Wrapper that provides default implementations and manages extension lifecycle.

class ExtensionProxy:
    """
    Wrapper providing default implementations for extensions.
    """
    
    def __init__(self, extension, package_name=None, package_version=None):
        """
        Create extension proxy.
        
        Parameters:
        - extension: Extension instance to wrap
        - package_name (str, optional): Package name for metadata
        - package_version (str, optional): Package version for metadata
        """

Converter System

Convert custom Python types to/from ASDF-serializable representations.

class Converter:
    """
    Abstract base class for type converters.
    """
    
    def can_convert(self, obj) -> bool:
        """
        Check if this converter can handle the given object.
        
        Parameters:
        - obj: Object to check
        
        Returns:
        bool: True if this converter can handle the object
        """
    
    def convert(self, obj, **kwargs):
        """
        Convert object to ASDF-serializable form.
        
        Parameters:
        - obj: Object to convert
        - **kwargs: Additional conversion options
        
        Returns:
        ASDF-serializable representation
        """
    
    def can_convert_to_tree(self, obj_type) -> bool:
        """
        Check if this converter can convert objects of given type to tree form.
        
        Parameters:
        - obj_type: Type to check
        
        Returns:
        bool: True if type can be converted to tree
        """
    
    def convert_to_tree(self, obj, ctx):
        """
        Convert object to tree representation for YAML serialization.
        
        Parameters:
        - obj: Object to convert
        - ctx: Serialization context
        
        Returns:
        Tree representation suitable for YAML
        """
    
    def convert_from_tree(self, tree, ctx):
        """
        Convert tree representation back to Python object.
        
        Parameters:
        - tree: Tree representation from YAML
        - ctx: Deserialization context
        
        Returns:
        Reconstructed Python object
        """

class ConverterProxy:
    """
    Wrapper for converter instances providing metadata and lifecycle management.
    """
    
    def __init__(self, converter, tags, package_name=None, package_version=None):
        """
        Create converter proxy.
        
        Parameters:
        - converter: Converter instance to wrap
        - tags (list): YAML tags this converter handles
        - package_name (str, optional): Package name for metadata
        - package_version (str, optional): Package version for metadata
        """

Compression System

Custom compression algorithms for array data and file content.

class Compressor:
    """
    Abstract base class for compression algorithms.
    """
    
    @property
    def label(self) -> str:
        """
        Short label identifying this compression algorithm.
        """
    
    def compress(self, data, **kwargs):
        """
        Compress data using this algorithm.
        
        Parameters:
        - data (bytes): Data to compress
        - **kwargs: Algorithm-specific options
        
        Returns:
        bytes: Compressed data
        """
    
    def decompress(self, data, **kwargs):
        """
        Decompress data using this algorithm.
        
        Parameters:
        - data (bytes): Compressed data
        - **kwargs: Algorithm-specific options
        
        Returns:
        bytes: Decompressed data
        """

Validation System

Additional schema validation beyond core ASDF schemas.

class Validator:
    """
    Abstract base class for additional validation.
    """
    
    def validate(self, data, schema_uri, **kwargs):
        """
        Validate data against additional constraints.
        
        Parameters:
        - data: Data to validate
        - schema_uri (str): URI of schema being validated against
        - **kwargs: Validation options
        
        Raises:
        ValidationError: If validation fails
        """

Tag Definition

Define YAML tags for custom object serialization.

class TagDefinition:
    """
    Definition of a YAML tag for ASDF objects.
    """
    
    def __init__(self, tag_uri, schema_uris=None):
        """
        Create tag definition.
        
        Parameters:
        - tag_uri (str): URI of the YAML tag
        - schema_uris (list, optional): URIs of associated schemas
        """
    
    @property
    def tag_uri(self) -> str:
        """URI of the YAML tag."""
    
    @property
    def schema_uris(self) -> list:
        """List of schema URIs associated with this tag."""

Extension Management

System for managing collections of extensions and their interactions.

class ExtensionManager:
    """
    Manages collection of extensions and their interactions.
    """
    
    def get_extensions(self) -> list:
        """
        Get all managed extensions.
        
        Returns:
        list: All Extension objects under management
        """
    
    def get_converter_for_type(self, typ):
        """
        Find converter capable of handling given type.
        
        Parameters:
        - typ: Type to find converter for
        
        Returns:
        Converter: Converter capable of handling the type, or None
        """
    
    def get_validator_for_uri(self, uri):
        """
        Find validator for given schema URI.
        
        Parameters:
        - uri (str): Schema URI
        
        Returns:
        Validator: Validator for the URI, or None
        """

def get_cached_extension_manager(extensions=None):
    """
    Get cached extension manager instance.
    
    Parameters:
    - extensions (list, optional): Additional extensions to include
    
    Returns:
    ExtensionManager: Cached manager instance
    """

Serialization Context

Context information available during serialization and deserialization operations.

class SerializationContext:
    """
    Context information during serialization/deserialization.
    """
    
    @property
    def extension_manager(self) -> ExtensionManager:
        """Extension manager for this context."""
    
    @property
    def url_mapping(self) -> dict:
        """URL mapping for resolving references."""
    
    @property
    def block_manager(self):
        """Block manager for array data."""

Usage Examples

Creating a Custom Extension

from asdf.extension import Extension, Converter
import numpy as np

class ComplexNumber:
    """Custom complex number class with metadata."""
    def __init__(self, real, imag, precision="double"):
        self.real = real
        self.imag = imag 
        self.precision = precision

class ComplexConverter(Converter):
    """Converter for ComplexNumber objects."""
    
    def can_convert(self, obj):
        return isinstance(obj, ComplexNumber)
    
    def convert_to_tree(self, obj, ctx):
        return {
            'real': obj.real,
            'imag': obj.imag,
            'precision': obj.precision
        }
    
    def convert_from_tree(self, tree, ctx):
        return ComplexNumber(
            tree['real'], 
            tree['imag'], 
            tree.get('precision', 'double')
        )

class ComplexExtension(Extension):
    """Extension for complex number support."""
    
    extension_uri = "asdf://example.com/complex/extensions/complex-1.0.0"
    converters = [ComplexConverter()]
    tags = ["asdf://example.com/complex/tags/complex-1.0.0"]

# Use the extension
complex_num = ComplexNumber(3.0, 4.0, "single")
data = {"my_complex": complex_num}

af = asdf.AsdfFile(data, extensions=[ComplexExtension()])
af.write_to("complex_data.asdf")

# Read with extension
with asdf.open("complex_data.asdf", extensions=[ComplexExtension()]) as af:
    restored = af.tree["my_complex"]
    print(f"{restored.real} + {restored.imag}i")

Custom Compression

from asdf.extension import Compressor
import zlib

class CustomCompressor(Compressor):
    """Custom compression using high compression ratio."""
    
    label = "custom_zlib"
    
    def compress(self, data, level=9, **kwargs):
        return zlib.compress(data, level=level)
        
    def decompress(self, data, **kwargs):
        return zlib.decompress(data)

class CompressionExtension(Extension):
    """Extension providing custom compression."""
    
    extension_uri = "asdf://example.com/compression/extensions/custom-1.0.0"
    compressors = [CustomCompressor()]

# Use custom compression
import numpy as np

data = {"large_array": np.random.random(100000)}
af = asdf.AsdfFile(data, extensions=[CompressionExtension()])
af.write_to("compressed.asdf", all_array_compression="custom_zlib")

Validation Extension

from asdf.extension import Validator, ValidationError

class RangeValidator(Validator):
    """Validates that numeric values are within specified ranges."""
    
    def validate(self, data, schema_uri, **kwargs):
        if "range-check" in schema_uri:
            if isinstance(data, (int, float)):
                if not (0 <= data <= 100):
                    raise ValidationError(f"Value {data} outside range [0, 100]")

class ValidationExtension(Extension):
    """Extension providing range validation."""
    
    extension_uri = "asdf://example.com/validation/extensions/range-1.0.0"
    validators = [RangeValidator()]

# Use validation
data = {"percentage": 85}  # Valid
af = asdf.AsdfFile(data, extensions=[ValidationExtension()])

# This would raise ValidationError:
# data = {"percentage": 150}  # Invalid

Multi-Component Extension

from asdf.extension import Extension, Converter, Compressor, Validator
import json
import gzip

class JsonConverter(Converter):
    """Converter for JSON-serializable objects."""
    
    def can_convert(self, obj):
        try:
            json.dumps(obj)
            return True
        except (TypeError, ValueError):
            return False
    
    def convert_to_tree(self, obj, ctx):
        return {"json_data": json.dumps(obj)}
    
    def convert_from_tree(self, tree, ctx):
        return json.loads(tree["json_data"])

class GzipCompressor(Compressor):
    """Gzip compression for text data."""
    
    label = "gzip"
    
    def compress(self, data, **kwargs):
        return gzip.compress(data)
    
    def decompress(self, data, **kwargs):
        return gzip.decompress(data)

class JsonSizeValidator(Validator):
    """Validates JSON data size limits."""
    
    def validate(self, data, schema_uri, **kwargs):
        if "json-size" in schema_uri:
            json_str = json.dumps(data)
            if len(json_str) > 1000000:  # 1MB limit
                raise ValidationError("JSON data exceeds size limit")

class FullExtension(Extension):
    """Complete extension with converter, compressor, and validator."""
    
    extension_uri = "asdf://example.com/full/extensions/full-1.0.0"
    converters = [JsonConverter()]
    compressors = [GzipCompressor()]
    validators = [JsonSizeValidator()]
    tags = ["asdf://example.com/full/tags/json-1.0.0"]

# Use complete extension
complex_data = {
    "metadata": {"type": "experiment", "version": 1},
    "parameters": [{"name": "temp", "value": 25.0}],
    "results": list(range(1000))
}

af = asdf.AsdfFile(
    {"experiment": complex_data}, 
    extensions=[FullExtension()]
)
af.write_to("full_extension_example.asdf")

Extension Discovery and Management

# Get all available extensions
manager = asdf.get_cached_extension_manager()
extensions = manager.get_extensions()

print(f"Found {len(extensions)} extensions:")
for ext in extensions:
    print(f"  {ext.extension_uri}")
    print(f"    Converters: {len(ext.converters)}")
    print(f"    Compressors: {len(ext.compressors)}")
    print(f"    Validators: {len(ext.validators)}")

# Find converter for specific type
converter = manager.get_converter_for_type(ComplexNumber)
if converter:
    print(f"Found converter for ComplexNumber: {converter}")

Install with Tessl CLI