Weather Impact Scheduler
Technical Implementation
import pandas as pd
from datetime import date, timedelta
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class WeatherCondition(Enum):
CLEAR = "clear"
CLOUDY = "cloudy"
RAIN = "rain"
HEAVY_RAIN = "heavy_rain"
SNOW = "snow"
WIND = "wind"
EXTREME_HEAT = "extreme_heat"
EXTREME_COLD = "extreme_cold"
class ActivitySensitivity(Enum):
HIGH = "high" # Concrete, painting, roofing
MEDIUM = "medium" # Excavation, masonry
LOW = "low" # Indoor work
@dataclass
class WeatherForecast:
forecast_date: date
condition: WeatherCondition
high_temp: float
low_temp: float
precipitation_mm: float
wind_speed_kmh: float
@dataclass
class ScheduleActivity:
activity_id: str
name: str
start_date: date
end_date: date
sensitivity: ActivitySensitivity
outdoor: bool
can_work_in_rain: bool = False
min_temp: float = 5.0
max_temp: float = 35.0
max_wind: float = 50.0
@dataclass
class WeatherImpact:
activity_id: str
impact_date: date
reason: str
delay_hours: float
recommendation: str
class WeatherImpactScheduler:
def __init__(self, project_name: str):
self.project_name = project_name
self.activities: Dict[str, ScheduleActivity] = {}
self.forecasts: Dict[date, WeatherForecast] = {}
self.impacts: List[WeatherImpact] = []
def add_activity(self, activity_id: str, name: str, start_date: date,
end_date: date, sensitivity: ActivitySensitivity,
outdoor: bool = True, can_work_in_rain: bool = False) -> ScheduleActivity:
activity = ScheduleActivity(
activity_id=activity_id,
name=name,
start_date=start_date,
end_date=end_date,
sensitivity=sensitivity,
outdoor=outdoor,
can_work_in_rain=can_work_in_rain
)
self.activities[activity_id] = activity
return activity
def add_forecast(self, forecast_date: date, condition: WeatherCondition,
high_temp: float, low_temp: float,
precipitation_mm: float = 0, wind_speed_kmh: float = 0):
forecast = WeatherForecast(
forecast_date=forecast_date,
condition=condition,
high_temp=high_temp,
low_temp=low_temp,
precipitation_mm=precipitation_mm,
wind_speed_kmh=wind_speed_kmh
)
self.forecasts[forecast_date] = forecast
def analyze_impacts(self) -> List[WeatherImpact]:
self.impacts = []
for activity in self.activities.values():
if not activity.outdoor:
continue
current = activity.start_date
while current <= activity.end_date:
forecast = self.forecasts.get(current)
if forecast:
impact = self._check_impact(activity, forecast)
if impact:
self.impacts.append(impact)
current += timedelta(days=1)
return self.impacts
def _check_impact(self, activity: ScheduleActivity,
forecast: WeatherForecast) -> Optional[WeatherImpact]:
reasons = []
delay_hours = 0
# Check precipitation
if forecast.condition in [WeatherCondition.RAIN, WeatherCondition.HEAVY_RAIN]:
if not activity.can_work_in_rain:
if activity.sensitivity == ActivitySensitivity.HIGH:
reasons.append("Rain - high sensitivity activity")
delay_hours = 8
else:
reasons.append("Rain delays")
delay_hours = 4
# Check temperature
if forecast.low_temp < activity.min_temp:
reasons.append(f"Too cold ({forecast.low_temp}°C)")
delay_hours = max(delay_hours, 8 if activity.sensitivity == ActivitySensitivity.HIGH else 4)
if forecast.high_temp > activity.max_temp:
reasons.append(f"Too hot ({forecast.high_temp}°C)")
delay_hours = max(delay_hours, 4)
# Check wind
if forecast.wind_speed_kmh > activity.max_wind:
reasons.append(f"High wind ({forecast.wind_speed_kmh} km/h)")
delay_hours = max(delay_hours, 8)
if reasons:
return WeatherImpact(
activity_id=activity.activity_id,
impact_date=forecast.forecast_date,
reason="; ".join(reasons),
delay_hours=delay_hours,
recommendation=self._get_recommendation(activity, forecast)
)
return None
def _get_recommendation(self, activity: ScheduleActivity,
forecast: WeatherForecast) -> str:
if forecast.condition in [WeatherCondition.RAIN, WeatherCondition.HEAVY_RAIN]:
return "Reschedule or plan indoor work"
if forecast.low_temp < activity.min_temp:
return "Use heating blankets or delay start"
if forecast.high_temp > activity.max_temp:
return "Start early, plan heat breaks"
if forecast.wind_speed_kmh > activity.max_wind:
return "Secure materials, delay crane work"
return "Monitor conditions"
def get_total_delay_forecast(self) -> Dict[str, Any]:
total_hours = sum(i.delay_hours for i in self.impacts)
by_activity = {}
for impact in self.impacts:
act = impact.activity_id
by_activity[act] = by_activity.get(act, 0) + impact.delay_hours
return {
'total_impact_hours': total_hours,
'total_impact_days': round(total_hours / 8, 1),
'affected_activities': len(by_activity),
'by_activity': by_activity,
'impact_count': len(self.impacts)
}
def export_analysis(self, output_path: str):
data = [{
'Activity': i.activity_id,
'Date': i.impact_date,
'Reason': i.reason,
'Delay Hours': i.delay_hours,
'Recommendation': i.recommendation
} for i in self.impacts]
pd.DataFrame(data).to_excel(output_path, index=False)
Quick Start
scheduler = WeatherImpactScheduler("Office Tower")
# Add activities
scheduler.add_activity("CONC-001", "Pour Slab L3", date(2024, 3, 15),
date(2024, 3, 20), ActivitySensitivity.HIGH, outdoor=True)
# Add forecasts
scheduler.add_forecast(date(2024, 3, 17), WeatherCondition.RAIN, 15, 8, 25, 20)
# Analyze
impacts = scheduler.analyze_impacts()
summary = scheduler.get_total_delay_forecast()
print(f"Projected delay: {summary['total_impact_days']} days")
Resources
- DDC Book: Chapter 3.3 - Schedule Management
kbarbel640-del