Build Streaming Quality Feature Stores for Predictive Maintenance with Quix Streams and DuckDB

"""
Production implementation for Building Streaming Quality Feature Stores for Predictive Maintenance.
Provides secure, scalable operations with Quix Streams and DuckDB.
"""

from typing import Dict, Any, List, Generator
import os
import logging
import duckdb
import time
import json
import requests

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

class Config:
    database_url: str = os.getenv('DATABASE_URL', 'duckdb://:memory:')
    quix_api_url: str = os.getenv('QUIX_API_URL', 'https://api.quix.io')

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'sensor_id' not in data:
        raise ValueError('Missing sensor_id')
    if 'value' not in data:
        raise ValueError('Missing value')
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input data fields.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    return {key: str(value).strip() for key, value in data.items()}

def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize data for processing.
    
    Args:
        data: Input data to normalize
    Returns:
        Normalized data
    """
    try:
        data['value'] = float(data['value'])  # Ensure value is a float
    except ValueError:
        raise ValueError('Value must be a float')
    return data

def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform records for database insertion.
    
    Args:
        records: List of raw records
    Returns:
        List of transformed records
    """
    return [normalize_data(sanitize_fields(record)) for record in records]

def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from Quix API.
    
    Args:
        api_url: URL for the Quix API
    Returns:
        Fetched data
    """
    response = requests.get(api_url)
    response.raise_for_status()  # Raises an error for bad responses
    return response.json()  # Returns the JSON response

def save_to_db(data: List[Dict[str, Any]]) -> None:
    """Save data to DuckDB database.
    
    Args:
        data: List of records to save
    Raises:
        Exception: If database insertion fails
    """
    try:
        con = duckdb.connect(Config.database_url)
        con.execute("CREATE TABLE IF NOT EXISTS sensor_data (sensor_id VARCHAR, value DOUBLE)")
        for record in data:
            con.execute("INSERT INTO sensor_data VALUES (?, ?)", (record['sensor_id'], record['value']))
        con.close()
    except Exception as e:
        logger.error(f"Failed to save data: {e}")
        raise

def aggregate_metrics() -> Dict[str, Any]:
    """Aggregate metrics from the DuckDB database.
    
    Returns:
        Aggregated metrics
    """
    try:
        con = duckdb.connect(Config.database_url)
        result = con.execute("SELECT sensor_id, AVG(value) as avg_value FROM sensor_data GROUP BY sensor_id").fetchall()
        con.close()
        return [{'sensor_id': row[0], 'avg_value': row[1]} for row in result]
    except Exception as e:
        logger.error(f"Failed to aggregate metrics: {e}")
        return []  # Return empty list on failure

class FeatureStore:
    """Class to manage Feature Store operations for predictive maintenance.
    """  
    def __init__(self, api_url: str):
        self.api_url = api_url

    def run(self) -> None:
        """Run the feature store process.
        
        Raises:
            Exception: If the process fails
        """
        try:
            data = fetch_data(self.api_url)  # Fetch data from API
            validated_data = [validate_input(record) for record in data]  # Validate each record
            transformed_data = transform_records(validated_data)  # Transform data for DB
            save_to_db(transformed_data)  # Save data to DB
            metrics = aggregate_metrics()  # Aggregate metrics
            logger.info(f"Aggregated metrics: {metrics}")
        except Exception as e:
            logger.error(f"Error in feature store run: {e}")

if __name__ == '__main__':
    feature_store = FeatureStore(Config.quix_api_url)
    feature_store.run()  # Execute the feature store process