Imported from majiayu000/claude-skill-registry (skills/data/environmental-monitoring/SKILL.md). Install upstream with npx skills add majiayu000/claude-skill-registry --skill environmental-monitoring. Copyright stays with the author.

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)