Weather API Integration

Weather API integration is the subsystem that turns third-party meteorological feeds into validated, field-anchored records that irrigation scheduling, pest-pressure forecasting, and spray-window logic can safely act on. It sits inside the Farm Data Ingestion & Field Boundary Synchronization pipeline and answers one question before any forecast is allowed to drive a decision — does this gridded observation map, unambiguously and with a known schema, to a real polygon on a verified field, and is it fresh enough to trust? Get that wrong and the failure is quiet: a plausible-but-stale temperature silently green-lights a spray inside a temperature inversion, or a mis-projected grid cell attributes rainfall to the wrong block and skews a whole season of irrigation accounting.

Problem framing: forecast data is gridded, rate-limited, and units-ambiguous

Every weather provider returns data on its own grid, in its own units, on its own cadence, and behind its own request quota. Open-Meteo, Tomorrow.io, and the NOAA/NWS API each expose different JSON shapes, different null conventions for missing hours, and different coordinate-reference assumptions. The specific sub-problem this subsystem owns is turning that fragmented, quota-limited stream into one canonical, field-anchored observation schema deterministically, so a downstream agronomic model never has to reason about which vendor produced a record or whether a value is in millimetres or inches.

The failure cost is asymmetric. A weather record does not crash anything when it is wrong — it coerces cleanly into a valid-looking row and then misinforms an irrigation or application decision worth real money and real regulatory exposure. Three failure modes recur:

• Silent unit or projection drift. A provider returns wind in km/h where the pipeline assumes m/s, or grids on a projection that shifts a cell tens of metres, mis-attributing a frost event to the wrong field. Normalization and containment checks at the ingestion boundary reject the ambiguity before it is persisted.
• Stale-as-fresh substitution. An endpoint degrades and returns yesterday’s forecast with today’s fetch timestamp. A freshness contract that compares the observation time to the request time flags the staleness instead of letting it masquerade as live data.
• Quota exhaustion under fleet load. A multi-farm deployment stampedes the endpoint and gets throttled precisely when a frost alert matters most. Prioritized, rate-limited polling keeps the critical path inside quota; the mechanics live in Handling weather API rate limits for crop models.

This subsystem decides whether and how each weather payload becomes a trustworthy, field-anchored observation, quarantining anything it cannot validate rather than guessing optimistically. The downstream Weather Window Logic engine and the broader Crop Application Timing & Agronomic Validation reference consume what this pipeline produces — nothing they decide can be more correct than the observation delivered here.

Prerequisites and dependencies

The integration sits between raw acquisition and the field-anchored store, so it depends on a small pinned stack and three upstream contracts. Confirm all of them before wiring it into ingestion.

• A polling source and rate budget. Payloads are pulled on the cadence and backoff defined by the Async Polling Strategies loop; per-endpoint quota shaping and request prioritization are governed by Handling weather API rate limits for crop models.
• A canonical field registry. Each grid observation is resolved to a field zone against the polygon registry maintained by the parent Farm Data Ingestion & Field Boundary Synchronization pipeline. Boundary-file decoding, CRS enforcement, and topology repair for that registry live in Parsing ISOXML and Shapefile field boundaries.
• A versioned field contract. The canonical observation shape and its validation rules are governed by Field Schema Design, and any compliance fields a weather record must carry originate in the EPA/USDA Rule Mapping threshold set — never hard-coded here.

Library versions are pinned so ingestion is reproducible across worker nodes: pydantic>=2.5 for Rust-backed validation and field_validator, aiohttp>=3.9 for non-blocking fetches with connection pooling, shapely>=2.0 and an R-tree index (shapely.STRtree) for grid-to-polygon containment, and Python 3.10+ for match/case and the X | None union syntax used below. The governing external standards are ISO 8601 for timestamps and SI units at the validation boundary; latitude/longitude are held as WGS84 (EPSG:4326) and any provider grid on another projection is reprojected before containment. Downstream weather-driven application decisions must remain auditable against EPA pesticide application recordkeeping and USDA NRCS reporting requirements.

Architecture of this subsystem

The integration is a four-stage transform: a payload is fetched with backoff and a fallback endpoint, validated (units and timestamp normalized) into the canonical model, resolved to a field zone by spatial containment, then either committed as a field-anchored observation or quarantined for review. Every stage emits a structured log line carrying a request_id, so a single payload is traceable end to end, and every stage has an explicit conservative outcome — quarantine or a deterministic historical fallback, never a silent coercion — when it cannot proceed.

The inputs are a raw weather payload dict plus scalar context (endpoint URLs, API key, request_id) and the field-polygon index; the outputs are a validated FieldObservation or a quarantine record. The integration points are strictly one-directional: this service consumes payloads and the field registry and produces canonical observations that the boundary-synchronization, audit, and Weather Window Logic layers consume — it never writes back to the polling loop or the provider API.

Step-by-step implementation

The reference integration below runs against Python 3.10+ and passes a syntax and indentation check. It is built in three composable stages: fetch with a validated fallback chain, spatially resolve each observation to a field, then correlate the result with equipment telemetry.

Step 1 — Align grid observations to field polygons and validate the schema

The foundational step maps a provider’s grid coordinates to a canonical field zone and enforces the schema before anything is persisted. Coordinates are reprojected to WGS84, a prepared R-tree index resolves the containing polygon, and any observation whose cell intersects no active field is held as unattributed rather than forced to the nearest block. Malformed or out-of-range values trigger quarantine, not silent degradation.

Step 2 — Fetch asynchronously with a validated fallback chain

Weather APIs enforce strict request quotas, so synchronous blocking calls do not survive multi-farm load. Fetch with aiohttp connection pooling and exponential backoff, and route the payload through a primary→fallback endpoint chain. Strict pydantic validation runs on every response; a provider that returns a malformed or unit-inconsistent payload is treated as a failed attempt, not a trusted record. Rate-limiter placement and token-bucket sizing are covered in depth by Handling weather API rate limits for crop models; the fetch below assumes it runs inside that budget.

import asyncio
import logging
import aiohttp
from datetime import datetime, timezone
from typing import Any, Optional
from pydantic import BaseModel, ValidationError, field_validator

audit_logger = logging.getLogger("weather.ingestion.audit")
audit_logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
audit_logger.addHandler(handler)


class WeatherPayload(BaseModel):
    lat: float
    lon: float
    timestamp: datetime
    temperature_c: float
    precipitation_mm: float
    wind_speed_ms: float

    @field_validator("timestamp")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        # A naive timestamp cannot be compared to the fetch time for a freshness check.
        return v if v.tzinfo else v.replace(tzinfo=timezone.utc)


async def fetch_weather_with_fallback(
    session: aiohttp.ClientSession,
    primary_url: str,
    fallback_url: str,
    api_key: str,
    request_id: str,
    total_timeout_s: float = 10.0,
) -> Optional[WeatherPayload]:
    """
    Async fetch with strict validation, a primary->fallback endpoint chain,
    and an audit line per attempt. Returns None to signal quarantine.
    """
    headers = {"Authorization": f"Bearer {api_key}"}
    timeout = aiohttp.ClientTimeout(total=total_timeout_s)

    for attempt, url in enumerate([primary_url, fallback_url], start=1):
        try:
            audit_logger.info("REQ_START | id=%s url=%s attempt=%d", request_id, url, attempt)
            async with session.get(url, headers=headers, timeout=timeout) as resp:
                resp.raise_for_status()
                data: dict[str, Any] = await resp.json()

                # Field paths must match the provider's actual response shape (see note below).
                payload = WeatherPayload(
                    lat=data["coordinates"]["lat"],
                    lon=data["coordinates"]["lon"],
                    timestamp=datetime.fromisoformat(data["time"]),
                    temperature_c=data["current"]["temperature_2m"],
                    precipitation_mm=data["current"]["precipitation"],
                    wind_speed_ms=data["current"]["wind_speed_10m"],
                )
                audit_logger.info("REQ_SUCCESS | id=%s url=%s validated=true", request_id, url)
                return payload

        except aiohttp.ClientError as e:
            audit_logger.warning("REQ_FAIL | id=%s url=%s error=%s", request_id, url, str(e))
            if attempt == 2:
                audit_logger.critical("FALLBACK_EXHAUSTED | id=%s action=quarantine", request_id)
                return None
        except ValidationError as e:
            audit_logger.error("SCHEMA_VIOLATION | id=%s details=%s", request_id, e.json())
            return None
        except Exception as e:  # noqa: BLE001 - fail closed on unexpected shapes
            audit_logger.error("UNEXPECTED_ERROR | id=%s error=%s", request_id, str(e))
            return None

        await asyncio.sleep(2 ** attempt)  # Exponential backoff before the fallback endpoint
    return None

