Weather Impact Analysis

Weather Impact Analysis

Overview

This skill implements weather data analysis for construction project management. Integrate weather forecasts, historical data, and activity sensitivity to predict delays and optimize scheduling.

Capabilities:

• Weather forecast integration
• Activity weather sensitivity mapping
• Delay prediction and quantification
• Schedule optimization based on weather
• Historical weather impact analysis
• Risk factor calculation

Quick Start

from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import List, Dict, Optional
from enum import Enum
import requests

class WeatherCondition(Enum):
    CLEAR = "clear"
    CLOUDY = "cloudy"
    RAIN = "rain"
    HEAVY_RAIN = "heavy_rain"
    SNOW = "snow"
    FROST = "frost"
    HIGH_WIND = "high_wind"
    EXTREME_HEAT = "extreme_heat"
    EXTREME_COLD = "extreme_cold"

@dataclass
class WeatherDay:
    date: date
    condition: WeatherCondition
    temp_high: float
    temp_low: float
    precipitation_mm: float
    wind_speed_kmh: float
    humidity_pct: float

@dataclass
class ActivitySensitivity:
    activity_type: str
    min_temp: float
    max_temp: float
    max_wind: float
    max_precipitation: float
    can_work_in_rain: bool

def check_work_day(weather: WeatherDay, activity: ActivitySensitivity) -> Dict:
    """Check if work is possible for given weather and activity"""
    can_work = True
    reasons = []

    if weather.temp_low < activity.min_temp:
        can_work = False
        reasons.append(f"Temperature too low: {weather.temp_low}°C < {activity.min_temp}°C")

    if weather.temp_high > activity.max_temp:
        can_work = False
        reasons.append(f"Temperature too high: {weather.temp_high}°C > {activity.max_temp}°C")

    if weather.wind_speed_kmh > activity.max_wind:
        can_work = False
        reasons.append(f"Wind too strong: {weather.wind_speed_kmh} km/h > {activity.max_wind} km/h")

    if weather.precipitation_mm > activity.max_precipitation and not activity.can_work_in_rain:
        can_work = False
        reasons.append(f"Precipitation: {weather.precipitation_mm}mm")

    return {
        'date': weather.date,
        'can_work': can_work,
        'reasons': reasons,
        'productivity_factor': 1.0 if can_work else 0.0
    }

# Example
concrete_work = ActivitySensitivity(
    activity_type="concrete_placement",
    min_temp=5,
    max_temp=35,
    max_wind=40,
    max_precipitation=2,
    can_work_in_rain=False
)

today_weather = WeatherDay(
    date=date.today(),
    condition=WeatherCondition.RAIN,
    temp_high=15,
    temp_low=8,
    precipitation_mm=10,
    wind_speed_kmh=20,
    humidity_pct=80
)

result = check_work_day(today_weather, concrete_work)
print(f"Can work: {result['can_work']}, Reasons: {result['reasons']}")

Comprehensive Weather Analysis System

Weather Data Integration

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from typing import List, Dict, Optional, Tuple
from enum import Enum
import requests
import json

class WeatherSeverity(Enum):
    NORMAL = 1
    CAUTION = 2
    WARNING = 3
    SEVERE = 4
    EXTREME = 5

@dataclass
class HourlyWeather:
    datetime: datetime
    temperature: float
    feels_like: float
    humidity: float
    wind_speed: float
    wind_direction: float
    precipitation: float
    precipitation_probability: float
    condition: WeatherCondition
    visibility: float
    uv_index: float

@dataclass
class DailyForecast:
    date: date
    temp_high: float
    temp_low: float
    sunrise: datetime
    sunset: datetime
    precipitation_total: float
    precipitation_probability: float
    primary_condition: WeatherCondition
    hourly: List[HourlyWeather] = field(default_factory=list)
    severity: WeatherSeverity = WeatherSeverity.NORMAL

