Threshold Tuning

Threshold tuning is the calibration layer of a crop application timing engine: it turns a static agronomic rule — “apply when soil moisture is between X and Y” — into a versioned, context-aware band that shifts with crop stage, weather, and field history, then gates every application against it. This subsystem sits inside the Crop Application Timing & Agronomic Validation engine and answers a narrow question before any equipment is cleared: given the conditions active right now, is this reading inside the exact band its agronomic and regulatory constraints require — and can we prove which band was in force? Getting the band wrong does not crash the applicator; it authorizes a plausible-but-wrong application that surfaces months later in a yield gap or a nutrient-management audit.

Problem framing: a threshold is not a constant

An application threshold looks like a fixed pair of numbers, but treating it that way is the root cause of most silent agronomic failures. A soil-moisture band that is correct at emergence over-fertilizes at grain fill; a canopy-temperature ceiling that flags stress in July is meaningless in May; a nitrate limit that holds on a sandy loam leaches on a heavier soil after irrigation. The band that gates a decision has to move with crop phenology, real-time weather, soil texture, and the operator’s own post-application feedback — and it has to move without a code deploy, because the people who recalibrate it are agronomists, not engineers.

The failure cost is asymmetric and quiet. Set a band too loose and the engine authorizes an application the crop cannot use, wasting the input, risking phytotoxicity, and — for nutrients near a waterway — risking a leaching event that becomes a compliance finding. Set it too tight and the engine holds applications the crop genuinely needs, stranding the agronomic window until operators lose trust and start overriding the gate entirely. This subsystem exists to resolve the correct band deterministically for the current context, evaluate the reading against it, stamp the decision with the exact threshold version that produced it, and default to the conservative outcome whenever confidence drops — before dispatch, not after the tank is empty. The tighter, edge-case treatment of soil-water bands for nutrient timing lives in Tuning moisture thresholds for fertilizer application; this page is the end-to-end reference for the tuning and evaluation service itself.

Prerequisites and dependencies

The service depends on three upstream data contracts and a small, pinned stack. Confirm all of them before wiring the evaluation gate into dispatch.

• Normalized field telemetry. Soil volumetric water content, canopy temperature, and nutrient indices arrive as validated, UTC-timestamped payloads from the Farm Data Ingestion & Field Boundary Synchronization pipeline, with raw device streams already resolved by Equipment Telemetry Parsing. This service does not talk to sensors directly.
• A confirmed growth stage. Band selection reads a BBCH or Zadoks stage supplied by Growth Stage Mapping. When the stage is absent, the service selects the most restrictive band on file rather than guessing optimistically.
• Live microclimate signals. Wind, temperature-inversion state, and impending precipitation come from Weather Window Logic, which draws on the upstream Weather API Integration. Weather constraints act as hard overrides, not as inputs to be averaged. The base band limits themselves are versioned in the EPA/USDA Rule Mapping threshold set and are never hard-coded here.

Library versions are pinned so evaluation is reproducible across nodes: pandas>=2.1 for time-series alignment, pydantic>=2.5 for boundary schema validation, and Python 3.10+ for match/case and the X | None union syntax used below. Regulatory inputs are the product label, EPA nutrient pollution guidance, and applicable USDA NRCS nutrient management (Practice 590) standards, cited inline wherever a limit is set.

Architecture of this subsystem

The service is a three-stage transform: raw telemetry is normalized and smoothed into a validated reading, a threshold resolver selects the active band for the current stage and weather context from the versioned threshold set, then a deterministic evaluator classifies the reading against that band and freezes the decision into the audit ledger before handing a directive to dispatch. Each stage emits a structured log line carrying a correlation_id, so a single request is traceable end to end, and each stage has an explicit conservative fallback rather than an exception that halts a live application window.

The inputs are a raw sensor payload plus scalar context (growth stage, weather flags, threshold-set version); the output is a compliance decision object — state, action, reason, and the band version that produced it. The integration points are strictly one-directional: this service consumes stage and weather signals and produces decisions that the dispatch and audit layers consume. It never writes back to the sibling subsystems, and the base limits it reads are owned by the rule-mapping threshold set, so a limit change is a data edit, not a redeploy.

Step-by-step implementation

Step 1 — Ingest and normalize the reading

Standardize heterogeneous telemetry — MQTT soil-probe payloads, REST canopy feeds, scout uploads — into a single validated, smoothed reading. Validate the payload schema at the boundary with pydantic, convert the timestamp to UTC, and apply a rolling-window mean to suppress high-frequency sensor noise. A malformed payload or a network partition must degrade to a flagged fallback state, never crash the worker.

import logging
import pandas as pd
from pydantic import BaseModel, ValidationError
from typing import Any

audit_logger = logging.getLogger("ag.threshold.audit")
audit_logger.setLevel(logging.INFO)


class SensorPayload(BaseModel):
    sensor_id: str
    metric: str
    value: float
    timestamp: str
    correlation_id: str


def ingest_and_normalize(
    raw_payload: dict[str, Any],
    historical_window: pd.Series,
    smoothing_window: int = 5,
) -> dict[str, Any]:
    """Validate, UTC-normalize, and rolling-smooth one telemetry reading.

    Degrades to a flagged fallback state on validation or runtime failure
    so an ingestion worker is never brought down by a bad payload.
    """
    corr_id = raw_payload.get("correlation_id", "UNKNOWN")
    try:
        payload = SensorPayload(**raw_payload)
        ts = pd.to_datetime(payload.timestamp, utc=True)

        if len(historical_window) >= smoothing_window:
            combined = pd.concat([historical_window, pd.Series([payload.value], index=[ts])])
            smoothed = float(combined.rolling(window=smoothing_window).mean().iloc[-1])
        else:
            smoothed = payload.value
            audit_logger.warning(
                "Insufficient history for smoothing; using raw value. corr_id=%s", corr_id
            )

        return {
            "sensor_id": payload.sensor_id,
            "metric": payload.metric,
            "normalized_value": smoothed,
            "timestamp": ts.isoformat(),
            "correlation_id": corr_id,
            "status": "VALID",
        }
    except ValidationError as ve:
        audit_logger.error("Schema validation failed. corr_id=%s err=%s", corr_id, ve)
        return {"status": "INVALID", "correlation_id": corr_id, "fallback_applied": True}
    except Exception as e:
        audit_logger.critical("Unexpected ingestion failure. corr_id=%s err=%s", corr_id, e)
        return {"status": "ERROR", "correlation_id": corr_id, "fallback_applied": True}

Expected log output on the cold-start (insufficient history) path:

WARNING ag.threshold.audit Insufficient history for smoothing; using raw value. corr_id=7c1a…

Step 2 — Resolve the active threshold band

The band is not a literal in the evaluator; it is resolved for the current context from the versioned threshold set. Select the band keyed on (metric, growth_stage), apply any weather-driven tightening, and return the limits together with the threshold_set_version that produced them so the decision stays reproducible after a later recalibration. When the stage is missing, fall back to the most restrictive band on file.

from dataclasses import dataclass


@dataclass(frozen=True)
class ThresholdConfig:
    lower_bound: float
    upper_bound: float
    version: str
    drift_tolerance: float = 0.05
    reference_value: float = 0.0


class ThresholdSet:
    """Append-only registry mapping (metric, stage) -> ThresholdConfig."""

    def __init__(self, version: str, bands: dict[tuple[str, str], ThresholdConfig]):
        self.version = version
        self._bands = bands

    def resolve(self, metric: str, growth_stage: str | None) -> ThresholdConfig | None:
        if growth_stage is not None and (metric, growth_stage) in self._bands:
            return self._bands[(metric, growth_stage)]
        # No stage → pick the tightest band registered for this metric.
        candidates = [c for (m, _), c in self._bands.items() if m == metric]
        if not candidates:
            return None
        return min(candidates, key=lambda c: c.upper_bound - c.lower_bound)


