Detect Industrial Sensor Faults with dtaianomaly and Darts

"""
Production implementation for detecting industrial sensor faults using dtaianomaly and Darts.
Provides secure, scalable operations with data validation and logging.
"""

from typing import List, Dict, Any, Optional
import os
import logging
import time
import numpy as np
import pandas as pd
from dtaidistance import dtw
from darts import TimeSeries
from darts.models import ExponentialSmoothing
from sqlalchemy import create_engine, text

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

class Config:
    """
    Configuration class to hold environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///sensors.db')


def validate_input(data: Dict[str, Any]) -> bool:
    """
    Validate incoming sensor data.
    
    Args:
        data: Dictionary containing sensor data
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'sensor_id' not in data:
        raise ValueError('Missing sensor_id')
    if not isinstance(data['readings'], list):
        raise ValueError('Readings must be a list')
    return True


def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Sanitize input data fields to prevent injection attacks.
    
    Args:
        data: Incoming data dictionary
    Returns:
        Sanitized data dictionary
    """
    sanitized_data = {k: str(v).strip() for k, v in data.items()}
    return sanitized_data


def fetch_data(sensor_id: str) -> Optional[List[float]]:
    """
    Fetch sensor data from the database.
    
    Args:
        sensor_id: Unique identifier for the sensor
    Returns:
        List of sensor readings or None if not found
    """
    engine = create_engine(Config.database_url)
    with engine.connect() as connection:
        result = connection.execute(text('SELECT readings FROM sensor_data WHERE sensor_id = :id'), {'id': sensor_id})
        readings = result.fetchone()
    return readings[0] if readings else None


def save_to_db(sensor_id: str, readings: List[float]) -> None:
    """
    Save processed sensor data into the database.
    
    Args:
        sensor_id: Unique identifier for the sensor
        readings: List of processed readings
    """
    engine = create_engine(Config.database_url)
    with engine.connect() as connection:
        connection.execute(text('INSERT INTO sensor_data (sensor_id, readings) VALUES (:id, :readings)'), {'id': sensor_id, 'readings': readings})


def process_batch(sensor_data: List[Dict[str, Any]]) -> None:
    """
    Process a batch of sensor data for fault detection.
    
    Args:
        sensor_data: List of sensor data dictionaries
    """
    for data in sensor_data:
        try:
            validate_input(data)  # Validate input data
            sanitized_data = sanitize_fields(data)  # Sanitize fields
            readings = fetch_data(sanitized_data['sensor_id'])  # Fetch readings
            if readings is None:
                logger.warning(f'Sensor {sanitized_data["sensor_id"]} not found.')  # Log warning
                continue  # Skip if no readings found

            # Process readings using Darts for time series forecasting
            series = TimeSeries.from_values(readings)
            model = ExponentialSmoothing()
            model.fit(series)
            forecast = model.predict(10)  # Predict next 10 time steps

            # Check for anomalies using dtaianomaly
            distance = dtw.warping_distance(readings, forecast.values())
            if distance > 10:  # Threshold for anomaly detection
                logger.error(f'Anomaly detected for sensor {sanitized_data["sensor_id"]}!')
                save_to_db(sanitized_data['sensor_id'], readings)  # Save readings
        except Exception as e:
            logger.error(f'Error processing data for {data}: {str(e)}')  # Log error


def handle_errors(func):
    """
    Decorator to handle errors in function execution.
    
    Args:
        func: Function to wrap
    Returns:
        Wrapped function
    """
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in function {func.__name__}: {str(e)}')
            return None
    return wrapper


@handle_errors
def main(sensor_data: List[Dict[str, Any]]) -> None:
    """
    Main function to orchestrate the sensor fault detection process.
    
    Args:
        sensor_data: List of sensor data dictionaries
    """
    logger.info('Starting sensor fault detection process...')
    process_batch(sensor_data)


if __name__ == '__main__':
    # Example usage
    test_data = [
        {'sensor_id': 'sensor_1', 'readings': [1.0, 1.1, 1.2]},
        {'sensor_id': 'sensor_2', 'readings': [0.9, 1.0, 10.0]},  # This one will trigger an anomaly
    ]
    main(test_data)  # Run the main function