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