from __future__ import annotations from collections import defaultdict from datetime import date, timedelta from statistics import mean from typing import Any from .evidence import make_evidence from .http_client import get_json from .models import EvidenceItem FORECAST_URL = "https://api.open-meteo.com/v1/forecast" ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive" def fetch_climate_bundle(lat: float, lon: float) -> tuple[dict[str, Any], list[EvidenceItem]]: evidence: list[EvidenceItem] = [] bundle: dict[str, Any] = {"forecast": None, "recent_historical": None, "climate_normal": None} try: forecast = get_json( FORECAST_URL, { "latitude": lat, "longitude": lon, "current": [ "temperature_2m", "relative_humidity_2m", "precipitation", "wind_speed_10m", "wind_direction_10m", ], "daily": [ "temperature_2m_max", "temperature_2m_min", "precipitation_sum", "wind_speed_10m_max", "wind_direction_10m_dominant", ], "forecast_days": 7, "timezone": "auto", }, ) bundle["forecast"] = _summarize_forecast(forecast) evidence.append( make_evidence( category="Climate", finding="Forecast/current weather data was retrieved for immediate site-visit context.", source_name="Open-Meteo Forecast API", source_url="https://open-meteo.com/en/docs", source_type="public API", resolution_or_scope="anchor coordinate; 7-day forecast/current model output", confidence="medium", limitation="Forecast/modelled current conditions, not long-term climate or on-site measurement.", design_implication="Use for immediate site-visit timing and short-term comfort/drainage awareness.", verification_needed="Check weather locally before visiting.", output_label="public_data", ) ) except Exception as exc: # noqa: BLE001 - surfaced as evidence for user evidence.append(_failure_evidence("Forecast/current climate", exc)) try: recent = _fetch_archive_summary(lat, lon, years=1) bundle["recent_historical"] = recent evidence.append( make_evidence( category="Climate", finding="Recent historical weather summary was computed from archive data.", source_name="Open-Meteo Historical Weather API", source_url="https://open-meteo.com/en/docs/historical-weather-api", source_type="public API", resolution_or_scope="anchor coordinate; previous 12 months", confidence="medium", limitation="Reanalysis/modelled historical data, not plot-level measurement.", design_implication="Use for recent rainfall, heat, humidity, and wind tendencies.", verification_needed="Ask locals and verify waterlogging/heat exposure on site.", output_label="public_data", ) ) except Exception as exc: # noqa: BLE001 evidence.append(_failure_evidence("Recent historical climate", exc)) try: normal = _fetch_archive_summary(lat, lon, years=10, full_calendar_years=True) bundle["climate_normal"] = normal evidence.append( make_evidence( category="Climate", finding="Climate-normal style monthly summary was computed from multi-year historical archive data.", source_name="Open-Meteo Historical Weather API", source_url="https://open-meteo.com/en/docs/historical-weather-api", source_type="public API", resolution_or_scope="anchor coordinate; approximate last 10 complete calendar years", confidence="medium", limitation="This is a lightweight design-context summary, not an official climatological normal.", design_implication="Use for early design assumptions around heat, rainfall, humidity, and ventilation.", verification_needed="Confirm with studio-required climate sources if your submission needs official normals.", output_label="public_data", ) ) except Exception as exc: # noqa: BLE001 evidence.append(_failure_evidence("Climate-normal style summary", exc)) return bundle, evidence def _fetch_archive_summary(lat: float, lon: float, years: int, full_calendar_years: bool = False) -> dict[str, Any]: if full_calendar_years: end = date(date.today().year - 1, 12, 31) start = date(end.year - years + 1, 1, 1) else: end = date.today() - timedelta(days=5) start = end - timedelta(days=365 * years) data = get_json( ARCHIVE_URL, { "latitude": lat, "longitude": lon, "start_date": start.isoformat(), "end_date": end.isoformat(), "daily": [ "temperature_2m_mean", "precipitation_sum", "relative_humidity_2m_mean", "wind_speed_10m_mean", "wind_direction_10m_dominant", ], "timezone": "auto", }, timeout=35, ) summary = _summarize_archive(data) summary["period"] = f"{start.isoformat()} to {end.isoformat()}" return summary def _summarize_forecast(data: dict[str, Any]) -> dict[str, Any]: current = data.get("current") or {} daily = data.get("daily") or {} return { "current_temperature_c": current.get("temperature_2m"), "current_humidity_pct": current.get("relative_humidity_2m"), "current_precipitation_mm": current.get("precipitation"), "current_wind_speed_kmh": current.get("wind_speed_10m"), "current_wind_direction_deg": current.get("wind_direction_10m"), "forecast_avg_max_c": _avg(daily.get("temperature_2m_max")), "forecast_avg_min_c": _avg(daily.get("temperature_2m_min")), "forecast_total_precip_mm": _sum(daily.get("precipitation_sum")), "forecast_avg_wind_kmh": _avg(daily.get("wind_speed_10m_max")), } def _summarize_archive(data: dict[str, Any]) -> dict[str, Any]: daily = data.get("daily") or {} dates = daily.get("time") or [] monthly: dict[int, dict[str, list[float]]] = defaultdict(lambda: defaultdict(list)) monthly_precip_by_year: dict[int, dict[int, list[float]]] = defaultdict(lambda: defaultdict(list)) for idx, day in enumerate(dates): year = int(day[:4]) month = int(day[5:7]) for key in ( "temperature_2m_mean", "relative_humidity_2m_mean", "wind_speed_10m_mean", "wind_direction_10m_dominant", ): values = daily.get(key) or [] if idx < len(values) and values[idx] is not None: monthly[month][key].append(float(values[idx])) precip_values = daily.get("precipitation_sum") or [] if idx < len(precip_values) and precip_values[idx] is not None: monthly_precip_by_year[month][year].append(float(precip_values[idx])) month_rows = [] for month in range(1, 13): row = {"month": month} bucket = monthly.get(month, {}) precip_year_totals = [ total for values in monthly_precip_by_year.get(month, {}).values() if (total := _sum(values)) is not None ] row["temperature_c"] = _avg(bucket.get("temperature_2m_mean")) row["precipitation_mm"] = _avg(precip_year_totals) row["humidity_pct"] = _avg(bucket.get("relative_humidity_2m_mean")) row["wind_speed_kmh"] = _avg(bucket.get("wind_speed_10m_mean")) row["wind_direction_deg"] = _avg(bucket.get("wind_direction_10m_dominant")) month_rows.append(row) return { "months": month_rows, "avg_temperature_c": _avg([r["temperature_c"] for r in month_rows if r["temperature_c"] is not None]), "total_precipitation_mm": _sum([r["precipitation_mm"] for r in month_rows if r["precipitation_mm"] is not None]), "avg_humidity_pct": _avg([r["humidity_pct"] for r in month_rows if r["humidity_pct"] is not None]), "avg_wind_speed_kmh": _avg([r["wind_speed_kmh"] for r in month_rows if r["wind_speed_kmh"] is not None]), } def _failure_evidence(label: str, exc: Exception) -> EvidenceItem: return make_evidence( category="Climate", finding=f"{label} could not be retrieved.", source_name="Open-Meteo", source_url="https://open-meteo.com/", source_type="public API", resolution_or_scope="not available", confidence="low", limitation=f"API request failed: {type(exc).__name__}.", design_implication="Do not make climate claims from this missing layer.", verification_needed="Retry later or use a studio-approved climate source.", output_label="site_visit_required", ) def _avg(values: list[float] | None) -> float | None: valid = [float(v) for v in values or [] if v is not None] return round(mean(valid), 2) if valid else None def _sum(values: list[float] | None) -> float | None: valid = [float(v) for v in values or [] if v is not None] return round(sum(valid), 2) if valid else None