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: