#!/usr/bin/env python3 """Independent, pipeline-free reproduction of TaxpayerScope's published real spending per resident for one Florida county (METH-05, D-10). **Standard library plus `openpyxl` only — this script never imports anything from `pipeline/`.** An AST-level test (`pipeline/tests/publish/test_reproduce_published_figure.py`, Test 6) enforces that boundary. Run it either with a plain `pip install openpyxl` environment (`python3 reproduce_published_figure.py ...`), or, to reuse the pipeline project's already-pinned `openpyxl` without a second install, from a `pipeline/` checkout: uv run --frozen --directory pipeline python ../scripts/reproduce_published_figure.py \\ --candidate ../apps/web/dist This script is Florida-specific by design (EDR sheet layout, fund-type column labels, and the October-September fiscal year) — it is a reproduction tool proving one state adapter's published figures, never core platform code (mirrors `infrastructure/distribution/render_template.py`'s "never imports pipeline" docstring). **What it proves.** For one government (Orange County by default, `--government-id` to check any other published government), it independently recomputes nominal spending, population, the fiscal-year CPI mean, real (inflation-adjusted) spending, and real spending per resident for the government's latest published fiscal year and for the ten-year window's start fiscal year (`summary.json`'s own `real_spending_per_resident.fiscal_year` / `.start_fiscal_year`) — and compares every one of them, to the cent, against the published `spending.json` and `summary.json` figures. It is read-only and deterministic: it writes nothing, and repeated or parallel runs over the same inputs print the same output and return the same exit code. **The eight documented steps**, each re-derived from the archived files and the published JSON's own methodology fields — never read from pipeline output, never taking a rounded published figure as an input to another figure's computation: 1. Load the government's published `summary.json`, `spending.json`, and `sources.json`, and the state's `counties.json` (for the government's display name) — either from a local release directory (`--candidate`) or a live site (`--base-url`). 2. For each fiscal year being checked, fetch the archived EDR expenditures workbook, every archived BLS CPI-U window, and the archived Census PEP file(s) that file's `sources.json` entry names via its `archive_url` (D-08's content-addressed mirror path). 3. Recompute each fetched file's SHA-256 and compare it to `sources.json`'s own `sha256` before using its bytes for anything — a mismatch is an `ERROR` (an unverifiable input), never a `MISMATCH` (a disagreement between two verified values). 4. Locate the EDR workbook's fiscal-year sheet (the sheet is literally named the fiscal year, e.g. `"2024"`) and its Governmental Funds columns by their row-3 group label and row-4 fund-type labels — `General`, `Special Revenue`, `Debt Service`, `Capital Projects`, `Permanent` — never a fixed column index (`pipeline/adapters/florida/parsers/edr_workbook.py`). 5. Sum every leaf account-code row's governmental-fund cells — a row with no numeric account code in its label columns is a rollup, not a leaf, and is skipped; an empty or blank leaf cell reads as $0.00 (ADR-0010 Q1) — excluding transfer-out codes 581, 586, and 588 (`docs/decisions/0001-total-expenditures-definition.md`). This is nominal spending, exact to the cent. 6. Merge every BLS CPI-U monthly value across every archived BLS source entry `sources.json` names, and sum October of the prior calendar year through September of the fiscal year (Florida's fiscal_year_start_month is 10 — hardcoded here, since this script is Florida-specific by design) — twelve months exactly, or this fails closed (`pipeline/normalize/inflation.py`). 7. Select the archived Census PEP file that carries `POPESTIMATE{fiscal_year}` and has the highest such vintage among the candidates (`pipeline/normalize/population.py`'s D-08 rule), and read that county's row where `SUMLEV` is `"050"`, `STNAME` is `"Florida"`, and `CTYNAME` is `"{name} County"`. 8. Compute, in this exact order (`pipeline/metrics/compute.py`, `pipeline/metrics/exact.py::truncate10`, `pipeline/publish/write_government_data.py`): a. Nominal spending: sum leaf cells at cents precision (each cell independently rounded to the nearest cent, `ROUND_HALF_UP`, before summing) — exact, never rounded again until step (e). b. The CPI ratio between the checked fiscal year and the government's dollar year (its latest covered year) is the ratio of the two years' unrounded twelve-month sums — this equals the ratio of the two years' means, since both share the same twelve-month divisor, so the mean's own division by 12 never has to happen before the ratio is taken. c. Real spending = nominal spending (exact) * that ratio, truncated toward zero at ten decimal places (`truncate10`). d. Real spending per resident = real spending (from (c), still exact) divided by population, truncated toward zero at ten decimal places. e. Every whole-dollar total (nominal spending, real spending) rounds `ROUND_HALF_UP` to the nearest dollar exactly once; every per-resident figure (real spending per resident) rounds `ROUND_HALF_UP` to the nearest cent exactly once; the CPI mean (fiscal-year sum divided by 12, computed at 50-digit precision) rounds `ROUND_HALF_UP` to three decimal places exactly once. No figure is ever rounded twice, and no figure is ever computed from an already-rounded published value. Prints one line per step taken and one line per compared figure — `OK: ...` on a match, `MISMATCH: ...` naming the figure on a disagreement, `ERROR: ...` naming the missing input or the SHA-256 mismatch. Exit codes: `0` when every figure matches, `1` on any mismatch, `2` on a missing input or a SHA-256 mismatch. Never uses a tolerance, never reads pipeline code or intermediate pipeline outputs, and never resolves a disagreement by changing a published figure to match this script's recomputation. """ from __future__ import annotations import argparse import csv import hashlib import io import json import re import sys import urllib.error import urllib.request from dataclasses import dataclass from decimal import ROUND_HALF_UP, Decimal, localcontext from fractions import Fraction from pathlib import Path from typing import Any import openpyxl DEFAULT_GOVERNMENT_ID = "US-FL-COUNTY-ORANGE" DEFAULT_STATE = "florida" # Florida's fiscal_year_start_month (October-September) — this script is Florida-specific # by design (docstring above), so this is a literal, not a value read from anywhere. FISCAL_YEAR_START_MONTH = 10 # docs/decisions/0001-total-expenditures-definition.md — the excluded transfer-out codes. EXCLUDED_EXPENDITURE_CODES = frozenset({"581", "586", "588"}) GOVERNMENTAL_FUND_GROUP_LABEL = "Governmental Funds" GOVERNMENTAL_FUND_LABELS = ( "General", "Special Revenue", "Debt Service", "Capital Projects", "Permanent", ) FUND_GROUP_ROW = 3 FUND_TYPE_ROW = 4 FIRST_ACCOUNT_ROW = 5 _CENTS = Decimal("0.01") _CPI_PLACES = Decimal("0.001") _WHOLE_DOLLAR = Decimal(1) _TEN_PLACES = Decimal("1E-10") _RATIO_SCALE = 10**10 _REQUEST_TIMEOUT_SECONDS = 30 _VINTAGE_RE = re.compile(r"co-est(\d{4})-alldata\.csv$") _POPESTIMATE_RE = re.compile(r"^POPESTIMATE(\d{4})$") _M13_ANNUAL_AVERAGE_PERIOD = "M13" class ReproductionError(Exception): """A missing input or a SHA-256 mismatch (exit code 2) — never a disagreement between two verified, present values (that is a `MISMATCH`, exit code 1, handled separately).""" # --- fetching ------------------------------------------------------------------------ class Fetcher: """Reads a release's files either from a local candidate directory or a live site, uniformly. Never writes anything.""" def __init__(self, *, candidate: Path | None, base_url: str | None) -> None: if (candidate is None) == (base_url is None): raise ValueError("exactly one of candidate/base_url must be set") self.candidate = candidate self.base_url = base_url.rstrip("/") if base_url is not None else None def read_json(self, relative_path: str) -> Any: return json.loads(self._read_bytes(relative_path).decode("utf-8")) def read_archived(self, archive_url: str, expected_sha256: str, label: str) -> bytes: """Fetch `archive_url`'s bytes and verify them against `expected_sha256` before returning — every caller uses the returned bytes only after this check passes.""" data = self._read_bytes(archive_url) actual = hashlib.sha256(data).hexdigest() if actual != expected_sha256: raise ReproductionError( f"{label}: fetched {archive_url} but it hashes to {actual}, not the " f"sources.json sha256 {expected_sha256} — refusing to use unverifiable bytes" ) return data def _read_bytes(self, relative_path: str) -> bytes: relative_path = relative_path.lstrip("/") if self.candidate is not None: path = self.candidate / relative_path if not path.is_file(): raise ReproductionError(f"missing input: {path} does not exist") return path.read_bytes() url = f"{self.base_url}/{relative_path}" request = urllib.request.Request( url, headers={"User-Agent": "taxpayerscope-reproduce-published-figure/1"} ) try: with urllib.request.urlopen(request, timeout=_REQUEST_TIMEOUT_SECONDS) as response: return response.read() except (OSError, urllib.error.URLError) as exc: raise ReproductionError(f"missing input: could not fetch {url}: {exc}") from exc def _validate_base_url(base_url: str) -> None: if base_url.startswith("https://"): return if base_url.startswith("http://127.0.0.1") or base_url.startswith("http://localhost"): return raise ReproductionError( f"--base-url must be https://, or http://127.0.0.1/http://localhost for tests, " f"got {base_url!r}" ) # --- EDR workbook parsing (step 4-5) -------------------------------------------------- def _forward_fill(row: list[Any]) -> list[Any]: last = None filled = [] for v in row: if v is not None: last = v filled.append(last) return filled def _leaf_account_code(label_values: list[Any]) -> str | None: """The first numeric (non-bool) value in a row's label-area cells, formatted exactly as printed (a whole-number float drops its `.0`) — or `None` when the row carries no account code at all, meaning it is a category-rollup row, not a leaf.""" for value in label_values: if isinstance(value, bool): continue if isinstance(value, (int, float)): if isinstance(value, float) and value.is_integer(): return str(int(value)) return str(value) return None def _to_decimal_cents(value: Any) -> Decimal: """D-02/ADR-0010 Q1: an empty leaf cell is $0.00. Otherwise, round to the nearest cent (`ROUND_HALF_UP`) exactly as the pipeline's own `_to_decimal_amount` does.""" if value == "": return Decimal("0.00") if value is None: value = 0 raw = Decimal(repr(float(value))) return raw.quantize(_CENTS, rounding=ROUND_HALF_UP) def _governmental_columns(rows: list[tuple[Any, ...]]) -> list[int]: def cell(row_idx: int, col_idx: int) -> Any: row = rows[row_idx - 1] if 1 <= row_idx <= len(rows) else () return row[col_idx - 1] if 1 <= col_idx <= len(row) else None max_col = max((len(r) for r in rows), default=0) group_row = _forward_fill([cell(FUND_GROUP_ROW, c) for c in range(1, max_col + 1)]) type_row = [cell(FUND_TYPE_ROW, c) for c in range(1, max_col + 1)] columns = [ idx for idx, (group, label) in enumerate(zip(group_row, type_row), start=1) if group == GOVERNMENTAL_FUND_GROUP_LABEL and label in GOVERNMENTAL_FUND_LABELS ] if not columns: raise ReproductionError( "could not resolve any Governmental Funds column via the row-3/row-4 labels" ) return columns def _sum_governmental_leaf_cells(ws: Any) -> Decimal: """Step 4-5: sum every leaf account-code row's governmental-fund cells, excluding transfer-out codes 581/586/588, at cents precision.""" rows = list( ws.iter_rows(min_row=1, max_row=ws.max_row, min_col=1, max_col=ws.max_column, values_only=True) ) def cell(row_idx: int, col_idx: int) -> Any: row = rows[row_idx - 1] if 1 <= row_idx <= len(rows) else () return row[col_idx - 1] if 1 <= col_idx <= len(row) else None governmental_cols = _governmental_columns(rows) first_fund_col = min(governmental_cols) label_cols = list(range(1, first_fund_col)) total = Decimal("0.00") for r in range(FIRST_ACCOUNT_ROW, len(rows) + 1): label_values = [cell(r, c) for c in label_cols] code = _leaf_account_code(label_values) if code is None: continue if code in EXCLUDED_EXPENDITURE_CODES: continue for c in governmental_cols: total += _to_decimal_cents(cell(r, c)) return total def _recompute_nominal_expenditures( fetcher: Fetcher, edr_source: dict, fiscal_year: int ) -> Decimal: data = fetcher.read_archived(edr_source["archive_url"], edr_source["sha256"], edr_source["source_id"]) workbook = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True) sheet_name = str(fiscal_year) if sheet_name not in workbook.sheetnames: raise ReproductionError( f"{edr_source['source_id']}: no sheet named {sheet_name!r} (sheets: " f"{workbook.sheetnames})" ) return _sum_governmental_leaf_cells(workbook[sheet_name]) # --- BLS CPI-U merge (step 6) ---------------------------------------------------------- def _fiscal_year_months(fiscal_year: int) -> list[tuple[int, int]]: year = fiscal_year - 1 month = FISCAL_YEAR_START_MONTH months: list[tuple[int, int]] = [] for _ in range(12): months.append((year, month)) month += 1 if month > 12: month = 1 year += 1 return months def _recompute_cpi_fy_sum( fetcher: Fetcher, bls_sources: list[dict], series_id: str, fiscal_year: int ) -> Decimal: merged: dict[tuple[int, int], Decimal] = {} for source in bls_sources: data = fetcher.read_archived(source["archive_url"], source["sha256"], source["source_id"]) payload = json.loads(data.decode("utf-8")) series_list = payload.get("Results", {}).get("series", []) series = next((s for s in series_list if s.get("seriesID") == series_id), None) if series is None: continue for entry in series.get("data", []): period = entry.get("period", "") if period == _M13_ANNUAL_AVERAGE_PERIOD: continue if not (period.startswith("M") and period[1:].isdigit()): raise ReproductionError(f"{source['source_id']}: unrecognized period {period!r}") month = int(period[1:]) year = int(entry["year"]) value = Decimal(str(entry["value"])).quantize(_CPI_PLACES) key = (year, month) if key in merged and merged[key] != value: raise ReproductionError( f"{series_id}: conflicting CPI value for {year}-{month:02d}: " f"{merged[key]} vs {value} (from {source['source_id']})" ) merged[key] = value months = _fiscal_year_months(fiscal_year) total = Decimal("0.000") missing = [f"{y}-{m:02d}" for y, m in months if (y, m) not in merged] if missing: raise ReproductionError(f"FY{fiscal_year}: missing CPI month(s) {missing}") for key in months: total += merged[key] return total.quantize(_CPI_PLACES, rounding=ROUND_HALF_UP) def _cpi_mean(cpi_fy_sum: Decimal, month_count: int = 12) -> Decimal: with localcontext() as ctx: ctx.prec = 50 return Decimal(cpi_fy_sum) / Decimal(month_count) # --- Census PEP selection (step 7) ------------------------------------------------------ def _recompute_population( fetcher: Fetcher, pep_sources: list[dict], county_name: str, fiscal_year: int ) -> int: target_column = f"POPESTIMATE{fiscal_year}" target_cty_name = f"{county_name} County" candidates: list[tuple[int, list[dict[str, str]], str]] = [] for source in pep_sources: data = fetcher.read_archived(source["archive_url"], source["sha256"], source["source_id"]) text = data.decode("latin-1") reader = csv.DictReader(io.StringIO(text)) fieldnames = reader.fieldnames or [] popestimate_years = [ int(match.group(1)) for name in fieldnames if (match := _POPESTIMATE_RE.fullmatch(name)) is not None ] if fiscal_year not in popestimate_years: continue rows = list(reader) candidates.append((max(popestimate_years), rows, source["source_id"])) if not candidates: raise ReproductionError( f"FY{fiscal_year}: no archived Census PEP source contains {target_column}" ) candidates.sort(key=lambda c: c[0]) _, rows, source_id = candidates[-1] matching = [ row for row in rows if row.get("SUMLEV") == "050" and row.get("STNAME") == "Florida" and row.get("CTYNAME") == target_cty_name ] if not matching: raise ReproductionError(f"{source_id}: no row for SUMLEV 050 / Florida / {target_cty_name!r}") raw_value = matching[0][target_column].strip() try: population = int(raw_value) except ValueError as exc: raise ReproductionError( f"{source_id}: {target_column} value {raw_value!r} is not an integer" ) from exc if population <= 0: raise ReproductionError(f"{source_id}: {target_column} value {population} must be positive") return population # --- rounding (step 8e) ------------------------------------------------------------------ def _round_int(value: Decimal) -> int: return int(Decimal(value).quantize(_WHOLE_DOLLAR, rounding=ROUND_HALF_UP)) def _round_cents(value: Decimal) -> Decimal: return Decimal(value).quantize(_CENTS, rounding=ROUND_HALF_UP) def _round_cpi(value: Decimal) -> Decimal: return Decimal(value).quantize(_CPI_PLACES, rounding=ROUND_HALF_UP) def _truncate10(value: Fraction) -> Decimal: scaled = int(value * _RATIO_SCALE) return Decimal(scaled).scaleb(-10).quantize(_TEN_PLACES) # --- comparison helpers ------------------------------------------------------------------- @dataclass class Outcome: matched: bool def _compare_int(label: str, published: int, recomputed: int, outcomes: list[Outcome]) -> None: if published == recomputed: print(f"OK: {label} = {recomputed} (matches published)") outcomes.append(Outcome(True)) else: print(f"MISMATCH: {label} — published {published}, recomputed {recomputed}") outcomes.append(Outcome(False)) def _compare_decimal( label: str, published: float, recomputed: Decimal, outcomes: list[Outcome] ) -> None: published_decimal = Decimal(str(published)) if published_decimal == recomputed: print(f"OK: {label} = {recomputed} (matches published)") outcomes.append(Outcome(True)) else: print(f"MISMATCH: {label} — published {published_decimal}, recomputed {recomputed}") outcomes.append(Outcome(False)) # --- orchestration ------------------------------------------------------------------------- def _find_one(sources: list[dict], predicate) -> dict: matching = [s for s in sources if predicate(s)] if not matching: raise ReproductionError("no matching source found in sources.json for a required input") return matching[0] def _government_name(counties_doc: dict, government_id: str) -> str: for county in counties_doc.get("counties", []): if county.get("government_id") == government_id: return str(county["name"]) raise ReproductionError(f"{government_id}: not found in counties.json") def run(fetcher: Fetcher, government_id: str, state: str) -> int: print(f"Step 1: loading published JSON for {government_id}") gov_dir = f"data/v1/{state}/{government_id}" summary = fetcher.read_json(f"{gov_dir}/summary.json") spending = fetcher.read_json(f"{gov_dir}/spending.json") sources_doc = fetcher.read_json(f"{gov_dir}/sources.json") counties_doc = fetcher.read_json(f"data/v1/{state}/counties.json") name = _government_name(counties_doc, government_id) print(f" {government_id} is {name!r} County") all_sources = sources_doc["sources"] edr_source = _find_one( all_sources, lambda s: s["source_id"].startswith("fl-edr-") and "expenditures" in s["source_id"] ) bls_sources = [s for s in all_sources if s["source_id"].startswith("bls-cpi-")] pep_sources = [s for s in all_sources if s["source_id"].startswith("census-pep-")] if not bls_sources: raise ReproductionError("sources.json has no bls-cpi- source entries") if not pep_sources: raise ReproductionError("sources.json has no census-pep- source entries") series_id = sources_doc["cpi_series"] dollar_year = summary["dollar_year"] latest_fy = summary["real_spending_per_resident"]["fiscal_year"] start_fy = summary["real_spending_per_resident"]["start_fiscal_year"] published_years = {y["fiscal_year"]: y for y in spending["years"]} checked_years = sorted({start_fy, latest_fy}) cpi_years = sorted({dollar_year, start_fy, latest_fy}) print(f"Step 2-3: fetching and verifying archived files for FY{checked_years} (dollar_year FY{dollar_year})") cpi_fy_sum_by_year: dict[int, Decimal] = {} for fiscal_year in cpi_years: cpi_fy_sum_by_year[fiscal_year] = _recompute_cpi_fy_sum(fetcher, bls_sources, series_id, fiscal_year) nominal_by_year: dict[int, Decimal] = {} population_by_year: dict[int, int] = {} for fiscal_year in checked_years: print(f"Step 4-5: summing governmental leaf cells for FY{fiscal_year} (excluding 581/586/588)") nominal_by_year[fiscal_year] = _recompute_nominal_expenditures(fetcher, edr_source, fiscal_year) print(f"Step 7: selecting the Census PEP row for FY{fiscal_year}") population_by_year[fiscal_year] = _recompute_population(fetcher, pep_sources, name, fiscal_year) dollar_year_cpi_sum = cpi_fy_sum_by_year[dollar_year] outcomes: list[Outcome] = [] for fiscal_year in checked_years: year_row = published_years.get(fiscal_year) if year_row is None: raise ReproductionError(f"FY{fiscal_year}: not present in spending.json") nominal_exact = nominal_by_year[fiscal_year] population = population_by_year[fiscal_year] cpi_fy_sum = cpi_fy_sum_by_year[fiscal_year] cpi_fy_mean_rounded = _round_cpi(_cpi_mean(cpi_fy_sum)) print(f"Step 8: computing real spending for FY{fiscal_year}") cpi_ratio = Fraction(dollar_year_cpi_sum) / Fraction(cpi_fy_sum) real_fraction = Fraction(nominal_exact) * cpi_ratio real_truncated = _truncate10(real_fraction) real_per_capita_truncated = _truncate10(real_fraction / Fraction(population)) _compare_int( f"FY{fiscal_year} nominal total_expenditures", year_row["total_expenditures"]["value"], _round_int(nominal_exact), outcomes, ) _compare_int( f"FY{fiscal_year} population", year_row["population"]["value"], population, outcomes, ) _compare_decimal( f"FY{fiscal_year} cpi_fy_mean", year_row["cpi_fy_mean"]["value"], cpi_fy_mean_rounded, outcomes, ) _compare_int( f"FY{fiscal_year} real_expenditures", year_row["real_expenditures"]["value"], _round_int(real_truncated), outcomes, ) real_per_resident_rounded = _round_cents(real_per_capita_truncated) _compare_decimal( f"FY{fiscal_year} real_spending_per_resident (spending.json)", year_row["real_spending_per_resident"]["value"], real_per_resident_rounded, outcomes, ) summary_published = ( summary["real_spending_per_resident"]["value"] if fiscal_year == latest_fy else summary["real_spending_per_resident"]["start_value"] ) _compare_decimal( f"FY{fiscal_year} real_spending_per_resident (summary.json)", summary_published, real_per_resident_rounded, outcomes, ) return 0 if all(o.matched for o in outcomes) else 1 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Independently reproduce TaxpayerScope's published real spending per " "resident for one Florida county from its archived source files (METH-05, D-10)" ) location = parser.add_mutually_exclusive_group(required=True) location.add_argument("--candidate", type=Path, help="A local release directory") location.add_argument("--base-url", help="A live site's base URL (https:// required)") parser.add_argument("--government-id", default=DEFAULT_GOVERNMENT_ID) parser.add_argument("--state", default=DEFAULT_STATE) args = parser.parse_args(argv) try: if args.base_url is not None: _validate_base_url(args.base_url) fetcher = Fetcher(candidate=args.candidate, base_url=args.base_url) return run(fetcher, args.government_id, args.state) except ReproductionError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 except (OSError, json.JSONDecodeError, KeyError) as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())