Detect Time-Series Motifs and Anomalies in Conveyor Sensor Data with STUMPY and scikit-learn

"""
Production implementation for detecting time-series motifs and anomalies in conveyor sensor data.
Utilizes STUMPY for motif discovery and scikit-learn for anomaly detection.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import numpy as np
import pandas as pd
import stumpy
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import MinMaxScaler

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

class Config:
    """Configuration class for environment variables."""
    data_source: str = os.getenv('DATA_SOURCE')

def validate_input(data: List[float]) -> bool:
    """Validate input data for anomalies.
    
    Args:
        data: List of sensor data to validate.
    Returns:
        True if valid.
    Raises:
        ValueError: If validation fails.
    """  
    if not data or len(data) < 2:
        raise ValueError('Input data should contain at least two values.')  # Ensure sufficient data
    return True

def sanitize_fields(data: List[float]) -> List[float]:
    """Sanitize input data by removing NaNs and infs.
    
    Args:
        data: Input sensor data.
    Returns:
        Cleaned data list.
    """  
    cleaned_data = [x for x in data if pd.notnull(x) and np.isfinite(x)]  # Remove invalid entries
    return cleaned_data

def normalize_data(data: List[float]) -> np.ndarray:
    """Normalize data to [0, 1] range.
    
    Args:
        data: List of numeric sensor data.
    Returns:
        Normalized data as a NumPy array.
    """  
    scaler = MinMaxScaler()
    normalized_data = scaler.fit_transform(np.array(data).reshape(-1, 1))
    return normalized_data.flatten()  # Return as flat array

def transform_records(data: List[float]) -> np.ndarray:
    """Transform records for motif discovery using STUMPY.
    
    Args:
        data: Sensor data for transformation.
    Returns:
        Transformed data for motif analysis.
    """  
    return normalize_data(data)

def fetch_data(source: str) -> List[float]:
    """Fetch data from the defined source.
    
    Args:
        source: Data source (e.g., file, API).
    Returns:
        List of sensor data.
    """  
    # Placeholder for fetching data
    logger.info('Fetching data from %s', source)
    return [1.0, 2.0, 3.0, np.nan, 4.0, 5.0]  # Example static data

def save_to_db(data: List[float]) -> None:
    """Save processed data to the database.
    
    Args:
        data: Data to save.
    Returns:
        None
    """  
    logger.info('Saving data to database')  # Simulated save operation

def handle_errors(err: Exception) -> None:
    """Log errors gracefully.
    
    Args:
        err: Exception to handle.
    Returns:
        None
    """  
    logger.error('An error occurred: %s', err)  # Log the error

class AnomalyDetector:
    """Main class for anomaly detection logic."""

    def __init__(self, data_source: str) -> None:
        self.data_source = data_source

    def detect_anomalies(self) -> Tuple[List[float], List[int]]:
        """Detect anomalies in the sensor data.
        
        Returns:
            Tuple of detected anomalies and their indices.
        """  
        try:
            raw_data = fetch_data(self.data_source)  # Fetch data
            validated_data = sanitize_fields(raw_data)  # Sanitize input
            validated_data = validate_input(validated_data)  # Validate input
            transformed_data = transform_records(validated_data)  # Transform data
            motifs = stumpy.stump(time_series_a=transformed_data, m=3)  # STUMPY for motif discovery
            model = IsolationForest(contamination=0.1)
            model.fit(transformed_data.reshape(-1, 1))  # Train anomaly detection model
            anomalies = model.predict(transformed_data.reshape(-1, 1))
            data_with_anomalies = [val for idx, val in enumerate(validated_data) if anomalies[idx] == -1]  # Extract anomalies
            return data_with_anomalies, [idx for idx, val in enumerate(anomalies) if val == -1]  # Return anomalies and indices
        except Exception as e:
            handle_errors(e)  # Handle errors gracefully
            return [], []  # Return empty lists on error

if __name__ == '__main__':
    # Example usage
    config = Config()
    detector = AnomalyDetector(data_source=config.data_source)
    anomalies, indices = detector.detect_anomalies()  # Detect anomalies
    logger.info('Detected anomalies: %s at indices: %s', anomalies, indices)  # Log detected anomalies