From e48a8365d7ed70e30b8e01e698aa672b06fb385c Mon Sep 17 00:00:00 2001 From: Ben Young Date: Thu, 22 Oct 2026 14:07:51 -0400 Subject: [PATCH 1/2] feat(analysis): add Cornerstone GHG FBS pin operator toolkit Preflight Energy FBS + NowcastMUT vintages, gas x sector compare, and prepare_pin draft for year-stem pin bumps without changing committed pins. --- bedrock/analysis/ghg_fbs_pin/README.md | 57 +++ bedrock/analysis/ghg_fbs_pin/__init__.py | 1 + .../ghg_fbs_pin/check_upstream_inputs.py | 476 ++++++++++++++++++ bedrock/analysis/ghg_fbs_pin/compare_gas_e.py | 113 +++++ bedrock/analysis/ghg_fbs_pin/prepare_pin.py | 181 +++++++ .../ghg_fbs_pin/upstream_inputs_pin.json | 111 ++++ 6 files changed, 939 insertions(+) create mode 100644 bedrock/analysis/ghg_fbs_pin/README.md create mode 100644 bedrock/analysis/ghg_fbs_pin/__init__.py create mode 100644 bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py create mode 100644 bedrock/analysis/ghg_fbs_pin/compare_gas_e.py create mode 100644 bedrock/analysis/ghg_fbs_pin/prepare_pin.py create mode 100644 bedrock/analysis/ghg_fbs_pin/upstream_inputs_pin.json diff --git a/bedrock/analysis/ghg_fbs_pin/README.md b/bedrock/analysis/ghg_fbs_pin/README.md new file mode 100644 index 00000000..afb26666 --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/README.md @@ -0,0 +1,57 @@ +# Cornerstone GHG FBS pin bump + +Operator toolkit for rebuilding the Cornerstone GHG FBS and drafting a pin +bump (upload + JSON promote remain manual). See also +`bedrock/utils/snapshots/README.md` (Cornerstone GHG FBS pin). + +## Gas totals vs current pin + +```powershell +$env:Path = "$env:USERPROFILE\.local\bin;$env:Path" +uv run python -m bedrock.analysis.ghg_fbs_pin.compare_gas_e --rebuild +``` + +## Upstream input pin (before rebuild) + +Locks (1) exact Energy FBS filenames and (2) NowcastMUT vintage per inventory +year. Nowcast Use FBAs are not pinned: they are melted from MUT tables under +``flowsa/NowcastMUT``, and FBS regen uses ``latest_nowcast_mut_vintage`` (not +USAConfig), so the preflight requires the pinned vintage to be what GCS would +serve as latest and that no stale Nowcast Use FBA sits in ``FBA_DIR``. + +2017 pins MUT ``v0.5.0_011f41f`` (compensation fix, #1123); 2018-2024 pin +``v0.5.0_a166975``. Edit [`upstream_inputs_pin.json`](upstream_inputs_pin.json) +when refreshing Energy inputs or MUT vintages. + +```powershell +# Stewi / facilitymatcher (also invoked by the upstream check when enabled) +uv run python -m bedrock.utils.snapshots.stewi_facility_pin + +# Energy FBS + NowcastMUT; fails on missing, rivals, stale Nowcast FBAs, or +# latest MUT vintage mismatch +uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs + +# Optional: fetch missing pinned objects from GCS (still fails on rivals) +# uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs --download + +# Move Energy FBS rivals, other MUT vintages, and Nowcast Use FBAs to _quarantine/ +# uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs --quarantine-rivals + +# Optional: fill empty Energy FBS filenames from newest local match, then save +# uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs --discover --write +``` + +## Rebuild + draft pin JSON + +```powershell +uv run python -m bedrock.analysis.ghg_fbs_pin.prepare_pin +``` + +Writes under `output/` (gitignored if covered by analysis patterns; otherwise local-only): + +- the rebuilt `{method}_v{version}_{hash}.parquet` +- `cornerstone_ghg_fbs_2024_pin.DRAFT.json` with `method`, `filename`, `sha256` + +Then upload the parquet to `gs://cornerstone-default/transform/output_data/`, +promote the draft pin, and set `cornerstone_ghg_fbs_filename` on the canonical +USA config (and matching waterfall pins) to the new year-stem filename. diff --git a/bedrock/analysis/ghg_fbs_pin/__init__.py b/bedrock/analysis/ghg_fbs_pin/__init__.py new file mode 100644 index 00000000..15498f37 --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/__init__.py @@ -0,0 +1 @@ +"""Operator toolkit for Cornerstone GHG FBS pin bumps.""" diff --git a/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py b/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py new file mode 100644 index 00000000..bef00423 --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py @@ -0,0 +1,476 @@ +"""Preflight: pinned Energy FBS + NowcastMUT vintages for a GHG rebuild. + +``find_file`` loads the newest ``ctime`` match for a stem under ``FBS_DIR``. +Energy FBS pins lock exact filenames. Nowcast Use attribution is melted from +NowcastMUT (``flowsa/NowcastMUT``); the FBA parse path uses +``latest_nowcast_mut_vintage`` and ignores USAConfig, so this pin locks the +MUT vintage (and requires local Nowcast FBAs for that year to be absent so +regen cannot reuse a stale melt). + +Usage:: + + uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs + uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs \\ + --method GHG_national_Cornerstone_2024 + uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs \\ + --download + uv run python -m bedrock.analysis.ghg_fbs_pin.check_upstream_inputs \\ + --quarantine-rivals +""" + +from __future__ import annotations + +import argparse +import json +import shutil +from pathlib import Path +from typing import Any + +from bedrock.extract.iot.nowcast_mut_storage import ( + GCS_NOWCAST_MUT_DIR, + STORED_MUT_TABLES, + latest_nowcast_mut_vintage, + nowcast_mut_artifact_name, + nowcast_mut_artifact_stem, +) +from bedrock.utils.config.settings import FBA_DIR, FBS_DIR + +DEFAULT_PIN = Path(__file__).resolve().parent / 'upstream_inputs_pin.json' + +_DEFAULT_GCS_FBS = 'transform/output_data' +_DEFAULT_GCS_MUT = GCS_NOWCAST_MUT_DIR + + +def load_pin(path: Path | None = None) -> dict[str, Any]: + pin_path = path or DEFAULT_PIN + raw = json.loads(pin_path.read_text(encoding='utf-8')) + if not isinstance(raw, dict) or 'methods' not in raw: + raise ValueError(f'upstream inputs pin must have a methods object: {pin_path}') + return raw + + +def _stem_rivals(directory: Path, stem: str, pinned_name: str) -> list[str]: + if not directory.is_dir(): + return [] + return sorted( + p.name + for p in directory.glob(f'{stem}_v*.parquet') + if p.name != pinned_name + ) + + +def _newest_local(directory: Path, stem: str) -> str | None: + if not directory.is_dir(): + return None + matches = sorted( + directory.glob(f'{stem}_v*.parquet'), + key=lambda p: p.stat().st_ctime, + reverse=True, + ) + return matches[0].name if matches else None + + +def _iter_fbs_entries( + method_block: dict[str, Any], +) -> list[tuple[str, str | None]]: + """Yield (stem, filename_or_none) for Energy FBS pins.""" + block = method_block.get('fbs') or {} + if not isinstance(block, dict): + raise ValueError("'fbs' must be an object of stem -> filename") + out: list[tuple[str, str | None]] = [] + for stem, filename in block.items(): + name = None if filename in (None, '') else str(filename) + out.append((str(stem), name)) + return out + + +def _gcs_fbs_prefix(pin: dict[str, Any]) -> str: + return str(pin.get('gcs_sub_bucket_fbs') or _DEFAULT_GCS_FBS).strip('/') + + +def _gcs_mut_prefix(pin: dict[str, Any]) -> str: + return str(pin.get('gcs_sub_bucket_mut') or _DEFAULT_GCS_MUT).strip('/') + + +def _mut_spec(method_block: dict[str, Any]) -> dict[str, Any] | None: + raw = method_block.get('nowcast_mut') + if raw in (None, {}): + return None + if not isinstance(raw, dict): + raise ValueError("'nowcast_mut' must be an object") + for key in ('year', 'stage', 'vintage'): + if key not in raw: + raise ValueError(f"'nowcast_mut' missing {key!r}") + tables = raw.get('tables') or list(STORED_MUT_TABLES) + return { + 'year': int(raw['year']), + 'stage': str(raw['stage']), + 'vintage': str(raw['vintage']), + 'tables': [str(t) for t in tables], + } + + +def _local_nowcast_fbas(year: int) -> list[Path]: + """Cached Nowcast Use FBAs for *year* under FBA_DIR (poison for MUT pin).""" + if not Path(FBA_DIR).is_dir(): + return [] + return sorted(Path(FBA_DIR).glob(f'Nowcast_Detail_Use_AfterRedef_{year}_v*.parquet')) + + +def _mut_rivals(year: int, stage: str, table: str, pinned_name: str) -> list[str]: + stem = nowcast_mut_artifact_stem(table, year=year, stage=stage) # type: ignore[arg-type] + directory = Path(FBS_DIR) + if not directory.is_dir(): + return [] + return sorted( + p.name + for p in directory.glob(f'{stem}_v*.parquet') + if p.name != pinned_name + ) + + +def check_nowcast_mut( + method: str, + spec: dict[str, Any], + *, + pin: dict[str, Any], + download: bool = False, +) -> list[str]: + """Require pinned MUT tables locally; fail on wrong latest vintage or FBAs.""" + problems: list[str] = [] + year = spec['year'] + stage = spec['stage'] + vintage = spec['vintage'] + gcs_prefix = _gcs_mut_prefix(pin) + directory = Path(FBS_DIR) + + try: + latest = latest_nowcast_mut_vintage(year=year, stage=stage) # type: ignore[arg-type] + except Exception as exc: + problems.append( + f'{method}: could not resolve latest NowcastMUT vintage for ' + f'{year}/{stage}: {exc}' + ) + latest = None + if latest is not None and latest != vintage: + problems.append( + f'{method}: pinned NowcastMUT vintage {vintage!r} but ' + f'latest_nowcast_mut_vintage({year}, {stage!r}) is {latest!r}; ' + f'FBA regen would melt {latest!r}' + ) + + for table in spec['tables']: + filename = nowcast_mut_artifact_name( + table, # type: ignore[arg-type] + year=year, + stage=stage, # type: ignore[arg-type] + vintage=vintage, + ) + dest = directory / filename + gcs_uri = f'gs://cornerstone-default/{gcs_prefix}/{filename}' + if not dest.is_file(): + if download: + try: + from bedrock.utils.io.gcp import download_gcs_file # noqa: PLC0415 + + dest.parent.mkdir(parents=True, exist_ok=True) + download_gcs_file(filename, gcs_prefix, str(dest)) + except Exception as exc: + problems.append( + f'{method}: missing MUT {dest} (pin {gcs_uri}); ' + f'download failed: {exc}' + ) + continue + if not dest.is_file(): + problems.append( + f'{method}: missing MUT {dest} after download ({gcs_uri})' + ) + continue + else: + problems.append(f'{method}: missing MUT {dest} (pin {gcs_uri})') + continue + rivals = _mut_rivals(year, stage, table, filename) + if rivals: + problems.append( + f'{method}: competing local NowcastMUT for {table} {year}/{stage} ' + f'in {directory}: pin={filename} also={rivals}' + ) + + fbas = _local_nowcast_fbas(year) + if fbas: + problems.append( + f'{method}: local Nowcast Use FBAs for {year} would be reused instead ' + f'of melting pinned MUT {vintage!r}: {[p.name for p in fbas]} ' + f'(run --quarantine-rivals)' + ) + return problems + + +def check_method( + method: str, + method_block: dict[str, Any], + *, + pin: dict[str, Any] | None = None, + download: bool = False, +) -> list[str]: + """Return human-readable problems for one method (empty means OK).""" + pin = pin or {} + problems: list[str] = [] + gcs_fbs = _gcs_fbs_prefix(pin) + for stem, filename in _iter_fbs_entries(method_block): + directory = Path(FBS_DIR) + if not filename: + problems.append( + f'{method}: fbs stem {stem!r} has no pinned filename ' + f'(run --discover --write)' + ) + continue + dest = directory / filename + gcs_uri = f'gs://cornerstone-default/{gcs_fbs}/{filename}' + if not dest.is_file(): + if download: + try: + from bedrock.utils.io.gcp import download_gcs_file # noqa: PLC0415 + + dest.parent.mkdir(parents=True, exist_ok=True) + download_gcs_file(filename, gcs_fbs, str(dest)) + except Exception as exc: + problems.append( + f'{method}: missing {dest} (pin {gcs_uri}); ' + f'download failed: {exc}' + ) + continue + if not dest.is_file(): + problems.append( + f'{method}: missing {dest} after download ({gcs_uri})' + ) + continue + else: + problems.append(f'{method}: missing {dest} (pin {gcs_uri})') + continue + rivals = _stem_rivals(directory, stem, filename) + if rivals: + problems.append( + f'{method}: competing local fbs for {stem} in {directory}: ' + f'pin={filename} also={rivals}' + ) + + mut = _mut_spec(method_block) + if mut is not None: + problems.extend(check_nowcast_mut(method, mut, pin=pin, download=download)) + return problems + + +def check_upstream_inputs( + *, + pin: dict[str, Any] | None = None, + methods: list[str] | None = None, + download: bool = False, +) -> list[str]: + """Return all problems across selected methods (+ optional stewi check).""" + pin = pin if pin is not None else load_pin() + method_map: dict[str, Any] = pin['methods'] + selected = methods if methods is not None else sorted(method_map) + problems: list[str] = [] + for method in selected: + if method not in method_map: + problems.append(f'unknown method {method!r} (not in pin methods)') + continue + problems.extend( + check_method(method, method_map[method], pin=pin, download=download) + ) + if pin.get('require_stewi_facility_pin', True): + from bedrock.utils.snapshots.stewi_facility_pin import ( # noqa: PLC0415 + check_stewi_facility_pin, + ensure_pinned_inventories, + pin_enforced, + ) + + if pin_enforced(): + if download: + try: + ensure_pinned_inventories(download=True) + except FileNotFoundError as exc: + problems.append(str(exc)) + problems.extend(check_stewi_facility_pin()) + return problems + + +def discover_filenames( + pin: dict[str, Any], + *, + methods: list[str] | None = None, + refresh: bool = False, +) -> dict[str, Any]: + """Fill null/empty Energy FBS filenames from newest local match.""" + method_map: dict[str, Any] = pin['methods'] + selected = methods if methods is not None else sorted(method_map) + for method in selected: + block = method_map[method] + entries = block.setdefault('fbs', {}) + directory = Path(FBS_DIR) + for stem, filename in list(entries.items()): + if filename and not refresh: + continue + newest = _newest_local(directory, stem) + entries[stem] = newest if newest is not None else filename + return pin + + +def quarantine_rivals( + pin: dict[str, Any], + *, + methods: list[str] | None = None, + quarantine_dirname: str = '_quarantine', +) -> list[Path]: + """Move Energy FBS rivals, other MUT vintages, and Nowcast Use FBAs aside.""" + method_map: dict[str, Any] = pin['methods'] + selected = methods if methods is not None else sorted(method_map) + moved: list[Path] = [] + + def _move(src: Path, qdir: Path) -> None: + nonlocal moved + if not src.is_file(): + return + qdir.mkdir(parents=True, exist_ok=True) + dest = qdir / src.name + if dest.exists(): + dest = qdir / f'{src.stem}__{src.stat().st_mtime_ns}{src.suffix}' + shutil.move(str(src), str(dest)) + moved.append(dest) + meta = src.with_name(f'{src.stem}_metadata.json') + if meta.is_file(): + meta_dest = qdir / meta.name + if meta_dest.exists(): + meta_dest = ( + qdir / f'{meta.stem}__{meta.stat().st_mtime_ns}{meta.suffix}' + ) + shutil.move(str(meta), str(meta_dest)) + moved.append(meta_dest) + + for method in selected: + block = method_map[method] + for stem, filename in _iter_fbs_entries(block): + if not filename: + continue + directory = Path(FBS_DIR) + for rival in _stem_rivals(directory, stem, filename): + _move(directory / rival, directory / quarantine_dirname) + + mut = _mut_spec(block) + if mut is None: + continue + year = mut['year'] + stage = mut['stage'] + vintage = mut['vintage'] + for table in mut['tables']: + pinned = nowcast_mut_artifact_name( + table, # type: ignore[arg-type] + year=year, + stage=stage, # type: ignore[arg-type] + vintage=vintage, + ) + for rival in _mut_rivals(year, stage, table, pinned): + _move(Path(FBS_DIR) / rival, Path(FBS_DIR) / quarantine_dirname) + for fba in _local_nowcast_fbas(year): + _move(fba, Path(FBA_DIR) / quarantine_dirname) + return moved + + +def _print_ok(pin: dict[str, Any], methods: list[str]) -> None: + for method in methods: + block = pin['methods'][method] + print(f'OK {method}') + for stem, filename in _iter_fbs_entries(block): + print(f' fbs {stem}: {filename}') + mut = _mut_spec(block) + if mut is not None: + print( + f' nowcast_mut {mut["year"]}/{mut["stage"]}: {mut["vintage"]} ' + f'tables={mut["tables"]}' + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=(__doc__ or '').splitlines()[0]) + parser.add_argument( + '--pin', + type=Path, + default=DEFAULT_PIN, + help='Path to upstream_inputs_pin.json', + ) + parser.add_argument( + '--method', + action='append', + dest='methods', + default=None, + help='Method stem to check (repeatable; default: all in pin)', + ) + parser.add_argument( + '--download', + action='store_true', + help='Download missing pinned objects from GCS (still fails on rivals)', + ) + parser.add_argument( + '--discover', + action='store_true', + help='Fill empty Energy FBS filenames from newest local parquet', + ) + parser.add_argument( + '--write', + action='store_true', + help='With --discover, write updated filenames back to the pin JSON', + ) + parser.add_argument( + '--refresh', + action='store_true', + help='With --discover, overwrite existing filenames with newest local', + ) + parser.add_argument( + '--quarantine-rivals', + action='store_true', + help=( + 'Move Energy FBS rivals, other MUT vintages, and Nowcast Use FBAs ' + 'into _quarantine/ then re-check' + ), + ) + args = parser.parse_args() + + pin = load_pin(args.pin) + methods = args.methods or sorted(pin['methods']) + + if args.discover: + pin = discover_filenames(pin, methods=methods, refresh=args.refresh) + if args.write: + args.pin.write_text( + json.dumps(pin, indent=2) + '\n', + encoding='utf-8', + ) + print(f'Wrote {args.pin}') + else: + print(json.dumps(pin, indent=2)) + print('(pass --write to save)') + + if args.quarantine_rivals: + moved = quarantine_rivals(pin, methods=methods) + print(f'Quarantined {len(moved)} files under */_quarantine/') + for path in moved: + print(f' {path}') + + problems = check_upstream_inputs( + pin=pin, + methods=methods, + download=args.download, + ) + if problems: + print('Upstream input pin check FAILED:') + for line in problems: + print(f' - {line}') + raise SystemExit(1) + _print_ok(pin, methods) + if pin.get('require_stewi_facility_pin', True): + print('stewi_facility_pin: OK') + + +if __name__ == '__main__': + main() diff --git a/bedrock/analysis/ghg_fbs_pin/compare_gas_e.py b/bedrock/analysis/ghg_fbs_pin/compare_gas_e.py new file mode 100644 index 00000000..19f585fd --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/compare_gas_e.py @@ -0,0 +1,113 @@ +"""Compare gas E from a rebuilt FBS against the pinned parquet. + +Loads the pinned Cornerstone GHG FBS and a regenerated FBS for the same year +stem, then checks that ``load_E_from_flowsa``-equivalent pivots match +(Flowable x SectorProducedBy sums). Run after a local rebuild, before bumping +``cornerstone_ghg_fbs_filename``. + +Usage:: + + uv run python -m bedrock.analysis.ghg_fbs_pin.compare_gas_e \\ + [--retained path/to/retained.parquet] [--method GHG_national_Cornerstone_2024] +""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +import pandas as pd + +REL_TOL = 1e-6 +# Floor so near-zero float dust (kg) does not flag thousands of empty cells. +ABS_TOL_KG = 1.0 + + +def _gas_sector_totals(fbs: pd.DataFrame) -> pd.Series: + return ( + fbs.groupby(['Flowable', 'SectorProducedBy'], observed=True)['FlowAmount'] + .sum() + .sort_index() + ) + + +def compare(reference: pd.DataFrame, retained: pd.DataFrame) -> pd.DataFrame: + """Return rows where gas x sector totals differ beyond relative/abs tol.""" + totals = pd.concat( + { + 'reference': _gas_sector_totals(reference), + 'retained': _gas_sector_totals(retained), + }, + axis=1, + ).fillna(0.0) + diff = totals['retained'] - totals['reference'] + scale = totals['reference'].abs().clip(lower=1.0) + threshold = (REL_TOL * scale).clip(lower=ABS_TOL_KG) + # Filter after assign: empty.loc[...].assign(series) reindexes to the + # full series and invents NaN rows (pandas), which looked like 4099 diffs. + return totals.assign(diff=diff).loc[diff.abs() > threshold] + + +def main() -> None: + parser = argparse.ArgumentParser(description=(__doc__ or '').splitlines()[0]) + parser.add_argument( + '--method', + default=None, + help='GHG FBS method stem (default: active config year stem).', + ) + parser.add_argument( + '--retained', + type=Path, + default=None, + help='Retained-activity parquet. If omitted, rebuild with generateFlowBySector.', + ) + parser.add_argument( + '--rebuild', + action='store_true', + help='Force generateFlowBySector even when --retained is set.', + ) + args = parser.parse_args() + + from bedrock.transform.allocation.derived import ( # noqa: PLC0415 + _load_cornerstone_ghg_fbs_from_gcs, + _select_cornerstone_ghg_fbs_base_name, + ) + from bedrock.transform.flowbysector import FlowBySector # noqa: PLC0415 + + method = args.method or _select_cornerstone_ghg_fbs_base_name() + reference = _load_cornerstone_ghg_fbs_from_gcs(base_name=method) + + if args.retained is not None and not args.rebuild: + retained = pd.read_parquet(args.retained) + else: + retained = pd.DataFrame( + FlowBySector.generateFlowBySector( + method, + download_sources_ok=True, + retain_activity_columns=True, + ) + ) + + if 'ActivityProducedBy' not in retained.columns: + raise SystemExit( + 'Retained FBS lacks ActivityProducedBy; method YAML retain flag ' + 'or retain_activity_columns=True did not reach the output' + ) + + bad = compare(reference, retained) + print( + f'{method}: {len(bad)} of ' + f'{reference.groupby(["Flowable", "SectorProducedBy"]).ngroups} ' + 'Flowable x SectorProducedBy totals differ' + ) + if not bad.empty: + print((bad / 1e9).round(6).to_string()) + raise SystemExit(1) + print('Gas x sector totals match the pinned FBS within tolerance.') + activities = retained.groupby('MetaSources')['ActivityProducedBy'].nunique() + print('Activities per MetaSources (top 10):') + print(activities.sort_values(ascending=False).head(10).to_string()) + + +if __name__ == '__main__': + main() diff --git a/bedrock/analysis/ghg_fbs_pin/prepare_pin.py b/bedrock/analysis/ghg_fbs_pin/prepare_pin.py new file mode 100644 index 00000000..d26f9874 --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/prepare_pin.py @@ -0,0 +1,181 @@ +"""Rebuild a Cornerstone GHG FBS and draft the pin bump. + +Steps this script covers locally (upload is still manual):: + +1. ``FlowBySector.generateFlowBySector`` for the year stem. +2. Confirm gas x sector totals match the currently pinned parquet (and that + ``ActivityProducedBy`` is present when the method retains activities). +3. Write SHA256 + a draft ``cornerstone_ghg_fbs_2024_pin.json`` beside the + parquet, and print the GCS upload / config paths to edit. + +Usage:: + + uv run python -m bedrock.analysis.ghg_fbs_pin.prepare_pin + uv run python -m bedrock.analysis.ghg_fbs_pin.prepare_pin \\ + --method GHG_national_Cornerstone_2024 --skip-gas-check +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import shutil +import time +from pathlib import Path + +from bedrock.analysis.ghg_fbs_pin.compare_gas_e import compare + +OUTPUT_DIR = Path(__file__).resolve().parent / 'output' +PIN_TEMPLATE = { + 'method': '', + 'gcs_sub_bucket': 'transform/output_data', + 'filename': '', + 'sha256': '', +} + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open('rb') as f: + for chunk in iter(lambda: f.read(65536), b''): + digest.update(chunk) + return digest.hexdigest() + + +def _configs_pinning_stem(method: str) -> list[str]: + """Config YAML paths whose cornerstone_ghg_fbs_filename mentions *method* or NFT.""" + configs = Path(__file__).resolve().parents[2] / 'utils' / 'config' / 'configs' + repo_root = Path(__file__).resolve().parents[3] + hits: list[str] = [] + # Canonical / NFT 2024 pins that the year-stem retained rebuild replaces. + needles = ( + f'{method}_v', + 'GHG_national_Cornerstone_nowcast_facilities_transportation_2024_v', + ) + for path in sorted(configs.glob('*.yaml')): + text = path.read_text(encoding='utf-8') + if 'cornerstone_ghg_fbs_filename' not in text: + continue + if any(n in text for n in needles): + hits.append(str(path.relative_to(repo_root))) + return hits + + +def main() -> None: + parser = argparse.ArgumentParser(description=(__doc__ or '').splitlines()[0]) + parser.add_argument( + '--method', + default=None, + help='GHG FBS method stem (default: active config year stem).', + ) + parser.add_argument( + '--skip-gas-check', + action='store_true', + help='Skip compare against the currently pinned parquet.', + ) + parser.add_argument( + '--skip-rebuild', + action='store_true', + help='Use the newest local FBS_DIR parquet for --method instead of regenerating.', + ) + args = parser.parse_args() + + from bedrock.transform.allocation.derived import ( # noqa: PLC0415 + _load_cornerstone_ghg_fbs_from_gcs, + _select_cornerstone_ghg_fbs_base_name, + ) + from bedrock.transform.flowbysector import FlowBySector # noqa: PLC0415 + from bedrock.utils.config.settings import FBS_DIR # noqa: PLC0415 + + method = args.method or _select_cornerstone_ghg_fbs_base_name() + OUTPUT_DIR.mkdir(exist_ok=True) + + if args.skip_rebuild: + candidates = sorted( + FBS_DIR.glob(f'{method}_v*.parquet'), + key=lambda p: p.stat().st_mtime, + ) + if not candidates: + raise SystemExit(f'No local parquet for {method!r} under {FBS_DIR}') + written = candidates[-1] + print(f'Using existing {written.name}') + else: + started = time.time() + FlowBySector.generateFlowBySector(method, download_sources_ok=True) + print(f'{method} rebuilt in {time.time() - started:.0f}s') + written_candidates = [ + p + for p in FBS_DIR.glob(f'{method}_v*') + if p.stat().st_mtime >= started and p.suffix == '.parquet' + ] + if not written_candidates: + raise SystemExit( + f'Rebuild finished but no new {method}_v*.parquet under {FBS_DIR}' + ) + written = max(written_candidates, key=lambda p: p.stat().st_mtime) + + retained = __import__('pandas').read_parquet(written) + if 'ActivityProducedBy' not in retained.columns: + raise SystemExit( + f'{written.name} lacks ActivityProducedBy; retain_activity_columns ' + 'did not reach the output' + ) + n_act = retained['ActivityProducedBy'].nunique(dropna=True) + print(f'ActivityProducedBy: {n_act} distinct values, {len(retained)} rows') + + if not args.skip_gas_check: + reference = _load_cornerstone_ghg_fbs_from_gcs(base_name=method) + bad = compare(reference, retained) + if not bad.empty: + out = OUTPUT_DIR / f'{method}_gas_sector_diffs.csv' + (bad / 1e9).round(6).to_csv(out) + raise SystemExit( + f'Gas x sector totals differ from the current pin ' + f'({len(bad)} cells). Wrote {out}' + ) + print('Gas x sector totals match the current pin.') + + dest = OUTPUT_DIR / written.name + if written.resolve() != dest.resolve(): + shutil.copy2(written, dest) + meta = written.with_suffix('.json') + if meta.is_file(): + shutil.copy2(meta, OUTPUT_DIR / meta.name) + digest = _sha256(dest) + + pin = dict(PIN_TEMPLATE) + pin['method'] = method + pin['filename'] = dest.name + pin['sha256'] = digest + pin_path = OUTPUT_DIR / 'cornerstone_ghg_fbs_2024_pin.DRAFT.json' + pin_path.write_text(json.dumps(pin, indent=2) + '\n', encoding='utf-8') + + print() + print('Draft pin:') + print(json.dumps(pin, indent=2)) + print() + print('Next:') + print( + f' 1. Upload {dest.name} (+ metadata JSON if present) to ' + 'gs://cornerstone-default/transform/output_data/' + ) + print( + ' 2. Copy draft pin over bedrock/utils/snapshots/' + 'cornerstone_ghg_fbs_2024_pin.json' + ) + print( + ' 3. Set cornerstone_ghg_fbs_filename on canonical + waterfall configs ' + f'to {dest.name!r}. Candidates:' + ) + for rel in _configs_pinning_stem(method): + print(f' - {rel}') + print( + ' 4. uv run pytest bedrock/transform/__tests__/test_fbs.py ' + '-m fbs_regeneration -v' + ) + print(f'Draft written to {pin_path}') + + +if __name__ == '__main__': + main() diff --git a/bedrock/analysis/ghg_fbs_pin/upstream_inputs_pin.json b/bedrock/analysis/ghg_fbs_pin/upstream_inputs_pin.json new file mode 100644 index 00000000..843aa429 --- /dev/null +++ b/bedrock/analysis/ghg_fbs_pin/upstream_inputs_pin.json @@ -0,0 +1,111 @@ +{ + "require_stewi_facility_pin": true, + "gcs_sub_bucket_fbs": "transform/output_data", + "gcs_sub_bucket_mut": "flowsa/NowcastMUT", + "methods": { + "GHG_national_Cornerstone_2017": { + "fbs": { + "Energy_manufacturing_national_nowcast_2017": "Energy_manufacturing_national_nowcast_2017_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2017": "Energy_highway_fuel_shares_national_2017_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2017": "Energy_industrial_distillate_national_2017_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2017, + "stage": "after", + "vintage": "v0.5.0_011f41f", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2018": { + "fbs": { + "Energy_manufacturing_national_nowcast_2018": "Energy_manufacturing_national_nowcast_2018_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2018": "Energy_highway_fuel_shares_national_2018_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2018": "Energy_industrial_distillate_national_2018_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2018, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2019": { + "fbs": { + "Energy_manufacturing_national_nowcast_2019": "Energy_manufacturing_national_nowcast_2019_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2019": "Energy_highway_fuel_shares_national_2019_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2019": "Energy_industrial_distillate_national_2019_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2019, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2020": { + "fbs": { + "Energy_manufacturing_national_nowcast_2020": "Energy_manufacturing_national_nowcast_2020_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2020": "Energy_highway_fuel_shares_national_2020_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2020": "Energy_industrial_distillate_national_2020_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2020, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2021": { + "fbs": { + "Energy_manufacturing_national_nowcast_2021": "Energy_manufacturing_national_nowcast_2021_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2021": "Energy_highway_fuel_shares_national_2021_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2021": "Energy_industrial_distillate_national_2021_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2021, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2022": { + "fbs": { + "Energy_manufacturing_national_nowcast_2022": "Energy_manufacturing_national_nowcast_2022_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2022": "Energy_highway_fuel_shares_national_2022_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2022": "Energy_industrial_distillate_national_2022_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2022, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2023": { + "fbs": { + "Energy_manufacturing_national_nowcast_2023": "Energy_manufacturing_national_nowcast_2023_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2023": "Energy_highway_fuel_shares_national_2023_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2023": "Energy_industrial_distillate_national_2023_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2023, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + }, + "GHG_national_Cornerstone_2024": { + "fbs": { + "Energy_manufacturing_national_nowcast_2024": "Energy_manufacturing_national_nowcast_2024_v0.5.0_a166975.parquet", + "Energy_highway_fuel_shares_national_2024": "Energy_highway_fuel_shares_national_2024_v0.6.0_635777c.parquet", + "Energy_industrial_distillate_national_2024": "Energy_industrial_distillate_national_2024_v0.6.0_635777c.parquet" + }, + "nowcast_mut": { + "year": 2024, + "stage": "after", + "vintage": "v0.5.0_a166975", + "tables": ["Make", "Use", "Import", "Margins"] + } + } + } +} From 8c8235a52ea3f4125173dbfcbb03324d0bb3e6d5 Mon Sep 17 00:00:00 2001 From: Ben Young Date: Thu, 22 Oct 2026 15:28:54 -0400 Subject: [PATCH 2/2] fix(analysis): black and drop unused type ignores in pin check Co-authored-by: Cursor --- .../ghg_fbs_pin/check_upstream_inputs.py | 22 +++++++++---------- 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py b/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py index bef00423..d0c63bfa 100644 --- a/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py +++ b/bedrock/analysis/ghg_fbs_pin/check_upstream_inputs.py @@ -53,9 +53,7 @@ def _stem_rivals(directory: Path, stem: str, pinned_name: str) -> list[str]: if not directory.is_dir(): return [] return sorted( - p.name - for p in directory.glob(f'{stem}_v*.parquet') - if p.name != pinned_name + p.name for p in directory.glob(f'{stem}_v*.parquet') if p.name != pinned_name ) @@ -114,7 +112,9 @@ def _local_nowcast_fbas(year: int) -> list[Path]: """Cached Nowcast Use FBAs for *year* under FBA_DIR (poison for MUT pin).""" if not Path(FBA_DIR).is_dir(): return [] - return sorted(Path(FBA_DIR).glob(f'Nowcast_Detail_Use_AfterRedef_{year}_v*.parquet')) + return sorted( + Path(FBA_DIR).glob(f'Nowcast_Detail_Use_AfterRedef_{year}_v*.parquet') + ) def _mut_rivals(year: int, stage: str, table: str, pinned_name: str) -> list[str]: @@ -123,9 +123,7 @@ def _mut_rivals(year: int, stage: str, table: str, pinned_name: str) -> list[str if not directory.is_dir(): return [] return sorted( - p.name - for p in directory.glob(f'{stem}_v*.parquet') - if p.name != pinned_name + p.name for p in directory.glob(f'{stem}_v*.parquet') if p.name != pinned_name ) @@ -145,7 +143,7 @@ def check_nowcast_mut( directory = Path(FBS_DIR) try: - latest = latest_nowcast_mut_vintage(year=year, stage=stage) # type: ignore[arg-type] + latest = latest_nowcast_mut_vintage(year=year, stage=stage) except Exception as exc: problems.append( f'{method}: could not resolve latest NowcastMUT vintage for ' @@ -161,9 +159,9 @@ def check_nowcast_mut( for table in spec['tables']: filename = nowcast_mut_artifact_name( - table, # type: ignore[arg-type] + table, year=year, - stage=stage, # type: ignore[arg-type] + stage=stage, vintage=vintage, ) dest = directory / filename @@ -365,9 +363,9 @@ def _move(src: Path, qdir: Path) -> None: vintage = mut['vintage'] for table in mut['tables']: pinned = nowcast_mut_artifact_name( - table, # type: ignore[arg-type] + table, year=year, - stage=stage, # type: ignore[arg-type] + stage=stage, vintage=vintage, ) for rival in _mut_rivals(year, stage, table, pinned):