class WeatherDataService:
    """Weather data integration service"""

    def __init__(self, api_key: str = None, provider: str = "openweathermap"):
        self.api_key = api_key
        self.provider = provider
        self.cache: Dict[str, Dict] = {}
        self.cache_duration = timedelta(hours=1)

    def get_forecast(self, latitude: float, longitude: float,
                    days: int = 14) -> List[DailyForecast]:
        """Get weather forecast for location"""
        cache_key = f"{latitude},{longitude}"

        if cache_key in self.cache:
            cached = self.cache[cache_key]
            if datetime.now() - cached['timestamp'] < self.cache_duration:
                return cached['data']

        if self.provider == "openweathermap":
            forecast = self._fetch_openweathermap(latitude, longitude, days)
        else:
            forecast = self._generate_sample_forecast(days)

        self.cache[cache_key] = {
            'timestamp': datetime.now(),
            'data': forecast
        }

        return forecast

    def _fetch_openweathermap(self, lat: float, lon: float,
                              days: int) -> List[DailyForecast]:
        """Fetch from OpenWeatherMap API"""
        url = f"https://api.openweathermap.org/data/2.5/forecast"
        params = {
            'lat': lat,
            'lon': lon,
            'appid': self.api_key,
            'units': 'metric'
        }

        try:
            response = requests.get(url, params=params)
            data = response.json()
            return self._parse_openweathermap(data)
        except Exception as e:
            print(f"Weather API error: {e}")
            return self._generate_sample_forecast(days)

    def _parse_openweathermap(self, data: Dict) -> List[DailyForecast]:
        """Parse OpenWeatherMap response"""
        forecasts = []
        daily_data = {}

        for item in data.get('list', []):
            dt = datetime.fromtimestamp(item['dt'])
            day = dt.date()

            if day not in daily_data:
                daily_data[day] = {
                    'temps': [],
                    'precipitation': 0,
                    'conditions': [],
                    'hourly': []
                }

            daily_data[day]['temps'].append(item['main']['temp'])
            daily_data[day]['precipitation'] += item.get('rain', {}).get('3h', 0)

            condition = self._map_condition(item['weather'][0]['main'])
            daily_data[day]['conditions'].append(condition)

            daily_data[day]['hourly'].append(HourlyWeather(
                datetime=dt,
                temperature=item['main']['temp'],
                feels_like=item['main']['feels_like'],
                humidity=item['main']['humidity'],
                wind_speed=item['wind']['speed'] * 3.6,  # m/s to km/h
                wind_direction=item['wind'].get('deg', 0),
                precipitation=item.get('rain', {}).get('3h', 0),
                precipitation_probability=item.get('pop', 0) * 100,
                condition=condition,
                visibility=item.get('visibility', 10000) / 1000,
                uv_index=0
            ))

        for day, data in daily_data.items():
            primary_condition = max(set(data['conditions']), key=data['conditions'].count)

            forecasts.append(DailyForecast(
                date=day,
                temp_high=max(data['temps']),
                temp_low=min(data['temps']),
                sunrise=datetime.combine(day, datetime.min.time().replace(hour=6)),
                sunset=datetime.combine(day, datetime.min.time().replace(hour=18)),
                precipitation_total=data['precipitation'],
                precipitation_probability=max(h.precipitation_probability for h in data['hourly']),
                primary_condition=primary_condition,
                hourly=data['hourly'],
                severity=self._calculate_severity(primary_condition, data)
            ))

        return sorted(forecasts, key=lambda x: x.date)

    def _map_condition(self, condition_str: str) -> WeatherCondition:
        """Map API condition to enum"""
        mapping = {
            'Clear': WeatherCondition.CLEAR,
            'Clouds': WeatherCondition.CLOUDY,
            'Rain': WeatherCondition.RAIN,
            'Drizzle': WeatherCondition.RAIN,
            'Thunderstorm': WeatherCondition.HEAVY_RAIN,
            'Snow': WeatherCondition.SNOW,
            'Mist': WeatherCondition.CLOUDY,
            'Fog': WeatherCondition.CLOUDY
        }
        return mapping.get(condition_str, WeatherCondition.CLEAR)

    def _calculate_severity(self, condition: WeatherCondition,
                           data: Dict) -> WeatherSeverity:
        """Calculate weather severity"""
        max_temp = max(data['temps'])
        min_temp = min(data['temps'])
        precip = data['precipitation']

        if condition in [WeatherCondition.HEAVY_RAIN, WeatherCondition.SNOW]:
            if precip > 50:
                return WeatherSeverity.EXTREME
            elif precip > 25:
                return WeatherSeverity.SEVERE

        if max_temp > 40 or min_temp < -15:
            return WeatherSeverity.SEVERE

        if max_temp > 35 or min_temp < -5:
            return WeatherSeverity.WARNING

        if condition == WeatherCondition.RAIN:
            return WeatherSeverity.CAUTION

        return WeatherSeverity.NORMAL

    def _generate_sample_forecast(self, days: int) -> List[DailyForecast]:
        """Generate sample forecast for testing"""
        import random
        forecasts = []

        for i in range(days):
            day = date.today() + timedelta(days=i)
            temp_base = 15 + random.uniform(-5, 10)
            condition = random.choice(list(WeatherCondition))

            forecasts.append(DailyForecast(
                date=day,
                temp_high=temp_base + random.uniform(3, 8),
                temp_low=temp_base - random.uniform(3, 8),
                sunrise=datetime.combine(day, datetime.min.time().replace(hour=6)),
                sunset=datetime.combine(day, datetime.min.time().replace(hour=18)),
                precipitation_total=random.uniform(0, 20) if condition == WeatherCondition.RAIN else 0,
                precipitation_probability=random.uniform(0, 100) if condition == WeatherCondition.RAIN else 10,
                primary_condition=condition,
                severity=WeatherSeverity.NORMAL
            ))

        return forecasts

