Rotation Constraint Modeling

Rotation constraint modeling is the validation layer of the planning platform: it turns the informal agronomy an operator carries in their head — don’t put sugarbeet where atrazine ran last year, break the soybean cyst-nematode cycle, don’t grow corn on corn on corn forever — into machine-readable, versioned constraints that a proposed multi-year rotation can be checked against deterministically. This subsystem lives inside Season Planning & Crop Rotation Optimization and answers a single question before any plan is committed: does this sequence of crops, on these fields, violate any rule we are obligated to honor — and if so, which rule, and where? A plan that passes silently but breaks a herbicide plant-back interval does not fail loudly at planning time; it fails eighteen months later as a stunted replant crop and a rejected conservation payment.

Problem framing: rotation rules are heterogeneous, versioned, and enforceable

The rules that govern a legal, agronomically sound rotation come from incompatible sources and change on their own schedules. A herbicide plant-back interval is printed on an EPA-approved product label and is legally binding. A disease-break sequence is agronomic best practice, revised as pathogen pressure shifts. A maximum-consecutive-crop limit and a cover-crop window are often written into an NRCS conservation plan the grower has signed. A nitrogen cap comes from a nutrient management plan. Encoding these as scattered if statements across planning code guarantees that when a label revision or a plan renewal lands, some checks are updated and others are quietly missed.

The sub-problem this subsystem owns is representing every one of those rules as an addressable, versioned object — a constraint with a stable rule_id, a citation to the authority that issued it, and an effective date — so that validating a rotation is a deterministic pass over a known rule set rather than a walk through hand-written logic. The output is not a boolean. It is a list of typed violations, each naming the field, the season, the rule it broke, and the source that rule came from, so an agronomist can act on it and an auditor can trace it. This is the planning-time analogue of the runtime checks in EPA/USDA Rule Mapping: the same rules that reject a non-compliant application at spray time should reject a non-compliant sequence at plan time, before a seed is ordered.

Prerequisites and dependencies

The validator sits between a rotation optimizer and the plan store, so it depends on three upstream contracts. Confirm each before wiring it in.

A crop calendar. Each field-season carries a crop code and, where known, a planting date. Growth-stage semantics and the BBCH scale that anchor timing come from Growth Stage Mapping; the validator consumes crop identity, not phenology, but the two share the same canonical crop vocabulary.
A versioned rule registry. Plant-back tables, disease-break intervals, consecutive limits, and nutrient caps are governed centrally and stamped with a ruleset_version, never hard-coded inline. This mirrors the threshold registry that EPA/USDA Rule Mapping maintains for application-time compliance.
An application history. Prior herbicide applications per field-season are needed to evaluate carryover; those records originate in the ingestion and audit trail rather than being re-entered by the planner.

Library versions are pinned so evaluation is reproducible across planner nodes: pydantic>=2.5 for typed models with Decimal rate fields, Python 3.10+ for match/case and X | None unions, and the standard-library decimal and logging modules. The two focused guides below extend this base — one models herbicide carryover restrictions in depth against a real plant-back table, and one uses a CP-SAT solver to model cover-crop rotations under the same conservation constraints.

Architecture of this subsystem

The validator is a deterministic three-part transform: a proposed rotation (typed field-season rows) is checked by an evaluator that loads a versioned rule set, each rule inspects the ordered sequence and emits zero or more typed violations, and the collected violations are sorted into a stable report carrying a pass/fail verdict. Every rule evaluation and the final verdict write a structured line to an append-only audit ledger, so a plan’s approval is reconstructable from the exact rule set that judged it.

The inputs are an ordered list of typed field-season rows plus a rule set selected by version; the outputs are a ValidationReport carrying a verdict and a sorted list of Violation objects. The integration is one-directional: this service reads the proposed rotation and the rule registry and produces a report that the optimizer and the plan-approval workflow consume — it never mutates the plan or the rules.

Step-by-step implementation

The reference validator below models the proposed rotation, encodes each rule as a small addressable class, and evaluates them deterministically into typed violations. It targets Python 3.10+ and passes a syntax and indentation check.

Step 1 — Model the proposed rotation as typed field-seasons

Each field-season is a pydantic model carrying the crop, any herbicides applied that season, the planned nitrogen rate as a Decimal (never a float — a regulated rate must round predictably), and the cover crop if one is scheduled. The rules operate on the ordered sequence of these rows per field.

Parameter	Type	Default	Purpose
`field_id`	`str`	—	Stable field identifier; rules group and order rows by it.
`year`	`int`	—	Season year; the ordering key within a field.
`crop`	`str`	—	Canonical crop code (e.g. `corn`, `soybean`, `sugarbeet`).
`herbicides`	`list[str]`	`[]`	Active-ingredient codes applied that season; drives carryover checks.
`n_applied_lb_ac`	`Decimal`	`0`	Planned nitrogen rate in lb/ac, checked against the 590 cap.
`cover_crop`	`str \| None`	`None`	Cover species if scheduled; absence in a fallow window is flagged.

Step 2 — Encode each rule with a `rule_id` and a citation

Every rule is a subclass of RotationRule with a stable rule_id, a citation URL pointing at the issuing authority, and an evaluate method that returns typed Violation objects. A rule that finds nothing returns an empty list; it never mutates the plan and never raises for a mere policy breach.

Parameter	Type	Default	Purpose
`PLANT_BACK_MONTHS`	`dict`	label table	Active ingredient → rotational crop → months that must elapse.
`HOST_BREAK`	`dict`	agronomic table	Crop → (disease family, minimum break years).
`limits`	`dict[str, int]`	per-crop	Maximum consecutive years a crop may occupy a field.
`crop_n_cap_lb_ac`	`dict[str, Decimal]`	590 plan	Per-crop nitrogen ceiling from the nutrient management plan.

Step 3 — Evaluate deterministically and emit typed violations

The validator sorts the input, runs each rule, collects the violations, and sorts the result so an identical plan always yields an identical report. The verdict is ok only when no violation carries BLOCK severity; WARN violations surface agronomic risk without rejecting the plan outright.

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from enum import Enum
from typing import Sequence

from pydantic import BaseModel, Field

logger = logging.getLogger("rotation.constraints")


class Severity(str, Enum):
    BLOCK = "block"   # rotation is rejected until the violation is resolved
    WARN = "warn"     # agronomic risk; admissible only with a documented override


class CropSeason(BaseModel):
    """One field-season in a proposed multi-year rotation."""

    field_id: str
    year: int
    crop: str                                   # canonical crop code
    planted_on: date | None = None
    herbicides: list[str] = Field(default_factory=list)
    cover_crop: str | None = None
    n_applied_lb_ac: Decimal = Decimal("0")     # Decimal, never float, for a regulated rate


@dataclass(frozen=True)
class Violation:
    rule_id: str
    severity: Severity
    field_id: str
    year: int
    message: str
    citation: str        # authoritative URL the rule derives from


class RotationRule:
    """Base class: inspect the ordered field-season sequence, return typed violations."""

    rule_id: str = "BASE"
    citation: str = ""

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        raise NotImplementedError

    @staticmethod
    def _by_field(seasons: Sequence[CropSeason]) -> dict[str, list[CropSeason]]:
        grouped: dict[str, list[CropSeason]] = {}
        for s in seasons:
            grouped.setdefault(s.field_id, []).append(s)
        return {fid: sorted(rows, key=lambda s: s.year) for fid, rows in grouped.items()}


class PlantBackIntervalRule(RotationRule):
    """EPA label carryover: a plant-back interval must elapse before a rotational crop."""

    rule_id = "CARRYOVER-PBI"
    citation = "https://www.epa.gov/pesticide-labels"

    # active ingredient -> {rotational crop: months that must elapse after application}
    PLANT_BACK_MONTHS = {
        "atrazine": {"soybean": 12, "sugarbeet": 24, "canola": 24},
        "imazethapyr": {"corn": 9, "canola": 26, "sugarbeet": 40},
        "clopyralid": {"soybean": 10, "sugarbeet": 18, "canola": 18},
    }

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        out: list[Violation] = []
        for field_id, rows in self._by_field(seasons).items():
            for prior, following in zip(rows, rows[1:]):
                for ai in prior.herbicides:
                    required = self.PLANT_BACK_MONTHS.get(ai, {}).get(following.crop)
                    if required is None:
                        continue
                    elapsed_months = (following.year - prior.year) * 12
                    if elapsed_months < required:
                        out.append(Violation(
                            self.rule_id, Severity.BLOCK, field_id, following.year,
                            f"{following.crop} planted {elapsed_months} mo after {ai}; "
                            f"label requires a {required} mo plant-back interval",
                            self.citation,
                        ))
        return out


