Generating EPA biennial hazardous waste reports

Problem statement #

You need a Python routine that reads a report year out of the immutable waste manifest ledger, aggregates the shipped quantities by EPA hazardous-waste code and management method, normalizes every mixed unit onto a single canonical basis, reconciles the totals against the ledger’s hash chain, and emits a reproducible submission file matching the structure of the EPA Hazardous Waste Report (Form 8700-13 A/B) — so that regenerating last cycle’s report from the same ledger produces a byte-identical artifact and the same content hash.

This task sits under the parent guide on EPA RCRA Hazardous Waste Tracking, part of the broader Hazardous Material & Chemical Inventory Compliance practice. It is the aggregation counterpart to tracking RCRA hazardous-waste accumulation dates: where that page enforces the clock on open containers, this one rolls the closed, shipped history into the federal report a large quantity generator files by March 1 of each even-numbered year under 40 CFR 262.41.

Prerequisites #

Before deploying the report generator, confirm the environment and policy configuration:

• Python 3.10+ (the code uses datetime.now(timezone.utc), Decimal, and modern type hints).
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] to read the ledger; the standard library covers hashing, JSON, and decimal arithmetic. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials):
  • WASTE_DB_URL — the SQLAlchemy connection string for the manifest ledger.
• Policy config: a version-controlled unit-conversion table (pounds, kilograms, and gallons to a canonical kilogram basis, with a per-waste-stream density where a volume is reported) and the site’s EPA ID, kept alongside the ledger schema from the parent guide. The reproducibility and hashing discipline follows the immutable audit report generation subsystem.

Step-by-step implementation #

The flow below is enforced by the generator: shipped events for the report year are loaded from the immutable ledger, filtered to reportable waste, aggregated by EPA waste code and management method, unit-normalized to kilograms, reconciled against the ledger chain, and emitted as a reproducible submission file carrying its own artifact hash.

Figure: the report is a pure fold over the immutable ledger, so the same year always renders the same artifact.

Step 1 — Load and filter the report year from the ledger #

The generator reads only shipped events — the disposition that counts toward the biennial report — for the report year, straight from the append-only ledger. Reading from the immutable source rather than a mutable summary table is what guarantees the report matches what an auditor would independently reconstruct.

from __future__ import annotations

import hashlib
import json
import logging
from collections import defaultdict
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP

from sqlalchemy import String, DateTime, Numeric, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

logging.basicConfig(level=logging.INFO, format="%(asctime)s [BIENNIAL] %(message)s")
logger = logging.getLogger(__name__)


class Base(DeclarativeBase):
    pass


class ShipmentLedger(Base):
    __tablename__ = "waste_shipment_ledger"
    event_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    container_id: Mapped[str] = mapped_column(String(64))
    waste_code: Mapped[str] = mapped_column(String(4))      # e.g. D001, F003
    mgmt_method: Mapped[str] = mapped_column(String(4))     # EPA method code, e.g. H040
    quantity: Mapped[Decimal] = mapped_column(Numeric(14, 4))
    unit: Mapped[str] = mapped_column(String(4))            # lb, kg, gal
    shipped_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    content_hash: Mapped[str] = mapped_column(String(64), unique=True)
    prev_hash: Mapped[str] = mapped_column(String(64))


def load_report_year(session: Session, report_year: int) -> list[ShipmentLedger]:
    """Load every shipped-waste event in the report year from the ledger."""
    start = datetime(report_year, 1, 1, tzinfo=timezone.utc)
    end = datetime(report_year + 1, 1, 1, tzinfo=timezone.utc)
    rows = session.execute(
        select(ShipmentLedger)
        .where(ShipmentLedger.shipped_at >= start)
        .where(ShipmentLedger.shipped_at < end)
        .order_by(ShipmentLedger.shipped_at, ShipmentLedger.event_id)
    ).scalars().all()
    logger.info("loaded %d shipped events for %d", len(rows), report_year)
    return rows

Step 2 — Normalize units onto a canonical basis #

The GM (Waste Generation and Management) form reports mass, but a laboratory ships in pounds, kilograms, and gallons. Every quantity is converted to kilograms with exact Decimal arithmetic before it is summed; a gallon requires the waste stream’s density, which comes from the version-controlled policy table so a conversion is auditable rather than a magic number in code.

# Canonical basis is kilograms. Volume conversions require a per-stream density
# (kg/gal) from the policy config; a missing density is an error, never a guess.
_TO_KG: dict[str, Decimal] = {
    "kg": Decimal("1"),
    "lb": Decimal("0.45359237"),  # exact international pound
}


def to_kilograms(quantity: Decimal, unit: str,
                 density_kg_per_gal: dict[str, Decimal],
                 waste_code: str) -> Decimal:
    """Convert a shipped quantity to kilograms with exact decimal arithmetic."""
    if unit in _TO_KG:
        return quantity * _TO_KG[unit]
    if unit == "gal":
        density = density_kg_per_gal.get(waste_code)
        if density is None:
            raise ValueError(
                f"no density for {waste_code}; cannot convert gallons to mass"
            )
        return quantity * density
    raise ValueError(f"unsupported unit: {unit}")

Step 3 — Aggregate by waste code and management method #

A container can carry several EPA waste codes, and the GM form is organized by waste stream and its management method. The aggregation folds the normalized events into a total per (waste_code, mgmt_method) pair using exact decimals, and the fold is deterministic because the events were loaded in a stable order.