The field paths (data["coordinates"]["lat"], data["current"]["temperature_2m"], and the rest) are provider-specific and must be validated against your endpoint’s schema before deployment — response structure varies between Open-Meteo, Tomorrow.io, and NOAA/NWS, and a mismatched path is exactly the kind of silent drift the quarantine path exists to catch. Expected log output for a healthy fetch, then a failover, then an exhausted chain:

2026-07-02 06:00:01 | INFO     | REQ_START | id=obs-4417 url=https://api.open-meteo.com/... attempt=1
2026-07-02 06:00:01 | INFO     | REQ_SUCCESS | id=obs-4417 url=https://api.open-meteo.com/... validated=true
2026-07-02 06:05:03 | WARNING  | REQ_FAIL | id=obs-4418 url=https://api.open-meteo.com/... error=429 Too Many Requests
2026-07-02 06:05:05 | INFO     | REQ_START | id=obs-4418 url=https://api.tomorrow.io/... attempt=2
2026-07-02 06:05:05 | INFO     | REQ_SUCCESS | id=obs-4418 url=https://api.tomorrow.io/... validated=true
2026-07-02 06:10:12 | CRITICAL | FALLBACK_EXHAUSTED | id=obs-4419 action=quarantine

Step 3 — Correlate the observation with equipment telemetry

Raw meteorology gains operational value only when contextualized against machinery state and soil-sensor networks. Correlating weather with Equipment Telemetry Parsing output lets decision engines adjust implement depth, spray-drift parameters, and harvest timing against a real-time microclimate. When live data is unavailable, the pipeline applies a deterministic historical fallback rather than emitting a null-state record — and it stamps that substitution so no downstream model mistakes an interpolated value for a live one. The layered fallback ordering itself is governed by the Fallback Routing Logic contract.

import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Optional


@dataclass
class CorrelatedRecord:
    field_id: str
    timestamp: str
    temperature_c: float
    soil_moisture_pct: Optional[float]
    equipment_status: str
    data_source: str
    fallback_applied: bool


def correlate_weather_with_telemetry(
    weather_payload: Optional["WeatherPayload"],
    telemetry_stream: list[dict[str, Any]],
    historical_baseline: dict[str, float],
    audit_logger: logging.Logger,
) -> CorrelatedRecord:
    """
    Correlate a live observation with the latest equipment telemetry, applying a
    deterministic historical fallback and marking the substitution for the audit trail.
    """
    match weather_payload:
        case None:
            audit_logger.warning("WEATHER_MISSING | applying_historical_fallback")
            temp = historical_baseline.get("temperature_c", 18.5)
            source, fallback = "historical_baseline", True
            ts = datetime.now(timezone.utc).isoformat()
        case payload:
            temp = payload.temperature_c
            source, fallback = "live_api", False
            ts = payload.timestamp.isoformat()

    latest = telemetry_stream[-1] if telemetry_stream else {}
    record = CorrelatedRecord(
        field_id=latest.get("field_id", "unassigned"),
        timestamp=ts,
        temperature_c=temp,
        soil_moisture_pct=latest.get("soil_moisture_pct"),
        equipment_status=latest.get("status", "unknown"),
        data_source=source,
        fallback_applied=fallback,
    )

    audit_logger.info(
        "CORRELATION_COMMIT | field=%s source=%s fallback=%s",
        record.field_id, source, fallback,
    )
    return record

Edge cases and known failure modes

Compliance and audit integration

Every weather-driven decision is a potential audit artifact, so each stage feeds the append-only audit ledger defined by the parent ingestion pipeline. Three facts must survive from fetch to inspection: the provenance of the value (live_api versus historical_baseline, with the endpoint that served it), the validation and freshness outcome carried by the request_id, and the field zone the observation resolved to. Because a downstream application record can hinge on a temperature or wind reading — think a drift-restricted spray gated on wind speed — the observation that justified the decision must be reconstructable long after the fact.

