Ingest Factory Telemetry into Streaming Iceberg Tables with Quix Streams and PyIceberg

"""
Production implementation for ingesting factory telemetry into streaming Iceberg tables using Quix Streams and PyIceberg.
Provides secure, scalable operations with robust error handling and logging.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import time
from pydantic import BaseModel, ValidationError
from quixstreams import QuixClient
from pyiceberg import Table

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class to manage environment variables.
    """
    quix_api_key: str = os.getenv('QUIX_API_KEY')
    iceberg_table_path: str = os.getenv('ICEBERG_TABLE_PATH')

config = Config()

class TelemetryData(BaseModel):
    """
    Data model for factory telemetry.
    """
    machine_id: str
    temperature: float
    pressure: float
    timestamp: str

def validate_input(data: Dict[str, Any]) -> None:
    """
    Validate incoming telemetry data.
    
    Args:
        data: Dictionary containing telemetry data
    Raises:
        ValueError: If validation fails
    """
    try:
        TelemetryData(**data)  # Validate and parse
    except ValidationError as e:
        logger.error(f'Validation error: {e}');
        raise ValueError('Invalid telemetry data')

def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normalize the telemetry data for processing.
    
    Args:
        data: Raw telemetry data
    Returns:
        Normalized telemetry data
    """
    normalized = {
        'machine_id': data['machine_id'],
        'temperature': round(data['temperature'], 2),  # Round temperature
        'pressure': round(data['pressure'], 2),
        'timestamp': data['timestamp']
    }
    return normalized

def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Transform raw records into a format suitable for Iceberg.
    
    Args:
        records: List of raw telemetry records
    Returns:
        List of transformed records
    """
    return [normalize_data(record) for record in records]

def fetch_data() -> List[Dict[str, Any]]:
    """
    Fetch telemetry data from the source.
    
    Returns:
        List of telemetry data
    """
    # Simulated data fetch - replace with actual data source call
    data = [
        {'machine_id': 'M1', 'temperature': 76.5, 'pressure': 1.2, 'timestamp': '2023-10-01T12:00:00Z'},
        {'machine_id': 'M1', 'temperature': 75.3, 'pressure': 1.1, 'timestamp': '2023-10-01T12:01:00Z'},
    ]
    return data

def save_to_db(records: List[Dict[str, Any]]) -> None:
    """
    Save processed records to Iceberg table.
    
    Args:
        records: List of records to save
    """
    client = QuixClient(api_key=config.quix_api_key)
    table = Table(config.iceberg_table_path)
    for record in records:
        # Assume record is a dictionary that matches Iceberg schema
        table.insert(record)
    logger.info(f'Successfully saved {len(records)} records to Iceberg table.')

def process_batch(batch: List[Dict[str, Any]]) -> None:
    """
    Process a batch of telemetry data.
    
    Args:
        batch: List of raw telemetry data
    """
    try:
        transformed = transform_records(batch)  # Transform records
        save_to_db(transformed)  # Save to Iceberg
    except Exception as e:
        logger.error(f'Error processing batch: {e}')

def handle_errors(func):
    """
    Decorator to handle errors for functions.
    
    Args:
        func: Function to wrap
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')
            raise
    return wrapper

@handle_errors
def main():
    """
    Main function to orchestrate the ingestion process.
    """
    while True:
        logger.info('Fetching data...')
        raw_data = fetch_data()  # Fetch data
        for record in raw_data:
            validate_input(record)  # Validate each record
        process_batch(raw_data)  # Process the batch
        time.sleep(60)  # Wait before next fetch

if __name__ == '__main__':
    # Example usage
    main()  # Start ingestion process