def aggregate(rows: list[ShipmentLedger],
              density_kg_per_gal: dict[str, Decimal]
              ) -> dict[tuple[str, str], Decimal]:
    """Sum shipped mass in kilograms per (waste_code, management method)."""
    totals: dict[tuple[str, str], Decimal] = defaultdict(lambda: Decimal("0"))
    for r in rows:
        kg = to_kilograms(r.quantity, r.unit, density_kg_per_gal, r.waste_code)
        totals[(r.waste_code, r.mgmt_method)] += kg
    # Quantize once at the end to avoid compounding rounding across the sum.
    return {k: v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
            for k, v in totals.items()}

Step 4 — Reconcile against the ledger chain #

Before the report is emitted, the loaded rows are proven to be an unbroken, untampered slice of the ledger by recomputing each content_hash from its prev_hash. A reconciliation failure halts the report — a biennial submission built on a broken chain is not defensible.

def _event_hash(row: ShipmentLedger, prev_hash: str) -> str:
    event = {"container_id": row.container_id, "waste_code": row.waste_code,
             "mgmt_method": row.mgmt_method, "quantity": str(row.quantity),
             "unit": row.unit, "shipped_at": row.shipped_at.isoformat()}
    canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(f"{prev_hash}:{canonical}".encode()).hexdigest()


def reconcile(rows: list[ShipmentLedger]) -> None:
    """Raise if the loaded slice is not a valid hash chain."""
    for row in rows:
        if _event_hash(row, row.prev_hash) != row.content_hash:
            raise ValueError(f"ledger chain broken at {row.event_id[:12]}")
    logger.info("reconciled %d events against the hash chain", len(rows))

Step 5 — Emit a reproducible submission file #

The submission is serialized with sorted keys so the same ledger always yields the same bytes, and an artifact_hash over that canonical form is embedded so the file is self-verifying. The structure mirrors the GM/WR sections of Form 8700-13 A/B: one GM line per waste stream and management method, under a site header carrying the EPA ID and report year.

def build_submission(session: Session, report_year: int, epa_id: str,
                     density_kg_per_gal: dict[str, Decimal]) -> dict:
    rows = load_report_year(session, report_year)
    reconcile(rows)
    totals = aggregate(rows, density_kg_per_gal)

    gm_lines = [
        {"form": "GM", "waste_code": code, "management_method": method,
         "quantity_kg": str(kg)}
        for (code, method), kg in sorted(totals.items())
    ]
    body = {
        "form": "8700-13",
        "epa_id": epa_id,
        "report_year": report_year,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "gm_lines": gm_lines,
        "total_kg": str(sum((Decimal(l["quantity_kg"]) for l in gm_lines),
                            Decimal("0"))),
    }
    # Hash the canonical body WITHOUT generated_at so the artifact hash depends
    # only on the reported data, not on when the file was rendered.
    canonical = json.dumps({k: v for k, v in body.items() if k != "generated_at"},
                           sort_keys=True, separators=(",", ":"))
    body["artifact_hash"] = hashlib.sha256(canonical.encode()).hexdigest()
    logger.info("built 8700-13 for %d: %d GM lines, hash %s",
                report_year, len(gm_lines), body["artifact_hash"][:12])
    return body

Schema and field reference #

Verification #

1. Reproduce the artifact. Run build_submission twice for the same year and confirm both artifact_hash values match. Because the hash excludes generated_at, a re-render on a different day must produce the identical hash.
2. Reconcile the total. Confirm total_kg equals the sum of the GM line quantities, and independently sum the normalized ledger rows to the same figure. A gap means an event was dropped or double-counted.
3. Chain verification. Call reconcile on the loaded slice; a clean pass proves the report is built on an untampered ledger segment.
4. Unit-conversion spot check. Convert one known shipment by hand — 100 lb must render as 45.36 kg — and confirm the aggregated figure matches. A gallon line with no density in the policy table must raise, not silently zero.

Troubleshooting #

Three failure modes specific to assembling the biennial report:

• Totals drift because of unit conversion. Mixing pounds and kilograms with floating-point math compounds rounding across a large ledger. Use exact Decimal arithmetic, quantize only once at the end of the sum, and require a per-stream density for every gallon line — a missing density must raise rather than default, because a silent zero understates the reported mass.
• Episodic generation crosses the reporting boundary. A one-time cleanout shipped on December 31 belongs to that report year; the same waste generated in December but shipped January 2 belongs to the next. Filter strictly by shipped_at within the year’s UTC bounds, and never reclassify a shipment date to smooth a year — the ledger date is the reportable date.
• A multi-code container is counted more than once. A single drum carrying D001 and F003 legitimately appears under both codes on the GM form, but its mass must not be double-summed into a facility total as though it were two drums. Aggregate mass per (waste_code, management method) for the GM lines, but reconcile the facility total from distinct shipment events so shared-container mass is counted once.

Frequently asked questions #

Related #

• Up to the parent guide: EPA RCRA Hazardous Waste Tracking
• Tracking RCRA Hazardous Waste Accumulation Dates in Python — the clock counterpart that feeds shipped containers into this report
• Immutable Audit Report Generation — the reproducibility and hashing discipline applied here
• Hazardous Material & Chemical Inventory Compliance — the parent practice