def resolve_band(
    metric: str,
    growth_stage: str | None,
    threshold_set: ThresholdSet,
    precip_imminent: bool = False,
) -> ThresholdConfig | None:
    band = threshold_set.resolve(metric, growth_stage)
    if band is None:
        audit_logger.error("No band registered for metric=%s", metric)
        return None
    if growth_stage is None:
        audit_logger.warning(
            "Growth stage absent; using most restrictive band. metric=%s version=%s",
            metric, band.version,
        )
    if precip_imminent:
        # Weather is a hard tightening override, applied on top of the base band.
        narrowed = band.upper_bound - 0.15 * (band.upper_bound - band.lower_bound)
        band = ThresholdConfig(
            band.lower_bound, narrowed, band.version,
            band.drift_tolerance, band.reference_value,
        )
        audit_logger.info("Precip imminent; upper bound narrowed. version=%s", band.version)
    return band

Step 3 — Evaluate the reading and freeze the decision

The evaluator is a deterministic state machine. It classifies the normalized reading against the resolved band, treats sensor drift as a defer-and-recalibrate condition, and defaults to a conservative hold whenever telemetry is missing, invalid, or the engine itself faults. Every branch returns the same decision shape carrying the band version, so the audit ledger records exactly which limits cleared or blocked the pass.

from enum import Enum


class ComplianceState(Enum):
    APPROVED = "APPROVED"
    DEFERRED = "DEFERRED"
    REJECTED = "REJECTED"
    CONSERVATIVE_FALLBACK = "CONSERVATIVE_FALLBACK"


def evaluate_threshold(
    reading: dict[str, Any],
    band: ThresholdConfig | None,
    drift_detected: bool = False,
) -> dict[str, Any]:
    """Classify a reading against its band with strict conservative fallbacks."""
    corr_id = reading.get("correlation_id", "UNKNOWN")
    value = reading.get("normalized_value")
    version = band.version if band else "NONE"

    def decision(state: ComplianceState, action: str, reason: str) -> dict[str, Any]:
        return {
            "state": state.value,
            "action": action,
            "reason": reason,
            "threshold_set_version": version,
            "correlation_id": corr_id,
        }

    if band is None or value is None or reading.get("status") != "VALID":
        audit_logger.warning("Invalid telemetry/band; holding. corr_id=%s", corr_id)
        return decision(ComplianceState.CONSERVATIVE_FALLBACK, "HOLD_APPLICATION", "INVALID_TELEMETRY")

    try:
        span = band.upper_bound - band.lower_bound
        drifted = drift_detected or (
            span > 0 and abs(value - band.reference_value) > span * band.drift_tolerance
        )
        match (drifted, band.lower_bound <= value <= band.upper_bound):
            case (True, _):
                audit_logger.info("Drift tolerance exceeded; deferring. corr_id=%s", corr_id)
                return decision(ComplianceState.DEFERRED, "RECALIBRATE_SENSOR", "DRIFT_DETECTED")
            case (False, True):
                audit_logger.info("Within band; approving. corr_id=%s", corr_id)
                return decision(ComplianceState.APPROVED, "AUTHORIZE_DISPENSE", "WITHIN_SPEC")
            case (False, False):
                audit_logger.warning(
                    "Value %.4f outside [%.4f, %.4f]; rejecting. corr_id=%s",
                    value, band.lower_bound, band.upper_bound, corr_id,
                )
                return decision(ComplianceState.REJECTED, "HALT_DISPENSE", "OUT_OF_BOUNDS")
    except Exception as e:
        audit_logger.critical("Evaluator fault; conservative hold. corr_id=%s err=%s", corr_id, e)
        return decision(ComplianceState.CONSERVATIVE_FALLBACK, "HOLD_APPLICATION", "ENGINE_EXCEPTION")

Expected log output when a reading sits inside the resolved band:

INFO ag.threshold.audit Within band; approving. corr_id=7c1a…

When an application must be blocked, the directive routes into the deterministic recovery sequence documented in Fallback Application Chains rather than failing open.

Edge cases and known failure modes

Compliance and audit integration

Every evaluation is a compliance artifact, so each stage feeds the append-only audit ledger described in the parent engine. Three facts must survive from decision to inspection: the resolved band and its threshold_set_version, the normalized reading and the signals that shaped the band, and the resulting compliance state and action. Because the base limits originate in the EPA/USDA Rule Mapping threshold set and are stamped onto the record, retuning a band never rewrites history — a pass evaluated last season replays against the limits that were in force at the time.

The immutability guarantee is what makes a record defensible. Ledger entries are hash-anchored and append-only; a decision is never edited in place, only superseded by a new evaluation with its own correlation_id. Idempotent evaluation, keyed on sensor and timestamp, guarantees a network retry cannot emit a second dispense directive. Each nutrient limit cites its authority inline — the product label, EPA nutrient pollution guidance, and the applicable USDA NRCS nutrient management standard — so an auditor can trace any approval back to the rule that permitted it. Structured decisions serialize cleanly for cloud-native log aggregation; see Python’s logging documentation for the JSON-formatter pattern these workers use.

Verification

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

1. Golden band. Feed a reading known to sit at the midpoint of a fixed band and assert the decision is APPROVED with the expected threshold_set_version. A version-skew regression fails this immediately.
2. Inject a missing stage. Call Step 2 with growth_stage=None and assert the tightest registered band is returned and a fallback warning is logged. A regression shows up as a looser band being selected.
3. Inject drift. Evaluate a value inside the numeric bounds but with drift_detected=True; assert the state resolves to DEFERRED / RECALIBRATE_SENSOR, not APPROVED.
4. Corrupt the payload. Send a payload missing value; assert Step 1 returns status: INVALID and the evaluator returns CONSERVATIVE_FALLBACK / HOLD_APPLICATION rather than approving on partial data.

A run passes only when the injected fault produces the expected degraded-but-safe result and the ledger contains one traceable decision per reading. Deeper soil-water and nutrient-timing cases are covered in Tuning moisture thresholds for fertilizer application.

Frequently Asked Questions

Why resolve the band at evaluation time instead of hard-coding limits in the evaluator? Because the people who recalibrate agronomic limits are agronomists, not engineers. Resolving the band from a versioned, append-only registry lets drift sensitivity and nutrient ceilings be retuned as data, without a code deploy, while the stamped threshold_set_version keeps every past decision reproducible.

What happens when the growth-stage or weather feed is unavailable? The resolver selects the most restrictive band on file and logs the fallback, and the evaluator holds on any invalid telemetry. A missing signal can only tighten the gate or defer the pass — it can never loosen a band or approve on partial data.

How is a decision kept reproducible after thresholds are retuned? Every decision stores the resolved limits, the threshold_set_version, and the signals that shaped the band. Threshold sets are append-only and versioned, so replaying an old request reconstructs the exact band that cleared or blocked the pass even after the weights have been recalibrated.

Why treat weather as a hard override rather than one more input to average? Averaging a drift or runoff risk against favourable soil readings can approve an application that a single adverse condition should block. Impending precipitation, high wind, and inversions narrow or veto the band directly, so one hazardous condition is sufficient to defer or reject regardless of the other metrics.

Related

• Tuning moisture thresholds for fertilizer application — soil-water and nutrient-timing edge cases for this service.
• Growth Stage Mapping — supplies the BBCH/Zadoks stage that selects the active band.
• Weather Window Logic — supplies the wind, inversion, and precipitation overrides that tighten or veto a band.
• Buffer Zone Calculations — a downstream consumer whose base setback and drift multipliers this service versions.
• Fallback Application Chains — executes the deterministic recovery sequence when the evaluator blocks a pass.
• EPA/USDA Rule Mapping — the versioned threshold set that provides the authoritative band limits.

Up: Crop Application Timing & Agronomic Validation