sensor-data-aggregator

name	sensor-data-aggregator
description	Aggregate and analyze IoT sensor data from construction sites. Collect data from multiple sensor types, detect anomalies, and trigger alerts for safety and quality monitoring.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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: 
    unit: 
    quality: DataQuality = DataQuality.GOOD
    location: [] = 
    metadata:  = field(default_factory=)


 :
    : 
    name: 
    sensor_type: SensorType
    unit: 
    location:   
    thresholds:   
    calibration_date: datetime
    battery_level:  = 
    status:  = 


 :
    : 
    sensor_id: 
    sensor_type: SensorType
    severity: AlertSeverity
    timestamp: datetime
    value: 
    threshold: 
    message: 
    acknowledged:  = 
    resolved:  = 


 :
    sensor_type: SensorType
    period_start: datetime
    period_end: datetime
    readings_count: 
    min_value: 
    max_value: 
    avg_value: 
    std_dev: 
    alerts_triggered: 

 :
    

    
    DEFAULT_THRESHOLDS = {
        SensorType.TEMPERATURE: {: , : , : },
        SensorType.HUMIDITY: {: , : , : },
        SensorType.VIBRATION: {: , : , : },
        SensorType.NOISE: {: , : , : },
        SensorType.AIR_QUALITY: {: , : , : },
        SensorType.DUST: {: , : , : },
        SensorType.GAS: {: , : , : },
    }

     ():
        .site_name = site_name
        .sensors: [, Sensor] = {}
        .readings: [SensorReading] = []
        .alerts: [Alert] = []
        .alert_handlers: [] = []

     () -> Sensor:
        
         thresholds  :
            thresholds = .DEFAULT_THRESHOLDS.get(sensor_type, {})

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

     () -> SensorReading:
        
         sensor_id   .sensors:
             ValueError()

        sensor = .sensors[sensor_id]

        
        quality = ._validate_reading(sensor, value)

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

        .readings.append(reading)

        
         quality == DataQuality.GOOD:
            ._check_thresholds(sensor, reading)

         reading

     () -> :
        
        count = 
         r  readings:
            :
                .ingest_reading(
                    sensor_id=r[],
                    value=r[],
                    timestamp=r.get(, datetime.now()),
                    metadata=r.get()
                )
                count += 
             Exception:
                  
         count

     () -> DataQuality:
        
        thresholds = sensor.thresholds

        
           thresholds  value < thresholds[]:
             DataQuality.SUSPECT
           thresholds  value > thresholds[]:
             DataQuality.SUSPECT

        
        recent = .get_recent_readings(sensor., minutes=)
         (recent) >= :
            avg = statistics.mean([r.value  r  recent])
             (value - avg) > avg * :  
                 DataQuality.SUSPECT

         DataQuality.GOOD

     ():
        
        thresholds = sensor.thresholds

           thresholds  reading.value >= thresholds[]:
            ._create_alert(sensor, reading, AlertSeverity.CRITICAL)
           thresholds  reading.value >= thresholds[]:
            ._create_alert(sensor, reading, AlertSeverity.WARNING)

     ():
        
        threshold = sensor.thresholds.get(severity.value, )

        alert = Alert(
            =,
            sensor_id=sensor.,
            sensor_type=sensor.sensor_type,
            severity=severity,
            timestamp=reading.timestamp,
            value=reading.value,
            threshold=threshold,
            message=
        )

        .alerts.append(alert)

        
         handler  .alert_handlers:
            :
                handler(alert)
             Exception:
                

     ():
        
        .alert_handlers.append(handler)

     () -> [SensorReading]:
        
        cutoff = datetime.now() - timedelta(minutes=minutes)
         [r  r  .readings
                 r.sensor_id == sensor_id  r.timestamp > cutoff]

     () -> [SensorReading]:
        
        readings = [r  r  .readings  r.sensor_type == sensor_type]

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

         readings

     () -> [AggregatedMetric]:
        
        readings = .get_readings_by_type(sensor_type)

          readings:
             []

        
        periods: [datetime, [SensorReading]] = {}
         r  readings:
            
            period_start = r.timestamp.replace(
                minute=(r.timestamp.minute // period_minutes) * period_minutes,
                second=,
                microsecond=
            )
             period_start   periods:
                periods[period_start] = []
            periods[period_start].append(r)

        
        aggregates = []
         period_start, period_readings  (periods.items()):
            values = [r.value  r  period_readings]

            
            period_end = period_start + timedelta(minutes=period_minutes)
            period_alerts = ([a  a  .alerts
                                 a.sensor_type == sensor_type
                                 period_start <= a.timestamp < period_end])

            aggregates.append(AggregatedMetric(
                sensor_type=sensor_type,
                period_start=period_start,
                period_end=period_end,
                readings_count=(values),
                min_value=(values),
                max_value=(values),
                avg_value=statistics.mean(values),
                std_dev=statistics.stdev(values)  (values) >   ,
                alerts_triggered=period_alerts
            ))

         aggregates

     () -> []:
        
        cutoff = datetime.now() - timedelta(hours=lookback_hours)
        readings = [r  r  .readings
                    r.sensor_id == sensor_id  r.timestamp > cutoff]

         (readings) < :
             []

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

        anomalies = []
         r  readings:
            
             std > :
                z_score = (r.value - avg) / std
                 z_score > :  
                    anomalies.append({
                        : r.timestamp,
                        : r.value,
                        : avg,
                        : z_score,
                        : 
                    })

         anomalies

     () -> []:
        
        health = []
        now = datetime.now()

         sensor  .sensors.values():
            recent = .get_recent_readings(sensor., minutes=)

            
              recent:
                status = 
             sensor.battery_level < :
                status = 
             (r.quality != DataQuality.GOOD  r  recent[-:]):
                status = 
            :
                status = 

            health.append({
                : sensor.,
                : sensor.name,
                : sensor.sensor_type.value,
                : status,
                : sensor.battery_level,
                : recent[-].timestamp  recent  ,
                : (recent)
            })

         (health, key= x: x[] != , reverse=)

     () -> :
        
        zone_sensors = [s  s  .sensors.values()
                        s.location.get() == zone]

          zone_sensors:
             {: zone, : }

        summary = {
            : zone,
            : (zone_sensors),
            : {}
        }

         sensor  zone_sensors:
            recent = .get_recent_readings(sensor., minutes=)
              recent:
                

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

             sensor_type   summary[]:
                summary[][sensor_type] = {
                    : values[-]  values  ,
                    : statistics.mean(values)  values  ,
                    : sensor.unit,
                    : 
                }

                
                thresholds = sensor.thresholds
                current = values[-]
                   thresholds  current >= thresholds[]:
                    summary[][sensor_type][] = 
                   thresholds  current >= thresholds[]:
                    summary[][sensor_type][] = 

         summary

     () -> :
        
        lines = [
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            
        ]

        health = .get_sensor_health()
         h  health:
            status_icon =   h[] ==     h[] ==   
            lines.append(
                
            )

        
        recent_alerts = [a  a  .alerts
                        a.timestamp > datetime.now() - timedelta(hours=)]

         recent_alerts:
            lines.extend([
                ,
                ,
                ,
                ,
                
            ])

             alert  (recent_alerts, key= x: x.timestamp, reverse=)[:]:
                sev_icon =   alert.severity == AlertSeverity.CRITICAL  
                lines.append(
                    
                    
                )

        
        lines.extend([
            ,
            ,
            
        ])

         sensor_type  SensorType:
            readings = .get_readings_by_type(sensor_type)
              readings:
                

            recent = [r  r  readings
                      r.timestamp > datetime.now() - timedelta(minutes=)]
              recent:
                

            values = [r.value  r  recent]
            lines.append(
                
                
                
                
            )

         .join(lines)

Quick Start

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(, )  
aggregator.ingest_reading(, )
aggregator.ingest_reading(, )


readings = [
    {: , : },
    {: , : },
    {: , : }
]
aggregator.ingest_batch(readings)


health = aggregator.get_sensor_health()
 h  health:
    ()


summary = aggregator.get_zone_summary()
()


anomalies = aggregator.detect_anomalies()
()


(aggregator.generate_report())

Requirements

pip install (no external dependencies)