weather-impact-analysis

name	weather-impact-analysis
description	Analyze weather data impact on construction schedules. Predict weather delays, optimize work scheduling based on forecasts, and calculate weather-related risk factors for project planning.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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:
    

     ():
        .api_key = api_key
        .provider = provider
        .cache: [, ] = {}
        .cache_duration = timedelta(hours=)

     () -> [DailyForecast]:
        
        cache_key = 

         cache_key  .cache:
            cached = .cache[cache_key]
             datetime.now() - cached[] < .cache_duration:
                 cached[]

         .provider == :
            forecast = ._fetch_openweathermap(latitude, longitude, days)
        :
            forecast = ._generate_sample_forecast(days)

        .cache[cache_key] = {
            : datetime.now(),
            : forecast
        }

         forecast

     () -> [DailyForecast]:
        
        url = 
        params = {
            : lat,
            : lon,
            : .api_key,
            : 
        }

        :
            response = requests.get(url, params=params)
            data = response.json()
             ._parse_openweathermap(data)
         Exception  e:
            ()
             ._generate_sample_forecast(days)

     () -> [DailyForecast]:
        
        forecasts = []
        daily_data = {}

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

             day   daily_data:
                daily_data[day] = {
                    : [],
                    : ,
                    : [],
                    : []
                }

            daily_data[day][].append(item[][])
            daily_data[day][] += item.get(, {}).get(, )

            condition = ._map_condition(item[][][])
            daily_data[day][].append(condition)

            daily_data[day][].append(HourlyWeather(
                datetime=dt,
                temperature=item[][],
                feels_like=item[][],
                humidity=item[][],
                wind_speed=item[][] * ,  
                wind_direction=item[].get(, ),
                precipitation=item.get(, {}).get(, ),
                precipitation_probability=item.get(, ) * ,
                condition=condition,
                visibility=item.get(, ) / ,
                uv_index=
            ))

         day, data  daily_data.items():
            primary_condition = ((data[]), key=data[].count)

            forecasts.append(DailyForecast(
                date=day,
                temp_high=(data[]),
                temp_low=(data[]),
                sunrise=datetime.combine(day, datetime..time().replace(hour=)),
                sunset=datetime.combine(day, datetime..time().replace(hour=)),
                precipitation_total=data[],
                precipitation_probability=(h.precipitation_probability  h  data[]),
                primary_condition=primary_condition,
                hourly=data[],
                severity=._calculate_severity(primary_condition, data)
            ))

         (forecasts, key= x: x.date)

     () -> WeatherCondition:
        
        mapping = {
            : WeatherCondition.CLEAR,
            : WeatherCondition.CLOUDY,
            : WeatherCondition.RAIN,
            : WeatherCondition.RAIN,
            : WeatherCondition.HEAVY_RAIN,
            : WeatherCondition.SNOW,
            : WeatherCondition.CLOUDY,
            : WeatherCondition.CLOUDY
        }
         mapping.get(condition_str, WeatherCondition.CLEAR)

     () -> WeatherSeverity:
        
        max_temp = (data[])
        min_temp = (data[])
        precip = data[]

         condition  [WeatherCondition.HEAVY_RAIN, WeatherCondition.SNOW]:
             precip > :
                 WeatherSeverity.EXTREME
             precip > :
                 WeatherSeverity.SEVERE

         max_temp >   min_temp < -:
             WeatherSeverity.SEVERE

         max_temp >   min_temp < -:
             WeatherSeverity.WARNING

         condition == WeatherCondition.RAIN:
             WeatherSeverity.CAUTION

         WeatherSeverity.NORMAL

     () -> [DailyForecast]:
        
         random
        forecasts = []

         i  (days):
            day = date.today() + timedelta(days=i)
            temp_base =  + random.uniform(-, )
            condition = random.choice((WeatherCondition))

            forecasts.append(DailyForecast(
                date=day,
                temp_high=temp_base + random.uniform(, ),
                temp_low=temp_base - random.uniform(, ),
                sunrise=datetime.combine(day, datetime..time().replace(hour=)),
                sunset=datetime.combine(day, datetime..time().replace(hour=)),
                precipitation_total=random.uniform(, )  condition == WeatherCondition.RAIN  ,
                precipitation_probability=random.uniform(, )  condition == WeatherCondition.RAIN  ,
                primary_condition=condition,
                severity=WeatherSeverity.NORMAL
            ))

         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 .rain_sensitive  ,
                WeatherCondition.SNOW: ,
                WeatherCondition.FROST:   .frost_sensitive  ,
                WeatherCondition.HIGH_WIND: ,
                WeatherCondition.EXTREME_HEAT: ,
                WeatherCondition.EXTREME_COLD: 
            }

 :
    

    
    ACTIVITY_TEMPLATES = {
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(min_temp=, max_temp=, max_precipitation=, max_wind=),
            rain_sensitive=,
            frost_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(max_wind=, max_precipitation=),
            rain_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(min_temp=, max_precipitation=, max_wind=),
            rain_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(min_temp=-, max_precipitation=),
            rain_sensitive=,
            frost_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(min_temp=, max_temp=, max_precipitation=, max_wind=),
            rain_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(min_temp=, max_temp=, max_precipitation=),
            rain_sensitive=,
            frost_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(max_wind=, min_visibility=),
            rain_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(max_precipitation=),
            rain_sensitive=
        ),
        : ConstructionActivity(
            activity_id=,
            activity_name=,
            category=,
            thresholds=WeatherThresholds(),
            indoor=,
            rain_sensitive=
        )
    }

     ():
        .activities = (.ACTIVITY_TEMPLATES)

     ():
        
        .activities[activity.activity_id] = activity

     () -> [, ]:
        
        results = {}

         activity_id  activities:
            activity = .activities.get(activity_id)
              activity:
                

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

         results

     () -> :
        
         activity.indoor:
             {
                : ,
                : ,
                : [],
                : []
            }

        issues = []
        productivity = 

        
         weather.temp_low < activity.thresholds.min_temp:
            issues.append()
             activity.frost_sensitive:
                productivity *= 
            :
                productivity *= 

         weather.temp_high > activity.thresholds.max_temp:
            issues.append()
            productivity *= 

        
         weather.precipitation_total > activity.thresholds.max_precipitation:
            issues.append()
             activity.rain_sensitive:
                productivity *= 
            :
                productivity *= 

        
        condition_factor = activity.productivity_factors.get(
            weather.primary_condition, 
        )
        productivity *= condition_factor

        
        recommendations = []
         productivity <   productivity > :
            recommendations.append()
         weather.temp_low < activity.thresholds.min_temp + :
            recommendations.append()
         weather.precipitation_probability > :
            recommendations.append()

        can_work = productivity > 

         {
            : activity.activity_name,
            : can_work,
            : (productivity, ),
            : issues,
            : recommendations,
            : weather.primary_condition.value,
            : 
        }

     () -> [date]:
        
        activity = .activities.get(activity_id)
          activity:
             []

        optimal = []
         day  forecast:
            impact = ._calculate_impact(day, activity)
             impact[] >= min_productivity:
                optimal.append(day.date)

         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': [],
            : []
        }

         activity  schedule:
            activity_analysis = ._analyze_activity(
                activity, forecast_dict
            )
            analysis[].append(activity_analysis)

             activity_analysis[] > :
                analysis[] += activity_analysis[]

            analysis[].extend(activity_analysis[])

        
         analysis[] > :
            analysis[].append(
                
                
            )

         analysis

     () -> :
        
        result = {
            : activity.activity_id,
            : activity.activity_name,
            : activity.planned_start,
            : activity.planned_end,
            : [],
            : [],
            : ,
            : 
        }

        current_date = activity.planned_start
        productivities = []
        delay_days = 

         current_date <= activity.planned_end:
             current_date  forecast:
                weather = forecast[current_date]
                impact = .analyzer._calculate_impact(
                    weather,
                    .analyzer.activities.get(activity.activity_type,
                        .analyzer.ACTIVITY_TEMPLATES.get())
                )

                productivities.append(impact[])

                day_info = {
                    : current_date,
                    : impact[],
                    : impact[],
                    : weather.primary_condition.value
                }
                result[].append(day_info)

                  impact[]:
                    delay_days += 
                    result[].append({
                        : current_date,
                        : activity.activity_name,
                        : weather.primary_condition.value
                    })
                 impact[] < :
                    delay_days += ( - impact[])

            current_date += timedelta(days=)

        result[] = (delay_days)
        result[] = (productivities) / (productivities)  productivities  

         result

     () -> [date]:
        
        forecast = .weather.get_forecast(location[], location[])

        best_start = 
        best_avg_productivity = 

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

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

             productivities:
                avg = (productivities) / (productivities)
                 avg > best_avg_productivity:
                    best_avg_productivity = avg
                    best_start = test_start

         best_start  best_start != activity.planned_start:
             best_start
         

     () -> :
        
        analysis = .analyze_schedule(schedule, location)

         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary = pd.DataFrame([{
                : (schedule),
                : analysis[],
                : (analysis[]),
                : (analysis[])
            }])
            summary.to_excel(writer, sheet_name=, index=)

            
            activity_data = []
             act  analysis[]:
                activity_data.append({
                    : act[],
                    : act[],
                    : act[],
                    : ,
                    : 
                })
            pd.DataFrame(activity_data).to_excel(writer, sheet_name=, index=)

            
             analysis[]:
                pd.DataFrame(analysis[]).to_excel(
                    writer, sheet_name=, index=
                )

         output_path

Quick Reference

Activity Type	Min Temp	Max Precip	Max Wind	Rain Sensitive
Concrete	5°C	2mm	40 km/h	Yes
Steel Erection	-10°C	10mm	35 km/h	No
Roofing	0°C	0mm	30 km/h	Yes
Excavation	-5°C	25mm	50 km/h	Partial
Exterior Painting	10°C	0mm	25 km/h	Yes
Masonry	5°C	5mm	40 km/h	Yes
Crane Operations	-15°C	20mm	30 km/h	No

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