tools-only 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:
pass
return count
def _check_limits(self, station_id: str, parameter: ParameterType, value: float):
"""Check value against regulatory limits."""
# Get applicable limits
limits = self.REGULATORY_LIMITS.get(parameter, [])
limits.extend(self.custom_limits.get(parameter, []))
for limit in limits:
if limit.averaging_period_hours == 0:
# Instantaneous limit
check_value = value
else:
# Time-weighted average
check_value = self._calculate_twa(
station_id, parameter, limit.averaging_period_hours
)
if check_value is None:
continue
# Check against limit
if check_value >= limit.limit_value:
self._create_alert(
station_id, parameter, check_value, limit
)
elif check_value >= limit.limit_value * 0.8:
# Warning at 80% of limit
self._create_alert(
station_id, parameter, check_value, limit,
is_warning=True
)
def _calculate_twa(self, station_id: str, parameter: ParameterType,
hours: float) -> Optional[float]:
"""Calculate time-weighted average."""
cutoff = datetime.now() - timedelta(hours=hours)
readings = [r for r in self.readings
if r.station_id == station_id
and r.parameter == parameter
and r.timestamp > cutoff]
if not readings:
return None
return statistics.mean([r.value for r in readings])
def _create_alert(self, station_id: str, parameter: ParameterType,
value: float, limit: RegulatoryLimit,
is_warning: bool = False):
"""Create environmental alert."""
# Avoid duplicate alerts
recent_alerts = [a for a in self.alerts
if a.station_id == station_id
and a.parameter == parameter
and not a.resolved
and (datetime.now() - a.timestamp).total_seconds() < 3600]
if recent_alerts:
return
alert_type = (AlertType.THRESHOLD_WARNING if is_warning
else AlertType.THRESHOLD_EXCEEDANCE)
station = self.stations.get(station_id)
alert = EnvironmentalAlert(
id=f"ENV-{len(self.alerts)+1:05d}",
alert_type=alert_type,
parameter=parameter,
station_id=station_id,
timestamp=datetime.now(),
value=value,
threshold=limit.limit_value,
message=f"{parameter.value} {'approaching' if is_warning else 'exceeds'} "
f"{limit.regulation} limit ({limit.limit_value} {limit.unit}) "
f"at {station.name if station else station_id}"
)
self.alerts.append(alert)
def get_current_conditions(self, station_id: str = None) -> Dict:
"""Get current environmental conditions."""
conditions = {}
stations = ([self.stations[station_id]] if station_id
else self.stations.values())
for station in stations:
station_conditions = {}
for param in station.parameters:
# Get latest reading
readings = [r for r in self.readings
if r.station_id == station.id
and r.parameter == param]
if readings:
latest = max(readings, key=lambda r: r.timestamp)
station_conditions[param.value] = {
"value": latest.value,
"unit": latest.unit,
"timestamp": latest.timestamp,
"status": self._get_compliance_status(param, latest.value)
}
conditions[station.id] = {
"name": station.name,
"location": station.location,
"parameters": station_conditions
}
return conditions
def _get_compliance_status(self, parameter: ParameterType,
value: float) -> ComplianceStatus:
"""Determine compliance status for value."""
limits = self.REGULATORY_LIMITS.get(parameter, [])
limits.extend(self.custom_limits.get(parameter, []))
# Check instantaneous limits
instant_limits = [l for l in limits if l.averaging_period_hours == 0]
for limit in instant_limits:
if value >= limit.limit_value:
return ComplianceStatus.EXCEEDANCE
elif value >= limit.limit_value * 0.9:
return ComplianceStatus.WARNING
return ComplianceStatus.COMPLIANT
def check_compliance(self, start_date: datetime,
end_date: datetime) -> List[ComplianceRecord]:
"""Check compliance for period."""
records = []
for station in self.stations.values():
for param in station.parameters:
limits = self.REGULATORY_LIMITS.get(param, [])
limits.extend(self.custom_limits.get(param, []))
for limit in limits:
# Calculate average for period
readings = [r for r in self.readings
if r.station_id == station.id
and r.parameter == param
and start_date <= r.timestamp <= end_date]
if not readings:
continue
avg_value = statistics.mean([r.value for r in readings])
max_value = max(r.value for r in readings)
# Check appropriate value
if limit.averaging_period_hours == 0:
check_value = max_value
period_str = "Instantaneous"
else:
check_value = avg_value
period_str = f"{limit.averaging_period_hours:.0f}-hour avg"
# Determine status
if check_value >= limit.limit_value:
status = ComplianceStatus.EXCEEDANCE
elif check_value >= limit.limit_value * 0.9:
status = ComplianceStatus.WARNING
else:
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
))
return records
def get_exceedance_summary(self, days: int = 30) -> Dict:
"""Get summary of exceedances."""
cutoff = datetime.now() - timedelta(days=days)
recent_alerts = [a for a in self.alerts
if a.timestamp > cutoff
and a.alert_type == AlertType.THRESHOLD_EXCEEDANCE]
summary = {
"period_days": days,
"total_exceedances": len(recent_alerts),
"by_parameter": {},
"by_station": {},
"recent_events": []
}
for alert in recent_alerts:
# By parameter
param = alert.parameter.value
summary["by_parameter"][param] = summary["by_parameter"].get(param, 0) + 1
# By station
station = alert.station_id
summary["by_station"][station] = summary["by_station"].get(station, 0) + 1
# Recent events
summary["recent_events"] = sorted(
recent_alerts, key=lambda a: a.timestamp, reverse=True
)[:10]
return summary
def generate_daily_report(self, date: datetime = None) -> DailyReport:
"""Generate daily environmental report."""
if date is None:
date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
next_day = date + timedelta(days=1)
# Filter readings
day_readings = [r for r in self.readings
if date <= r.timestamp < next_day]
# Filter alerts
day_alerts = [a for a in self.alerts
if date <= a.timestamp < next_day]
# Check compliance
compliance_records = self.check_compliance(date, next_day)
exceedances = [r for r in compliance_records
if r.status == ComplianceStatus.EXCEEDANCE]
# Overall status
if exceedances:
overall_status = ComplianceStatus.EXCEEDANCE
elif any(r.status == ComplianceStatus.WARNING for r in compliance_records):
overall_status = ComplianceStatus.WARNING
else:
overall_status = ComplianceStatus.COMPLIANT
# Summary by parameter
param_summary = {}
for param in ParameterType:
param_readings = [r for r in day_readings if r.parameter == param]
if param_readings:
values = [r.value for r in param_readings]
param_summary[param.value] = {
"count": len(values),
"min": min(values),
"max": max(values),
"avg": statistics.mean(values),
"exceedances": len([r for r in compliance_records
if r.parameter == param
and r.status == ComplianceStatus.EXCEEDANCE])
}
return DailyReport(
date=date,
site_name=self.site_name,
parameters_monitored=len(set(r.parameter for r in day_readings)),
readings_collected=len(day_readings),
exceedances=len(exceedances),
alerts_triggered=len(day_alerts),
compliance_status=overall_status,
summary=param_summary
)
def generate_report(self) -> str:
"""Generate environmental monitoring report."""
lines = [
"# Environmental Monitoring Report",
"",
f"**Site:** {self.site_name}",
f"**Report Date:** {datetime.now().strftime('%Y-%m-%d %H:%M')}",
"",
"## Monitoring Stations",
"",
"| Station | Location | Parameters | Status |",
"|---------|----------|------------|--------|"
]
for station in self.stations.values():
params = ", ".join([p.value for p in station.parameters])
lines.append(
f"| {station.name} | {station.location.get('description', '-')} | "
f"{params} | {station.status} |"
)
# Current conditions
conditions = self.get_current_conditions()
lines.extend([
"",
"## Current Conditions",
""
])
for station_id, data in conditions.items():
lines.append(f"### {data['name']}")
lines.append("")
lines.append("| Parameter | Value | Status |")
lines.append("|-----------|-------|--------|")
for param, values in data['parameters'].items():
status_icon = ("✅" if values['status'] == ComplianceStatus.COMPLIANT
else "⚠️" if values['status'] == ComplianceStatus.WARNING
else "🔴")
lines.append(
f"| {param} | {values['value']:.1f} {values['unit']} | "
f"{status_icon} {values['status'].value} |"
)
lines.append("")
# Exceedance summary
exceedance_summary = self.get_exceedance_summary(30)
lines.extend([
"## 30-Day Exceedance Summary",
"",
f"**Total Exceedances:** {exceedance_summary['total_exceedances']}",
""
])
if exceedance_summary['by_parameter']:
lines.append("By Parameter:")
for param, count in exceedance_summary['by_parameter'].items():
lines.append(f"- {param}: {count}")
# Active alerts
active_alerts = [a for a in self.alerts if not a.resolved]
if active_alerts:
lines.extend([
"",
f"## Active Alerts ({len(active_alerts)})",
"",
"| Time | Parameter | Station | Value | Threshold |",
"|------|-----------|---------|-------|-----------|"
])
for alert in sorted(active_alerts, key=lambda a: a.timestamp, reverse=True)[:10]:
lines.append(
f"| {alert.timestamp.strftime('%Y-%m-%d %H:%M')} | "
f"{alert.parameter.value} | {alert.station_id} | "
f"{alert.value:.1f} | {alert.threshold} |"
)
return "\n".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": "dBA"},
{"station_id": "STA-001", "parameter": "pm25", "value": 31.5, "unit": "µg/m³"},
{"station_id": "STA-002", "parameter": "noise", "value": 88.0, "unit": "dBA"}
]
monitor.record_batch(readings)
# Get current conditions
conditions = monitor.get_current_conditions()
for station, data in conditions.items():
print(f"\n{data['name']}:")
for param, values in data['parameters'].items():
print(f" {param}: {values['value']} {values['unit']} - {values['status'].value}")
# Check compliance
compliance = monitor.check_compliance(
datetime.now() - timedelta(days=1),
datetime.now()
)
for record in compliance:
if record.status != ComplianceStatus.COMPLIANT:
print(f"⚠️ {record.parameter.value}: {record.measured_value} vs limit {record.limit_value}")
# Generate daily report
report = monitor.generate_daily_report()
print(f"\nDaily Status: {report.compliance_status.value}")
print(f"Exceedances: {report.exceedances}")
# Full report
print(monitor.generate_report())
Requirements
pip install (no external dependencies)