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: float
unit: str
averaging_period_hours: float # e.g., 8 for 8-hour TWA
regulation: str # e.g., "OSHA", "EPA"
description: str
@dataclass
class EnvironmentalReading:
station_id: str
parameter: ParameterType
timestamp: datetime
value: float
unit: str
quality_flag: str = "valid"
@dataclass
class MonitoringStation:
id: str
name: str
location: Dict # {lat, lon, description}
parameters: List[ParameterType]
installation_date: datetime
last_calibration: datetime
status: str = "active"
@dataclass
class ComplianceRecord:
parameter: ParameterType
regulation: str
limit_value: float
measured_value: float
averaging_period: str
status: ComplianceStatus
timestamp: datetime
location: str
@dataclass
class EnvironmentalAlert:
id: str
alert_type: AlertType
parameter: ParameterType
station_id: str
timestamp: datetime
value: float
threshold: float
message: str
acknowledged: bool = False
resolved: bool = False
resolution_notes: str = ""
@dataclass
class DailyReport:
date: datetime
site_name: str
parameters_monitored: int
readings_collected: int
exceedances: int
alerts_triggered: int
compliance_status: ComplianceStatus
summary: Dict[str, Dict]
class EnvironmentalMonitor:
"""Monitor environmental conditions on construction sites."""
# Default regulatory limits
REGULATORY_LIMITS = {
ParameterType.NOISE: [
RegulatoryLimit(ParameterType.NOISE, 85, "dBA", 8.0, "OSHA", "8-hour TWA"),
RegulatoryLimit(ParameterType.NOISE, 90, "dBA", 8.0, "OSHA", "Action level"),
RegulatoryLimit(ParameterType.NOISE, 115, "dBA", 0.25, "OSHA", "15-min max"),
],
ParameterType.PM25: [
RegulatoryLimit(ParameterType.PM25, 35, "µg/m³", 24.0, "EPA", "24-hour standard"),
RegulatoryLimit(ParameterType.PM25, 12, "µg/m³", 8760.0, "EPA", "Annual standard"),
],
ParameterType.PM10: [
RegulatoryLimit(ParameterType.PM10, 150, "µg/m³", 24.0, "EPA", "24-hour standard"),
],
ParameterType.CO: [
RegulatoryLimit(ParameterType.CO, 50, "ppm", 0.0, "OSHA", "Ceiling limit"),
RegulatoryLimit(ParameterType.CO, 35, "ppm", 8.0, "OSHA", "8-hour TWA"),
],
ParameterType.VIBRATION: [
RegulatoryLimit(ParameterType.VIBRATION, 25, "mm/s", 0.0, "ISO 4866", "Structural damage threshold"),
RegulatoryLimit(ParameterType.VIBRATION, 5, "mm/s", 0.0, "DIN 4150", "Sensitive structures"),
],
}
def __init__(self, site_name: str):
self.site_name = site_name
self.stations: Dict[str, MonitoringStation] = {}
self.readings: List[EnvironmentalReading] = []
self.alerts: List[EnvironmentalAlert] = []
self.custom_limits: Dict[ParameterType, List[RegulatoryLimit]] = {}
def add_station(self, id: str, name: str, location: Dict,
parameters: List[ParameterType]) -> MonitoringStation:
"""Add monitoring station."""
station = MonitoringStation(
id=id,
name=name,
location=location,
parameters=parameters,
installation_date=datetime.now(),
last_calibration=datetime.now()
)
self.stations[id] = station
return station
def add_custom_limit(self, parameter: ParameterType, limit_value: float,
unit: str, averaging_hours: float, regulation: str,
description: str):
"""Add custom regulatory limit."""
limit = RegulatoryLimit(
parameter=parameter,
limit_value=limit_value,
unit=unit,
averaging_period_hours=averaging_hours,
regulation=regulation,
description=description
)
if parameter not in self.custom_limits:
self.custom_limits[parameter] = []
self.custom_limits[parameter].append(limit)
def record_reading(self, station_id: str, parameter: ParameterType,
value: float, unit: str,
timestamp: datetime = None) -> EnvironmentalReading:
"""Record environmental reading."""
if station_id not in self.stations:
raise ValueError(f"Unknown station: {station_id}")
reading = EnvironmentalReading(
station_id=station_id,
parameter=parameter,
timestamp=timestamp or datetime.now(),
value=value,
unit=unit
)
self.readings.append(reading)
# Check against limits
self._check_limits(station_id, parameter, value)
return reading
def record_batch(self, readings: List[Dict]) -> int:
"""Record multiple readings."""
count = 0
for r in readings:
try:
self.record_reading(
station_id=r['station_id'],
parameter=ParameterType(r['parameter']),
value=r['value'],
unit=r['unit'],
timestamp=r.get('timestamp')
)
count += 1
except Exception: