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

python
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 >= thresholds['warning']:
            self._create_alert(sensor, reading, AlertSeverity.WARNING)

    def _create_alert(self, sensor: Sensor, reading: SensorReading,
                     severity: AlertSeverity):
        """Create and dispatch alert."""
        threshold = sensor.thresholds.get(severity.value, 0)

        alert = Alert(
            id=f"ALERT-{len(self.alerts)+1:06d}",
            sensor_id=sensor.id,
            sensor_type=sensor.sensor_type,
            severity=severity,
            timestamp=reading.timestamp,
            value=reading.value,
            threshold=threshold,
            message=f"{sensor.name}: {reading.value} {reading.unit} exceeds {severity.value} threshold ({threshold})"
        )

        self.alerts.append(alert)

        # Dispatch to handlers
        for handler in self.alert_handlers:
            try:
                handler(alert)
            except Exception:
                pass

    def register_alert_handler(self, handler: Callable):
        """Register alert callback handler."""
        self.alert_handlers.append(handler)

    def get_recent_readings(self, sensor_id: str,
                           minutes: int = 60) -> List[SensorReading]:
        """Get recent readings for sensor."""
        cutoff = datetime.now() - timedelta(minutes=minutes)
        return [r for r in self.readings
                if r.sensor_id == sensor_id and r.timestamp > cutoff]

    def get_readings_by_type(self, sensor_type: SensorType,
                            start: datetime = None,
                            end: datetime = None) -> List[SensorReading]:
        """Get readings by sensor type."""
        readings = [r for r in self.readings if r.sensor_type == sensor_type]

        if start:
            readings = [r for r in readings if r.timestamp >= start]
        if end:
            readings = [r for r in readings if r.timestamp <= end]

        return readings

    def aggregate_by_period(self, sensor_type: SensorType,
                           period_minutes: int = 60) -> List[AggregatedMetric]:
        """Aggregate readings into time periods."""
        readings = self.get_readings_by_type(sensor_type)

        if not readings:
            return []

        # Group by period
        periods: Dict[datetime, List[SensorReading]] = {}
        for r in readings:
            # Round to period start
            period_start = r.timestamp.replace(
                minute=(r.timestamp.minute // period_minutes) * period_minutes,
                second=0,
                microsecond=0
            )
            if period_start not in periods:
                periods[period_start] = []
            periods[period_start].append(r)

        # Calculate aggregates
        aggregates = []
        for period_start, period_readings in sorted(periods.items()):
            values = [r.value for r in period_readings]

            # Count alerts in period
            period_end = period_start + timedelta(minutes=period_minutes)
            period_alerts = len([a for a in self.alerts
                                if a.sensor_type == sensor_type
                                and period_start <= a.timestamp < period_end])

            aggregates.append(AggregatedMetric(
                sensor_type=sensor_type,
                period_start=period_start,
                period_end=period_end,
                readings_count=len(values),
                min_value=min(values),
                max_value=max(values),
                avg_value=statistics.mean(values),
                std_dev=statistics.stdev(values) if len(values) > 1 else 0,
                alerts_triggered=period_alerts
            ))

        return aggregates

    def detect_anomalies(self, sensor_id: str,
                        lookback_hours: int = 24) -> List[Dict]:
        """Detect anomalies in sensor data."""
        cutoff = datetime.now() - timedelta(hours=lookback_hours)
        readings = [r for r in self.readings
                   if r.sensor_id == sensor_id and r.timestamp > cutoff]

        if len(readings) < 10:
            return []

        values = [r.value for r in readings]
        avg = statistics.mean(values)
        std = statistics.stdev(values)

        anomalies = []
        for r in readings:
            # Z-score based anomaly detection
            if std > 0:
                z_score = abs(r.value - avg) / std
                if z_score > 3:  # 3 standard deviations
                    anomalies.append({
                        "timestamp": r.timestamp,
                        "value": r.value,
                        "expected": avg,
                        "z_score": z_score,
                        "type": "statistical_outlier"
                    })

        return anomalies

    def get_sensor_health(self) -> List[Dict]:
        """Get health status of all sensors."""
        health = []
        now = datetime.now()

        for sensor in self.sensors.values():
            recent = self.get_recent_readings(sensor.id, minutes=30)

            # Determine status
            if not recent:
                status = "offline"
            elif sensor.battery_level < 20:
                status = "low_battery"
            elif any(r.quality != DataQuality.GOOD for r in recent[-5:]):
                status = "degraded"
            else:
                status = "healthy"

            health.append({
                "sensor_id": sensor.id,
                "sensor_name": sensor.name,
                "type": sensor.sensor_type.value,
                "status": status,
                "battery": sensor.battery_level,
                "last_reading": recent[-1].timestamp if recent else None,
                "readings_30min": len(recent)
            })

        return sorted(health, key=lambda x: x['status'] != 'healthy', reverse=True)

    def get_zone_summary(self, zone: str) -> Dict:
        """Get summary for specific zone."""
        zone_sensors = [s for s in self.sensors.values()
                       if s.location.get('zone') == zone]

        if not zone_sensors:
            return {"zone": zone, "error": "No sensors in zone"}

        summary = {
            "zone": zone,
            "sensor_count": len(zone_sensors),
            "by_type": {}
        }

        for sensor in zone_sensors:
            recent = self.get_recent_readings(sensor.id, minutes=15)
            if not recent:
                continue

            values = [r.value for r in recent]
            sensor_type = sensor.sensor_type.value

            if sensor_type not in summary["by_type"]:
                summary["by_type"][sensor_type] = {
                    "current": values[-1] if values else None,
                    "avg": statistics.mean(values) if values else None,
                    "unit": sensor.unit,
                    "status": "normal"
                }

                # Check status
                thresholds = sensor.thresholds
                current = values[-1]
                if 'critical' in thresholds and current >= thresholds['critical']:
                    summary["by_type"][sensor_type]["status"] = "critical"
                elif 'warning' in thresholds and current >= thresholds['warning']:
                    summary["by_type"][sensor_type]["status"] = "warning"

        return summary

    def generate_report(self) -> str:
        """Generate sensor data report."""
        lines = [
            "# Sensor Data Report",
            "",
            f"**Site:** {self.site_name}",
            f"**Report Date:** {datetime.now().strftime('%Y-%m-%d %H:%M')}",
            "",
            "## Sensor Inventory",
            "",
            f"| Sensor | Type | Location | Status |",
            f"|--------|------|----------|--------|"
        ]

        health = self.get_sensor_health()
        for h in health:
            status_icon = "✅" if h['status'] == 'healthy' else "⚠️" if h['status'] == 'degraded' else "🔴"
            lines.append(
                f"| {h['sensor_name']} | {h['type']} | - | {status_icon} {h['status']} |"
            )

        # Recent alerts
        recent_alerts = [a for a in self.alerts
                       if a.timestamp > datetime.now() - timedelta(hours=24)]

        if recent_alerts:
            lines.extend([
                "",
                f"## Alerts (Last 24h) - {len(recent_alerts)} total",
                "",
                "| Time | Sensor | Severity | Value | Threshold |",
                "|------|--------|----------|-------|-----------|"
            ])

            for alert in sorted(recent_alerts, key=lambda x: x.timestamp, reverse=True)[:20]:
                sev_icon = "🔴" if alert.severity == AlertSeverity.CRITICAL else "🟡"
                lines.append(
                    f"| {alert.timestamp.strftime('%H:%M')} | {alert.sensor_id} | "
                    f"{sev_icon} {alert.severity.value} | {alert.value:.1f} | {alert.threshold} |"
                )

        # Current readings by type
        lines.extend([
            "",
            "## Current Readings by Type",
            ""
        ])

        for sensor_type in SensorType:
            readings = self.get_readings_by_type(sensor_type)
            if not readings:
                continue

            recent = [r for r in readings
                     if r.timestamp > datetime.now() - timedelta(minutes=15)]
            if not recent:
                continue

            values = [r.value for r in recent]
            lines.append(
                f"**{sensor_type.value}**: "
                f"Avg={statistics.mean(values):.1f}, "
                f"Min={min(values):.1f}, "
                f"Max={max(values):.1f}"
            )

        return "\n".join(lines)

Quick Start

python
from datetime import datetime, timedelta

# Initialize aggregator
aggregator = SensorDataAggregator("Construction Site A")

# Register sensors
aggregator.register_sensor(
    "TEMP-001", "Zone A Temperature",
    SensorType.TEMPERATURE, "°C",
    location={"zone": "A", "floor": 1, "x": 10, "y": 20},
    thresholds={"warning": 32, "critical": 38, "min": -10, "max": 50}
)

aggregator.register_sensor(
    "VIB-001", "Foundation Vibration",
    SensorType.VIBRATION, "mm/s",
    location={"zone": "Foundation", "floor": 0}
)

aggregator.register_sensor(
    "DUST-001", "Dust Monitor",
    SensorType.DUST, "mg/m³",
    location={"zone": "A", "floor": 1}
)

# Register alert handler
def handle_alert(alert):
    print(f"ALERT: {alert.severity.value} - {alert.message}")

aggregator.register_alert_handler(handle_alert)

# Ingest readings
aggregator.ingest_reading("TEMP-001", 28.5)
aggregator.ingest_reading("TEMP-001", 33.0)  # Warning!
aggregator.ingest_reading("VIB-001", 5.2)
aggregator.ingest_reading("DUST-001", 2.1)

# Batch ingest
readings = [
    {"sensor_id": "TEMP-001", "value": 29.0},
    {"sensor_id": "VIB-001", "value": 4.8},
    {"sensor_id": "DUST-001", "value": 2.5}
]
aggregator.ingest_batch(readings)

# Check sensor health
health = aggregator.get_sensor_health()
for h in health:
    print(f"{h['sensor_name']}: {h['status']}")

# Get zone summary
summary = aggregator.get_zone_summary("A")
print(f"Zone A: {summary}")

# Detect anomalies
anomalies = aggregator.detect_anomalies("TEMP-001")
print(f"Anomalies found: {len(anomalies)}")

# Generate report
print(aggregator.generate_report())

Requirements

bash
pip install (no external dependencies)