environmental-monitoring

name	environmental-monitoring
description	Monitor environmental conditions on construction sites. Track air quality, noise levels, vibration, dust, and weather to ensure compliance and worker safety.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Environmental Monitoring

Overview

Monitor and analyze environmental conditions on construction sites including air quality, noise, vibration, dust, and weather. Support regulatory compliance, worker safety, and community relations through real-time environmental tracking.

Environmental Monitoring System

┌─────────────────────────────────────────────────────────────────┐
│                ENVIRONMENTAL MONITORING                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  SENSORS                   MONITORING            COMPLIANCE     │
│  ───────                   ──────────            ──────────     │
│                                                                  │
│  💨 Air Quality  ───┐                            ✅ OSHA limits  │
│  🔊 Noise Level  ───┼─────→ Real-time   ────────→ ✅ EPA limits  │
│  📊 Vibration    ───┤       Dashboard            ✅ Local codes │
│  🌫️ Dust/PM     ───┤       Alerts               ✅ Permits     │
│  🌡️ Weather     ───┘       Reports              ✅ Neighbors   │
│                                                                  │
│  THRESHOLDS:                                                    │
│  • Noise: 85 dB (OSHA 8hr TWA)                                 │
│  • PM2.5: 35 µg/m³ (EPA 24hr)                                  │
│  • Vibration: 25 mm/s (structural)                             │
│  • CO: 50 ppm (OSHA ceiling)                                   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

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

class ParameterType(Enum):
    NOISE = "noise"
    PM25 = "pm25"
    PM10 = "pm10"
    CO = "co"
    CO2 = "co2"
    VOC = "voc"
    VIBRATION = "vibration"
    TEMPERATURE = "temperature"
    HUMIDITY = "humidity"
    WIND_SPEED = "wind_speed"
    WIND_DIRECTION = "wind_direction"
    RAINFALL = "rainfall"

class ComplianceStatus(Enum):
    COMPLIANT = "compliant"
    WARNING = "warning"
    EXCEEDANCE = "exceedance"
    CRITICAL = "critical"

class AlertType(Enum):
    THRESHOLD_WARNING = "threshold_warning"
    THRESHOLD_EXCEEDANCE = "threshold_exceedance"
    EQUIPMENT_MALFUNCTION = "equipment_malfunction"
    WEATHER_ALERT = "weather_alert"
    COMMUNITY_COMPLAINT = "community_complaint"

