Input Inventory Ingestion

Input inventory ingestion is the accounting spine of the platform: it turns purchase orders, receiving tickets, and product-label metadata for seed, fertilizer, and pesticide into a normalized, quantity-accurate stock ledger, then holds that ledger accountable against every application recorded in the field. This subsystem lives inside the Farm Data Ingestion & Field Boundary Synchronization pipeline, and it answers a question no agronomic model can answer for itself — did the product we applied actually come from a lot we received, in a quantity our books can reconcile? When that answer is anything but an unambiguous yes, a chemical application cannot be traced to its source lot during a recall, and an inspector’s request for the paper trail behind a restricted-use pesticide has no defensible reply.

Problem framing: receipts are messy, units disagree, and stock must balance

A working farm buys inputs from a dozen channels. A cooperative emails a PDF invoice, a chemical distributor pushes an EDI receipt, a seed dealer hands over a paper ticket that someone keys in later, and each of them expresses quantity in whatever unit was on the bill of lading — gallons here, pounds there, US short tons for bulk fertilizer, thousand-seed counts for a planter. The lot or batch number that makes a product traceable is printed on the container but is not always transcribed onto the purchase record, and the EPA registration number that FIFRA requires on every pesticide is sometimes dropped entirely because whoever entered the receipt did not know it was mandatory.

The specific problem this subsystem owns is converting that fragmented intake into one canonical inventory model — deterministic units, exact quantities, a lot number on every line, and an EPA registration number on every pesticide — and then keeping the resulting on-hand balance honest as applications draw it down. The failure cost is asymmetric and quiet, exactly like the telemetry side handled by Equipment Telemetry Parsing: a receipt entered in gallons but stored as if it were litres inflates on-hand stock by a factor of nearly four, so a later draw-down looks well within budget when the tank is actually running dry, and no exception is ever raised. Worse, a pesticide received without its lot number cannot be tied to an application later, which means a manufacturer recall cannot tell you whether the suspect batch ever touched your ground.

Two focused problems fall out of this and get their own guides below. Linking each application back to a specific received batch for recall traceability and FIFRA recordkeeping is covered in Tracking Pesticide Lot Numbers for Traceability; decomposing a mixed spray batch into per-product draw-downs and flagging usage variance is covered in Reconciling Tank-Mix Batches Against Inventory.

Prerequisites and dependencies

This service sits between raw procurement records and the compliance ledger, so it depends on a small pinned stack and three upstream contracts. Confirm each before wiring it into ingestion.

• A receipt source. Purchase orders and receiving records arrive as structured rows — from an ERP export, an EDI feed, or a keyed-in ticket. This subsystem does not scrape PDFs; it consumes already-tabular line items and is the first place their contents are trusted.
• A product-label reference. Every pesticide line is checked for a well-formed EPA registration number, and the product’s class (seed, fertilizer, or restricted-use pesticide) is resolved against a label database so the correct recordkeeping rules apply. The rules themselves come from the EPA/USDA Rule Mapping threshold set and are never hard-coded here.
• A versioned field contract. The canonical lot and reconciliation shapes are governed by Field Schema Design, so a change to how a lot is represented re-normalizes historical inventory the same way rather than forking the model.

Library versions are pinned so quantity arithmetic is reproducible across worker nodes: pydantic>=2.5 for Rust-backed validation and model_validator, Python’s standard-library decimal for exact regulated quantities (never float), and structlog/logging for machine-readable audit lines. Python 3.10+ supplies the match/case and X | None syntax used below. The governing compliance references are the federal FIFRA pesticide worker-safety and recordkeeping rules from the EPA and the USDA-administered federal pesticide recordkeeping program for restricted-use products; both expect a continuous trail from a received lot to the acre it was applied on.

Architecture of this subsystem

The service is a four-stage transform: a receipt line is validated, its quantity is normalized to a canonical unit as an exact Decimal, the resulting lot is posted to a FIFO ledger keyed by lot number, and the ledger’s book balance is periodically reconciled against a measured on-hand count with any variance beyond tolerance flagged for review. Every stage emits a structured log line, and every stage has an explicit conservative outcome — hold for review, never a silent default — when it cannot proceed. A pesticide line with no EPA registration number is rejected at stage one; it is never assigned a placeholder and carried forward.

The inputs are a receipt line dict plus scalar context (a tolerance and a canonical-unit table); the outputs are a validated NormalizedLot posted to the ledger, or an exception record. Integration is one-directional: this service consumes procurement records and produces a reconciled stock ledger that the compliance and reporting layers of the parent pipeline read — it never writes back to the ERP.

Step-by-step implementation

The reference service below is one ingest-and-reconcile module with three composable stages. It targets Python 3.10+ and passes a syntax and indentation check.

Step 1 — Validate each receipt line

Procurement rows are the first place a lot number or an EPA registration is trusted, so validation is strict: a pesticide line without a registration number is rejected outright, quantities must be positive, and timestamps must be timezone-aware. Nothing is defaulted into place.

Step 2 — Normalize units and quantities to canonical Decimals

Each source unit is mapped to a canonical unit (mass in kilograms, liquid volume in litres, count in seeds) through an exact Decimal conversion factor, then quantized with an auditable rounding rule. This is where a gallon becomes 3.785411784 litres deterministically instead of being stored as an ambiguous number.

Step 3 — Post lots to the ledger and reconcile against draw-down

Normalized lots are posted to a FIFO ledger keyed by lot number; applications draw stock down oldest-lot-first so each consumed unit is attributed to a specific received batch. A periodic reconciliation compares the ledger’s book balance against a measured physical count and flags any variance beyond a relative tolerance.

import logging
import uuid
from collections import defaultdict, deque
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

logger = logging.getLogger("inventory.ingest")


class ProductClass(str, Enum):
    SEED = "seed"
    FERTILIZER = "fertilizer"
    PESTICIDE = "pesticide"


# Canonical units: mass in kilograms, liquid volume in litres, count in seeds.
# Every factor is an exact Decimal; a regulated quantity never touches float.
UNIT_TO_CANONICAL: dict[str, tuple[str, Decimal]] = {
    "kg": ("kg", Decimal("1")),
    "lb": ("kg", Decimal("0.45359237")),
    "oz": ("kg", Decimal("0.028349523125")),
    "ton": ("kg", Decimal("907.18474")),        # US short ton
    "l": ("L", Decimal("1")),
    "gal": ("L", Decimal("3.785411784")),        # US liquid gallon
    "qt": ("L", Decimal("0.946352946")),
    "fl_oz": ("L", Decimal("0.0295735295625")),
    "seed": ("seed", Decimal("1")),
    "k_seed": ("seed", Decimal("1000")),
}


class ReceiptLine(BaseModel):
    """One validated procurement line before unit normalization."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    receipt_id: str
    product_name: str
    product_class: ProductClass
    epa_reg_no: str | None = None
    lot_number: str
    quantity: Decimal = Field(gt=0)
    unit: str
    received_at: datetime

    @field_validator("received_at")
    @classmethod
    def _tz_aware_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("received_at must be timezone-aware")
        return v.astimezone(timezone.utc)

    @field_validator("unit")
    @classmethod
    def _known_unit(cls, v: str) -> str:
        key = v.strip().lower()
        if key not in UNIT_TO_CANONICAL:
            raise ValueError(f"unrecognized unit: {v!r}")
        return key

    @model_validator(mode="after")
    def _epa_reg_required_for_pesticide(self) -> "ReceiptLine":
        # FIFRA recordkeeping requires the EPA registration number on every
        # pesticide receipt; a missing value is a hard reject, not a default.
        if self.product_class is ProductClass.PESTICIDE and not self.epa_reg_no:
            raise ValueError("pesticide receipt missing EPA registration number")
        return self


class NormalizedLot(BaseModel):
    """A receipt after conversion to a canonical unit and exact quantity."""

    model_config = ConfigDict(frozen=True)

    lot_number: str
    product_name: str
    product_class: ProductClass
    epa_reg_no: str | None
    canonical_unit: str
    canonical_qty: Decimal
    received_at: datetime


def normalize_receipt(line: ReceiptLine) -> NormalizedLot:
    """Convert one receipt line to a canonical-unit, quantized Decimal lot."""
    canonical_unit, factor = UNIT_TO_CANONICAL[line.unit]
    # Quantize in the canonical unit; ROUND_HALF_UP is deterministic and auditable.
    qty = (line.quantity * factor).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
    logger.info(
        '{"event":"receipt_normalized","lot":"%s","from":"%s %s","to":"%s %s"}',
        line.lot_number, line.quantity, line.unit, qty, canonical_unit,
    )
    return NormalizedLot(
        lot_number=line.lot_number,
        product_name=line.product_name,
        product_class=line.product_class,
        epa_reg_no=line.epa_reg_no,
        canonical_unit=canonical_unit,
        canonical_qty=qty,
        received_at=line.received_at,
    )


class StockReconciliation(BaseModel):
    """Book-versus-measured comparison for one product's on-hand balance."""

    product_name: str
    canonical_unit: str
    book_on_hand: Decimal
    measured_on_hand: Decimal
    variance: Decimal
    relative_variance: Decimal
    within_tolerance: bool