The immutability guarantee is what makes that reconstruction defensible. Ledger entries are hash-anchored and append-only; an observation is never edited in place, only superseded by a newer fetch with its own request_id, and idempotent writes prevent a network retry from duplicating a record. Fallback activations, schema violations, and quarantine events are all captured, giving an auditor a continuous trail from an EPA pesticide-use record or USDA NRCS report back to the exact meteorological value — and its provenance flag — that informed the pass. Structured logs should stream to a centralized observability platform so validation-failure rate, quarantine volume, fallback frequency, and fetch-latency percentiles are monitored as pipeline-health signals; a rising fallback rate is an early warning that a provider is degrading before it fails outright.

Verification

Confirm correct operation in staging with a reproducible fault-injection scenario before promoting a change:

1. Golden observation. Feed a well-formed payload with a timezone-aware ISO 8601 timestamp inside a known field polygon and assert fetch_weather_with_fallback returns a validated WeatherPayload, the record resolves to the expected field_id, and a REQ_SUCCESS line is written.
2. Force the fallback endpoint. Make the primary URL return 429 and assert the fetch fails over to the secondary, a single REQ_FAIL warning precedes a REQ_SUCCESS on attempt 2, and the value round-trips correctly.
3. Exhaust the chain. Point both endpoints at an always-erroring stub and assert fetch_weather_with_fallback returns None, FALLBACK_EXHAUSTED is logged, and correlate_weather_with_telemetry emits a record with data_source=historical_baseline and fallback_applied=True.
4. Inject unit drift. Send a payload with wind in km/h and assert the validation boundary rejects or normalizes it rather than persisting a raw value, producing a SCHEMA_VIOLATION (or a logged conversion) instead of a silently wrong record.
5. Inject an out-of-bounds cell. Send an observation whose coordinates fall outside every field polygon and assert it is held as unattributed, not forced to the nearest block.

A run passes only when each injected fault produces the expected degraded-but-safe result and the ledger contains one traceable entry per stage. The rate-limit behaviour that shapes fetch cadence is verified separately in Handling weather API rate limits for crop models.

Frequently Asked Questions

Which weather provider should the pipeline target? The pipeline is provider-agnostic by design: the canonical WeatherPayload model and the primary→fallback chain let you run Open-Meteo, Tomorrow.io, or NOAA/NWS behind the same validation boundary. Pin one primary for cost and coverage, pick a second provider on a different infrastructure for the fallback, and keep the field-path mapping for each in a per-provider profile so a silent schema change is caught as a SCHEMA_VIOLATION rather than coerced.

How does the pipeline avoid acting on stale forecast data? Every payload carries a timezone-aware observation timestamp that is compared to the fetch time. When the gap exceeds the freshness window the record is quarantined instead of persisted, and the require_tz_aware validator guarantees the comparison is never silently wrong because a provider returned a naive timestamp.

What happens when both weather endpoints are unavailable? The fetch chain returns None, and correlation applies a deterministic historical baseline that is explicitly marked with data_source=historical_baseline and fallback_applied=True. No null-state record reaches downstream models, and the substitution is auditable, so a decision made on a fallback value is never mistaken for one made on live data.

How is a weather observation tied to the right field? Grid coordinates are reprojected to WGS84 and resolved against a prepared R-tree index of active field polygons. An observation whose cell intersects no field is held as unattributed rather than forced to the nearest block, which prevents rainfall or a frost event from being mis-attributed to acreage it never fell on.

Related

• Handling weather API rate limits for crop models — token-bucket quota shaping and request prioritization for the fetch loop above.
• Async Polling Strategies — the fetch cadence, backoff, and circuit-breaker layer that drives this integration.
• Equipment Telemetry Parsing — the machine-state stream this subsystem correlates weather against.
• Weather Window Logic — consumes these field-anchored observations to compute safe application windows.
• Fallback Routing Logic — governs the ordering of the live→historical fallback applied during correlation.
• Field Schema Design — governs the canonical observation contract this integration targets.

Up: Farm Data Ingestion & Field Boundary Synchronization