weather-impact-analysis

What this skill does

# 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

```python
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

```python
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: