Machine Data APIs for Management-System Integration

Machine data arrives through a grower’s management system account — John Deere Operations Center, Climate FieldView, Trimble Ag Software, CNH AFS Connect — as a paginated JSON API behind OAuth 2. The output of this pipeline is three tables in your own store: canonical fields with boundaries, operations with their type, crop, timing and machine, and the measurement series sampled along each operation’s path. Everything downstream on this site depends on those three: yield monitor data cleaning and telemetry QA consumes the third, spatial interpolation for yield data consumes its cleaned form, and every zonal statistic ever computed is grouped by the first. This topic sits under farm data platform engineering, which sets out the manifest and storage patterns the code here assumes.

The integration is deceptively simple to prototype and unpleasant to run. A single grower with one machine works on the first afternoon. What breaks in production is the combination of delayed uploads, rotating credentials, per-grower rate limits, boundaries that change mid-season, and unit fields nobody read.

Prerequisites

• Python 3.11+, httpx 0.27., tenacity 8.5., geopandas 1.0., shapely 2.0., SQLAlchemy 2.0.* with psycopg 3.2.*
• A developer application registered with the provider, holding a client identifier and secret, and a redirect URI you control
• Per-grower authorisation already granted through the provider’s consent screen — a machine-to-machine credential does not exist for grower-owned data
• PostGIS 3.4+ for the canonical field and manifest tables described in PostGIS schema design for farm data
• Boundary payloads are GeoJSON in EPSG:4326; all area comparison happens in a metre-based CRS chosen per field as in understanding CRS in precision agriculture

1. Concept and Contract

The resource tree is deeper than it looks

Every provider models the same hierarchy with different nouns: an organisation (the grower’s account) contains farms, which contain fields, which have boundaries with a validity period. Separately, operations — planting, application, harvest, tillage — reference a field and a season, and carry a link to their measurement series, the per-second samples recorded by the display. Machines and implements are a fourth branch that operations reference by identifier.

Two structural facts drive the design. First, the boundary is versioned independently of the field: the same field identifier can return a different polygon in July than it did in April, and the API will not tell you it changed unless you compare. Second, the measurement series is not returned inline — it is a separate, often large, download, sometimes as a file rather than JSON, and it is the only part of the tree where you should expect timeouts.

Authorisation is per grower, and the refresh token is the asset

There is no service credential that reads a grower’s data. The grower authorises your application once through a browser redirect, you exchange the resulting code for an access token and a refresh token, and from then on the refresh token is the relationship. Access tokens live for minutes to an hour; refresh tokens live for months but are commonly rotated on each exchange, which creates the sharpest failure mode in the whole integration: exchange a refresh token, crash before persisting the new one, and the grower must re-authorise by hand.

The rule that avoids it is to treat the exchange and the persistence as one atomic operation, and to refresh proactively on a clock rather than reactively on a 401. Authenticating with a machine data API in Python walks the full flow with a token store.

Incremental sync keys on modification time

The naive incremental query — “operations that started since my last run” — fails because upload lag is unbounded. A combine running in a river valley with no connectivity may upload a week of harvest in one burst on the drive home. The correct high-water mark is the provider’s record modification timestamp, and the correct query is modifiedSince = high_water - overlap, where the overlap is a few hours of slack for clock skew between their servers and yours. Records are then deduplicated by identity rather than by time, which is safe because the manifest makes re-ingestion a no-op.

2. Step-by-Step Implementation

Step 1 — A client that retries the things worth retrying

Rate limits and transient gateway errors are normal, not exceptional. Everything else should fail loudly and immediately.

import httpx
from tenacity import (retry, retry_if_exception_type, stop_after_attempt,
                      wait_exponential_jitter)

RETRYABLE_STATUS = {429, 500, 502, 503, 504}


class TransientApiError(RuntimeError):
    """A response worth retrying — rate limit or upstream failure."""


class FmisClient:
    def __init__(self, base_url: str, token_provider, timeout: float = 30.0):
        self._token_provider = token_provider           # callable → fresh access token
        self._client = httpx.Client(
            base_url=base_url,
            timeout=httpx.Timeout(timeout, connect=10.0),
            limits=httpx.Limits(max_connections=8, max_keepalive_connections=4),
            headers={"Accept": "application/json"},
        )

    @retry(
        retry=retry_if_exception_type(
            (TransientApiError, httpx.TransportError)),
        wait=wait_exponential_jitter(initial=1, max=60),
        stop=stop_after_attempt(6),
        reraise=True,
    )
    def get(self, path: str, **params) -> dict:
        resp = self._client.get(
            path, params=params or None,
            headers={"Authorization": f"Bearer {self._token_provider()}"},
        )
        if resp.status_code in RETRYABLE_STATUS:
            # Honour an explicit Retry-After before falling back to backoff.
            wait = resp.headers.get("Retry-After")
            raise TransientApiError(
                f"{resp.status_code} on {path}"
                + (f" (Retry-After: {wait}s)" if wait else "")
            )
        resp.raise_for_status()
        return resp.json()

    def close(self) -> None:
        self._client.close()

Validate the step by asserting the client survives a rate limit rather than by watching it work:

# In the test suite — a stub that returns 429 twice, then 200, must yield the payload.
assert client.get("/organizations")["values"], "client should recover from a 429 burst"

Step 2 — Walk the collections with a cursor, not a page number

Providers paginate with either a nextPage link or an offset-and-limit pair, and the cursor expires. Consume it immediately and never store it between runs.

from typing import Iterator

def iter_collection(client: FmisClient, path: str, page_size: int = 100, **params) -> Iterator[dict]:
    """Yield every item of a paginated collection, following the provider's cursor."""
    next_path, next_params = path, {**params, "pageSize": page_size}
    pages = 0
    while next_path:
        payload = client.get(next_path, **next_params)
        items = payload.get("values") or payload.get("items") or []
        for item in items:
            yield item
        pages += 1
        assert pages < 1000, f"pagination did not terminate on {path} — cursor loop?"
        link = next((l["uri"] for l in payload.get("links", []) if l.get("rel") == "nextPage"), None)
        next_path, next_params = (link, {}) if link else (None, {})

The assertion on page count is not paranoia. A cursor that returns itself is a real provider bug, and without the guard the loop consumes your rate-limit budget for the day in about ninety seconds.

Step 3 — Match boundaries to canonical fields spatially

An incoming field carries the provider’s identifier and a GeoJSON boundary. Resolve it against your canonical fields by geometry, not by name, and record the external identifier in the crosswalk.

import geopandas as gpd
from shapely.geometry import shape

MATCH_THRESHOLD = 0.90      # IoU above this is the same field
REVIEW_THRESHOLD = 0.30     # between the two, a human decides


def match_field(incoming_geojson: dict, canonical: gpd.GeoDataFrame, utm_epsg: int):
    """Return (field_id, iou, verdict) for an incoming boundary."""
    geom = shape(incoming_geojson)
    inc = gpd.GeoSeries([geom], crs="EPSG:4326").to_crs(epsg=utm_epsg).iloc[0]
    assert inc.is_valid, "incoming boundary is not a valid geometry — repair before matching"

    local = canonical.to_crs(epsg=utm_epsg)
    candidates = local[local.intersects(inc)]
    if candidates.empty:
        return None, 0.0, "new"

    inter = candidates.geometry.intersection(inc).area
    union = candidates.geometry.union(inc).area
    iou = (inter / union).sort_values(ascending=False)
    best_idx, best_iou = iou.index[0], float(iou.iloc[0])

    if best_iou >= MATCH_THRESHOLD:
        return candidates.loc[best_idx, "field_id"], best_iou, "matched"
    if best_iou >= REVIEW_THRESHOLD:
        return candidates.loc[best_idx, "field_id"], best_iou, "review"
    return None, best_iou, "new"