class DiseaseBreakRule(RotationRule):
    """Disease-break sequencing: a host crop needs an N-year break from its family."""

    rule_id = "DISEASE-BREAK"
    citation = "https://www.nrcs.usda.gov/resources/guides-and-instructions/conservation-crop-rotation-ac-code-328"

    # crop -> (disease family, minimum break years between plantings)
    HOST_BREAK = {
        "soybean": ("legume-scn", 2),          # soybean cyst nematode
        "canola": ("brassica-sclerotinia", 3),
        "sugarbeet": ("beet-rhizomania", 3),
    }

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        out: list[Violation] = []
        for field_id, rows in self._by_field(seasons).items():
            last_seen: dict[str, int] = {}
            for s in rows:
                spec = self.HOST_BREAK.get(s.crop)
                if spec is None:
                    continue
                family, break_years = spec
                prev = last_seen.get(family)
                if prev is not None and (s.year - prev) < break_years:
                    out.append(Violation(
                        self.rule_id, Severity.BLOCK, field_id, s.year,
                        f"{s.crop} replanted {s.year - prev} yr after {family}; "
                        f"needs a {break_years} yr disease break",
                        self.citation,
                    ))
                last_seen[family] = s.year
        return out


class MaxConsecutiveCropRule(RotationRule):
    """Limit how many years a crop may occupy a field back to back."""

    rule_id = "MAX-CONSECUTIVE"
    citation = "https://www.nrcs.usda.gov/resources/guides-and-instructions/conservation-crop-rotation-ac-code-328"

    def __init__(self, limits: dict[str, int]) -> None:
        self.limits = limits

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        out: list[Violation] = []
        for field_id, rows in self._by_field(seasons).items():
            run_crop: str | None = None
            run_len = 0
            for s in rows:
                run_len = run_len + 1 if s.crop == run_crop else 1
                run_crop = s.crop
                limit = self.limits.get(s.crop)
                if limit is not None and run_len > limit:
                    out.append(Violation(
                        self.rule_id, Severity.WARN, field_id, s.year,
                        f"{s.crop} grown {run_len} yr running; plan limit is {limit}",
                        self.citation,
                    ))
        return out


class CoverCropWindowRule(RotationRule):
    """Flag a bare-fallow window that a cover crop should occupy (NRCS 340)."""

    rule_id = "NRCS-340-COVER"
    citation = "https://www.nrcs.usda.gov/resources/guides-and-instructions/cover-crop-ac-code-340"

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        out: list[Violation] = []
        for field_id, rows in self._by_field(seasons).items():
            for prior, following in zip(rows, rows[1:]):
                gap = following.year - prior.year
                if gap > 1 and prior.cover_crop is None:
                    out.append(Violation(
                        self.rule_id, Severity.WARN, field_id, prior.year,
                        f"{gap - 1} yr bare-fallow window after {prior.crop} "
                        f"with no cover crop scheduled",
                        self.citation,
                    ))
        return out


class NutrientLimitRule(RotationRule):
    """Cap planned nitrogen at the NRCS 590 nutrient management plan rate."""

    rule_id = "NRCS-590-N"
    citation = "https://www.nrcs.usda.gov/resources/guides-and-instructions/nutrient-management-ac-code-590"

    def __init__(self, crop_n_cap_lb_ac: dict[str, Decimal]) -> None:
        self.caps = crop_n_cap_lb_ac

    def evaluate(self, seasons: Sequence[CropSeason]) -> list[Violation]:
        out: list[Violation] = []
        for s in seasons:
            cap = self.caps.get(s.crop)
            if cap is not None and s.n_applied_lb_ac > cap:
                out.append(Violation(
                    self.rule_id, Severity.BLOCK, s.field_id, s.year,
                    f"planned N {s.n_applied_lb_ac} lb/ac on {s.crop} exceeds "
                    f"the 590 plan cap of {cap} lb/ac",
                    self.citation,
                ))
        return out


@dataclass
class ValidationReport:
    ok: bool
    ruleset_version: str
    violations: list[Violation]

    @property
    def blocking(self) -> list[Violation]:
        return [v for v in self.violations if v.severity is Severity.BLOCK]


class RotationValidator:
    """Deterministically evaluate a proposed rotation against a versioned rule set."""

    def __init__(self, rules: Sequence[RotationRule], ruleset_version: str) -> None:
        self.rules = rules
        self.ruleset_version = ruleset_version

    def validate(self, seasons: Sequence[CropSeason]) -> ValidationReport:
        # Sort input so an identical plan always yields an identical report.
        ordered = sorted(seasons, key=lambda s: (s.field_id, s.year))
        violations: list[Violation] = []
        for rule in self.rules:
            found = rule.evaluate(ordered)
            violations.extend(found)
            logger.info(
                '{"event":"rule_evaluated","rule_id":"%s","ruleset":"%s","violations":%d}',
                rule.rule_id, self.ruleset_version, len(found),
            )
        # Stable report order regardless of rule execution order.
        violations.sort(key=lambda v: (v.field_id, v.year, v.rule_id))
        ok = not any(v.severity is Severity.BLOCK for v in violations)
        logger.info(
            '{"event":"rotation_validated","ruleset":"%s","ok":%s,"violations":%d}',
            self.ruleset_version, str(ok).lower(), len(violations),
        )
        return ValidationReport(ok, self.ruleset_version, violations)


def build_default_validator() -> RotationValidator:
    return RotationValidator(
        rules=[
            PlantBackIntervalRule(),
            DiseaseBreakRule(),
            MaxConsecutiveCropRule(limits={"corn": 3, "soybean": 1}),
            CoverCropWindowRule(),
            NutrientLimitRule(crop_n_cap_lb_ac={
                "corn": Decimal("200"), "soybean": Decimal("20"),
            }),
        ],
        ruleset_version="2026.02",
    )


if __name__ == "__main__":
    plan = [
        CropSeason(field_id="N-12", year=2026, crop="corn",
                   herbicides=["atrazine"], n_applied_lb_ac=Decimal("185")),
        CropSeason(field_id="N-12", year=2027, crop="sugarbeet"),  # 24-mo plant-back, 12 elapsed
    ]
    report = build_default_validator().validate(plan)
    for v in report.blocking:
        print(f"[{v.severity.value}] {v.rule_id} {v.field_id}/{v.year}: {v.message} -> {v.citation}")

Expected output when the sugarbeet follows atrazine inside the plant-back window:

[block] CARRYOVER-PBI N-12/2027: sugarbeet planted 12 mo after atrazine; label requires a 24 mo plant-back interval -> https://www.epa.gov/pesticide-labels

The corresponding audit lines emitted during that run:

{"event": "rule_evaluated", "rule_id": "CARRYOVER-PBI", "ruleset": "2026.02", "violations": 1}
{"event": "rotation_validated", "ruleset": "2026.02", "ok": false, "violations": 1}

Edge cases and known failure modes

Condition	Symptom	Fix
Plant-back interval measured in years, not seasons	A 12-month interval passes when the crops are only one growing season apart	Track the application month, not just the year; the herbicide carryover guide refines the interval to real dates and rainfall thresholds.
Nitrogen rate entered as a float	Rate compares as `199.99999` and slips under a `200` cap	Bind `n_applied_lb_ac` as `Decimal`; the model rejects a lossy float before comparison.
Field rows arrive out of year order	Consecutive-crop and disease-break runs miscount	The validator sorts per field before evaluating; never assume caller ordering.
Two rules disagree on the same season	A plan is both `WARN` and `BLOCK` for one field-year	Report keeps every violation; the verdict is driven only by `BLOCK`, so a warning never masks a hard stop.
Rule set updated mid-season	An approved plan re-validates differently later	Stamp every report with `ruleset_version`; re-validation runs against the version in force at approval, not today’s.
Crop code not present in any rule table	A genuinely risky sequence passes silently	Treat an unknown crop as unmodeled, not safe: emit a `WARN` from a catch-all rule so the gap is visible rather than assumed clean.
Cover-crop window flagged on an intentional set-aside	A CRP or fallow enrollment trips `NRCS-340-COVER`	Carry a `set_aside` flag on the field-season and skip the cover-crop rule for enrolled acres.

Compliance and audit integration

Every rule the validator carries traces to a document the grower is answerable to, and the report is the artifact that proves the plan honored it. Herbicide plant-back intervals are legally binding pesticide label restrictions; the citation on a CARRYOVER-PBI violation points at the label system so a reviewer can confirm the exact interval. Consecutive-crop limits and disease breaks are typically written into an NRCS Conservation Crop Rotation (Code 328) practice standard, cover-crop windows into Cover Crop (Code 340), and nitrogen caps into Nutrient Management (Code 590). Because each rule stamps its own citation onto the violation, a rejected plan is self-documenting: an agronomist sees which authority the sequence offends without hunting through a policy binder.

The audit contract is the same one the runtime enforcement layer honors in EPA/USDA Rule Mapping. Three facts must survive from validation to inspection: the ruleset_version the plan was judged against, the verdict, and every violation with its rule_id and citation. Ledger entries are append-only, so a plan approved under 2026.02 is never retroactively re-judged by a later rule set — it replays against the version that was in force. This is what makes a conservation-payment claim or an EPA recordkeeping response defensible: the plan was validated, the verdict was recorded, and the rules that produced it are pinned in time.

Verification

Confirm correct operation in staging with reproducible fault injection before promoting a change to the rule set:

Clean plan. Feed a compliant four-year corn-soybean-corn-soybean rotation with in-cap nitrogen and assert validate returns ok=True with zero violations and one rotation_validated ledger entry marked ok: true.
Inject a carryover break. Apply atrazine in year one and plant sugarbeet in year two and assert a single CARRYOVER-PBI BLOCK violation with the 24-month message and the EPA citation, and ok=False.
Inject a disease-break break. Plant soybean in consecutive years on one field and assert a DISEASE-BREAK violation naming legume-scn with the required break years.
Inject an over-cap nitrogen rate. Set a corn season to 250 lb/ac against a 200 cap and assert an NRCS-590-N BLOCK violation citing the 590 standard.
Confirm determinism. Shuffle the input row order and assert the report — verdict and violation list — is byte-identical across runs.

A run passes only when each injected fault produces the expected typed violation, the verdict flips only on BLOCK severity, and the ledger holds one traceable entry per rule plus the final verdict. Deeper carryover modeling and solver-based cover-crop selection are verified in their own guides below.

Frequently Asked Questions

Why return typed violations instead of a boolean pass/fail? Because a planner needs to know what to change, and an auditor needs to know which rule was enforced. A boolean tells you a plan failed; a Violation names the field, the season, the rule_id, and the citation, so the optimizer can revise exactly that field-year and the record proves the plan was checked against a real authority.

How does versioning the rule set matter at planning time? Rules change — a label revision shortens a plant-back interval, a renewed conservation plan tightens a nitrogen cap. Stamping every report with a ruleset_version means a plan approved months ago is re-validated against the rules that were in force when it was approved, not against today’s, so an approval is never silently invalidated by an unrelated policy update.

What is the difference between a BLOCK and a WARN violation? A BLOCK is a hard stop — a legal plant-back interval or an over-cap nitrogen rate — that makes the plan invalid until resolved. A WARN flags agronomic risk, such as one extra consecutive year of corn, that a grower may knowingly accept. The verdict is driven only by BLOCK, so a warning never blocks a plan and a block is never downgraded by a warning.

Can the same rules run at application time, not just at planning? Yes, and they should. The plant-back and nutrient rules encoded here are the planning-time mirror of the runtime checks in the EPA/USDA rule mapping layer. Encoding them once as versioned constraints lets the same authority govern both the sequence you plan and the application you make.

Encoding Herbicide Carryover Restrictions as Constraints — the plant-back interval check refined against a real atrazine, imazethapyr, and clopyralid table.
Modeling Cover-Crop Rotations with Constraint Solvers — CP-SAT species selection under termination-window, diversity, and NRCS conservation constraints.
EPA/USDA Rule Mapping — the runtime threshold registry these planning rules mirror.
Growth Stage Mapping — the BBCH crop-calendar semantics the field-season vocabulary shares.

Up: Season Planning & Crop Rotation Optimization

Rotation Constraint Modeling

Subtopics