class LotLedger:
    """On-hand stock per product, tracked as FIFO layers keyed by received lot."""

    def __init__(self) -> None:
        # product_name -> queue of [lot_number, remaining_qty, canonical_unit]
        self._layers: dict[str, deque[list[Any]]] = defaultdict(deque)
        self._received: dict[str, NormalizedLot] = {}

    def receive(self, lot: NormalizedLot) -> None:
        if lot.lot_number in self._received:
            # Idempotent: a replayed receipt reconciles to the same lot, never a duplicate.
            logger.warning('{"event":"duplicate_receipt","lot":"%s"}', lot.lot_number)
            return
        self._received[lot.lot_number] = lot
        self._layers[lot.product_name].append(
            [lot.lot_number, lot.canonical_qty, lot.canonical_unit]
        )

    def on_hand(self, product_name: str) -> Decimal:
        return sum((layer[1] for layer in self._layers[product_name]), Decimal("0"))

    def draw_down(self, product_name: str, qty: Decimal) -> list[tuple[str, Decimal]]:
        """Consume qty FIFO across lot layers; return (lot_number, amount) allocations."""
        if qty <= 0:
            raise ValueError("draw-down quantity must be positive")
        if qty > self.on_hand(product_name):
            raise ValueError(f"{product_name}: draw-down {qty} exceeds on-hand stock")
        allocations: list[tuple[str, Decimal]] = []
        layers = self._layers[product_name]
        remaining = qty
        while remaining > 0:
            layer = layers[0]  # oldest lot first
            take = min(layer[1], remaining)
            allocations.append((layer[0], take))
            layer[1] -= take
            remaining -= take
            if layer[1] == 0:
                layers.popleft()  # lot fully consumed, drop the layer
        return allocations

    def reconcile_stock(
        self,
        product_name: str,
        measured_on_hand: Decimal,
        tolerance: Decimal = Decimal("0.02"),
    ) -> StockReconciliation:
        """Compare the book balance to a measured count; flag variance beyond tolerance."""
        correlation_id = str(uuid.uuid4())
        book = self.on_hand(product_name)
        unit = self._layers[product_name][0][2] if self._layers[product_name] else "n/a"
        variance = (book - measured_on_hand).copy_abs()
        relative = (variance / book) if book else Decimal("0")
        ok = relative <= tolerance
        logger.info(
            '{"event":"stock_reconciled","cid":"%s","product":"%s","book":"%s",'
            '"measured":"%s","variance":"%s","within_tolerance":%s}',
            correlation_id, product_name, book, measured_on_hand,
            variance, str(ok).lower(),
        )
        return StockReconciliation(
            product_name=product_name,
            canonical_unit=unit,
            book_on_hand=book,
            measured_on_hand=measured_on_hand,
            variance=variance,
            relative_variance=relative,
            within_tolerance=ok,
        )


def ingest_receipt(ledger: LotLedger, raw: dict[str, Any]) -> NormalizedLot | None:
    """Validate, normalize, and post one receipt; return None if it is an exception."""
    try:
        line = ReceiptLine(**raw)
    except Exception as exc:  # pydantic ValidationError or a value guard
        logger.error(
            '{"event":"receipt_exception","receipt":"%s","reason":"%s"}',
            raw.get("receipt_id", "unknown"), str(exc).replace('"', "'"),
        )
        return None  # held for review; never coerced into the ledger
    lot = normalize_receipt(line)
    ledger.receive(lot)
    return lot

Expected log output when a fertilizer receipt in US short tons is ingested and later reconciled cleanly:

{"event": "receipt_normalized", "lot": "FERT-22B", "from": "12 ton", "to": "10886.2169 kg"}
{"event": "stock_reconciled", "cid": "b91c…", "product": "urea-46-0-0", "book": "10886.2169", "measured": "10870.0000", "variance": "16.2169", "within_tolerance": true}

Expected log output when a pesticide receipt arrives without its EPA registration number:

{"event": "receipt_exception", "receipt": "RCV-4471", "reason": "pesticide receipt missing EPA registration number"}

Edge cases and known failure modes

Compliance and audit integration

Every normalized lot and every reconciliation is a compliance artifact, so each stage feeds the append-only inventory ledger the parent pipeline maintains. Three facts must survive from receipt to inspection: the product’s EPA registration and lot number as received, the canonical quantity and the exact conversion that produced it, and the draw-down allocations that tie each application back to a specific lot. Because the recordkeeping rules originate in the EPA/USDA Rule Mapping threshold set rather than being hard-coded here, a rule change never rewrites history — a lot received last season reconciles against the contract that was in force when it arrived.

The federal FIFRA framework and the EPA’s pesticide worker-safety and recordkeeping rules require that a restricted-use pesticide application be traceable to the product actually used, and the USDA-administered pesticide recordkeeping program expects that trail to be producible on request. That is precisely what the lot ledger guarantees: because draw-downs consume specific lot layers FIFO, an auditor asking which batch of a product landed on a given field gets a deterministic answer, and a manufacturer recall can be scoped to exactly the fields whose applications drew from the suspect lot. Ledger entries are hash-anchored and never edited in place; a correction is a new entry, so the quantity history stays defensible. Structured logs should stream to a centralized observability platform so exception rate, reconciliation-variance distribution, and negative-stock attempts are monitored as pipeline-health signals.

Verification

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

1. Golden receipt. Ingest a well-formed fertilizer line in US short tons and assert on_hand returns the exact Decimal kilogram equivalent and a receipt_normalized ledger entry is written. A unit-factor regression fails here immediately.
2. Inject a missing EPA registration. Submit a pesticide line with epa_reg_no=None and assert ingest_receipt returns None, logs receipt_exception, and posts nothing — the line is held, not defaulted.
3. Inject an unknown unit. Submit a receipt with unit="drums" and assert it is held as an exception rather than stored with an assumed conversion.
4. Draw down and reconcile. Receive two lots, draw an application FIFO across them, then reconcile the book balance against a measured count inside and outside the 2% tolerance; assert within_tolerance flips exactly at the boundary and the allocations attribute each consumed unit to the correct lot.

A run passes only when each injected fault produces the expected held-but-safe result and the ledger contains one traceable entry per stage. The recall-traceability and tank-mix reconciliation cases are covered in depth in the two guides below.

Frequently Asked Questions

Why store quantities as Decimal instead of float? Because inventory has to balance to the unit, and binary floating point cannot represent common decimal quantities exactly, so float accumulates rounding drift that makes a reconciliation never quite close. Every quantity in this service is a Decimal with an explicit quantize step, so a receipt in gallons converts to litres by an exact factor and the books balance deterministically across worker nodes.

What happens to a pesticide receipt with no EPA registration number? It is held as an exception, not stored with a placeholder. FIFRA recordkeeping requires the registration number on every pesticide, so the model_validator rejects the line and it is routed for review, where the number is re-sourced from the product label. Assigning a default would produce a record that looks compliant but cannot be tied to a registered product during an audit.

How does the ledger tie an application to a specific received lot? Stock is stored as FIFO layers keyed by lot number, and a draw-down consumes the oldest layer first, returning the exact (lot_number, amount) allocations it used. So every consumed unit is attributed to a real received batch, which is what makes recall scoping and pesticide lot traceability possible.

How is a slow inventory discrepancy caught? By reconciling the ledger’s book balance against a measured physical count and flagging any relative variance beyond tolerance. A small persistent gap usually means unrecorded applications or shrinkage; catching it before the season’s carry-forward keeps the next year’s opening balance honest.

Related

• Tracking Pesticide Lot Numbers for Traceability — the lot-ledger and draw-down accounting that links every application to a received batch.
• Reconciling Tank-Mix Batches Against Inventory — decomposing a spray batch into per-product draw-downs and flagging usage variance.
• Equipment Telemetry Parsing — the sibling subsystem that turns machine payloads into the application events this ledger draws down against.
• EPA/USDA Rule Mapping — the versioned rule set supplying the recordkeeping fields each lot carries.
• Field Schema Design — governs the canonical lot and reconciliation contract this service targets.

Up: Farm Data Ingestion & Field Boundary Synchronization