Route Schema Design for Waste Management Logistics

The route schema is the immutable data contract between municipal dispatch systems and the fleet optimization engine. When it is loose — optional fields, implicit types, unbounded coordinates — the solver ingests uncertainty and returns confidently wrong routes: a manifest with a silently coerced null weight, a service stop dropped because its timestamp was naive rather than UTC, a facility node that no longer exists. This reference defines the contract that prevents those failures, so that every route object reaching the solver has already been proven structurally valid, regulator-aligned, and byte-for-byte reproducible. It sits directly under the Core Architecture & Compliance Mapping framework, which requires that the same input always produces the same audit hash; a schema that permits ambiguity makes that guarantee impossible.

The design divides the payload into four isolated domains — geospatial nodes, service intervals, asset telemetry, and compliance metadata — each with explicit type constraints and cross-referenced validation rules so that state corruption cannot leak from one domain into another during optimization.

Prerequisites

This page targets Python 3.10+ and models the contract with Pydantic v2 in strict, frozen mode so that validated route objects are immutable graph inputs. Pin the following before proceeding:

python -m pip install \
  pydantic==2.7.1 \
  python-dateutil==2.9.0.post0 \
  jsonschema==4.22.0

Every field below has a defined unit, range, and nullability. The validated object is the only thing the downstream OR-Tools Implementation is permitted to read; anything that fails validation is rejected at the ingestion edge and never becomes a graph vertex. The four domains carry these fields:

Coordinates require a minimum of six decimal places (roughly 0.11 m precision) so that dense urban navigation is not thrown off by GPS drift; the reconciliation of raw pings into stable coordinates happens upstream in the Telematics & Sensor Data Ingestion pipeline before a node ever reaches this schema, and the range checks here are the last line of defense against a drifted fix slipping through.

Core Implementation

The contract is built in four steps: define the typed domains and enums, assemble the RouteNode model with field validators, wrap the whole route in a container that enforces cross-node invariants and a payload hash, then run every incoming payload through a single validation gate that logs structured diagnostics.

Step 1 — Typed domains and node enums

Node type and waste stream are closed enumerations, not free strings. A closed set makes an illegal value impossible to represent rather than merely discouraged.

from enum import Enum


class NodeType(str, Enum):
    DEPOT = "depot"
    SERVICE_POINT = "service_point"
    WEIGH_STATION = "weigh_station"
    DISPOSAL_FACILITY = "disposal_facility"


class WasteStream(str, Enum):
    MSW = "msw"            # municipal solid waste
    RECYCLING = "recycling"
    ORGANICS = "organics"
    HAZARDOUS = "hazardous"  # EPA RCRA-regulated
    MEDICAL = "medical"

Step 2 — The RouteNode model with field validators

Each node is strict and frozen: extra keys are rejected, types are never coerced, and a validated node cannot be mutated after construction. The latitude, longitude, and UTC validators reject the three failure modes that cause the most production incidents — out-of-range coordinates, naive timestamps, and non-UTC offsets.

from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict, field_validator, model_validator


class RouteNode(BaseModel):
    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    sequence_index: int
    node_type: NodeType
    latitude: float
    longitude: float
    elevation_meters: float | None = None
    service_window_start: datetime
    service_window_end: datetime
    waste_stream_code: WasteStream
    max_payload_kg: float
    collected_mass_kg: float = 0.0

    @field_validator("latitude")
    @classmethod
    def validate_latitude(cls, v: float) -> float:
        if not -90.0 <= v <= 90.0:
            raise ValueError("Latitude out of WGS84 range [-90, 90]")
        return v

    @field_validator("longitude")
    @classmethod
    def validate_longitude(cls, v: float) -> float:
        if not -180.0 <= v <= 180.0:
            raise ValueError("Longitude out of WGS84 range [-180, 180]")
        return v

    @field_validator("service_window_start", "service_window_end")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.utcoffset() != timezone.utc.utcoffset(v):
            raise ValueError("Temporal fields must be timezone-aware (UTC)")
        return v

    @model_validator(mode="after")
    def check_window_and_payload(self) -> "RouteNode":
        if self.service_window_end <= self.service_window_start:
            raise ValueError("service_window_end must be after service_window_start")
        if not 0.0 <= self.collected_mass_kg <= self.max_payload_kg:
            raise ValueError("collected_mass_kg must be within [0, max_payload_kg]")
        return self

Step 3 — The Route container: cross-node invariants and payload hash

A route is more than a list of valid nodes; the sequence must be contiguous, capacity must accumulate monotonically without exceeding the vehicle rating, and the whole payload must carry a version and a cryptographic hash so that a downgrade or a mid-flight mutation is caught at deserialization. The bind between accumulated mass and the vehicle rating is the schema-level expression of the Capacity & Weight Limits constraint the solver later enforces as a cumulative dimension.

import hashlib
import json
from typing import List


class Route(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    schema_version: str
    route_id: str
    vehicle_gvwr_kg: float
    nodes: List[RouteNode]

    @model_validator(mode="after")
    def check_topology(self) -> "Route":
        # Contiguous, zero-based, strictly increasing sequence — no orphaned nodes.
        expected = list(range(len(self.nodes)))
        actual = [n.sequence_index for n in self.nodes]
        if actual != expected:
            raise ValueError(f"Non-contiguous sequence: expected {expected}, got {actual}")

        # Capacity accumulates monotonically along the path and never exceeds GVWR.
        cumulative = 0.0
        for node in self.nodes:
            cumulative += node.collected_mass_kg
            if cumulative > self.vehicle_gvwr_kg:
                raise ValueError(
                    f"Cumulative load {cumulative}kg exceeds GVWR "
                    f"{self.vehicle_gvwr_kg}kg at stop {node.sequence_index}"
                )
        return self

    def payload_hash(self) -> str:
        """Deterministic SHA-256 over the canonical serialization."""
        canonical = self.model_dump(mode="json", exclude={"nodes": {"__all__": {"collected_mass_kg"}}})
        blob = json.dumps(canonical, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(blob.encode("utf-8")).hexdigest()

The cumulative-load check restates the gross-weight rule enforced in graph construction. Written as an inequality, the schema guarantees that for a route of $k$ stops with tare mass $W_{\text{tare}}$:

$$W_{\text{gross}} = W_{\text{tare}} + \sum_{i=1}^{k} m_i \le \text{GVWR}$$

where $m_i$ is the mass collected at the $i$-th stop. The schema cannot evaluate per-axle distribution — that requires wheelbase geometry the solver owns — but it guarantees the gross bound holds before the solver is invoked, so an infeasible-by-weight route never consumes solver time.

Step 4 — The validation gate with structured logging

Every payload passes through one gate. On failure it returns a granular diagnostic naming the exact field, expected type, and violation reason; success emits a hash-stamped record. The JSONFormatter below is the custom formatter used site-wide — the standard logging module has no built-in JSON formatter, so structured audit output requires one.

import logging
from pydantic import ValidationError


class JSONFormatter(logging.Formatter):
    """Custom JSON formatter — the standard logging module has no built-in JSONFormatter."""
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
        }
        log_obj.update({k: v for k, v in record.__dict__.items()
                        if k not in logging.LogRecord("", 0, "", 0, "", (), None).__dict__})
        return json.dumps(log_obj)


logger = logging.getLogger("route_schema")
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)
logger.setLevel(logging.INFO)


def validate_route(payload: dict) -> Route | None:
    try:
        route = Route.model_validate(payload)
    except ValidationError as exc:
        for err in exc.errors():
            logger.error(
                "schema_rejection",
                extra={
                    "route_id": payload.get("route_id", "unknown"),
                    "field": ".".join(str(p) for p in err["loc"]),
                    "violation": err["type"],
                    "detail": err["msg"],
                },
            )
        return None
    logger.info("schema_accepted", extra={
        "route_id": route.route_id,
        "node_count": len(route.nodes),
        "payload_hash": route.payload_hash(),
    })
    return route

Serialization back onto the wire uses model_dump(exclude_unset=True, by_alias=True) to strip fields the producer never set, keeping API payloads small during peak collection hours without dropping any field the solver requires.

Regulatory Mapping

Each schema field maps to a specific federal or state rule; the mapping is what lets an auditor trace a dispatched route back to the authority that governs it. These citations belong in the contract itself, not in prose commentary:

