Detect Anomalies in Streaming IIoT Sensor Data with River and Polars

"""
Production implementation for detecting anomalies in streaming IIoT sensor data.
Utilizes River for machine learning and Polars for data manipulation.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import time
import polars as pl
from river import anomaly
from river import stream

# Configure logging for monitoring and debugging purposes
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables and settings
class Config:
    database_url: str = os.getenv('DATABASE_URL')
    sensor_data_stream: str = os.getenv('SENSOR_DATA_STREAM')

# Function to validate input data
async 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')  # Ensure sensor_id is present
    if 'value' not in data:
        raise ValueError('Missing value')  # Ensure value is present
    return True

# Function to sanitize fields in the data
async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields.
    Args:
        data: Raw input data
    Returns:
        Sanitized data
    """
    return {k: str(v).strip() for k, v in data.items()}  # Strip whitespace

# Function to normalize data for processing
async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize data fields.
    Args:
        data: Input data to normalize
    Returns:
        Normalized data
    """
    data['value'] = float(data['value'])  # Ensure value is a float
    return data

# Function to transform records into Polars DataFrame
async def transform_records(records: List[Dict[str, Any]]) -> pl.DataFrame:
    """Transform records into Polars DataFrame.
    Args:
        records: List of records to transform
    Returns:
        Polars DataFrame
    """
    return pl.DataFrame(records)  # Create DataFrame from records

# Function to process a batch of data
async def process_batch(data: List[Dict[str, Any]]) -> None:
    """Process a batch of sensor data.
    Args:
        data: List of sensor data dictionaries
    """
    # Validate and transform data
    for record in data:
        await validate_input(record)  # Validate each record
        sanitized = await sanitize_fields(record)  # Sanitize fields
        normalized = await normalize_data(sanitized)  # Normalize data
        # Log the processed data
        logger.info(f'Processed record: {normalized}')  # Log processed data

# Function to fetch data from the sensor stream
async def fetch_data() -> List[Dict[str, Any]]:
    """Fetch data from the sensor stream.
    Returns:
        List of sensor data dictionaries
    """
    # Simulate fetching data from a sensor stream
    return [{'sensor_id': 'sensor1', 'value': 23.5}, {'sensor_id': 'sensor2', 'value': 19.0}]  # Example data

# Function to save data to the database
async def save_to_db(data: pl.DataFrame) -> None:
    """Save processed data to the database.
    Args:
        data: Polars DataFrame to save
    """
    # Simulate saving data to a database
    logger.info(f'Saving to database: {data}')  # Log saving action

# Function to format output for display
def format_output(data: Any) -> str:
    """Format output for display.
    Args:
        data: Data to format
    Returns:
        Formatted string
    """
    return str(data)  # Simple string conversion

# Main orchestrator class for anomaly detection
class AnomalyDetector:
    def __init__(self) -> None:
        self.model = anomaly.AnomalyDetector()  # Initialize anomaly detection model

    async def run_detection(self) -> None:
        """Run anomaly detection workflow.
        """
        while True:
            try:
                data = await fetch_data()  # Fetch data from source
                await process_batch(data)  # Process the fetched data
                df = await transform_records(data)  # Transform into DataFrame
                # Check for anomalies
                for index, row in df.iterrows():
                    if self.model.predict(row['value']):  # Check for anomaly
                        logger.warning(f'Anomaly detected in sensor {row['sensor_id']} with value {row['value']}')  # Log anomaly
                await save_to_db(df)  # Save processed data to DB
                time.sleep(2)  # Simulate delay between fetches
            except Exception as e:
                logger.error(f'Error in workflow: {e}')  # Log errors

# Main block to execute the anomaly detection
if __name__ == '__main__':
    detector = AnomalyDetector()  # Create detector instance
    try:
        detector.run_detection()  # Start detection process
    except KeyboardInterrupt:
        logger.info('Stopping anomaly detection...')  # Graceful shutdown