Sensor Data Aggregator

Overview

Collect, aggregate, and analyze data from IoT sensors deployed across construction sites. Support real-time monitoring of environmental conditions, equipment status, structural integrity, and worker safety through unified data processing.

IoT Sensor Architecture

┌─────────────────────────────────────────────────────────────────┐
│                  SENSOR DATA AGGREGATION                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  SENSORS                    AGGREGATOR            OUTPUTS       │
│  ───────                    ──────────            ───────       │
│                                                                  │
│  🌡️ Temperature  ─────┐                          📊 Dashboard   │
│  💧 Humidity     ─────┤    ┌──────────────┐      ⚠️ Alerts      │
│  📊 Vibration    ─────┼───→│  AGGREGATE   │───→  📈 Analytics   │
│  🔊 Noise        ─────┤    │  PROCESS     │      📋 Reports     │
│  💨 Air Quality  ─────┤    │  ANALYZE     │      🔄 API         │
│  📍 Location     ─────┘    └──────────────┘                     │
│                                                                  │
│  DATA FLOW:                                                     │
│  Raw → Validate → Transform → Store → Analyze → Alert          │
│                                                                  │
│  ANALYSIS:                                                      │
│  • Real-time monitoring                                         │
│  • Trend detection                                              │
│  • Anomaly identification                                       │
│  • Threshold alerting                                           │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable, Tuple
from datetime import datetime, timedelta
from enum import Enum
import statistics
import json

class SensorType(Enum):
    TEMPERATURE = "temperature"
    HUMIDITY = "humidity"
    VIBRATION = "vibration"
    NOISE = "noise"
    AIR_QUALITY = "air_quality"
    DUST = "dust"
    GAS = "gas"
    PRESSURE = "pressure"
    STRAIN = "strain"
    TILT = "tilt"
    GPS = "gps"
    PROXIMITY = "proximity"

class AlertSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    EMERGENCY = "emergency"

class DataQuality(Enum):
    GOOD = "good"
    SUSPECT = "suspect"
    BAD = "bad"
    MISSING = "missing"

@dataclass
class SensorReading:
    sensor_id: str
    sensor_type: SensorType
    timestamp: datetime
    value: float
    unit: str
    quality: DataQuality = DataQuality.GOOD
    location: Optional[Dict] = None
    metadata: Dict = field(default_factory=dict)

@dataclass
class Sensor:
    id: str
    name: str
    sensor_type: SensorType
    unit: str
    location: Dict  # {zone, floor, coordinates}
    thresholds: Dict  # {warning, critical, min, max}
    calibration_date: datetime
    battery_level: float = 100.0
    status: str = "active"

@dataclass
class Alert:
    id: str
    sensor_id: str
    sensor_type: SensorType
    severity: AlertSeverity
    timestamp: datetime
    value: float
    threshold: float
    message: str
    acknowledged: bool = False
    resolved: bool = False

@dataclass
class AggregatedMetric:
    sensor_type: SensorType
    period_start: datetime
    period_end: datetime
    readings_count: int
    min_value: float
    max_value: float
    avg_value: float
    std_dev: float
    alerts_triggered: int

class SensorDataAggregator:
    """Aggregate and analyze IoT sensor data."""

    # Default thresholds by sensor type
    DEFAULT_THRESHOLDS = {
        SensorType.TEMPERATURE: {"warning": 35, "critical": 40, "unit": "°C"},
        SensorType.HUMIDITY: {"warning": 80, "critical": 90, "unit": "%"},
        SensorType.VIBRATION: {"warning": 10, "critical": 25, "unit": "mm/s"},
        SensorType.NOISE: {"warning": 85, "critical": 100, "unit": "dB"},
        SensorType.AIR_QUALITY: {"warning": 100, "critical": 150, "unit": "AQI"},
        SensorType.DUST: {"warning": 3, "critical": 10, "unit": "mg/m³"},
        SensorType.GAS: {"warning": 20, "critical": 50, "unit": "ppm"},
    }

    def __init__(self, site_name: str):
        self.site_name = site_name
        self.sensors: Dict[str, Sensor] = {}
        self.readings: List[SensorReading] = []
        self.alerts: List[Alert] = []
        self.alert_handlers: List[Callable] = []

    def register_sensor(self, id: str, name: str, sensor_type: SensorType,
                       unit: str, location: Dict,
                       thresholds: Dict = None) -> Sensor:
        """Register a new sensor."""
        if thresholds is None:
            thresholds = self.DEFAULT_THRESHOLDS.get(sensor_type, {})

        sensor = Sensor(
            id=id,
            name=name,
            sensor_type=sensor_type,
            unit=unit,
            location=location,
            thresholds=thresholds,
            calibration_date=datetime.now()
        )
        self.sensors[id] = sensor
        return sensor

    def ingest_reading(self, sensor_id: str, value: float,
                      timestamp: datetime = None,
                      metadata: Dict = None) -> SensorReading:
        """Ingest a sensor reading."""
        if sensor_id not in self.sensors:
            raise ValueError(f"Unknown sensor: {sensor_id}")

        sensor = self.sensors[sensor_id]

        # Validate data quality
        quality = self._validate_reading(sensor, value)

        reading = SensorReading(
            sensor_id=sensor_id,
            sensor_type=sensor.sensor_type,
            timestamp=timestamp or datetime.now(),
            value=value,
            unit=sensor.unit,
            quality=quality,
            location=sensor.location,
            metadata=metadata or {}
        )

        self.readings.append(reading)

        # Check thresholds
        if quality == DataQuality.GOOD:
            self._check_thresholds(sensor, reading)

        return reading

    def ingest_batch(self, readings: List[Dict]) -> int:
        """Ingest multiple readings at once."""
        count = 0
        for r in readings:
            try:
                self.ingest_reading(
                    sensor_id=r['sensor_id'],
                    value=r['value'],
                    timestamp=r.get('timestamp', datetime.now()),
                    metadata=r.get('metadata')
                )
                count += 1
            except Exception:
                pass  # Log error but continue
        return count

    def _validate_reading(self, sensor: Sensor, value: float) -> DataQuality:
        """Validate reading quality."""
        thresholds = sensor.thresholds

        # Check if value is within physical limits
        if 'min' in thresholds and value < thresholds['min']:
            return DataQuality.SUSPECT
        if 'max' in thresholds and value > thresholds['max']:
            return DataQuality.SUSPECT

        # Check for sudden spikes (compare with recent readings)
        recent = self.get_recent_readings(sensor.id, minutes=5)
        if len(recent) >= 3:
            avg = statistics.mean([r.value for r in recent])
            if abs(value - avg) > avg * 0.5:  # 50% deviation
                return DataQuality.SUSPECT

        return DataQuality.GOOD

    def _check_thresholds(self, sensor: Sensor, reading: SensorReading):
        """Check if reading exceeds thresholds."""
        thresholds = sensor.thresholds

        if 'critical' in thresholds and reading.value >= thresholds['critical']:
            self._create_alert(sensor, reading, AlertSeverity.CRITICAL)
        elif 'warning' in thresholds and reading.value >= threshol