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: