Equipment Telemetry Parsing

Equipment telemetry parsing is the translation layer of the ingestion platform: it turns raw CAN bus frames, ISO 11783 (ISOBUS) task data, and proprietary OEM API payloads into structured, validated agronomic events that every downstream planning and compliance calculation can trust. This subsystem sits inside the Farm Data Ingestion & Field Boundary Synchronization pipeline and answers one question before a record is allowed to influence input tracking — does this machine payload map, unambiguously and with a known schema version, to a real operation on a verified field? Get that wrong and the failure is not a crash; it is a plausible-looking record that silently misattributes a chemical application to the wrong acreage and surfaces during an audit.

Problem framing: machine output is noisy, versioned, and unforgiving

A modern sprayer, planter, or combine emits telemetry from a dozen electronic control units at once: high-frequency implement engagement and section-control states interleaved with lower-frequency engine diagnostics, GPS fixes, and fuel-rate counters. The formats are heterogeneous and none of them are stable. CAN frames follow the equipment’s J1939/ISOBUS profile, ISOBUS task controllers export ISOXML that varies by firmware, and OEM cloud APIs return JSON whose shape shifts between releases. The specific sub-problem this subsystem owns is turning that fragmented, drifting stream into one canonical event schema deterministically, so a downstream agronomic model never has to reason about which tractor brand or firmware produced a record.

The failure cost is asymmetric and quiet. Manufacturers routinely change payload structures on a firmware update without bumping any version field the consumer can see — this telemetry schema drift renames fuel_usage to fuel_rate_lph, changes a timestamp from epoch seconds to an ISO 8601 string, or drops a field entirely. A parser that assumes a fixed shape will coerce garbage into a valid-looking record, and every gallon of chemical or pound of fertilizer downstream is then accounted for against the wrong pass. The Async Polling Strategies subsystem decides when to pull data; this subsystem decides whether and how each payload becomes a trustworthy event, quarantining anything it cannot map rather than guessing optimistically.

Prerequisites and dependencies

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

• A polling or streaming source. Payloads arrive from the Async Polling Strategies loop or from OEM webhooks. Direct John Deere Operations Center integration — OAuth token lifecycle, pagination, and idempotent dedup — is covered in Connecting John Deere API to Python backend.
• A canonical field registry. Parsed GPS geometry is resolved to a field zone against the polygon registry maintained by the parent pillar. Boundary-file decoding for that registry — CRS enforcement and multipart topology repair — lives in Parsing ISOXML and Shapefile field boundaries.
• A versioned field contract. The canonical event shape and its validation rules are governed by Field Schema Design, and the compliance fields each event must carry originate in the EPA/USDA Rule Mapping threshold set — never hard-coded here.

Library versions are pinned so parsing is reproducible across worker nodes: pydantic>=2.5 (Rust-backed validation and field_validator), aiohttp>=3.9 for non-blocking fetches, structlog>=24.1 for machine-readable audit lines, and Python 3.10+ for match/case and the X | None union syntax used below. The governing external standards are ISO 11783 (ISOBUS) for the wire format and SAE J1939 for the underlying CAN transport; SI units and ISO 8601 timestamps are enforced at the validation boundary, and downstream compliance tagging follows EPA pesticide application recordkeeping and USDA NRCS reporting requirements.

Architecture of this subsystem

The parser is a four-stage transform: a payload is fetched with backoff, fingerprinted against a known schema version, validated (with a fallback chain for legacy shapes) into the canonical model, then either committed as an agronomic event or quarantined for review. Every stage emits a structured log line carrying a correlation_id, so a single payload is traceable end to end, and every stage has an explicit conservative outcome — quarantine, not a silent coercion — when it cannot proceed.

The inputs are a raw payload dict plus scalar context (endpoint, retry budget); the outputs are a validated TelemetryPayload and a metadata block, or a quarantine record. The integration points are strictly one-directional: this service consumes payloads and the field registry and produces canonical events that the boundary-synchronization and audit layers of the parent pillar consume — it never writes back to the polling loop or the OEM API.

Step-by-step implementation

The reference parser below is a single ingestion pipeline class with three composable stages. It runs against Python 3.10+ and passes a syntax and indentation check as shown.

Step 1 — Fetch each payload with backoff and a circuit breaker

Rural connectivity and OEM rate limits make naive request loops the primary source of dropped records. Fetch with exponential backoff and randomized jitter so simultaneously reconnecting fleets do not stampede the endpoint, and raise a bounded, logged error rather than retrying forever.

Step 2 — Validate with a version-aware fallback chain

The canonical model is defined once with pydantic; a custom validator coerces legacy epoch and Z-suffixed timestamps into timezone-aware ISO 8601. When strict validation fails, the parser remaps known legacy keys and retries once before quarantining — it never invents missing fields.

Step 3 — Emit the canonical event or quarantine

A record that passes validation is logged as a success and returned for enrichment (GPS-to-field resolution and unit tagging). A record that survives neither the strict nor the fallback path is quarantined with its raw keys captured, so the drift can be diagnosed without losing the evidence.

import asyncio
import time
import uuid
from datetime import datetime, timezone
from typing import Any

import aiohttp
import structlog
from pydantic import BaseModel, ValidationError, field_validator

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)
logger = structlog.get_logger("telemetry_parser")


class TelemetryPayload(BaseModel):
    machine_id: str
    timestamp: datetime
    latitude: float
    longitude: float
    engine_rpm: int | None = None
    pto_engaged: bool = False
    fuel_rate_lph: float | None = None

    @field_validator("timestamp", mode="before")
    @classmethod
    def coerce_timestamp(cls, v: Any) -> Any:
        # Legacy OEMs send epoch seconds; ISOBUS exports send ISO 8601 with a Z suffix.
        if isinstance(v, (int, float)):
            return datetime.fromtimestamp(v, tz=timezone.utc)
        if isinstance(v, str):
            try:
                return datetime.fromisoformat(v.replace("Z", "+00:00"))
            except ValueError as exc:
                raise ValueError("Invalid timestamp format") from exc
        return v


class TelemetryIngestionPipeline:
    """Fetch, validate, and canonicalize equipment telemetry with an audit trail."""

    FALLBACK_MAP = {
        "mach_id": "machine_id",
        "ts": "timestamp",
        "lat": "latitude",
        "lon": "longitude",
        "rpm": "engine_rpm",
        "fuel_usage": "fuel_rate_lph",
    }

    def __init__(self, api_endpoint: str, max_retries: int = 3, backoff_cap_s: float = 30.0):
        self.api_endpoint = api_endpoint
        self.max_retries = max_retries
        self.backoff_cap_s = backoff_cap_s

    async def fetch_with_backoff(
        self, session: aiohttp.ClientSession, payload_id: str, correlation_id: str
    ) -> dict[str, Any]:
        """Fetch with exponential backoff, jitter, and a bounded circuit breaker."""
        for attempt in range(self.max_retries):
            try:
                async with session.get(f"{self.api_endpoint}/{payload_id}") as resp:
                    resp.raise_for_status()
                    return await resp.json()
            except aiohttp.ClientError as exc:
                delay = min(2 ** attempt + (time.time() % 1), self.backoff_cap_s)
                logger.warning(
                    "fetch_retry",
                    correlation_id=correlation_id,
                    attempt=attempt,
                    error=str(exc),
                    retry_delay=delay,
                )
                await asyncio.sleep(delay)
        # Circuit opens: bounded, logged failure rather than an unbounded retry loop.
        raise ConnectionError(f"Max retries exceeded for {payload_id}")

    def parse_with_fallback(
        self, raw_data: dict[str, Any], correlation_id: str
    ) -> TelemetryPayload:
        """Strict validation first; remap known legacy keys once; else quarantine."""
        try:
            return TelemetryPayload(**raw_data)
        except ValidationError as exc:
            logger.warning(
                "primary_validation_failed",
                correlation_id=correlation_id,
                errors=exc.errors(),
            )

        mapped = {self.FALLBACK_MAP.get(k, k): v for k, v in raw_data.items()}
        try:
            payload = TelemetryPayload(**mapped)
            logger.info("legacy_schema_remapped", correlation_id=correlation_id)
            return payload
        except ValidationError as exc:
            logger.warning(
                "fallback_validation_failed",
                correlation_id=correlation_id,
                errors=exc.errors(),
            )

        # No known schema version matched: preserve evidence, refuse to guess.
        logger.error(
            "payload_quarantined",
            correlation_id=correlation_id,
            raw_keys=sorted(raw_data.keys()),
        )
        raise ValueError("Payload does not match any known schema version")

    async def process_payload(self, payload_id: str) -> TelemetryPayload:
        """End-to-end: fetch, validate, and emit a canonical event with audit logging."""
        correlation_id = str(uuid.uuid4())
        async with aiohttp.ClientSession() as session:
            raw = await self.fetch_with_backoff(session, payload_id, correlation_id)
            try:
                parsed = self.parse_with_fallback(raw, correlation_id)
            except Exception as exc:
                logger.critical(
                    "ingestion_aborted",
                    correlation_id=correlation_id,
                    payload_id=payload_id,
                    error=str(exc),
                )
                raise
            logger.info(
                "telemetry_validated",
                correlation_id=correlation_id,
                machine_id=parsed.machine_id,
                status="success",
            )
            return parsed

Expected log output on the degraded (legacy remap) path:

{"event": "primary_validation_failed", "correlation_id": "7c1e…", "errors": [{"loc": ["machine_id"], "type": "missing"}]}
{"event": "legacy_schema_remapped", "correlation_id": "7c1e…"}
{"event": "telemetry_validated", "correlation_id": "7c1e…", "machine_id": "JD-8R-410", "status": "success"}

Expected log output when a payload matches no known schema version:

{"event": "primary_validation_failed", "correlation_id": "9a4f…", "errors": [{"loc": ["timestamp"], "type": "datetime_type"}]}
{"event": "fallback_validation_failed", "correlation_id": "9a4f…", "errors": [{"loc": ["machine_id"], "type": "missing"}]}
{"event": "payload_quarantined", "correlation_id": "9a4f…", "raw_keys": ["device", "epoch", "position"]}

Edge cases and known failure modes

Compliance and audit integration

Every parsed event is a compliance artifact, so each stage feeds the append-only audit ledger described in the parent pillar. Three facts must survive from parse to inspection: the raw payload’s schema-version fingerprint, the validation outcome (strict, remapped, or quarantined) with the correlation_id that produced it, and the canonical event itself. Because the required compliance fields originate in the EPA/USDA Rule Mapping threshold set and are stamped onto the record, a schema change never rewrites history — an event parsed last season replays against the contract that was in force at the time.

The immutability guarantee is what makes a record defensible during an inspection. Ledger entries are hash-anchored and append-only; an event is never edited in place, only superseded by a new parse with its own correlation_id, and idempotent writes keyed on machine_id + timestamp + payload_hash prevent a network retry from duplicating an application record. Coercion attempts, fallback activations, 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 machine frame that produced it. Structured logs should stream to a centralized observability platform so validation-failure rate, quarantine volume, and fetch-latency percentiles are monitored as pipeline-health signals.

Verification

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

1. Golden payload. Feed a well-formed record with an ISO 8601 timestamp and assert process_payload returns a TelemetryPayload with a timezone-aware timestamp and a telemetry_validated ledger entry. A timestamp-coercion regression fails here immediately.
2. Inject schema drift. Send a payload using only legacy keys (mach_id, ts, fuel_usage) and assert the record is recovered via the remap, the value round-trips correctly, and a legacy_schema_remapped entry is written.
3. Inject an unmappable payload. Send a record missing machine_id under any alias and assert it is quarantined, ValueError is raised, and payload_quarantined captures the raw keys — not a coerced record with a default identity.
4. Force the circuit breaker. Point api_endpoint at an endpoint that always errors and assert exactly max_retries fetch_retry warnings precede a bounded ConnectionError, with no unbounded retry loop.

A run passes only when each injected fault produces the expected degraded-but-safe result and the ledger contains one traceable entry per stage. Deeper boundary-decoding and OEM-authentication cases are covered in Parsing ISOXML and Shapefile field boundaries and Connecting John Deere API to Python backend.

Frequently Asked Questions

Why use a fallback key map instead of just versioning the payloads? Because most agricultural OEMs change payload structure on a firmware update without exposing any version field the consumer can read. The parser tries the strict canonical model first, then a single deterministic remap of known legacy keys, and quarantines anything else — so a silent rename is recovered and logged rather than coerced into a misattributed record.

What happens to a payload that matches no known schema version? It is quarantined, not dropped and not guessed. The raw keys are captured in a payload_quarantined audit entry with the request’s correlation_id, and a ValueError propagates so the ingestion loop can route the record for manual review without losing the evidence needed to diagnose the drift.

How does the parser avoid double-counting an application after a network retry? Writes are idempotent, keyed on a composite of machine_id, timestamp, and a payload_hash. A replayed payload reconciles to the same event, so a retry during a connectivity blip never produces two chemical-application records for one pass.

Which standards govern the wire format the parser accepts? Field telemetry rides SAE J1939 on the CAN transport and is exported as ISO 11783 (ISOBUS) task data; timestamps are normalized to ISO 8601 and units to SI at the validation boundary. Compliance fields required on each event come from the versioned EPA/USDA rule mapping rather than being hard-coded in the parser.

Related

• Parsing ISOXML and Shapefile field boundaries — CRS enforcement and topology repair for the boundary files this parser resolves GPS against.
• Connecting John Deere API to Python backend — OAuth lifecycle, pagination, and idempotent dedup for OEM cloud payloads.
• Async Polling Strategies — the fetch cadence and circuit-breaker layer that feeds this parser.
• Weather API Integration — enriches parsed machine states with microclimate context for downstream models.
• Field Schema Design — governs the canonical event contract this parser targets.
• EPA/USDA Rule Mapping — the versioned threshold set supplying the compliance fields each event carries.

Up: Farm Data Ingestion & Field Boundary Synchronization