The UTM zone must be chosen per field rather than globally — a farm straddling a zone boundary produces area errors of several percent if forced into one zone, which is exactly the kind of error validating coordinate systems for variable-rate maps exists to catch.

Step 4 — Poll operations by modification time

from datetime import datetime, timedelta, timezone

OVERLAP = timedelta(hours=6)


def sync_operations(client: FmisClient, org_id: str, high_water: datetime) -> list[dict]:
    """Fetch operations modified since the high-water mark, with an overlap window."""
    since = (high_water - OVERLAP).astimezone(timezone.utc)
    ops = list(iter_collection(
        client, f"/organizations/{org_id}/fieldOperations",
        modifiedSince=since.isoformat().replace("+00:00", "Z"),
    ))
    for op in ops:
        assert "id" in op and "fieldId" in op, f"operation missing identity: {sorted(op)[:6]}"
    return ops


def advance_high_water(ops: list[dict], previous: datetime) -> datetime:
    """New high-water mark: the newest modification time actually seen."""
    seen = [datetime.fromisoformat(o["modifiedTime"].replace("Z", "+00:00")) for o in ops]
    return max(seen + [previous])

Advance the mark from the data, never from datetime.now(). Using the wall clock means any record modified during the run — between the query and the mark being written — is skipped forever.

Step 5 — Normalise units on the way in

AREA_TO_HA = {"ha": 1.0, "hectare": 1.0, "ac": 0.40468564224, "acre": 0.40468564224}
MASS_TO_KG = {"kg": 1.0, "t": 1000.0, "lb": 0.45359237, "bu": None}   # bushels need a crop


def to_hectares(value: float, unit: str) -> float:
    factor = AREA_TO_HA.get(unit.strip().lower())
    assert factor is not None, f"unknown area unit {unit!r} — add it explicitly, never guess"
    return value * factor

Bushels are deliberately unmappable in the table above: a bushel is a volume-based trade unit whose mass depends on the crop and its moisture, so any conversion needs the crop and the test weight. Silently assuming maize at 56 lb per bushel is how a soybean field ends up with a 7% yield error that nobody can trace.

3. Key Parameters and Tuning

4. Edge Cases and Failure Modes

Boundaries change mid-season. A grower edits a headland in June and every acre-based rate you exported in May is now against a different denominator. Store boundaries with a validity interval, never update in place, and stamp each operation with the boundary version current at its start time.

The same operation appears twice with different identifiers. Displays occasionally re-upload after a firmware update, creating a second operation record with identical geometry and timing. Deduplicate on the natural key — field, operation type, start and end timestamps, machine — rather than the provider identifier, and keep both identifiers in the crosswalk.

Empty geometry on an operation. Some records carry an operation with no measurement series at all: a task was created on the display and never executed. These are legitimate and must not be treated as failures; skip them, but count them, because a sudden rise in empty operations usually means a display is misconfigured rather than that nothing happened.

Multi-polygon fields with a hole. Fields with a wetland, a pond or a pivot corner arrive as MultiPolygon with interior rings. Code that assumes Polygon and takes geom.exterior silently deletes the hole, inflating field area by the wetland’s size and applying fertiliser to open water in the resulting prescription. Handle MultiPolygon explicitly, and assert that area computed from the geometry agrees with the reported area.

Token revoked by the grower. Treat a 401 that survives a refresh as a business event, not an outage: mark the grower connection as needing re-authorisation, stop scheduling their syncs, and surface it — retrying every fifteen minutes for a month is how you get an application suspended.

Clock skew and daylight saving. Operation timestamps are commonly local time with an offset; measurement series timestamps sometimes are not. Convert everything to UTC at the boundary of your system and assert the timezone is present. A harvest that appears to run from 23:00 to 01:00 across a daylight-saving change and comes back with negative duration is this bug.

5. Verification and Output Validation

Three checks catch nearly every ingestion defect before it reaches an agronomist.

