Tracking RCRA hazardous waste accumulation dates in Python

Problem statement #

You need a Python routine that reads every open hazardous-waste container in the laboratory, derives each container’s regulatory disposal deadline from its accumulation start date and the institution’s generator category, and escalates containers that are within a warning window of — or already past — their 40 CFR Part 262 time limit, writing the result idempotently so a nightly re-run never double-escalates or corrupts the record.

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 deliberately narrow: it computes and escalates deadlines. It does not classify waste, ship it, or file the biennial report — it surfaces which containers are approaching a legal cliff so a human can act before the date on the label becomes a violation.

Prerequisites #

Before deploying the accumulation tracker, confirm the environment and policy configuration:

• Python 3.10+ (the code uses X | None unions and datetime.now(timezone.utc)).
• Libraries: pydantic>=2.5 for the container model and SQLAlchemy>=2.0 with psycopg[binary] for the idempotent write. Install with pip install "pydantic>=2.5" "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials):
  • WASTE_DB_URL — the SQLAlchemy connection string for the tracking store.
• Policy config: a version-controlled table of RCRA limits (90 days for a large quantity generator per 262.17; 180 or 270 days for a small quantity generator per 262.16; the 3-day satellite transfer clock per 262.15) and the institution’s current generator_status, kept alongside the deadline arithmetic developed in the parent guide’s implementation. The container events themselves are recorded by the same ledger described there.

Step-by-step implementation #

The flow below is enforced by the tracker: each container is modelled with its area_type, generator_status, and immutable accum_start_date; the deadline is computed by adding the applicable RCRA limit; days_remaining is measured against the current UTC instant; and any container inside the warning window is escalated and written idempotently to the tracking ledger.

Figure: the deadline is derived from the immutable start date; only the status changes as the clock advances.

Step 1 — Model the container as an immutable regulatory fact #

The Pydantic model is where the container’s regulatory identity is fixed. The accum_start_date is required and must be timezone-aware — for a central-area container it is the day accumulation began, and for a satellite container that has hit its cap it is the day the 262.15 transfer clock started. A frozen model prevents any later code from quietly editing the start date to buy time against the clock.

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator

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


class GeneratorStatus(str, Enum):
    VSQG = "VSQG"  # 40 CFR 262.14 — quantity limits, no accumulation clock
    SQG = "SQG"    # 40 CFR 262.16 — 180 days (270 if > 200 miles)
    LQG = "LQG"    # 40 CFR 262.17 — 90 days


class AreaType(str, Enum):
    SAA = "SAA"  # satellite accumulation area, 40 CFR 262.15
    CAA = "CAA"  # central accumulation area


class WasteContainer(BaseModel):
    # frozen=True makes the start date and identity immutable once constructed.
    model_config = ConfigDict(frozen=True)

    container_id: str = Field(..., min_length=3, max_length=64)
    generator_status: GeneratorStatus
    area_type: AreaType
    accum_start_date: datetime
    quantity: Decimal = Field(..., gt=0)
    unit: str = Field(..., pattern=r"^(gal|L|lb|kg|qt)$")
    saa_cap_reached: bool = False
    ships_over_200_miles: bool = False

    @field_validator("accum_start_date")
    @classmethod
    def must_be_utc_aware(cls, v: datetime) -> datetime:
        # The day a clock rolls over is legally significant; a naive datetime
        # would be interpreted in a local zone and could misdate a deadline.
        if v.tzinfo is None:
            raise ValueError("accum_start_date must be timezone-aware (UTC).")
        return v.astimezone(timezone.utc)

Step 2 — Resolve the RCRA limit for this container #

The limit is a function of (generator_status, area_type) and two flags. Keeping the day counts in a table means the regulation is legible in one place and a state overlay that tightens a clock is a data change, not a logic change. A satellite area returns no clock until its cap is reached; only then does the 3-day transfer window apply.

_CAA_LIMIT_DAYS: dict[GeneratorStatus, int] = {
    GeneratorStatus.LQG: 90,   # 262.17
    GeneratorStatus.SQG: 180,  # 262.16
}
_SQG_LONG_HAUL_DAYS = 270      # 262.16(b)(3), waste shipped > 200 miles
_SAA_TRANSFER_DAYS = 3         # 262.15(a)(6), once the volume cap is reached


def resolve_limit_days(c: WasteContainer) -> int | None:
    """Return the RCRA time limit in days, or None when no clock yet applies."""
    if c.area_type is AreaType.SAA:
        return _SAA_TRANSFER_DAYS if c.saa_cap_reached else None
    if c.generator_status is GeneratorStatus.VSQG:
        return None
    if c.generator_status is GeneratorStatus.SQG and c.ships_over_200_miles:
        return _SQG_LONG_HAUL_DAYS
    return _CAA_LIMIT_DAYS[c.generator_status]

Step 3 — Compute the deadline and days remaining #

The deadline is arithmetic on the immutable start date; days_remaining is a comparison against a single now captured once per run so every container in the batch is judged against the same instant. Passing now in — rather than calling the clock inside the loop — is what makes the evaluation reproducible for the audit step later.

class ContainerState(str, Enum):
    ACCUMULATING = "ACCUMULATING"
    WARN = "WARN"
    OVERDUE = "OVERDUE"