@dataclass
class RegulatoryLimit:
    parameter: ParameterType
    limit_value: 
    unit: 
    averaging_period_hours:   
    regulation:   
    description: 


 :
    station_id: 
    parameter: ParameterType
    timestamp: datetime
    value: 
    unit: 
    quality_flag:  = 


 :
    : 
    name: 
    location:   
    parameters: [ParameterType]
    installation_date: datetime
    last_calibration: datetime
    status:  = 


 :
    parameter: ParameterType
    regulation: 
    limit_value: 
    measured_value: 
    averaging_period: 
    status: ComplianceStatus
    timestamp: datetime
    location: 


 :
    : 
    alert_type: AlertType
    parameter: ParameterType
    station_id: 
    timestamp: datetime
    value: 
    threshold: 
    message: 
    acknowledged:  = 
    resolved:  = 
    resolution_notes:  = 


 :
    date: datetime
    site_name: 
    parameters_monitored: 
    readings_collected: 
    exceedances: 
    alerts_triggered: 
    compliance_status: ComplianceStatus
    summary: [, ]

 :
    

    
    REGULATORY_LIMITS = {
        ParameterType.NOISE: [
            RegulatoryLimit(ParameterType.NOISE, , , , , ),
            RegulatoryLimit(ParameterType.NOISE, , , , , ),
            RegulatoryLimit(ParameterType.NOISE, , , , , ),
        ],
        ParameterType.PM25: [
            RegulatoryLimit(ParameterType.PM25, , , , , ),
            RegulatoryLimit(ParameterType.PM25, , , , , ),
        ],
        ParameterType.PM10: [
            RegulatoryLimit(ParameterType.PM10, , , , , ),
        ],
        ParameterType.CO: [
            RegulatoryLimit(ParameterType.CO, , , , , ),
            RegulatoryLimit(ParameterType.CO, , , , , ),
        ],
        ParameterType.VIBRATION: [
            RegulatoryLimit(ParameterType.VIBRATION, , , , , ),
            RegulatoryLimit(ParameterType.VIBRATION, , , , , ),
        ],
    }

     ():
        .site_name = site_name
        .stations: [, MonitoringStation] = {}
        .readings: [EnvironmentalReading] = []
        .alerts: [EnvironmentalAlert] = []
        .custom_limits: [ParameterType, [RegulatoryLimit]] = {}

     () -> MonitoringStation:
        
        station = MonitoringStation(
            =,
            name=name,
            location=location,
            parameters=parameters,
            installation_date=datetime.now(),
            last_calibration=datetime.now()
        )
        .stations[] = station
         station

     ():
        
        limit = RegulatoryLimit(
            parameter=parameter,
            limit_value=limit_value,
            unit=unit,
            averaging_period_hours=averaging_hours,
            regulation=regulation,
            description=description
        )
         parameter   .custom_limits:
            .custom_limits[parameter] = []
        .custom_limits[parameter].append(limit)

     () -> EnvironmentalReading:
        
         station_id   .stations:
             ValueError()

        reading = EnvironmentalReading(
            station_id=station_id,
            parameter=parameter,
            timestamp=timestamp  datetime.now(),
            value=value,
            unit=unit
        )

        .readings.append(reading)

        
        ._check_limits(station_id, parameter, value)

         reading

     () -> :
        
        count = 
         r  readings:
            :
                .record_reading(
                    station_id=r[],
                    parameter=ParameterType(r[]),
                    value=r[],
                    unit=r[],
                    timestamp=r.get()
                )
                count += 
             Exception:
                
         count

     ():
        
        
        limits = .REGULATORY_LIMITS.get(parameter, [])
        limits.extend(.custom_limits.get(parameter, []))

         limit  limits:
             limit.averaging_period_hours == :
                
                check_value = value
            :
                
                check_value = ._calculate_twa(
                    station_id, parameter, limit.averaging_period_hours
                )
                 check_value  :
                    

            
             check_value >= limit.limit_value:
                ._create_alert(
                    station_id, parameter, check_value, limit
                )
             check_value >= limit.limit_value * :
                
                ._create_alert(
                    station_id, parameter, check_value, limit,
                    is_warning=
                )

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

          readings:
             

         statistics.mean([r.value  r  readings])

     ():
        
        
        recent_alerts = [a  a  .alerts
                         a.station_id == station_id
                         a.parameter == parameter
                          a.resolved
                         (datetime.now() - a.timestamp).total_seconds() < ]

         recent_alerts:
            

        alert_type = (AlertType.THRESHOLD_WARNING  is_warning
                      AlertType.THRESHOLD_EXCEEDANCE)

        station = .stations.get(station_id)

        alert = EnvironmentalAlert(
            =,
            alert_type=alert_type,
            parameter=parameter,
            station_id=station_id,
            timestamp=datetime.now(),
            value=value,
            threshold=limit.limit_value,
            message=
                    
                    
        )

        .alerts.append(alert)

     () -> :
        
        conditions = {}

        stations = ([.stations[station_id]]  station_id
                    .stations.values())

         station  stations:
            station_conditions = {}

             param  station.parameters:
                
                readings = [r  r  .readings
                            r.station_id == station.
                            r.parameter == param]

                 readings:
                    latest = (readings, key= r: r.timestamp)
                    station_conditions[param.value] = {
                        : latest.value,
                        : latest.unit,
                        : latest.timestamp,
                        : ._get_compliance_status(param, latest.value)
                    }

            conditions[station.] = {
                : station.name,
                : station.location,
                : station_conditions
            }

         conditions

     () -> ComplianceStatus:
        
        limits = .REGULATORY_LIMITS.get(parameter, [])
        limits.extend(.custom_limits.get(parameter, []))

        
        instant_limits = [l  l  limits  l.averaging_period_hours == ]
         limit  instant_limits:
             value >= limit.limit_value:
                 ComplianceStatus.EXCEEDANCE
             value >= limit.limit_value * :
                 ComplianceStatus.WARNING

         ComplianceStatus.COMPLIANT

     () -> [ComplianceRecord]:
        
        records = []

         station  .stations.values():
             param  station.parameters:
                limits = .REGULATORY_LIMITS.get(param, [])
                limits.extend(.custom_limits.get(param, []))

                 limit  limits:
                    
                    readings = [r  r  .readings
                                r.station_id == station.
                                r.parameter == param
                                start_date <= r.timestamp <= end_date]

                      readings:
                        

                    avg_value = statistics.mean([r.value  r  readings])
                    max_value = (r.value  r  readings)

                    
                     limit.averaging_period_hours == :
                        check_value = max_value
                        period_str = 
                    :
                        check_value = avg_value
                        period_str = 

                    
                     check_value >= limit.limit_value:
                        status = ComplianceStatus.EXCEEDANCE
                     check_value >= limit.limit_value * :
                        status = ComplianceStatus.WARNING
                    :
                        status = ComplianceStatus.COMPLIANT

                    records.append(ComplianceRecord(
                        parameter=param,
                        regulation=limit.regulation,
                        limit_value=limit.limit_value,
                        measured_value=check_value,
                        averaging_period=period_str,
                        status=status,
                        timestamp=end_date,
                        location=station.name
                    ))

         records

     () -> :
        
        cutoff = datetime.now() - timedelta(days=days)
        recent_alerts = [a  a  .alerts
                        a.timestamp > cutoff
                        a.alert_type == AlertType.THRESHOLD_EXCEEDANCE]

        summary = {
            : days,
            : (recent_alerts),
            : {},
            : {},
            : []
        }

         alert  recent_alerts:
            
            param = alert.parameter.value
            summary[][param] = summary[].get(param, ) + 

            
            station = alert.station_id
            summary[][station] = summary[].get(station, ) + 

        
        summary[] = (
            recent_alerts, key= a: a.timestamp, reverse=
        )[:]

         summary

     () -> DailyReport:
        
         date  :
            date = datetime.now().replace(hour=, minute=, second=, microsecond=)

        next_day = date + timedelta(days=)

        
        day_readings = [r  r  .readings
                        date <= r.timestamp < next_day]

        
        day_alerts = [a  a  .alerts
                      date <= a.timestamp < next_day]

        
        compliance_records = .check_compliance(date, next_day)
        exceedances = [r  r  compliance_records
                       r.status == ComplianceStatus.EXCEEDANCE]

        
         exceedances:
            overall_status = ComplianceStatus.EXCEEDANCE
         (r.status == ComplianceStatus.WARNING  r  compliance_records):
            overall_status = ComplianceStatus.WARNING
        :
            overall_status = ComplianceStatus.COMPLIANT

        
        param_summary = {}
         param  ParameterType:
            param_readings = [r  r  day_readings  r.parameter == param]
             param_readings:
                values = [r.value  r  param_readings]
                param_summary[param.value] = {
                    : (values),
                    : (values),
                    : (values),
                    : statistics.mean(values),
                    : ([r  r  compliance_records
                                        r.parameter == param
                                        r.status == ComplianceStatus.EXCEEDANCE])
                }

         DailyReport(
            date=date,
            site_name=.site_name,
            parameters_monitored=((r.parameter  r  day_readings)),
            readings_collected=(day_readings),
            exceedances=(exceedances),
            alerts_triggered=(day_alerts),
            compliance_status=overall_status,
            summary=param_summary
        )

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

         station  .stations.values():
            params = .join([p.value  p  station.parameters])
            lines.append(
                
                
            )

        
        conditions = .get_current_conditions()
        lines.extend([
            ,
            ,
            
        ])

         station_id, data  conditions.items():
            lines.append()
            lines.append()
            lines.append()
            lines.append()

             param, values  data[].items():
                status_icon = (  values[] == ComplianceStatus.COMPLIANT
                                 values[] == ComplianceStatus.WARNING
                               )
                lines.append(
                    
                    
                )

            lines.append()

        
        exceedance_summary = .get_exceedance_summary()
        lines.extend([
            ,
            ,
            ,
            
        ])

         exceedance_summary[]:
            lines.append()
             param, count  exceedance_summary[].items():
                lines.append()

        
        active_alerts = [a  a  .alerts   a.resolved]
         active_alerts:
            lines.extend([
                ,
                ,
                ,
                ,
                
            ])

             alert  (active_alerts, key= a: a.timestamp, reverse=)[:]:
                lines.append(
                    
                    
                    
                )

         .join(lines)