Activity Weather Sensitivity

@dataclass
class WeatherThresholds:
    min_temp: float = -10
    max_temp: float = 45
    max_wind: float = 50
    max_precipitation: float = 50
    max_snow_depth: float = 20
    min_visibility: float = 0.5  # km

@dataclass
class ConstructionActivity:
    activity_id: str
    activity_name: str
    category: str
    thresholds: WeatherThresholds
    indoor: bool = False
    rain_sensitive: bool = True
    frost_sensitive: bool = False
    productivity_factors: Dict[WeatherCondition, float] = field(default_factory=dict)

    def __post_init__(self):
        if not self.productivity_factors:
            self.productivity_factors = {
                WeatherCondition.CLEAR: 1.0,
                WeatherCondition.CLOUDY: 0.95,
                WeatherCondition.RAIN: 0.3 if self.rain_sensitive else 0.8,
                WeatherCondition.HEAVY_RAIN: 0.0 if self.rain_sensitive else 0.5,
                WeatherCondition.SNOW: 0.2,
                WeatherCondition.FROST: 0.5 if self.frost_sensitive else 0.8,
                WeatherCondition.HIGH_WIND: 0.3,
                WeatherCondition.EXTREME_HEAT: 0.6,
                WeatherCondition.EXTREME_COLD: 0.4
            }

