#!/usr/bin/env python3 """Reproduce the audited cardboard ledger's arithmetic and file consistency. Python 3.9+; standard library only. Run from any working directory: python reproduce_analysis.py python reproduce_analysis.py --data-dir /path/to/dataset/files This verifies the published input snapshot and derived results. It does not fetch new data, recreate AF&PA's national estimation program, independently audit facility submissions, or validate laws from CSV text. """ from __future__ import annotations import argparse import csv import json import math from pathlib import Path from typing import Any CSV_TABLES = { "cardboard-recycling-rate-ledger.csv": "national_rate_ledger", "us-cardboard-tonnage-production.csv": "national_tonnage_and_production", "epa-corrugated-material-balance.csv": "epa_material_balance", "wisconsin-mrf-materials-2015-2024.csv": "wisconsin_mrf_materials", "wisconsin-material-changes-2015-2024.csv": "wisconsin_decade_changes", "brown-county-recycling-transfer-station.csv": "brown_county_transfer_station", "occ-price-index-ppi.csv": "occ_price_index", "selected-state-cardboard-disposal-rules.csv": "selected_state_disposal_rules", } MATERIALS = ["occ_tons", "all_other_paper_tons", "aluminum_tons", "steel_bimetal_tons", "glass_tons", "plastic_1_7_tons"] def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def rounded_equal(actual: float, expected: float, label: str) -> None: require(math.isclose(float(actual), round(expected, 1), abs_tol=1e-9), f"{label}: stored {actual}; recalculated {round(expected, 1)}") def csv_value(value: Any) -> str: return "" if value is None else str(value) def verify(data_dir: Path) -> dict[str, Any]: data_path = data_dir / "cardboard-recycling-statistics-dataset.json" data = json.loads(data_path.read_text(encoding="utf-8")) tables = data["tables"] row_counts: dict[str, int] = {} for filename, key in CSV_TABLES.items(): path = data_dir / filename with path.open(encoding="utf-8", newline="") as handle: csv_rows = list(csv.DictReader(handle)) json_rows = tables[key] require(len(csv_rows) == len(json_rows), f"Row count mismatch: {filename}") for row_number, (csv_row, json_row) in enumerate(zip(csv_rows, json_rows), 2): require(set(csv_row) == set(json_row), f"Column mismatch: {filename}, row {row_number}") for field, value in json_row.items(): require(csv_row[field] == csv_value(value), f"Mismatch: {filename}, row {row_number}, {field}") row_counts[filename] = len(csv_rows) wi = tables["wisconsin_mrf_materials"] require([r["data_year"] for r in wi] == list(range(2015, 2025)), "Wisconsin series must cover 2015–2024 without gaps or duplicates") for index, row in enumerate(wi): occ = row["occ_tons"] total = row["total_tons"] other_paper = row["all_other_paper_tons"] rounded_equal(row["occ_share_of_total_pct"], 100 * occ / total, f"{row['data_year']} OCC output share") rounded_equal(row["occ_share_of_paper_fiber_pct"], 100 * occ / (occ + other_paper), f"{row['data_year']} OCC paper-fiber share") difference = sum(row[k] for k in MATERIALS) - total require(difference == row["component_sum_minus_published_total_tons"], "Published-total discrepancy field mismatch") require(abs(difference) <= 2, "Unexpected Wisconsin source-total discrepancy") if index == 0: require(row["occ_yoy_pct"] is None, "First year must have null YoY") else: rounded_equal(row["occ_yoy_pct"], 100 * (occ / wi[index-1]["occ_tons"] - 1), f"{row['data_year']} OCC annual change") changes = tables["wisconsin_decade_changes"] for row in changes: require(row["change_tons"] == row["end_tons"] - row["start_tons"], f"Tonnage change mismatch: {row['material']}") rounded_equal(row["change_pct"], 100 * (row["end_tons"] / row["start_tons"] - 1), row["material"]) non_occ = next(r for r in changes if r["material"] == "Everything except corrugated") require(non_occ["start_tons"] == wi[0]["total_tons"] - wi[0]["occ_tons"], "Non-OCC baseline must use published total minus OCC") require(non_occ["end_tons"] == wi[-1]["total_tons"] - wi[-1]["occ_tons"], "Non-OCC endpoint must use published total minus OCC") crossover = next(r["data_year"] for r in wi if r["occ_tons"] > r["all_other_paper_tons"]) require(crossover == 2019, "OCC/paper crossover should be 2019 in this series") require(max(wi, key=lambda r: r["total_tons"])["data_year"] == 2019, "Reported total-output peak mismatch") require(max(wi, key=lambda r: r["occ_tons"])["data_year"] == 2024, "Reported OCC-output peak mismatch") epa = {r["measure"]: r["value"] for r in tables["epa_material_balance"] if r["data_year"] == 2018} balanced = (epa["Corrugated boxes recycled"] + epa["Combusted with energy recovery"] + epa["Landfilled"]) require(balanced == epa["Corrugated boxes generated"], "EPA balance mismatch") rounded_equal(epa["Corrugated-box recycling rate"], 100 * epa["Corrugated boxes recycled"] / epa["Corrugated boxes generated"], "EPA rate arithmetic") ppi_rows = tables["occ_price_index"] ppi = {r["month"]: r["index_value"] for r in ppi_rows} require(len(ppi) == len(ppi_rows) == 11, "PPI snapshot must have eleven unique months") require(sorted(ppi)[-1] == "2026-08", "PPI latest observation mismatch") require(ppi["2026-02"] == 206.954, "February observation mismatch") require(ppi["2026-06"] == 241.197, "June vintage mismatch") require(ppi["2026-07"] == 242.294, "July vintage mismatch") require(ppi["2026-08"] == 254.991, "August observation mismatch") findings = data["derived_findings"] nov_change = 100 * (ppi["2026-08"] / ppi["2025-11"] - 1) jan_change = 100 * (ppi["2026-08"] / ppi["2026-01"] - 1) rounded_equal(findings["ppi_change_pct_november_2025_august_2026"], nov_change, "PPI November–August change") rounded_equal(findings["ppi_change_pct_january_2026_august_2026"], jan_change, "PPI January–August change") rounded_equal(findings["wisconsin_occ_tonnage_change_pct_2015_2024"], 100 * (wi[-1]["occ_tons"] / wi[0]["occ_tons"] - 1), "Wisconsin decade change") rounded_equal(findings["wisconsin_occ_output_share_change_percentage_points"], 100 * wi[-1]["occ_tons"] / wi[-1]["total_tons"] - 100 * wi[0]["occ_tons"] / wi[0]["total_tons"], "Wisconsin output-share percentage-point change") for row in tables["brown_county_transfer_station"]: require(row["cardboard_only_tons"] is None and row["cardboard_rate_pct"] is None, "Unmeasured Brown County cardboard fields must stay null") require(tables["brown_county_transfer_station"][-1]["single_stream_recyclables_tons"] == 24697, "Detailed Brown County 2025 value mismatch") require("24,500" in tables["brown_county_transfer_station"][-1]["source_discrepancy_note"], "Brown County source conflict must remain documented") return { "status": "passed", "dataset_version": data["version"], "csv_row_counts": row_counts, "epa_balance_short_tons": balanced, "wisconsin_occ_change_pct": changes[0]["change_pct"], "wisconsin_occ_output_share_2024_pct": wi[-1]["occ_share_of_total_pct"], "wisconsin_occ_share_change_percentage_points": findings["wisconsin_occ_output_share_change_percentage_points"], "wisconsin_crossover_year": crossover, "ppi_change_november_2025_to_august_2026_pct": round(nov_change, 1), "ppi_change_january_to_august_2026_pct": round(jan_change, 1), "scope": "Arithmetic and packaged-file consistency; not an independent audit of producers' original data." } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--data-dir", type=Path, default=Path(__file__).resolve().parent) args = parser.parse_args() try: result = verify(args.data_dir) except (OSError, ValueError, KeyError, TypeError, StopIteration) as error: parser.exit(1, f"Validation failed: {error}\n") print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())