Quick Start

from datetime import datetime, timedelta

# Initialize monitor
monitor = EnvironmentalMonitor("Downtown Construction Site")

# Add monitoring stations
monitor.add_station(
    "STA-001", "North Perimeter",
    location={"lat": 40.7128, "lon": -74.0060, "description": "North fence line"},
    parameters=[ParameterType.NOISE, ParameterType.PM25, ParameterType.PM10]
)

monitor.add_station(
    "STA-002", "Equipment Area",
    location={"lat": 40.7125, "lon": -74.0055, "description": "Near excavation"},
    parameters=[ParameterType.NOISE, ParameterType.VIBRATION, ParameterType.CO]
)

# Add custom limit for local ordinance
monitor.add_custom_limit(
    ParameterType.NOISE, 65, "dBA", 0,
    "Local Ordinance", "Residential boundary limit"
)

# Record readings
monitor.record_reading("STA-001", ParameterType.NOISE, 78.5, "dBA")
monitor.record_reading("STA-001", ParameterType.PM25, 28.3, "µg/m³")
monitor.record_reading("STA-002", ParameterType.VIBRATION, 8.2, "mm/s")

# Batch record
readings = [
    {"station_id": "STA-001", "parameter": "noise", "value": 82.0, "unit": },
    {: , : , : , : },
    {: , : , : , : }
]
monitor.record_batch(readings)


conditions = monitor.get_current_conditions()
 station, data  conditions.items():
    ()
     param, values  data[].items():
        ()


compliance = monitor.check_compliance(
    datetime.now() - timedelta(days=),
    datetime.now()
)
 record  compliance:
     record.status != ComplianceStatus.COMPLIANT:
        ()


report = monitor.generate_daily_report()
()
()


(monitor.generate_report())

Requirements

pip install (no external dependencies)