#!/usr/bin/env python3 """Reproduce the published pallet-reference arithmetic using only the standard library. Usage: python reproduce.py python reproduce.py --report validation-report.json Keep this script beside the CSV and JSON downloads. It checks the published aggregate inputs and calculations, not the researchers' original survey models. It never evaluates the human-readable formula strings as code. """ from __future__ import annotations import argparse import csv import json import math import sys from pathlib import Path from typing import Any RELEASE_DATE = "2026-09-11" FIELDS = [ "record_id", "series", "metric", "value", "value_operator", "unit", "observation_period", "geography", "population_or_denominator", "record_type", "source_ids", "source_url", "source_locator", "source_publication_date", "verified_date", "verification_tier", "formula", "input_record_ids", "qualification", ] def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def number(value: Any, label: str) -> float: require(isinstance(value, (int, float)) and not isinstance(value, bool), f"{label}: expected a numeric value") result = float(value) require(math.isfinite(result), f"{label}: value is not finite") return result def equal(actual: float, expected: float, label: str) -> None: require(math.isclose(actual, expected, rel_tol=1e-12, abs_tol=1e-10), f"{label}: found {actual!r}, expected {expected!r}") def validate(directory: Path) -> dict[str, Any]: json_path = directory / f"pallet-recycling-data-{RELEASE_DATE}.json" csv_path = directory / f"pallet-recycling-data-{RELEASE_DATE}.csv" data = json.loads(json_path.read_text(encoding="utf-8")) records = data["records"] require(isinstance(records, list) and bool(records), "No dataset records") indexed: dict[str, dict[str, Any]] = {} sources = {source["id"]: source for source in data["sources"]} for record in records: identifier = record["record_id"] require(identifier not in indexed, f"Duplicate record ID: {identifier}") require(set(record) == set(FIELDS), f"Unexpected fields: {identifier}") number(record["value"], identifier) require(record["value_operator"] in {"=", "<"}, f"Invalid operator: {identifier}") require(all(source_id in sources for source_id in record["source_ids"].split(";")), f"Unknown source ID: {identifier}") indexed[identifier] = record with csv_path.open(newline="", encoding="utf-8") as handle: reader = csv.DictReader(handle) require(reader.fieldnames == FIELDS, "CSV header does not match the documented fields") csv_rows = list(reader) require(len(csv_rows) == len(records), "CSV and JSON record counts differ") seen: set[str] = set() for row in csv_rows: identifier = row["record_id"] require(identifier in indexed and identifier not in seen, f"CSV record ID mismatch: {identifier}") seen.add(identifier) reference = indexed[identifier] for field in FIELDS: if field == "value": equal(float(row[field]), number(reference[field], identifier), f"CSV {identifier}") else: require(row[field] == str(reference[field]), f"CSV/JSON mismatch: {identifier}.{field}") computed = {record["record_id"]: number(record["value"], record["record_id"]) for record in records if not record["input_record_ids"]} results: list[dict[str, Any]] = [] checked: set[str] = set() for definition in data["calculation_definitions"]: identifier = definition["record_id"] require(identifier in indexed and identifier not in checked, f"Invalid or duplicate calculation: {identifier}") record = indexed[identifier] inputs = definition["input_record_ids"] require(inputs == record["input_record_ids"].split(";"), f"Input-list mismatch: {identifier}") require(all(item in computed for item in inputs), f"Missing or out-of-order input: {identifier}") require(all(indexed[item]["value_operator"] == "=" for item in inputs), f"A source bound cannot be treated as an exact value: {identifier}") values = [computed[item] for item in inputs] operation = definition["operation"] if operation == "sum": require(bool(values), f"Empty sum: {identifier}") actual = sum(values) elif operation == "share_percent": require(len(values) == 2 and values[1] != 0, f"Invalid ratio: {identifier}") actual = 100 * values[0] / values[1] elif operation == "change_percent": require(len(values) == 2 and values[0] != 0, f"Invalid change: {identifier}") actual = 100 * (values[1] - values[0]) / values[0] else: raise ValueError(f"Unknown calculation operation: {operation}") equal(actual, number(record["value"], identifier), identifier) equal(actual, number(definition["value"], identifier), f"Calculation definition {identifier}") expected_display = round(actual, 1) if record["unit"] == "percent" else actual equal(number(definition["display_value"], identifier), expected_display, f"Display value {identifier}") require(definition["formula"] == record["formula"], f"Formula-label mismatch: {identifier}") computed[identifier] = actual checked.add(identifier) results.append({"record_id": identifier, "recalculated_value": actual, "display_value": expected_display, "status": "pass"}) expected_derived = {record["record_id"] for record in records if record["input_record_ids"]} require(checked == expected_derived, "Not every derived record was reproduced") balance_checks = 0 for series in ("epa_wood", "epa_packaging"): for year in (2010, 2015, 2017, 2018): prefix = f"{series}-{year}-" equal(computed[prefix + "generated"], sum(computed[prefix + item] for item in ("recycled", "energy_recovery", "landfilled")), f"EPA mass balance {series} {year}") balance_checks += 1 for metric in ("pallet_tons", "stream_tons", "samples"): equal(computed[f"swaco-overall-{metric}"], computed[f"swaco-commercial-{metric}"] + computed[f"swaco-residential-{metric}"], f"SWACO sector sum {metric}") balance_checks += 1 bound = indexed["core-fate-2021-5"] require(bound["value_operator"] == "<" and bound["value"] == 1, "The incoming-core landfill bound must remain less than 1%, not exactly 1%") require(len(records) == 97 and len(checked) == 28, "Unexpected counts for this release") return { "release_version": data["release_version"], "status": "pass", "records_checked": len(records), "source_records": len(records) - len(checked), "derived_calculations_reproduced": len(checked), "accounting_checks": balance_checks, "csv_json_agreement": True, "source_bound_preserved": True, "scope": "Checks published aggregate arithmetic and file consistency, not original survey fieldwork or expansion models.", "calculations": results, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--directory", type=Path, default=Path(__file__).resolve().parent, help="Directory containing the matching CSV and JSON downloads") parser.add_argument("--report", type=Path, help="Write the validation results as JSON") arguments = parser.parse_args() try: report = validate(arguments.directory) if arguments.report: arguments.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") except (OSError, ValueError, KeyError, TypeError, csv.Error) as error: print(f"Validation failed: {error}", file=sys.stderr) return 1 print(f"PASS: {report['records_checked']} records; {report['derived_calculations_reproduced']} calculations; " f"{report['accounting_checks']} accounting checks; CSV/JSON agreement; strict upper bound preserved.") return 0 if __name__ == "__main__": raise SystemExit(main())