class ActivityWeatherAnalyzer:
    """Analyze weather impact on construction activities"""

    # Default activity definitions
    ACTIVITY_TEMPLATES = {
        'concrete_placement': ConstructionActivity(
            activity_id='ACT-001',
            activity_name='Concrete Placement',
            category='structural',
            thresholds=WeatherThresholds(min_temp=5, max_temp=35, max_precipitation=2, max_wind=40),
            rain_sensitive=True,
            frost_sensitive=True
        ),
        'steel_erection': ConstructionActivity(
            activity_id='ACT-002',
            activity_name='Steel Erection',
            category='structural',
            thresholds=WeatherThresholds(max_wind=35, max_precipitation=10),
            rain_sensitive=False
        ),
        'roofing': ConstructionActivity(
            activity_id='ACT-003',
            activity_name='Roofing',
            category='envelope',
            thresholds=WeatherThresholds(min_temp=0, max_precipitation=0, max_wind=30),
            rain_sensitive=True
        ),
        'excavation': ConstructionActivity(
            activity_id='ACT-004',
            activity_name='Excavation',
            category='earthwork',
            thresholds=WeatherThresholds(min_temp=-5, max_precipitation=25),
            rain_sensitive=True,
            frost_sensitive=True
        ),
        'painting_exterior': ConstructionActivity(
            activity_id='ACT-005',
            activity_name='Exterior Painting',
            category='finishing',
            thresholds=WeatherThresholds(min_temp=10, max_temp=35, max_precipitation=0, max_wind=25),
            rain_sensitive=True
        ),
        'masonry': ConstructionActivity(
            activity_id='ACT-006',
            activity_name='Masonry Work',
            category='structural',
            thresholds=WeatherThresholds(min_temp=5, max_temp=32, max_precipitation=5),
            rain_sensitive=True,
            frost_sensitive=True
        ),
        'crane_operations': ConstructionActivity(
            activity_id='ACT-007',
            activity_name='Crane Operations',
            category='equipment',
            thresholds=WeatherThresholds(max_wind=30, min_visibility=1.0),
            rain_sensitive=False
        ),
        'electrical_exterior': ConstructionActivity(
            activity_id='ACT-008',
            activity_name='Exterior Electrical',
            category='MEP',
            thresholds=WeatherThresholds(max_precipitation=0),
            rain_sensitive=True
        ),
        'interior_work': ConstructionActivity(
            activity_id='ACT-009',
            activity_name='Interior Work',
            category='finishing',
            thresholds=WeatherThresholds(),
            indoor=True,
            rain_sensitive=False
        )
    }

    def __init__(self):
        self.activities = dict(self.ACTIVITY_TEMPLATES)

    def add_activity(self, activity: ConstructionActivity):
        """Add custom activity"""
        self.activities[activity.activity_id] = activity

    def analyze_day(self, weather: DailyForecast,
                   activities: List[str]) -> Dict[str, Dict]:
        """Analyze weather impact for specific day and activities"""
        results = {}

        for activity_id in activities:
            activity = self.activities.get(activity_id)
            if not activity:
                continue

            impact = self._calculate_impact(weather, activity)
            results[activity_id] = impact

        return results

    def _calculate_impact(self, weather: DailyForecast,
                         activity: ConstructionActivity) -> Dict:
        """Calculate weather impact on activity"""
        if activity.indoor:
            return {
                'can_work': True,
                'productivity': 1.0,
                'issues': [],
                'recommendations': []
            }

        issues = []
        productivity = 1.0

        # Temperature check
        if weather.temp_low < activity.thresholds.min_temp:
            issues.append(f"Low temperature: {weather.temp_low}°C")
            if activity.frost_sensitive:
                productivity *= 0.0
            else:
                productivity *= 0.5

        if weather.temp_high > activity.thresholds.max_temp:
            issues.append(f"High temperature: {weather.temp_high}°C")
            productivity *= 0.6

        # Precipitation check
        if weather.precipitation_total > activity.thresholds.max_precipitation:
            issues.append(f"Precipitation: {weather.precipitation_total}mm")
            if activity.rain_sensitive:
                productivity *= 0.0
            else:
                productivity *= 0.7

        # Condition-based productivity
        condition_factor = activity.productivity_factors.get(
            weather.primary_condition, 1.0
        )
        productivity *= condition_factor

        # Generate recommendations
        recommendations = []
        if productivity < 0.5 and productivity > 0:
            recommendations.append("Consider rescheduling to more favorable day")
        if weather.temp_low < activity.thresholds.min_temp + 5:
            recommendations.append("Plan for cold weather precautions")
        if weather.precipitation_probability > 50:
            recommendations.append("Have rain contingency plan ready")

        can_work = productivity > 0

        return {
            'activity_name': activity.activity_name,
            'can_work': can_work,
            'productivity': round(productivity, 2),
            'issues': issues,
            'recommendations': recommendations,
            'weather_condition': weather.primary_condition.value,
            'temperature_range': f"{weather.temp_low}°C - {weather.temp_high}°C"
        }

    def find_optimal_days(self, forecast: List[DailyForecast],
                         activity_id: str,
                         min_productivity: float = 0.8) -> List[date]:
        """Find optimal days for an activity"""
        activity = self.activities.get(activity_id)
        if not activity:
            return []

        optimal = []
        for day in forecast:
            impact = self._calculate_impact(day, activity)
            if impact['productivity'] >= min_productivity:
                optimal.append(day.date)

        return optimal

Schedule Weather Integration

from datetime import date, timedelta
from typing import List, Dict
import pandas as pd

@dataclass
class ScheduledActivity:
    activity_id: str
    activity_name: str
    activity_type: str  # Maps to ACTIVITY_TEMPLATES
    planned_start: date
    planned_end: date
    duration_days: int
    is_critical: bool = False

