Weather Api

Fetch weather data for construction scheduling. Historical data, forecasts, and risk assessment for outdoor work.

Security Scan

The skill's stated purpose (fetch weather and assess construction risk) matches its instructions, but there are notable inconsistencies and missing dependency declarations that you should review before installing.

Purpose & Capability

Name and description align with the instructions: the SKILL.md implements weather fetches (Open-Meteo) and construction risk logic. Requesting python3 and granting filesystem+network access is reasonable for fetching data, processing it, and exporting results.

Instruction Scope

SKILL.md contains Python code that calls a public weather API and processes user-provided data; instructions.md explicitly says to only use data supplied by the user. The skill asks for file paths and export capabilities — acceptable for the stated purpose — but the instructions allow reading and writing files and making network calls, which creates an exfiltration vector if sensitive files are provided.

Install Mechanism

This is an instruction-only skill (no install spec), but the embedded Python code depends on third-party libraries (requests, pandas) that are not declared in the metadata. The skill only declares python3 as a required binary; it provides no guidance for installing required Python packages. That mismatch (code requiring packages that may not exist) is an incoherence and could cause the agent to attempt ad-hoc package installation or fail unpredictably.

Credentials

No environment variables or credentials are requested, and the code uses an unauthenticated public API (Open-Meteo). That is proportional. However, claw.json grants 'filesystem' and 'network' permissions — appropriate for the skill's functionality but worth noting because those permissions permit reading user files and making external requests.

Persistence & Privilege

The skill is not forced-always (always: false) and is user-invocable. It does not request persistent or cross-skill configuration changes. The main privilege note is the declared filesystem and network permissions which are reasonable for its role but should be accepted consciously.

What to consider before installing

This skill appears to implement the advertised weather + construction-risk features, but before installing: (1) confirm you are comfortable granting filesystem and network access (these let the skill read user files and make external requests); (2) confirm the runtime Python environment has requests and pandas (or plan to install them) because SKILL.md uses those libraries but the package dependencies are not declared; (3) avoid supplying sensitive files (credentials, private configs) as input — exports and network calls could leak data if misused; (4) verify the homepage/source (https://datadrivenconstruction.io) and prefer running the skill in a sandboxed environment until you’ve reviewed its full behavior. If you need higher assurance, request the author add an install spec (declaring pip dependencies) and make the code files explicit rather than only in documentation.

Weather API for Construction

Overview

Weather impacts 50% of construction activities. This skill fetches weather data for scheduling, risk assessment, and productivity adjustments.

Python Implementation

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)

Quick Start

# 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)

Common Use Cases

1. Schedule Planning

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 Analysis

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

Resources

• DDC Book: Chapter 2.2 - Open Data Integration
• Open-Meteo API: https://open-meteo.com/