• service_window_start / service_window_end encode municipal collection-window and noise ordinances (local code, typically enforced under a city’s noise-control chapter). The solver consumes these as Time Window Constraints; a 05:40 arrival in a 07:00 quiet-hour zone is rejected here, before it can become a citation.
• vehicle_gvwr_kg maps to Gross Vehicle Weight Rating under 49 CFR 571 (Federal Motor Vehicle Safety Standards) and to posted per-axle and bridge limits under 23 CFR 658. The schema enforces the gross bound; the bridge-formula axle check is bound as a separate solver dimension.
• Driver on-duty and driving-time ceilings (Hours of Service, 49 CFR 395) are not stored on the node directly but constrain the total route duration derived from these windows; the field-by-field translation lives in the DOT/FMCSA Rule Mapping reference, where each solver parameter traces back to its CFR paragraph.
• waste_stream_code = hazardous triggers the EPA hazardous-waste manifest requirement under 40 CFR 262 Subpart B, whose e-Manifest is EPA Form 8700-22. Hazardous routing restrictions additionally fall under 49 CFR 397. The exact field mappings for automated e-Manifest submission are specified in the JSON Schema for Municipal Disposal Tracking child reference.
• node_type = disposal_facility carries the facility’s posted acceptance hours and daily tonnage cap; a route referencing a closed or capped facility fails pre-flight, which keeps drivers from arriving where the payload cannot legally be tipped.

Validation & Verification

Confirm the contract fires correctly with unit tests that assert both acceptance and rejection, and check the structured log fields the gate emits. A route with a non-contiguous sequence or an over-GVWR accumulation must be rejected; a well-formed route must return an object whose payload_hash() is stable across runs.

import pytest
from datetime import datetime, timezone


def _node(idx: int, mass: float, stream: str = "msw") -> dict:
    return {
        "sequence_index": idx,
        "node_type": "service_point",
        "latitude": 34.052235,
        "longitude": -118.243683,
        "service_window_start": datetime(2026, 7, 2, 14, 0, tzinfo=timezone.utc).isoformat(),
        "service_window_end": datetime(2026, 7, 2, 16, 0, tzinfo=timezone.utc).isoformat(),
        "waste_stream_code": stream,
        "max_payload_kg": 12000.0,
        "collected_mass_kg": mass,
    }


BASE = {"schema_version": "1.2.0", "route_id": "RT-A1B2C3", "vehicle_gvwr_kg": 15000.0}


def test_valid_route_is_accepted():
    payload = {**BASE, "nodes": [_node(0, 5000.0), _node(1, 4000.0)]}
    route = validate_route(payload)
    assert route is not None
    assert route.payload_hash() == route.payload_hash()  # deterministic


def test_over_gvwr_is_rejected():
    payload = {**BASE, "nodes": [_node(0, 9000.0), _node(1, 9000.0)]}
    assert validate_route(payload) is None  # 18000kg > 15000kg GVWR


def test_non_contiguous_sequence_is_rejected():
    payload = {**BASE, "nodes": [_node(0, 1000.0), _node(2, 1000.0)]}
    assert validate_route(payload) is None


def test_naive_timestamp_is_rejected():
    bad = _node(0, 1000.0)
    bad["service_window_start"] = "2026-07-02T14:00:00"  # no tzinfo
    assert validate_route({**BASE, "nodes": [bad]}) is None

In production, alert on the schema_rejection log event grouped by field: a spike in rejections on one field is an early signal that an upstream producer has drifted from the contract. The same discipline is applied to raw coordinate streams in the Schema Validation Pipelines reference, which validates GPS fixes before they are promoted to route nodes.

Failure Modes & Edge Cases

When a payload is unsatisfiable, the gate rejects it — it never silently repairs data, because a repaired route is an unauditable route. Recovery is an explicit, logged decision made after rejection, and it happens in the Fallback Routing Logic layer, not inside the schema. The two most common recoverable cases are an over-capacity route and an infeasible time window; both are handled by relaxing a soft parameter while every hard compliance boundary stays intact.

def widen_window(node: RouteNode, minutes: int) -> RouteNode:
    """Widen a service window symmetrically — used only when the window is a soft
    SLA target, never when it encodes a legal quiet-hour ordinance."""
    from datetime import timedelta
    return node.model_copy(update={
        "service_window_start": node.service_window_start - timedelta(minutes=minutes),
        "service_window_end": node.service_window_end + timedelta(minutes=minutes),
    })