def evaluate(c: WasteContainer, *, warn_window_days: int = 14,
             now: datetime | None = None) -> dict[str, object]:
    """Return the deadline, days remaining, and state for one container."""
    now = now or datetime.now(timezone.utc)
    days = resolve_limit_days(c)
    if days is None:
        # Satellite area under its cap, or a VSQG stream: no clock is running.
        return {"container_id": c.container_id, "deadline": None,
                "days_remaining": None, "state": ContainerState.ACCUMULATING}

    deadline = c.accum_start_date + timedelta(days=days)
    days_remaining = (deadline - now).days
    if days_remaining < 0:
        state = ContainerState.OVERDUE
    elif days_remaining <= warn_window_days:
        state = ContainerState.WARN
    else:
        state = ContainerState.ACCUMULATING
    return {"container_id": c.container_id, "deadline": deadline,
            "days_remaining": days_remaining, "state": state}

Step 4 — Escalate and write idempotently #

Containers in WARN or OVERDUE state are escalated; every evaluation is written to the tracking ledger keyed on a content hash of the container and the evaluation date, so re-running the job on the same day is a no-op. Escalation reuses the same routing the lab already relies on for escalating overdue calibration to lab managers.

from sqlalchemy import String, DateTime, Integer, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class AccumulationLedger(Base):
    __tablename__ = "accumulation_ledger"
    eval_id: Mapped[str] = mapped_column(String(64), primary_key=True)
    container_id: Mapped[str] = mapped_column(String(64), index=True)
    deadline: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    days_remaining: Mapped[int | None] = mapped_column(Integer)
    state: Mapped[str] = mapped_column(String(16))
    evaluated_on: Mapped[str] = mapped_column(String(10))  # UTC date, YYYY-MM-DD


def _eval_id(container_id: str, evaluated_on: str, state: str) -> str:
    canonical = json.dumps(
        {"c": container_id, "d": evaluated_on, "s": state},
        sort_keys=True, separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode()).hexdigest()


def track(session: Session, containers: list[WasteContainer],
          escalate,  # callable: (result: dict) -> None
          *, warn_window_days: int = 14,
          now: datetime | None = None) -> dict[str, int]:
    """Evaluate, escalate, and write each container idempotently."""
    now = now or datetime.now(timezone.utc)
    evaluated_on = now.date().isoformat()
    stats = {"escalated": 0, "written": 0, "skipped": 0}

    for c in containers:
        result = evaluate(c, warn_window_days=warn_window_days, now=now)
        state = result["state"].value

        if result["state"] in (ContainerState.WARN, ContainerState.OVERDUE):
            escalate(result)
            stats["escalated"] += 1

        stmt = pg_insert(AccumulationLedger).values(
            eval_id=_eval_id(c.container_id, evaluated_on, state),
            container_id=c.container_id, deadline=result["deadline"],
            days_remaining=result["days_remaining"], state=state,
            evaluated_on=evaluated_on,
        ).on_conflict_do_nothing(index_elements=["eval_id"])
        if session.execute(stmt).rowcount:
            stats["written"] += 1
        else:
            stats["skipped"] += 1

    session.commit()
    logger.info("tracked %d containers: %s", len(containers), stats)
    return stats

Schema and field reference #

Verification #

1. Reproduce a deadline. Call evaluate(container, now=fixed_instant) in a REPL and confirm the returned deadline equals accum_start_date + limit. For an LQG central container started 2026-01-01, the deadline must be exactly 90 days later.
2. Null-clock invariant. Confirm a satellite container with saa_cap_reached=False returns deadline=None and state=ACCUMULATING, and that setting the flag returns a 3-day deadline. A satellite container that shows a clock before it is full is a bug.
3. Dry-run idempotency. Run track twice with the same containers and the same now. The second pass must report written=0 and skipped= the batch size — no duplicate ledger rows for the same container, date, and state.
4. Category-change recompute. Change generator_status from SQG to LQG and re-evaluate; the deadline must move from 180 to 90 days after the unchanged start date, and a formerly-safe container may now be OVERDUE.

Troubleshooting #

Three failure modes specific to accumulation-date tracking:

• A satellite container never escalates even when overflowing. The saa_cap_reached flag was never asserted, so resolve_limit_days returns None and no clock runs. Under 40 CFR 262.15 the satellite area has no time limit until the 55-gallon (or 1-quart acute) cap is reached — but once it is, three days is a hard clock. Assert the flag the moment the cap is observed; do not wait for a paperwork step.
• Deadlines shift when the generator status changes mid-year. Episodic generation is real: a site that runs a one-time cleanout can cross 1000 kg/month and become an LQG for that period under 262.13, shortening every open central-area clock from 180 to 90 days. Re-evaluate all open containers whenever the site’s category changes; because the deadline is derived from the immutable start date, the recompute is safe and never edits history.
• A container looks overdue in one environment and fine in another. The accum_start_date was stored or parsed as a naive datetime and interpreted in different local zones. The validator rejects naive datetimes for exactly this reason — the day a clock rolls over is legally significant, so persist and compare everything in UTC.

Frequently asked questions #

Related #

• Up to the parent guide: EPA RCRA Hazardous Waste Tracking
• Generating EPA Biennial Hazardous Waste Reports — aggregating the tracked containers into Form 8700-13
• Escalating Overdue Calibration to Lab Managers — the deadline-escalation pattern reused here
• Hazardous Material & Chemical Inventory Compliance — the parent practice