#!/usr/bin/env python """Clean-room verification that the food-processing package is self-contained. Reproduces what an EXTERNAL user (Claudio Beretta / ZHAW FCM) does — import into a FRESH Brightway project via the published loader, then prove from the SHIPPED artifact (not from trust): 1. OGD-only: NO ecoinvent / biosphere3 / EDB database leaks in (the headline assertion — this is what --ogd-strict guarantees at export). 2. Layers present and NOT collapsed (food_processing_replacements on top, bafu + bafu_biosphere at the bottom, tributary middle layers by name). 3. Transparency: a sample composite node has technosphere links + inline provenance/sediment; a sample direct-emission node carries biosphere flows. 4. Computability + sanity ranges: EF 3.1 Climate computes a plausible per-kg PROCESSING value on a hero row, and N of 16 EF 3.1 categories compute non-zero. Composite gate_to_gate nodes whose own processing emissions are zero (upstream unbound) are exercised for non-zero COMPUTE, not a number. Usage: python verify_self_contained.py """ from __future__ import annotations import sys from pathlib import Path HERE = Path(__file__).parent TOP_DB = "food_processing_replacements" TEST_PROJECT = "food_processing_verify_cleanroom" # Per-kg PROCESSING-ONLY plausibility band (kg CO2eq/kg). Gate-to-gate food # processing is energy-light: juice extraction / canning / drying land roughly # 0.01–5 kg CO2eq/kg (e.g. tomato canning ~0.082, spray-drying powders higher). # Cradle-to-gate proxies that bake in upstream can be higher; we only assert the # band on a known gate_to_gate hero row. PROC_BAND = (0.005, 6.0) def main() -> int: import bw2data import bw2calc pkg = HERE / "food_processing_inventory.bw2package" methods = HERE / "food_processing_methods_cf.json" for p in (pkg, methods): if not p.exists(): print(f"[verify] FAILED — missing {p.name}", file=sys.stderr) return 1 if TEST_PROJECT in bw2data.projects: bw2data.projects.delete_project(TEST_PROJECT, delete_dir=True) bw2data.projects.set_current(TEST_PROJECT) assert not list(bw2data.databases), "test project not empty" sys.argv = ["load", "--project", TEST_PROJECT, "--package", str(pkg), "--methods", str(methods)] sys.path.insert(0, str(HERE)) import load_food_processing if load_food_processing.main() != 0: print("[verify] FAILED — loader returned non-zero", file=sys.stderr) return 1 dbs = {d: len(bw2data.Database(d)) for d in bw2data.databases} print(f"[verify] layers: {dbs}") # 1. OGD-only — the headline. food_processing_agribalyse_lci is exempt # from the source-name patterns: it is this package's own layer for # Agribalyse-CONTEXT reconstructions, built from BAFU + open literature # only — the name follows the product semantics, not the data source. leak = [d for d in dbs if d != "food_processing_agribalyse_lci" and ("ecoinvent" in d.lower() or "biosphere3" in d.lower() or d in ("EDB", "EDB_remapped_final", "EDB_bafu_remapped") or "agribalu" in d.lower() or "agribalyse" in d.lower())] if leak: print(f"[verify] FAILED — licensed/non-OGD database leaked: {leak}", file=sys.stderr) return 1 print("[verify] OGD-only OK — no ecoinvent / biosphere3 / EDB layers present") # 2. Not collapsed. if TOP_DB not in dbs or dbs[TOP_DB] < 100: print(f"[verify] FAILED — {TOP_DB} missing or collapsed ({dbs.get(TOP_DB)})", file=sys.stderr) return 1 if "bafu" not in dbs or "bafu_biosphere" not in dbs: print("[verify] FAILED — closure layers (bafu / bafu_biosphere) missing", file=sys.stderr) return 1 top = bw2data.Database(TOP_DB) # 3a. Transparency — a composite node carries technosphere links + provenance. composite = next((a for a in top if list(a.technosphere()) and a.get("provenance")), None) if composite is None: print("[verify] FAILED — no composite node with technosphere links + " "provenance", file=sys.stderr) return 1 n_tech = len(list(composite.technosphere())) has_sed = bool(composite.get("sediment")) foodon = any(t.get("ns") == "foodon" for t in (composite.get("sediment", {}) or {}).get("terms", [])) print(f"[verify] sample composite '{composite.get('name')[:40]}': {n_tech} " f"technosphere links, sediment={has_sed}, FoodOn term={foodon}, " f"provenance keys={sorted((composite.get('provenance') or {}).keys())}") if n_tech < 1 or not composite.get("provenance"): print("[verify] FAILED — composite node not transparent", file=sys.stderr) return 1 # 3b. Transparency — a direct-emission node carries biosphere flows. direct = next((a for a in top if list(a.biosphere())), None) if direct is not None: n_bio = len(list(direct.biosphere())) print(f"[verify] sample direct-emission '{direct.get('name')[:40]}': " f"{n_bio} biosphere flows") # 4. Computability across method families. def score(act, method) -> float: lca = bw2calc.LCA({act: 1}, method) lca.lci(); lca.lcia() return lca.score ef_climate = ("EF 3.1 (BAFU)", "Climate Change") if ef_climate not in bw2data.methods: print("[verify] FAILED — EF 3.1 Climate Change not registered", file=sys.stderr) return 1 # Hero row: a known gate_to_gate composite/direct with real processing # emissions (tomato canning ~0.082). Prefer one whose own EF 3.1 climate > 0. hero = None for a in top: nm = (a.get("name") or "").lower() if "tomato" in nm and ("can" in nm or "peel" in nm): hero = a break if hero is None: # any node with a non-zero EF 3.1 climate for a in top: if abs(score(a, ef_climate)) > 0: hero = a break if hero is None: print("[verify] FAILED — no node computes a non-zero EF 3.1 Climate value", file=sys.stderr) return 1 hero_clim = score(hero, ef_climate) print(f"[verify] hero '{hero.get('name')[:40]}': EF 3.1 Climate = " f"{hero_clim:.4g} kg CO2eq/kg") if not (PROC_BAND[0] <= hero_clim <= PROC_BAND[1]): print(f"[verify] FAILED — hero EF 3.1 climate {hero_clim:.4g} outside " f"plausible processing band {PROC_BAND}", file=sys.stderr) return 1 # Count EF 3.1 categories that compute non-zero on the hero. ef31 = [m for m in bw2data.methods if m[:1] == ("EF 3.1 (BAFU)",) and len(m) == 2] ef_nonzero = sum(1 for m in ef31 if abs(score(hero, m)) > 0) print(f"[verify] EF 3.1: {len(ef31)}/16 registered, {ef_nonzero} compute " f"non-zero on the hero") if len(ef31) < 16: print(f"[verify] FAILED — only {len(ef31)}/16 EF 3.1 categories registered", file=sys.stderr) return 1 # UBP + GWP100 + a GLAM sample present and computing. extras = {} ubp = ("Ecological Scarcity 2021", "total", "UBP - BAFU") if ubp in bw2data.methods: extras["UBP"] = score(hero, ubp) gwp = next((m for m in bw2data.methods if "GWP100 - BAFU" in " | ".join(m)), None) if gwp: extras["GWP100"] = score(hero, gwp) glam = next((m for m in bw2data.methods if "glam" in " | ".join(m).lower()), None) if glam: extras["GLAM(sample)"] = score(hero, glam) for k, v in extras.items(): print(f"[verify] {k}: {v:.4g}") # Whole-package compute sanity: every node must COMPUTE (no crash); count # how many produce a non-zero EF 3.1 climate (zeros are legitimate for # upstream-unbound gate_to_gate composites). computed = nonzero = 0 for a in top: try: s = score(a, ef_climate) computed += 1 if abs(s) > 0: nonzero += 1 except Exception as e: print(f"[verify] FAILED — node {a.get('code')} did not compute: {e}", file=sys.stderr) return 1 print(f"[verify] whole-package compute: {computed}/{len(top)} nodes computed; " f"{nonzero} have non-zero EF 3.1 climate (zeros = upstream-unbound " f"gate_to_gate composites with no own processing emissions)") bw2data.projects.delete_project(TEST_PROJECT, delete_dir=True) print(f"\n[verify] PASS — self-contained, OGD-only, not collapsed, transparent; " f"hero EF 3.1 climate {hero_clim:.4g} kg CO2eq/kg, {len(ef31)}/16 EF 3.1 " f"categories registered, all {computed} nodes compute.") return 0 if __name__ == "__main__": sys.exit(main())