def validate_operation(op: dict, series: "gpd.GeoDataFrame", boundary) -> None:
    """Assert an operation and its telemetry are internally consistent."""
    # 1. Every sample falls inside (a small buffer around) the field boundary.
    inside = series.geometry.within(boundary.buffer(20))     # metres, in the field's UTM CRS
    share = float(inside.mean())
    assert share > 0.95, f"only {share:.1%} of samples fall inside the boundary — wrong field match?"

    # 2. Duration implied by the samples matches the operation record.
    span = (series["ts"].max() - series["ts"].min()).total_seconds()
    declared = (op["end"] - op["start"]).total_seconds()
    assert abs(span - declared) < 0.25 * max(declared, 1), (
        f"sample span {span:.0f}s disagrees with declared {declared:.0f}s")

    # 3. Working width and speed produce a plausible covered area.
    covered_ha = (series["speed_m_s"].clip(lower=0) * series["width_m"]).sum() / 10_000
    field_ha = boundary.area / 10_000
    assert 0.5 * field_ha < covered_ha < 3.0 * field_ha, (
        f"covered area {covered_ha:.1f} ha implausible for a {field_ha:.1f} ha field")

The third check is the one that earns its place. Overlap on headlands legitimately pushes covered area above field area, so the upper bound is generous — but a covered area of ten times the field means the width was recorded in centimetres, and a covered area of a tenth means it was recorded in metres where feet were expected. Both are common and neither raises an error anywhere else.

Give the validator a deliberately broken input in the test suite — a series shifted 500 m east, a width scaled by 100 — and assert it raises. A validation function that has never rejected anything is not evidence of clean data.

6. Integration with the Broader Pipeline

The three tables this topic produces are the entry point for most of the site. Telemetry flows into yield monitor data cleaning and telemetry QA, whose cleaned output is interpolated by interpolating sparse yield monitor data with kriging and classified into zones by management zone classification algorithms. Boundaries constrain every imagery read, including the windowed satellite reads in satellite imagery APIs and STAC catalogues, and they are the clip geometry in clipping rasters to field boundaries. The loop closes when a prescription generated by variable-rate export to ISOXML is pushed back to the same management system for the operator to execute.

The scheduling of all this — how often to poll, what to do during the six weeks when everything happens at once — belongs to orchestrating seasonal pipelines with Airflow, and the pagination and rate-limit mechanics get their own treatment in paginating and rate-limiting management-system pulls.

Frequently Asked Questions

Can I cache the field list and skip the walk? Cache it, but re-walk on a schedule — daily out of season, hourly during planting and harvest. Fields are created and renamed constantly, and a stale field list means new operations arrive referencing a field identifier you have never seen, which most pipelines handle by dropping the record.

Should telemetry be stored raw or cleaned? Both, in separate places. Raw samples are published data — the record of what the machine reported — and must never be edited. Cleaned samples are derived and are recomputed whenever the cleaning rules change. Storing only the cleaned form makes every future improvement to flow-delay correction unusable on historical seasons.

How large does the telemetry table get? Roughly one sample per second per machine while working. A combine harvesting a 45-hectare field at 6 km/h with a 9 m header takes about 8 hours and produces about 29,000 samples; a 400-field operation across planting, application and harvest lands in the low tens of millions of rows per season. That is comfortable for PostgreSQL with season partitioning and a GiST index, and uncomfortable without either.

This topic is part of Farm Data Platform Engineering: APIs, Storage & Orchestration — see there for the manifest, storage and scheduling patterns the code above assumes.

Related

• Authenticating with the John Deere Operations Center API — the authorisation code flow, token storage and proactive refresh in Python
• Paginating and Rate-Limiting Management-System Pulls — cursor semantics, backoff with jitter and bounded concurrency for a full-season backfill
• PostGIS Schema Design for Farm Data — the canonical field, crosswalk and manifest tables this sync writes into
• Yield Monitor Data Cleaning & Telemetry QA — what to do with the measurement series once it has landed
• Orchestrating Seasonal Pipelines with Airflow — polling cadence, catch-up semantics and per-grower task isolation