概述

天气影响 50% 的建筑活动。此技能可获取天气数据，用于调度安排、风险评估和生产率调整。

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)

常见用例

1. 施工进度规划

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}")

2. 历史数据分析

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}")

资源

• DDC 书籍：第 2.2 章 - 开放数据集成
• Open-Meteo API：https://open-meteo.com/