| 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
monitor = EnvironmentalMonitor("Downtown Construction Site")
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]
)
monitor.add_custom_limit(
ParameterType.NOISE, 65, "dBA", 0,
"Local Ordinance", "Residential boundary limit"
)
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")
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)
