ATON V2 Documentation

pip install aton-format

npm install aton-format
# or
yarn add aton-format

git clone https://github.com/dagoSte/aton-format.git
cd aton-format
pip install -e .

from aton_format import ATONEncoder, ATONDecoder, CompressionMode

# Initialize encoder with compression
encoder = ATONEncoder(
    compression=CompressionMode.BALANCED,
    optimize=True
)

# Your data
data = {
    "employees": [
        {"id": 1, "name": "Alice", "salary": 95000, "active": True},
        {"id": 2, "name": "Bob", "salary": 92000, "active": True},
        {"id": 3, "name": "Carol", "salary": 110000, "active": False}
    ]
}

# Encode to ATON
aton_text = encoder.encode(data)
print(aton_text)

# Decode back to original
decoder = ATONDecoder()
original = decoder.decode(aton_text)

assert data == original  # Perfect round-trip!

const ATON = require('aton-format');

// Initialize encoder
const encoder = new ATON.Encoder({
    compression: ATON.CompressionMode.BALANCED,
    optimize: true
});

// Your data
const data = {
    employees: [
        {id: 1, name: "Alice", salary: 95000, active: true},
        {id: 2, name: "Bob", salary: 92000, active: true},
        {id: 3, name: "Carol", salary: 110000, active: false}
    ]
};

// Encode to ATON
const atonText = encoder.encode(data);
console.log(atonText);

// Decode back
const decoder = new ATON.Decoder();
const original = decoder.decode(atonText);

console.assert(JSON.stringify(data) === JSON.stringify(original));

encoder = ATONEncoder(
    compression=CompressionMode.ADAPTIVE,
    optimize=True,
    queryable=True
)

result = encoder.encode_with_query(
    data,
    "employees WHERE salary > 100000 ORDER BY salary DESC"
)

decoder = ATONDecoder(validate=True)
data = decoder.decode(aton_text)

# Perfect round-trip guaranteed
assert original_data == data

stream_encoder = ATONStreamEncoder(
    chunk_size=1000,
    compression=CompressionMode.ULTRA
)

# Process 100K records in chunks
data = {"products": large_product_list}

for chunk_info in stream_encoder.stream_encode(data):
    print(f"Chunk {chunk_info['chunk_id']}/{chunk_info['total_chunks']}")
    print(f"Records: {chunk_info['metadata']['records_in_chunk']}")
    
    # Process chunk
    process_chunk(chunk_info['data'])
    
    # Memory stays constant!

from aton_format import ATONQueryEngine

engine = ATONQueryEngine()

# Complex query
results = engine.query(
    data,
    """
    products WHERE 
        (price BETWEEN 100 AND 500) 
        AND category IN ('Electronics', 'Computers')
        AND name LIKE '%Premium%'
    ORDER BY rating DESC
    LIMIT 20
    """
)

# Results are filtered, sorted, and paginated
for product in results:
    print(product['name'], product['price'])

encoder = ATONEncoder(compression=CompressionMode.FAST)
# Fastest encoding, moderate compression

encoder = ATONEncoder(compression=CompressionMode.BALANCED)
# Optimal balance, recommended default

encoder = ATONEncoder(compression=CompressionMode.ULTRA)
# Maximum compression for storage/batch

encoder = ATONEncoder(compression=CompressionMode.ADAPTIVE)
# Automatically chooses best strategy

[SELECT fields] table_name [WHERE conditions] [ORDER BY field direction] [LIMIT n] [OFFSET m]

# Simple condition
employees WHERE salary > 100000

# Multiple conditions
employees WHERE salary > 100000 AND active = true

# Complex logic
employees WHERE (role = 'Engineer' OR role = 'Manager') 
         AND salary > 80000

employees WHERE active = true

products WHERE price > 100 AND stock > 0 AND featured = true

customers WHERE email LIKE '%@gmail.com' AND name LIKE 'A%'

orders WHERE total BETWEEN 1000 AND 5000 AND created_date > '2024-01-01'