class ScheduleWeatherOptimizer:
    """Optimize construction schedule based on weather"""

    def __init__(self, weather_service: WeatherDataService,
                 activity_analyzer: ActivityWeatherAnalyzer):
        self.weather = weather_service
        self.analyzer = activity_analyzer

    def analyze_schedule(self, schedule: List[ScheduledActivity],
                        location: Tuple[float, float]) -> Dict:
        """Analyze schedule against weather forecast"""
        forecast = self.weather.get_forecast(location[0], location[1])
        forecast_dict = {f.date: f for f in forecast}

        analysis = {
            'activities': [],
            'weather_delays': 0,
            'risk_days': [],
            'recommendations': []
        }

        for activity in schedule:
            activity_analysis = self._analyze_activity(
                activity, forecast_dict
            )
            analysis['activities'].append(activity_analysis)

            if activity_analysis['expected_delay'] > 0:
                analysis['weather_delays'] += activity_analysis['expected_delay']

            analysis['risk_days'].extend(activity_analysis['risk_days'])

        # Generate overall recommendations
        if analysis['weather_delays'] > 5:
            analysis['recommendations'].append(
                f"Schedule shows {analysis['weather_delays']} potential weather delay days. "
                "Consider buffer time or alternative scheduling."
            )

        return analysis

    def _analyze_activity(self, activity: ScheduledActivity,
                         forecast: Dict[date, DailyForecast]) -> Dict:
        """Analyze single activity against weather"""
        result = {
            'activity_id': activity.activity_id,
            'activity_name': activity.activity_name,
            'planned_start': activity.planned_start,
            'planned_end': activity.planned_end,
            'day_analysis': [],
            'risk_days': [],
            'expected_delay': 0,
            'avg_productivity': 1.0
        }

        current_date = activity.planned_start
        productivities = []
        delay_days = 0

        while current_date <= activity.planned_end:
            if current_date in forecast:
                weather = forecast[current_date]
                impact = self.analyzer._calculate_impact(
                    weather,
                    self.analyzer.activities.get(activity.activity_type,
                        self.analyzer.ACTIVITY_TEMPLATES.get('interior_work'))
                )

                productivities.append(impact['productivity'])

                day_info = {
                    'date': current_date,
                    'can_work': impact['can_work'],
                    'productivity': impact['productivity'],
                    'weather': weather.primary_condition.value
                }
                result['day_analysis'].append(day_info)

                if not impact['can_work']:
                    delay_days += 1
                    result['risk_days'].append({
                        'date': current_date,
                        'activity': activity.activity_name,
                        'reason': weather.primary_condition.value
                    })
                elif impact['productivity'] < 0.7:
                    delay_days += (1 - impact['productivity'])

            current_date += timedelta(days=1)

        result['expected_delay'] = round(delay_days)
        result['avg_productivity'] = sum(productivities) / len(productivities) if productivities else 1.0

        return result

    def suggest_reschedule(self, activity: ScheduledActivity,
                          location: Tuple[float, float],
                          flexibility_days: int = 7) -> Optional[date]:
        """Suggest better start date for activity"""
        forecast = self.weather.get_forecast(location[0], location[1])

        best_start = None
        best_avg_productivity = 0

        for offset in range(-flexibility_days, flexibility_days + 1):
            test_start = activity.planned_start + timedelta(days=offset)
            test_end = test_start + timedelta(days=activity.duration_days - 1)

            productivities = []
            for f in forecast:
                if test_start <= f.date <= test_end:
                    act_template = self.analyzer.activities.get(activity.activity_type)
                    if act_template:
                        impact = self.analyzer._calculate_impact(f, act_template)
                        productivities.append(impact['productivity'])

            if productivities:
                avg = sum(productivities) / len(productivities)
                if avg > best_avg_productivity:
                    best_avg_productivity = avg
                    best_start = test_start

        if best_start and best_start != activity.planned_start:
            return best_start
        return None

    def generate_weather_report(self, schedule: List[ScheduledActivity],
                               location: Tuple[float, float],
                               output_path: str) -> str:
        """Generate weather impact report"""
        analysis = self.analyze_schedule(schedule, location)

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Summary
            summary = pd.DataFrame([{
                'Total Activities': len(schedule),
                'Weather Delay Days': analysis['weather_delays'],
                'High Risk Days': len(analysis['risk_days']),
                'Recommendations': len(analysis['recommendations'])
            }])
            summary.to_excel(writer, sheet_name='Summary', index=False)

            # Activity details
            activity_data = []
            for act in analysis['activities']:
                activity_data.append({
                    'Activity': act['activity_name'],
                    'Start': act['planned_start'],
                    'End': act['planned_end'],
                    'Avg Productivity': f"{act['avg_productivity']:.0%}",
                    'Expected Delay': f"{act['expected_delay']} days"
                })
            pd.DataFrame(activity_data).to_excel(writer, sheet_name='Activities', index=False)

            # Risk days
            if analysis['risk_days']:
                pd.DataFrame(analysis['risk_days']).to_excel(
                    writer, sheet_name='Risk_Days', index=False
                )

        return output_path

Quick Reference

Resources

• OpenWeatherMap API: https://openweathermap.org/api
• Weather Underground: https://www.wunderground.com/weather/api
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See 4d-simulation for schedule visualization
• See risk-assessment-ml for weather risk prediction
• See site-logistics-optimization for delivery scheduling