def split_over_capacity(route: Route) -> list[list[RouteNode]]:
    """Split an over-GVWR route into GVWR-legal segments, each ending at a
    disposal_facility tip. Preserves node order; never drops a stop."""
    segments: list[list[RouteNode]] = [[]]
    cumulative = 0.0
    for node in route.nodes:
        if cumulative + node.collected_mass_kg > route.vehicle_gvwr_kg:
            segments.append([])
            cumulative = 0.0
        segments[-1].append(node)
        cumulative += node.collected_mass_kg
    return segments

Two edge cases deserve explicit handling. First, a node whose service_window_end sits before service_window_start after a daylight-saving transition — this is why every timestamp must be UTC-normalized upstream, never local. Second, a facility node whose acceptance window closes mid-route: the schema accepts the node (it is structurally valid) but pre-flight against the daily acceptance matrix must reject the route, so acceptance-hour checks belong in pre-flight, not in the type contract.

Integration Checklist

• Pydantic pinned to 2.7.1 and models run with strict=True, frozen=True, extra="forbid".
• NodeType and WasteStream enumerations reviewed against the current EPA RCRA classification list.
• All incoming timestamps normalized to UTC at the ingestion edge before schema validation.
• validate_route is the only entry point into the solver; no path constructs a Route bypassing it.
• schema_version and payload_hash() written to the audit log on every accepted route.
• schema_rejection events alerted on, grouped by field, in the observability platform.
• Facility acceptance-hour and daily-tonnage checks wired into pre-flight, separate from the type contract.
• Unit tests cover accept, over-GVWR reject, non-contiguous reject, and naive-timestamp reject.
• Fallback relaxations gated so they never widen a legally binding quiet-hour window.

FAQ

Why enforce the schema as a hard gate instead of best-effort parsing?

Best-effort parsing coerces bad data into plausible-looking values — a missing weight becomes 0, a naive timestamp becomes local time — and the solver then optimizes over fiction. Because the Core Architecture & Compliance Mapping framework requires byte-identical output for identical input, any coercion breaks reproducibility and the audit hash. A hard gate rejects ambiguity at the edge so the solver only ever sees proven-valid input.

Should the route schema perform axle-load and bridge-formula checks?

No. The schema enforces the gross-weight bound because it can, using only accumulated mass and the vehicle rating. Per-axle distribution depends on wheelbase geometry and where mass sits in the hopper, which the solver owns as a separate cumulative dimension. Keeping axle logic out of the type contract avoids duplicating geometry the schema cannot fully model.

Where do timezone bugs actually come from?

Almost always from naive timestamps that get interpreted as local time, then shifted during a daylight-saving transition — producing a service_window_end earlier than its start. The enforce_utc validator rejects any timestamp lacking an explicit UTC offset, which is why normalization must happen in the Telematics & Sensor Data Ingestion pipeline before validation.

How does the payload hash prevent downgrade attacks?

Every route carries a schema_version and a SHA-256 payload_hash() over its canonical serialization. A consumer recomputes the hash on deserialization; any mismatch — a mutated field, a legacy parser re-emitting an older shape — rejects the object immediately, so a stale or tampered payload cannot be replayed against a newer solver.

What happens when a route fails validation mid-collection?

The gate returns None and logs a schema_rejection with the offending field; nothing corrupted reaches the dispatch queue. Recovery is an explicit decision in the Fallback Routing Logic layer, which may widen a soft window or split an over-capacity route — but only after relaxing soft parameters, never a legally binding constraint.

Up: Core Architecture & Compliance Mapping

Related

• JSON Schema for Municipal Disposal Tracking — Draft-07 field mappings for automated EPA e-Manifest submission.
• DOT/FMCSA Rule Mapping — translating federal HOS and weight mandates into solver constraints.
• Security & Access Boundaries — encryption, tokenization, and schema-versioning controls around manifest data.
• Fallback Routing Logic — deterministic degradation when a route fails pre-flight.
• Capacity & Weight Limits — binding GVWR as a cumulative dimension in the solver.