employees WHERE 
    (department = 'Engineering' AND salary > 90000)
    OR (department = 'Sales' AND commission > 50000)
    OR (role = 'Executive')
ORDER BY salary DESC
LIMIT 50

SELECT name, email, department 
FROM employees 
WHERE active = true
ORDER BY name ASC

products WHERE category = 'Electronics'
ORDER BY rating DESC
LIMIT 20 OFFSET 40  # Page 3 (40-60)

from aton_format import ATONStreamEncoder, CompressionMode

# Initialize stream encoder
stream_encoder = ATONStreamEncoder(
    chunk_size=1000,
    compression=CompressionMode.ULTRA
)

# Large dataset
data = {
    "transactions": [
        # ... 1 million records
    ]
}

# Stream encode
for chunk in stream_encoder.stream_encode(data):
    # Chunk structure
    print(f"Chunk {chunk['chunk_id']} of {chunk['total_chunks']}")
    print(f"First chunk: {chunk['is_first']}")
    print(f"Last chunk: {chunk['is_last']}")
    print(f"Records: {chunk['metadata']['records_in_chunk']}")
    
    # Send chunk to LLM or store
    send_to_llm(chunk['data'])

ATONError (base)
|-- ATONEncodingError
|-- ATONDecodingError
|-- ATONQueryError
`-- ATONCompressionError

from aton_format import (
    ATONEncoder, 
    ATONEncodingError, 
    ATONDecodingError,
    ATONQueryError
)

try:
    encoder = ATONEncoder(validate=True)
    result = encoder.encode(data)
    
except ATONEncodingError as e:
    print(f"Encoding failed: {e}")
    # Handle encoding error
    
except ATONQueryError as e:
    print(f"Query failed: {e}")
    # Handle query error
    
except Exception as e:
    print(f"Unexpected error: {e}")
    # Handle unexpected errors

# Enable validation (default)
encoder = ATONEncoder(validate=True)

# Validation checks:
# - Data structure
# - Type consistency
# - Required fields
# - Value ranges

# Real-time applications
encoder = ATONEncoder(compression=CompressionMode.FAST)

# General purpose (recommended)
encoder = ATONEncoder(compression=CompressionMode.BALANCED)

# Batch processing / storage
encoder = ATONEncoder(compression=CompressionMode.ULTRA)

# Mixed workloads
encoder = ATONEncoder(compression=CompressionMode.ADAPTIVE)

# DON'T: Load everything in memory
data = load_million_records()
encoded = encoder.encode(data)  # Memory spike!

# DO: Use streaming
stream_encoder = ATONStreamEncoder(chunk_size=1000)
for chunk in stream_encoder.stream_encode(data):
    process_chunk(chunk)  # Constant memory!

# DON'T: Send all data then filter
all_data = encoder.encode(large_dataset)
# LLM has to process everything...

# DO: Filter before encoding
filtered = encoder.encode_with_query(
    large_dataset,
    "employees WHERE salary > 100000 LIMIT 100"
)
# Only 100 relevant records sent!

# Always enable optimization for repeated structures
encoder = ATONEncoder(
    optimize=True,  # Enable schemas and defaults
    compression=CompressionMode.BALANCED
)

from aton_format import ATONError

try:
    result = encoder.encode(data)
except ATONError as e:
    # ATON-specific errors
    logger.error(f"ATON error: {e}")
    fallback_to_json(data)
except Exception as e:
    # Unexpected errors
    logger.critical(f"Critical error: {e}")
    raise

# Always validate in production
encoder = ATONEncoder(validate=True)
decoder = ATONDecoder(validate=True)

# Round-trip test
encoded = encoder.encode(data)
decoded = decoder.decode(encoded)
assert data == decoded

import time

# Test different modes
modes = [CompressionMode.FAST, CompressionMode.BALANCED, CompressionMode.ULTRA]

for mode in modes:
    encoder = ATONEncoder(compression=mode)
    
    start = time.time()
    result = encoder.encode(data)
    duration = time.time() - start
    
    print(f"{mode.value}: {duration:.3f}s, {len(result)} chars")