Fetch weather data for construction scheduling. Historical data, forecasts, and risk assessment for outdoor work.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
설치 방법 보기설치하기 전에 에이전트에게 전달되는 지시 내용을 확인할 수 있습니다.
Weather impacts 50% of construction activities. This skill fetches weather data for scheduling, risk assessment, and productivity adjustments.
import requests
import pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
class WeatherRisk(Enum):
"""Weather risk levels for construction."""
LOW = "low"
MODERATE = "moderate"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class WeatherCondition:
"""Weather condition at a point in time."""
timestamp: datetime
temperature: float # Celsius
humidity: float # Percent
wind_speed: float # m/s
precipitation: float # mm
conditions: str
@dataclass
class WorkabilityAssessment:
"""Assessment of weather workability."""
date: datetime
risk_level: WeatherRisk
workable_hours: int
affected_activities: List[str]
recommendations: List[str]
class WeatherAPIClient:
"""Client for weather APIs."""
# Free tier endpoints
OPEN_METEO_BASE = "https://api.open-meteo.com/v1"
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
def get_forecast(self, latitude: float, longitude: float,
days: int = 7) -> List[WeatherCondition]:
"""Get weather forecast."""
url = f"{self.OPEN_METEO_BASE}/forecast"
params = {
'latitude': latitude,
'longitude': longitude,
'hourly': 'temperature_2m,relative_humidity_2m,wind_speed_10m,precipitation',
'forecast_days': days
}
response = requests.get(url, params=params)
if response.status_code != 200:
raise Exception(f"API error: {response.status_code}")
data = response.json()
return self._parse_forecast(data)
def get_historical(self, latitude: float, longitude: float,
start_date: str, end_date: str) -> List[WeatherCondition]:
"""Get historical weather data."""
url = f"{self.OPEN_METEO_BASE}/archive"
params = {
'latitude': latitude,
'longitude': longitude,
'start_date': start_date,
'end_date': end_date,
'hourly': 'temperature_2m,relative_humidity_2m,wind_speed_10m,precipitation'
}
response = requests.get(url, params=params)
if response.status_code != 200:
raise Exception(f"API error: {response.status_code}")
data = response.json()
return self._parse_forecast(data)
def _parse_forecast(self, data: Dict) -> List[WeatherCondition]:
"""Parse API response to WeatherCondition list."""
conditions = []
hourly = data.get('hourly', {})
times = hourly.get('time', [])
temps = hourly.get('temperature_2m', [])
humidity = hourly.get('relative_humidity_2m', [])
wind = hourly.get('wind_speed_10m', [])
precip = hourly.get('precipitation', [])
for i in range(len(times)):
conditions.append(WeatherCondition(
timestamp=datetime.fromisoformat(times[i]),
temperature=temps[i] if i < len(temps) else 0,
humidity=humidity[i] if i < len(humidity) else 0,
wind_speed=wind[i] if i < len(wind) else 0,
precipitation=precip[i] if i < len(precip) else 0,
conditions=self._describe_conditions(
temps[i] if i < len(temps) else 0,
precip[i] if i < len(precip) else 0,
wind[i] if i < len(wind) else 0
)
))
return conditions
def _describe_conditions(self, temp: float, precip: float, wind: float) -> str:
"""Generate weather description."""
conditions = []
if temp < 0:
conditions.append("Freezing")
elif temp > 35:
conditions.append("Extreme heat")
elif temp > 30:
conditions.append("Hot")
elif temp < 10:
conditions.append("Cold")
if precip > 10:
conditions.append("Heavy rain")
elif precip > 2:
conditions.append("Rain")
elif precip > 0:
conditions.append("Light rain")
if wind > 15:
conditions.append("Strong winds")
elif wind > 10:
conditions.append("Windy")
return ", ".join(conditions) if conditions else "Clear"
def to_dataframe(self, conditions: List[WeatherCondition]) -> pd.DataFrame:
"""Convert conditions to DataFrame."""
data = [{
'timestamp': c.timestamp,
'temperature': c.temperature,
'humidity': c.humidity,
'wind_speed': c.wind_speed,
'precipitation': c.precipitation,
'conditions': c.conditions
} for c in conditions]
return pd.DataFrame(data)
class ConstructionWeatherRisk:
"""Assess weather risk for construction activities."""
# Activity-specific thresholds
THRESHOLDS = {
'concrete_pour': {
'min_temp': 5, 'max_temp': 35,
'max_wind': 12, 'max_precip': 0.5
},
'crane_work': {
'min_temp': -10, 'max_temp': 40,
'max_wind': 10, 'max_precip': 5
},
'exterior_paint': {
'min_temp': 10, 'max_temp': 35,
'max_wind': 8, 'max_precip': 0
},
'roofing': {
'min_temp': 5, 'max_temp': 38,
'max_wind': 12, 'max_precip': 0
},
'earthwork': {
'min_temp': -5, 'max_temp': 40,
'max_wind': 20, 'max_precip': 10
}
}
def assess_workability(self, condition: WeatherCondition,
activities: List[str] = None) -> WorkabilityAssessment:
"""Assess workability for given conditions."""
if activities is None:
activities = list(self.THRESHOLDS.keys())
affected = []
recommendations = []
for activity in activities:
if activity in self.THRESHOLDS:
thresh = self.THRESHOLDS[activity]
reasons = []
if condition.temperature < thresh['min_temp']:
reasons.append(f"Too cold ({condition.temperature}°C)")
if condition.temperature > thresh['max_temp']:
reasons.append(f"Too hot ({condition.temperature}°C)")
if condition.wind_speed > thresh['max_wind']:
reasons.append(f"High wind ({condition.wind_speed} m/s)")
if condition.precipitation > thresh['max_precip']:
reasons.append(f"Precipitation ({condition.precipitation} mm)")
if reasons:
affected.append(activity)
recommendations.append(f"{activity}: " + ", ".join(reasons))
# Determine overall risk level
if len(affected) >= len(activities) * 0.8:
risk = WeatherRisk.CRITICAL
workable = 0
elif len(affected) >= len(activities) * 0.5:
risk = WeatherRisk.HIGH
workable = 4
elif len(affected) > 0:
risk = WeatherRisk.MODERATE
workable = 6
else:
risk = WeatherRisk.LOW
workable = 8
return WorkabilityAssessment(
date=condition.timestamp,
risk_level=risk,
workable_hours=workable,
affected_activities=affected,
recommendations=recommendations
)
def weekly_forecast_risk(self, conditions: List[WeatherCondition],
activities: List[str] = None) -> pd.DataFrame:
"""Assess risk for week of weather data."""
# Group by date
daily_conditions = {}
for c in conditions:
date = c.timestamp.date()
if date not in daily_conditions:
daily_conditions[date] = []
daily_conditions[date].append(c)
assessments = []
for date, day_conditions in daily_conditions.items():
# Use midday condition as representative
midday = [c for c in day_conditions
if 10 <= c.timestamp.hour <= 16]
representative = midday[len(midday)//2] if midday else day_conditions[0]
assessment = self.assess_workability(representative, activities)
assessments.append({
'date': date,
'risk_level': assessment.risk_level.value,
'workable_hours': assessment.workable_hours,
'affected_count': len(assessment.affected_activities)
})
return pd.DataFrame(assessments)
# Initialize client
weather = WeatherAPIClient()
# Get forecast for site
conditions = weather.get_forecast(latitude=52.52, longitude=13.41, days=7)
df = weather.to_dataframe(conditions)
print(df.head())
# Assess construction risk
risk = ConstructionWeatherRisk()
weekly_risk = risk.weekly_forecast_risk(conditions)
print(weekly_risk)
conditions = weather.get_forecast(52.52, 13.41, days=14)
risk = ConstructionWeatherRisk()
# Check concrete pour window
for c in conditions:
assessment = risk.assess_workability(c, ['concrete_pour'])
if assessment.risk_level == WeatherRisk.LOW:
print(f"Good for concrete: {c.timestamp}")
historical = weather.get_historical(52.52, 13.41, '2024-01-01', '2024-03-31')
df = weather.to_dataframe(historical)
# Count rain days
rain_days = df[df['precipitation'] > 2]['timestamp'].dt.date.nunique()
print(f"Rain days in Q1: {rain_days}")
아직 리뷰가 없습니다. 사용해 본 소감을 남겨 주세요.
로그인 없이 쓸 수 있습니다. 사용 후기, 질문, 작동하지 않은 환경 등을 편하게 남겨 주세요.
아직 댓글이 없습니다. 첫 댓글을 남겨 보세요.
Create 4D construction simulations by linking BIM elements to project schedules. Generate time-based visualizations, sequence analysis, and construction phasing with Gantt integration.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Analyze 5000+ IFC and Revit projects at scale for patterns, benchmarks, and insights. Big data analysis for construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Check construction AI systems against the EU AI Act (Regulation 2024/1689): classify risk of estimation, scheduling, CV and agent tools, document transparency, keep human oversight. Use when deploying or auditing AI in construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Orchestrate multiple AI agents for construction workflows: estimator, scheduler, document, QA and safety agents coordinated by a supervisor agent, with human checkpoints. Use when automating end-to-end project processes with agentic AI.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Create Apache Airflow DAGs for construction data pipelines. Orchestrate ETL, validation, and reporting workflows.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Automate as-built documentation and digital handover for construction. Compile project records, generate O&M manuals, create asset databases, and ensure complete project closeout.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.