""" Stress test & edge case coverage for optimisation/optimisation.py Run with: python tests/test_optimisation_stress.py """ import sys, traceback, json from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from optimisation.optimisation import run_optimisation # ── Helpers ─────────────────────────────────────────────────────────────────── PASS = "\033[92mPASS\033[0m" FAIL = "\033[91mFAIL\033[0m" WARN = "\033[93mWARN\033[0m" results = [] def run_case(name, expected_success, **kwargs): try: ok, res = run_optimisation(**kwargs) if ok: total = res.get("total_removed", 0) gain = res.get("resource_actual_gain", {}) if expected_success: status = PASS detail = f"total={total:.4f} gain_keys={list(gain.keys())[:3]}" else: status = WARN detail = f"Expected failure but got success — total={total:.4f}" else: msg = res.get("message", "") if not expected_success: status = PASS detail = f"Correctly failed: {msg[:80]}" else: status = FAIL detail = f"Unexpected failure: {msg[:80]}" print(f" [{status}] {name}\n {detail}") results.append((name, status, detail)) except Exception as e: tb = traceback.format_exc().strip().split("\n")[-1] print(f" [{FAIL}] {name}\n EXCEPTION: {tb}") results.append((name, FAIL, f"EXCEPTION: {tb}")) # ── Realistic base fixtures ──────────────────────────────────────────────────── CAPS = { "Land": 100.0, "Water": 50.0, "Rock": 80.0, "Energy": 200.0, } COSTS = { "Afforestation": {"Land": 0.5, "Water": 0.2}, "Agroforestry": {"Land": 0.3, "Water": 0.1}, "Enhanced Weathering": {"Rock": 0.4, "Energy": 0.3}, "BioCCS": {"Energy": 0.6, "Water": 0.05}, } def base_constraints(cap_type="percent", cap_value=100): return {m: {"active": True, "cap_type": cap_type, "cap_value": cap_value} for m in COSTS} def inactive_constraints(): return {m: {"active": False, "cap_type": "percent", "cap_value": 100} for m in COSTS} # ══════════════════════════════════════════════════════════════════════════════ # SECTION 1 — Baseline & sanity # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 1 · Baseline & sanity ─────────────────────────────────────────") run_case("1.1 Normal run", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS) run_case("1.2 Single method, single resource", expected_success=True, resource_caps={"Land": 100.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Land": 0.5}}) run_case("1.3 All methods at 100% absolute cap", expected_success=True, resource_caps=CAPS, method_costs=COSTS, method_constraints=base_constraints("absolute", 1000.0)) run_case("1.4 Very large caps (numerical scale)", expected_success=True, resource_caps={k: v * 1e9 for k, v in CAPS.items()}, method_constraints=base_constraints(), method_costs=COSTS) run_case("1.5 Very small caps (near-zero precision)", expected_success=True, resource_caps={k: v * 1e-6 for k, v in CAPS.items()}, method_constraints=base_constraints(), method_costs=COSTS) run_case("1.6 All resources = 0", expected_success=True, resource_caps={k: 0.0 for k in CAPS}, method_constraints=base_constraints(), method_costs=COSTS) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 2 — Method constraint edge cases # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 2 · Method constraints ────────────────────────────────────────") run_case("2.1 All methods inactive", expected_success=False, resource_caps=CAPS, method_constraints=inactive_constraints(), method_costs=COSTS) run_case("2.2 One method active, others inactive", expected_success=True, resource_caps=CAPS, method_constraints={ "Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}, "Agroforestry": {"active": False, "cap_type": "percent", "cap_value": 100}, "Enhanced Weathering": {"active": False, "cap_type": "percent", "cap_value": 100}, "BioCCS": {"active": False, "cap_type": "percent", "cap_value": 100}, }, method_costs=COSTS) run_case("2.3 Percent cap = 0 on all methods", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints("percent", 0), method_costs=COSTS) run_case("2.4 Absolute cap = 0 on all methods", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints("absolute", 0.0), method_costs=COSTS) run_case("2.5 Absolute cap very tight (1e-10)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints("absolute", 1e-10), method_costs=COSTS) run_case("2.6 Mixed cap types", expected_success=True, resource_caps=CAPS, method_costs=COSTS, method_constraints={ "Afforestation": {"active": True, "cap_type": "percent", "cap_value": 50}, "Agroforestry": {"active": True, "cap_type": "absolute", "cap_value": 30.0}, "Enhanced Weathering": {"active": True, "cap_type": "percent", "cap_value": 100}, "BioCCS": {"active": True, "cap_type": "absolute", "cap_value": 1e-3}, }) run_case("2.7 Percent cap = 100 forces one method to take everything", expected_success=True, resource_caps={"Land": 100.0}, method_constraints={ "Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}, "Agroforestry": {"active": True, "cap_type": "percent", "cap_value": 5}, }, method_costs={"Afforestation": {"Land": 0.5}, "Agroforestry": {"Land": 0.3}}) run_case("2.8 Method missing from constraints dict", expected_success=True, resource_caps=CAPS, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs=COSTS) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 3 — Coefficient edge cases # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 3 · Coefficient edge cases ────────────────────────────────────") run_case("3.1 Method with all-zero coefficients", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={**COSTS, "Ghost": {"Land": 0.0, "Water": 0.0}}) run_case("3.2 Method with no resources in costs (empty dict)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={**COSTS, "Empty": {}}) run_case("3.3 All coefficients very large", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={"Afforestation": {"Land": 1e6, "Water": 1e6}}) run_case("3.4 Coefficient = 0 for one resource, positive for another", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={"Afforestation": {"Land": 0.0, "Water": 0.2}}) run_case("3.5 Resource in costs but not in resource_caps", expected_success=True, resource_caps={"Land": 100.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Land": 0.5, "MissingResource": 0.1}}) run_case("3.6 method_costs empty dict", expected_success=False, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={}) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 4 — enabled_methods # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 4 · enabled_methods ────────────────────────────────────────────") run_case("4.1 Disable one resource for one method", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods={"Land": {"Afforestation": False}}) run_case("4.2 Disable all resources for one method (effectively inactive)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods={"Land": {"Afforestation": False}, "Water": {"Afforestation": False}}) run_case("4.3 Disable all resources for all methods → all inactive", expected_success=False, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods={ "Land": {m: False for m in COSTS}, "Water": {m: False for m in COSTS}, "Rock": {m: False for m in COSTS}, "Energy": {m: False for m in COSTS}, }) run_case("4.4 enabled_methods = None (default)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods=None) run_case("4.5 enabled_methods references non-existent method", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods={"Land": {"NonExistentMethod": False}}) run_case("4.6 enabled_methods = True explicitly (already default)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, enabled_methods={"Land": {"Afforestation": True}}) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 5 — Custom resources (no overlap with standard) # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 5 · Custom resources (independent) ─────────────────────────────") custom_independent = [{ "name": "Mine A", "group": "NewGroup", "amount": 50.0, "unit": "t", "methods": { "Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}, } }] run_case("5.1 Custom resource, independent group", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=custom_independent) run_case("5.2 Custom resource, amount = 0", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Empty Mine", "group": "NewGroup", "amount": 0.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }]) run_case("5.3 Custom resource, no methods listed", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Orphan Batch", "group": "NewGroup", "amount": 50.0, "unit": "t", "methods": {}, }]) run_case("5.4 Custom resource, method not in method_costs", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Ghost Mine", "group": "NewGroup", "amount": 50.0, "unit": "t", "methods": {"NonExistentMethod": {"min": 0.1, "median": 0.2, "max": 0.3}}, }]) run_case("5.5 Multiple custom resources, different groups", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[ {"name": "Mine A", "group": "GroupA", "amount": 30.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}}, {"name": "Field B", "group": "GroupB", "amount": 60.0, "unit": "ha", "methods": {"Afforestation": {"min": 0.4, "median": 0.5, "max": 0.6}}}, ]) run_case("5.6 Custom resource = None", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=None) run_case("5.7 Custom resources = [] (empty list)", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[]) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 6 — Custom resources as substitutes (same group as standard) # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 6 · Custom resources (same group as standard) ──────────────────") custom_substitute = [{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }] run_case("6.1 Custom batch same group as standard, std cap > 0", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=custom_substitute) run_case("6.2 Custom batch same group, std cap = 0", expected_success=True, resource_caps={**CAPS, "Rock": 0.0}, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=custom_substitute) run_case("6.3 Custom batch same group, std cap = 0, batch cap = 0", expected_success=True, resource_caps={**CAPS, "Rock": 0.0}, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 0.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }]) run_case("6.4 Two custom batches, same group", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[ {"name": "Mine A", "group": "Rock", "amount": 30.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}}, {"name": "Mine B", "group": "Rock", "amount": 20.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.2, "median": 0.3, "max": 0.4}}}, ]) run_case("6.5 Custom batch same group, different coefficient than standard", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs={**COSTS, "Enhanced Weathering": {"Rock": 0.4, "Energy": 0.3, "Mine A": 0.1}}, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.05, "median": 0.1, "max": 0.15}}, }]) run_case("6.6 Custom batch, std disabled via enabled_methods", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=custom_substitute, enabled_methods={"Rock": {"Enhanced Weathering": False}}) run_case("6.7 Two methods use same custom batch", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": { "Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}, "BioCCS": {"min": 0.1, "median": 0.2, "max": 0.3}, }, }]) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 7 — Infeasible / degenerate LP # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 7 · Infeasible / degenerate LP ─────────────────────────────────") run_case("7.1 Empty resource_caps dict", expected_success=True, resource_caps={}, method_constraints=base_constraints(), method_costs=COSTS) run_case("7.2 resource_caps = None", expected_success=False, resource_caps=None, method_constraints=base_constraints(), method_costs=COSTS) run_case("7.3 Single method, single resource, coeff = 0 → inactive", expected_success=False, resource_caps={"Land": 100.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Land": 0.0}}) run_case("7.4 All caps = 0, all coeffs > 0", expected_success=True, resource_caps={k: 0.0 for k in CAPS}, method_constraints=base_constraints(), method_costs=COSTS) run_case("7.5 Percent caps sum to < 100 (feasible, just limits portfolio)", expected_success=True, resource_caps=CAPS, method_costs=COSTS, method_constraints={m: {"active": True, "cap_type": "percent", "cap_value": 10} for m in COSTS}) run_case("7.6 method_costs has method not in method_constraints", expected_success=True, resource_caps=CAPS, method_constraints={m: {"active": True, "cap_type": "percent", "cap_value": 100} for m in list(COSTS.keys())[:2]}, method_costs=COSTS) run_case("7.7 Negative cap_value absolute (should still run)", expected_success=True, resource_caps=CAPS, method_costs=COSTS, method_constraints={m: {"active": True, "cap_type": "absolute", "cap_value": -1.0} for m in COSTS}) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 8 — Result integrity checks # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 8 · Result integrity ───────────────────────────────────────────") def check_integrity(name, **kwargs): try: ok, res = run_optimisation(**kwargs) if not ok: print(f" [{WARN}] {name} — optimisation failed: {res.get('message','')[:60]}") results.append((name, WARN, "opt failed")) return total = res["total_removed"] method_usage = res["method_usage"] resource_usage = res["resource_usage"] resource_used_total = res["resource_used_total"] resource_remaining = res["resource_remaining"] errors = [] # total_removed >= 0 if total < -1e-9: errors.append(f"total_removed < 0: {total}") # sum(method_usage) ≈ total_removed method_sum = sum(method_usage.values()) if abs(method_sum - total) > 1e-6: errors.append(f"method_usage sum {method_sum:.6f} ≠ total {total:.6f}") # Build full cap dict: standard + custom batches full_caps = dict(kwargs["resource_caps"]) for b in (kwargs.get("custom_resources") or []): full_caps[b["name"]] = float(b.get("amount", 0)) # resource_remaining = cap - used (within tolerance) for r, remaining in resource_remaining.items(): cap = full_caps.get(r, 0) used = resource_used_total.get(r, 0) expected_remaining = cap - used if abs(remaining - expected_remaining) > 1e-6: errors.append(f"remaining[{r}] {remaining:.6f} ≠ cap-used {expected_remaining:.6f}") # no resource over-consumed for r, used in resource_used_total.items(): cap = full_caps.get(r, 0) if used > cap + 1e-6: errors.append(f"resource {r} over-consumed: used={used:.6f} cap={cap}") if errors: print(f" [{FAIL}] {name}") for e in errors: print(f" ✗ {e}") results.append((name, FAIL, "; ".join(errors))) else: print(f" [{PASS}] {name} total={total:.4f} methods={len(method_usage)}") results.append((name, PASS, f"total={total:.4f}")) except Exception as e: tb = traceback.format_exc().strip().split("\n")[-1] print(f" [{FAIL}] {name} — EXCEPTION: {tb}") results.append((name, FAIL, f"EXCEPTION: {tb}")) check_integrity("8.1 Integrity — normal run", resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS) check_integrity("8.2 Integrity — tight percent cap (10%)", resource_caps=CAPS, method_costs=COSTS, method_constraints={m: {"active": True, "cap_type": "percent", "cap_value": 10} for m in COSTS}) check_integrity("8.3 Integrity — absolute caps", resource_caps=CAPS, method_costs=COSTS, method_constraints=base_constraints("absolute", 20.0)) check_integrity("8.4 Integrity — with custom batch substitute", resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }]) check_integrity("8.5 Integrity — two batches same group", resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[ {"name": "Mine A", "group": "Rock", "amount": 30.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}}, {"name": "Mine B", "group": "Rock", "amount": 20.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.2, "median": 0.3, "max": 0.4}}}, ]) check_integrity("8.6 Integrity — std disabled for method with batch", resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }], enabled_methods={"Rock": {"Enhanced Weathering": False}}) check_integrity("8.7 Integrity — std cap = 0, batch present", resource_caps={**CAPS, "Rock": 0.0}, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 50.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }]) check_integrity("8.8 Integrity — large caps, many methods, mixed constraints", resource_caps={k: v * 1000 for k, v in CAPS.items()}, method_costs=COSTS, method_constraints={ "Afforestation": {"active": True, "cap_type": "percent", "cap_value": 30}, "Agroforestry": {"active": True, "cap_type": "absolute", "cap_value": 5000}, "Enhanced Weathering": {"active": True, "cap_type": "percent", "cap_value": 60}, "BioCCS": {"active": False, "cap_type": "absolute", "cap_value": 100}, }, custom_resources=[ {"name": "Mine A", "group": "Rock", "amount": 1000.0, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}}, ]) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 9 — Perturbation / resource_actual_gain checks # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 9 · Perturbation (resource_actual_gain) ────────────────────────") def check_gain(name, expect_binding_resources, **kwargs): try: ok, res = run_optimisation(**kwargs) if not ok: print(f" [{WARN}] {name} — failed") results.append((name, WARN, "opt failed")) return gain = res.get("resource_actual_gain", {}) binding = {r: v for r, v in gain.items() if v > 1e-9} non_negative = all(v >= -1e-9 for v in gain.values()) status = PASS if non_negative else FAIL note = "negative gain detected" if not non_negative else f"binding={list(binding.keys())}" print(f" [{status}] {name} {note}") results.append((name, status, note)) except Exception as e: tb = traceback.format_exc().strip().split("\n")[-1] print(f" [{FAIL}] {name} — EXCEPTION: {tb}") results.append((name, FAIL, f"EXCEPTION: {tb}")) check_gain("9.1 Gain non-negative — normal", [], resource_caps=CAPS, method_constraints=base_constraints(), method_costs=COSTS) check_gain("9.2 Gain — single binding resource", ["Land"], resource_caps={"Land": 1.0, "Water": 1e9, "Rock": 1e9, "Energy": 1e9}, method_constraints=base_constraints(), method_costs=COSTS) check_gain("9.3 Gain — all caps zero (all gains = 0)", [], resource_caps={k: 0.0 for k in CAPS}, method_constraints=base_constraints(), method_costs=COSTS) check_gain("9.4 Gain — custom batch binding", [], resource_caps={**CAPS, "Rock": 0.0}, method_constraints=base_constraints(), method_costs=COSTS, custom_resources=[{ "name": "Mine A", "group": "Rock", "amount": 0.1, "unit": "t", "methods": {"Enhanced Weathering": {"min": 0.3, "median": 0.4, "max": 0.5}}, }]) # ══════════════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════════════ print("\n── SUMMARY ────────────────────────────────────────────────────────────────") total_cases = len(results) passed = sum(1 for _, s, _ in results if "PASS" in s) failed = sum(1 for _, s, _ in results if "FAIL" in s) warned = sum(1 for _, s, _ in results if "WARN" in s) print(f" Total: {total_cases} PASS: {passed} FAIL: {failed} WARN: {warned}") if failed: print("\n Failed cases:") for name, status, detail in results: if "FAIL" in status: print(f" • {name}: {detail}") if warned: print("\n Warnings (unexpected behavior):") for name, status, detail in results: if "WARN" in status: print(f" • {name}: {detail}") print() # ══════════════════════════════════════════════════════════════════════════════ # SECTION 10 — "Other" method expansion (OR semantics) # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 10 · Other-method expansion ───────────────────────────────────") from utils.data_utils import is_other_method, get_other_method_variants, make_variant_name # ── 10.1 is_other_method detection ────────────────────────────────────────── def _check(name, condition, detail=""): icon = PASS if condition else FAIL print(f" [{icon}] {name}\n {detail or ('OK' if condition else 'FAILED')}") results.append((name, icon, detail)) _check("10.1a is_other_method('Enhanced weathering - Other mineral')", is_other_method("Enhanced weathering - Other mineral")) _check("10.1b is_other_method('Bio-char - Other biomass')", is_other_method("Bio-char - Other biomass")) _check("10.1c is_other_method('Afforestation') is False", not is_other_method("Afforestation")) _check("10.1d is_other_method('Mineral OAE / Ocean liming') is False", not is_other_method("Mineral OAE / Ocean liming")) # ── 10.2 get_other_method_variants ────────────────────────────────────────── _cr2 = [ {"name": "Ganite", "group": "Other mineral", "amount": 100.0, "unit": "Mt", "methods": {"Enhanced weathering - Other mineral": {"min": 1.0, "median": 2.5, "max": 4.0}}}, {"name": "Dunite", "group": "Other mineral", "amount": 80.0, "unit": "Mt", "methods": {"Enhanced weathering - Other mineral": {"min": 1.0, "median": 2.5, "max": 4.0}}}, ] v = get_other_method_variants("Enhanced weathering - Other mineral", _cr2) _check("10.2a Two batches → 2 variants", len(v) == 2, f"got {len(v)}") _check("10.2b Afforestation → 0 variants", len(get_other_method_variants("Afforestation", _cr2)) == 0) _check("10.2c Bio-char Other biomass (no batch) → 0", len(get_other_method_variants("Bio-char - Other biomass", _cr2)) == 0) _check("10.2d make_variant_name correct", make_variant_name("Enhanced weathering - Other mineral", "Ganite") == "Enhanced weathering - Other mineral: Ganite") # ── 10.3 LP: single Other method + single batch → bounded and correct ──────── run_case("10.3 Other method + 1 batch → bounded", expected_success=True, resource_caps={"Energy": 200.0}, method_constraints={ "Enhanced weathering - Other mineral - Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "Enhanced weathering - Other mineral - Ganite": {"Ganite": 2.5, "Energy": 0.1}, }, custom_resources=[], # batch already promoted to standard resource # Ganite cap comes from resource_caps # Note: in real app resource_caps_for_lp includes Ganite; we simulate that here ) # Simulated LP with Ganite in resource_caps (as the app does after expansion) run_case("10.3b Other method + 1 batch + batch in resource_caps → correct usage", expected_success=True, resource_caps={"Energy": 200.0, "Ganite": 100.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5, "Energy": 0.1}, }, custom_resources=[], ) # ── 10.4 LP: two variants, each limited by its own batch ──────────────────── run_case("10.4 Two variants independent — each limited by own batch", expected_success=True, resource_caps={"Energy": 1e9, "Ganite": 10.0, "Dunite": 20.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, "EW-Dunite": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5, "Energy": 0.0}, "EW-Dunite": {"Dunite": 2.5, "Energy": 0.0}, }, custom_resources=[], ) def check_result_10_4(): ok, res = run_optimisation( resource_caps={"Energy": 1e9, "Ganite": 10.0, "Dunite": 20.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, "EW-Dunite": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5, "Energy": 0.0}, "EW-Dunite": {"Dunite": 2.5, "Energy": 0.0}, }, custom_resources=[], ) if not ok: _check("10.4b Variant allocations correct", False, "LP failed") return allocs = res["method_usage"] ganite_co2 = allocs.get("EW-Ganite", 0) dunite_co2 = allocs.get("EW-Dunite", 0) expected_ganite = 10.0 / 2.5 # 4.0 expected_dunite = 20.0 / 2.5 # 8.0 _check("10.4b EW-Ganite uses Ganite cap fully", abs(ganite_co2 - expected_ganite) < 1e-4, f"got {ganite_co2:.4f}, expected {expected_ganite:.4f}") _check("10.4c EW-Dunite uses Dunite cap fully", abs(dunite_co2 - expected_dunite) < 1e-4, f"got {dunite_co2:.4f}, expected {expected_dunite:.4f}") _check("10.4d Variants independent (total = sum of individuals)", abs(res["total_removed"] - (expected_ganite + expected_dunite)) < 1e-4, f"total={res['total_removed']:.4f}, expected {expected_ganite + expected_dunite:.4f}") check_result_10_4() # ── 10.5 LP: variant inactive → only the other runs ───────────────────────── run_case("10.5 One variant inactive → only other runs", expected_success=True, resource_caps={"Ganite": 10.0, "Dunite": 20.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, "EW-Dunite": {"active": False, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5}, "EW-Dunite": {"Dunite": 2.5}, }, custom_resources=[], ) def check_10_5(): ok, res = run_optimisation( resource_caps={"Ganite": 10.0, "Dunite": 20.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, "EW-Dunite": {"active": False, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5}, "EW-Dunite": {"Dunite": 2.5}, }, custom_resources=[], ) if not ok: _check("10.5b Dunite inactive → only Ganite runs", False, "LP failed") return _check("10.5b Only EW-Ganite allocates", res["method_usage"].get("EW-Dunite", 0) == 0.0, f"Dunite alloc={res['method_usage'].get('EW-Dunite', 0)}") check_10_5() # ── 10.6 LP: variant with percent cap ─────────────────────────────────────── # Note: percent cap = 50% with two methods: EW-Ganite ≤ 50% of total(all methods). # With equal-cost methods, each gets 50% → each at its resource cap. def check_10_6(): # Two balanced variants: each limited to 50% of total portfolio. # With symmetric caps (Ganite=100, Dunite=100, coeff=2.5), optimal is 50/50 split. ok, res = run_optimisation( resource_caps={"Ganite": 100.0, "Dunite": 100.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 50}, "EW-Dunite": {"active": True, "cap_type": "percent", "cap_value": 50}, }, method_costs={ "EW-Ganite": {"Ganite": 2.5}, "EW-Dunite": {"Dunite": 2.5}, }, custom_resources=[], ) if not ok: _check("10.6 Variant percent cap (50%)", False, f"LP failed: {res.get('message', '')}") return alloc_g = res["method_usage"].get("EW-Ganite", 0) alloc_d = res["method_usage"].get("EW-Dunite", 0) total = alloc_g + alloc_d # Each should be ≤ 50% of total _check("10.6 Variant 50% percent cap — EW-Ganite ≤ 50% of total", total > 0 and alloc_g <= total * 0.5 + 1e-4, f"EW-Ganite={alloc_g:.4f}, total={total:.4f}") _check("10.6b Variant 50% percent cap — EW-Dunite ≤ 50% of total", total > 0 and alloc_d <= total * 0.5 + 1e-4, f"EW-Dunite={alloc_d:.4f}, total={total:.4f}") check_10_6() # ── 10.7 LP: no unbounded when batch in resource_caps (regression) ─────────── def check_10_7(): # This is the exact scenario that caused the unbounded LP bug. # Without the fix, custom_resources_for_lp containing the batch would # put it in custom_batch_names → excluded from standard_resources → no constraint. ok, res = run_optimisation( resource_caps={"Ganite": 50.0, "Dunite": 30.0}, method_constraints={ "EW-Ganite": {"active": True, "cap_type": "percent", "cap_value": 100}, "EW-Dunite": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Ganite": {"Ganite": 1.0}, "EW-Dunite": {"Dunite": 1.0}, }, custom_resources=[], # batches NOT in custom_resources → treated as standard ) bounded = ok and res.get("total_removed", 0) < 1e8 _check("10.7 No unbounded LP when batches are standard resources", bounded, f"total={res.get('total_removed', 'n/a')} ok={ok}") check_10_7() # ══════════════════════════════════════════════════════════════════════════════ # SECTION 11 — Usage threshold & tiny coefficients # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 11 · Usage threshold & tiny coefficients ──────────────────────") def check_11_1(): # Coefficient = 1e-11 (previously filtered by 1e-9 threshold → showed as 0 used). # Use Energy as the binding resource (prevents LP from going to 1e11) and # NonArableLand as the tiny-coeff resource to verify it's tracked in usage. ok, res = run_optimisation( resource_caps={"NonArableLand": 1.0, "Energy": 100.0}, method_constraints={"DACCS": {"active": True, "cap_type": "absolute", "cap_value": 100.0}}, method_costs={"DACCS": {"NonArableLand": 1e-11, "Energy": 1.0}}, custom_resources=[], ) if not ok: _check("11.1 Tiny coeff 1e-11 not filtered", False, f"LP failed: {res.get('message', '')}") return usage = res.get("resource_usage", {}).get("NonArableLand", {}).get("DACCS", 0) # DACCS produces 100 MtCO2 (bounded by Energy cap), usage = 100 * 1e-11 = 1e-9 _check("11.1 Tiny coeff 1e-11 tracked in resource_usage", usage > 0, f"usage={usage:.2e} (should be > 0)") check_11_1() def check_11_2(): # Coefficient exactly 0 → method should be neutralized ok, res = run_optimisation( resource_caps={"Land": 100.0}, method_constraints={"Method": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Method": {"Land": 0.0}}, custom_resources=[], ) _check("11.2 Coeff=0 → method neutralized (no result)", not ok or res.get("total_removed", 0) == 0, f"total={res.get('total_removed', 0):.4f}") check_11_2() def check_11_3(): # resource_remaining should be positive when tiny amount used ok, res = run_optimisation( resource_caps={"NonArableLand": 1.0, "Energy": 100.0}, method_constraints={"DACCS": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"DACCS": {"NonArableLand": 1e-11, "Energy": 1.0}}, custom_resources=[], ) if not ok: _check("11.3 resource_remaining correct for tiny usage", False, "LP failed") return remaining = res.get("resource_remaining", {}).get("NonArableLand", None) _check("11.3 resource_remaining ≥ 0 for tiny coeff", remaining is not None and remaining >= 0, f"remaining={remaining}") check_11_3() # ══════════════════════════════════════════════════════════════════════════════ # SECTION 12 — cap_type validation # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 12 · cap_type validation ──────────────────────────────────────") run_case("12.1 cap_type='percent' valid", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints("percent", 100), method_costs=COSTS) run_case("12.2 cap_type='absolute' valid", expected_success=True, resource_caps=CAPS, method_constraints=base_constraints("absolute", 500), method_costs=COSTS) run_case("12.3 cap_type='relative' invalid → ValueError caught", expected_success=False, resource_caps=CAPS, method_constraints={"Afforestation": {"active": True, "cap_type": "relative", "cap_value": 50}}, method_costs={"Afforestation": {"Land": 0.5}}) run_case("12.4 cap_type=None → ValueError caught", expected_success=False, resource_caps=CAPS, method_constraints={"Afforestation": {"active": True, "cap_type": None, "cap_value": 50}}, method_costs={"Afforestation": {"Land": 0.5}}) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 13 — Production data smoke test # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 13 · Production data smoke test ───────────────────────────────") import json, pandas as pd from utils.data_utils import strip_unit_suffix try: with open("data/cost_sliders.json") as f: cost_sliders_raw = json.load(f) df_csv = pd.read_csv("data/inputs.csv") # Build method_costs from medians, stripping unit suffixes so resource names # match resource_caps keys (e.g. "Electrical energy (TWh/MtCO₂)" → "Electrical energy"). prod_costs = {} for method, resources in cost_sliders_raw.items(): prod_costs[method] = {strip_unit_suffix(r): conf.get("median", 0.0) for r, conf in resources.items()} # Realistic resource caps (generous) prod_caps = { "Arable land": 50.0, "Non-arable land": 200.0, "Other land": 100.0, "Forest biomass": 500.0, "Non-forest biomass": 300.0, "Other biomass": 200.0, "Electrical energy": 5e6, "Thermal energy high grade": 2e6, "Thermal energy low grade": 3e6, "Thermal energy other grade": 1e6, "Geologic storage": 100.0, "Other chemical": 50.0, "Basalt mineral": 1000.0, "Olivine mineral": 500.0, "Wollastonite mineral": 200.0, "Other mineral": 300.0, "Salt water": 1000.0, "Other water": 500.0, "Shoreline": 10.0, } method_list = list(prod_costs.keys()) prod_constraints = {m: {"active": True, "cap_type": "percent", "cap_value": 100} for m in method_list} run_case("13.1 Production data — all methods active", expected_success=True, resource_caps=prod_caps, method_constraints=prod_constraints, method_costs=prod_costs) # 13.2 Only DACCS methods daccs = {m: c for m, c in prod_costs.items() if "DACCS" in m} daccs_constraints = {m: {"active": True, "cap_type": "percent", "cap_value": 100} for m in daccs} run_case("13.2 Production data — DACCS methods only", expected_success=True, resource_caps=prod_caps, method_constraints=daccs_constraints, method_costs=daccs) # 13.3 Tight energy cap → some DACCS infeasible tight_caps = dict(prod_caps) tight_caps["Electrical energy"] = 1.0 tight_caps["Thermal energy high grade"] = 1.0 tight_caps["Thermal energy low grade"] = 1.0 run_case("13.3 Production data — tight energy caps", expected_success=True, resource_caps=tight_caps, method_constraints=prod_constraints, method_costs=prod_costs) # 13.4 All caps = 0 → zero removal def check_13_4(): ok, res = run_optimisation( resource_caps={k: 0.0 for k in prod_caps}, method_constraints=prod_constraints, method_costs=prod_costs) _check("13.4 All caps=0 → zero removal", ok and res.get("total_removed", -1) == 0.0, f"total={res.get('total_removed', 'n/a')}") check_13_4() # 13.5 Absolute cap = 5 MtCO2 per method abs_constraints = {m: {"active": True, "cap_type": "absolute", "cap_value": 5.0} for m in method_list} run_case("13.5 Production data — 5 MtCO2 absolute cap per method", expected_success=True, resource_caps=prod_caps, method_constraints=abs_constraints, method_costs=prod_costs) # 13.6 Mixed cap types mixed_constraints = {} for i, m in enumerate(method_list): if i % 2 == 0: mixed_constraints[m] = {"active": True, "cap_type": "percent", "cap_value": 80} else: mixed_constraints[m] = {"active": True, "cap_type": "absolute", "cap_value": 10.0} run_case("13.6 Production data — mixed percent/absolute caps", expected_success=True, resource_caps=prod_caps, method_constraints=mixed_constraints, method_costs=prod_costs) except Exception as e: _check("13.x Production data load", False, str(e)) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 14 — smart_display_format (no floating-point noise) # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 14 · smart_display_format ─────────────────────────────────────") import sys; sys.path.insert(0, ".") from utils.ui_helpers import smart_display_format _fmt_cases = [ # (input, expected_output, description) (462961.1, "462961.1", "Large value from CSV — no float noise"), (462961.09999999997671693, "462961.1","Same float stored differently — same display"), (1e-11, "0.00000000001", "Tiny coefficient — fixed decimal not sci notation"), (1.1e-11, "0.000000000011", "Tiny coefficient with mantissa"), (2.5, "2.5", "Typical coefficient"), (0.5, "0.5", "Half"), (100.0, "100.0", "Integer-like float"), (1000.0, "1000.0", "Larger integer-like float"), (0.001, "0.001", "Small decimal"), (0.1234567890123456, "0.1234567890123456", "High-precision decimal"), (1.0, "1.0", "One"), (0.0, "", "Zero → empty string"), (5e-15, "", "Below 1e-14 threshold → empty string"), (9.99e-15, "", "Just below threshold → empty string"), (1e-14, "0.00000000000001", "Exactly at threshold → shown"), (2e-14, "0.00000000000002", "Just above threshold → shown"), (-0.5, "-0.5", "Negative value"), (-1e-11, "-0.00000000001", "Negative tiny coefficient"), ] for val, expected, desc in _fmt_cases: result = smart_display_format(val) ok = result == expected _check(f"14 · {desc}", ok, f"got {result!r}, expected {expected!r}") # ══════════════════════════════════════════════════════════════════════════════ # SECTION 15 — Batch shared between Other-method variant and non-Other method # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 15 · Shared batch: Other variant vs non-Other method ──────────") # Scenario: "Limestone" batch assigned to both: # - "EW-Other mineral" (Other → expanded to variant EW-Other mineral: Limestone) # - "Mineral OAE" (non-Other → should keep access to Limestone via type-A) # # EW-Other mineral coeff = 2.5 Mt/MtCO₂ → 4.2/2.5 = 1.68 MtCO₂ # Mineral OAE coeff = 1.743 Mt/MtCO₂ → 4.2/1.743 = 2.41 MtCO₂ (more efficient) # # Without the fix, Limestone is removed from custom_resources_for_lp but NOT # injected into Mineral OAE's applied_costs → Mineral OAE has no Limestone # constraint → LP allocates all Limestone to EW-Other mineral: Limestone → 1.68. # With the fix, Mineral OAE gets Limestone injected → LP prefers Mineral OAE → 2.41. _limestone_batch = { "name": "Limestone", "group": "Other mineral", "amount": 4.2, "unit": "Mt", "methods": { "EW-Other mineral": {"min": 0.5, "median": 2.5, "max": 25.0}, "Mineral OAE": {"min": 1.4, "median": 1.743, "max": 5.0}, }, } # Simulate what tab4 expansion does to applied_costs / resource_caps before # calling run_optimisation. After fix, Mineral OAE gets Limestone in its costs. def _simulate_expansion(base_costs, custom_res): """Replicate the tab4 "Other" expansion logic (the same code path).""" import sys, os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from utils.data_utils import is_other_method, get_other_method_variants, make_variant_name from config.config import hidden_resource_names as hidden applied = {m: dict(c) for m, c in base_costs.items()} resource_caps_lp = {"Other mineral": 0.0} # standard pool = 0 expanded_batch_names: set = set() for m in list(applied.keys()): if not is_other_method(m): continue batches = get_other_method_variants(m, custom_res) if not batches: continue all_names = {b["name"] for b in batches} for b in batches: vname = make_variant_name(m, b["name"]) vcosts = {r: c for r, c in applied[m].items() if r not in hidden and r not in all_names} vcosts[b["name"]] = b["methods"][m]["median"] applied[vname] = vcosts resource_caps_lp[b["name"]] = float(b["amount"]) expanded_batch_names.add(b["name"]) del applied[m] cr_for_lp = [b for b in custom_res if b["name"] not in expanded_batch_names] # THE FIX: inject batch coeff into non-Other methods for b in custom_res: if b["name"] not in expanded_batch_names: continue for mname, coeff_d in b.get("methods", {}).items(): if mname not in applied: continue applied[mname][b["name"]] = coeff_d.get("median", 0.0) return applied, resource_caps_lp, cr_for_lp def check_15_1(): """Mineral OAE should win over EW-Other mineral for Limestone (lower coeff = more efficient).""" base_costs = { "EW-Other mineral": {"Other mineral": 0.0}, # standard pool empty "Mineral OAE": {}, } applied, caps, cr_lp = _simulate_expansion(base_costs, [_limestone_batch]) # After expansion: variant "EW-Other mineral: Limestone" and "Mineral OAE" (with Limestone injected) _check("15.1a EW-Other mineral removed from applied_costs after expansion", "EW-Other mineral" not in applied) variant_name = "EW-Other mineral: Limestone" _check("15.1b Variant created in applied_costs", variant_name in applied, f"keys={list(applied.keys())}") _check("15.1c Mineral OAE has Limestone injected", "Limestone" in applied.get("Mineral OAE", {}), f"Mineral OAE costs={applied.get('Mineral OAE', {})}") _check("15.1d Limestone coeff for Mineral OAE = 1.743", abs(applied.get("Mineral OAE", {}).get("Limestone", 0) - 1.743) < 1e-9, f"got {applied.get('Mineral OAE', {}).get('Limestone', 'missing')}") _check("15.1e Limestone in resource_caps_lp with correct amount", abs(caps.get("Limestone", 0) - 4.2) < 1e-9, f"got {caps.get('Limestone', 'missing')}") _check("15.1f Limestone not in custom_resources_for_lp (no double constraint)", all(b["name"] != "Limestone" for b in cr_lp), f"cr_lp names={[b['name'] for b in cr_lp]}") check_15_1() def check_15_2(): """LP should allocate Limestone to Mineral OAE (coeff 1.743 < 2.5) — more efficient.""" # Simulate the state after expansion (with fix applied) ok, res = run_optimisation( resource_caps={"Limestone": 4.2}, method_constraints={ "EW-Other mineral: Limestone": {"active": True, "cap_type": "percent", "cap_value": 100}, "Mineral OAE": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Other mineral: Limestone": {"Limestone": 2.5}, "Mineral OAE": {"Limestone": 1.743}, }, custom_resources=[], ) if not ok: _check("15.2 LP: Mineral OAE preferred over EW-Other mineral", False, f"LP failed: {res.get('message', '')}") return alloc_oae = res["method_usage"].get("Mineral OAE", 0) alloc_ew = res["method_usage"].get("EW-Other mineral: Limestone", 0) expected_oae = 4.2 / 1.743 # ≈ 2.409 _check("15.2a Mineral OAE gets all Limestone (more efficient)", abs(alloc_oae - expected_oae) < 1e-3, f"Mineral OAE={alloc_oae:.4f}, expected≈{expected_oae:.4f}") _check("15.2b EW-Other mineral: Limestone gets 0 (LP correctly rejects it)", alloc_ew < 1e-6, f"EW alloc={alloc_ew:.6f}") _check("15.2c Total ≈ 2.409 MtCO₂ (Mineral OAE wins)", abs(res["total_removed"] - expected_oae) < 1e-3, f"total={res['total_removed']:.4f}") check_15_2() def check_15_3(): """Regression: without the fix, EW-Other mineral: Limestone would win (wrong).""" # Simulate WITHOUT fix: Mineral OAE has NO Limestone in costs ok, res = run_optimisation( resource_caps={"Limestone": 4.2}, method_constraints={ "EW-Other mineral: Limestone": {"active": True, "cap_type": "percent", "cap_value": 100}, "Mineral OAE": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Other mineral: Limestone": {"Limestone": 2.5}, "Mineral OAE": {}, # NO Limestone → old buggy state }, custom_resources=[], ) if not ok: _check("15.3 Regression check (old buggy behaviour)", False, f"LP failed") return alloc_ew = res["method_usage"].get("EW-Other mineral: Limestone", 0) # Without fix: Mineral OAE has no Limestone → EW wins (wrong, less efficient) _check("15.3 Without fix EW-Other gets Limestone (confirms the bug existed)", alloc_ew > 1.0, # EW-Other would get 4.2/2.5=1.68 f"EW alloc={alloc_ew:.4f} (expected ~1.68 in buggy state)") check_15_3() def check_15_4(): """Multiple non-Other methods sharing the same expanded batch all get the coeff injected.""" # Limestone shared between EW-Other mineral, Mineral OAE, AND Electrochemical OAE limestone_multi = { "name": "Limestone", "group": "Other mineral", "amount": 4.2, "unit": "Mt", "methods": { "EW-Other mineral": {"min": 0.5, "median": 2.5, "max": 25.0}, "Mineral OAE": {"min": 1.4, "median": 1.743, "max": 5.0}, "Electrochemical OAE": {"min": 2.0, "median": 2.95, "max": 3.5}, }, } base_costs = { "EW-Other mineral": {"Other mineral": 0.0}, "Mineral OAE": {}, "Electrochemical OAE": {}, } applied, caps, cr_lp = _simulate_expansion(base_costs, [limestone_multi]) _check("15.4a Mineral OAE gets Limestone injected", "Limestone" in applied.get("Mineral OAE", {})) _check("15.4b Electrochemical OAE gets Limestone injected", "Limestone" in applied.get("Electrochemical OAE", {})) _check("15.4c Electrochemical OAE coeff = 2.95", abs(applied.get("Electrochemical OAE", {}).get("Limestone", 0) - 2.95) < 1e-9, f"got {applied.get('Electrochemical OAE', {}).get('Limestone', 'missing')}") check_15_4() run_case("15.5 Three-way competition for Limestone — Mineral OAE wins (coeff 1.743 lowest)", expected_success=True, resource_caps={"Limestone": 4.2}, method_constraints={ "EW-Other: Limestone": {"active": True, "cap_type": "percent", "cap_value": 100}, "Mineral OAE": {"active": True, "cap_type": "percent", "cap_value": 100}, "Electrochem OAE": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "EW-Other: Limestone": {"Limestone": 2.5}, # 1.68 MtCO₂/4.2Mt "Mineral OAE": {"Limestone": 1.743}, # 2.41 MtCO₂/4.2Mt ← best "Electrochem OAE": {"Limestone": 2.95}, # 1.42 MtCO₂/4.2Mt }, custom_resources=[], ) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 16 — Multi-resource method with simultaneous custom-batch substitutes # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 16 · Multi-resource method (High-temp DACCS) — Leontief across roles ──") # Real coefficients from data/inputs.csv (median column), for # "High-temp DACCS (solid-sorbent/liquid-solvent)" — a genuine multi-resource # (Leontief/AND) method: every unit of CO2 removed needs ALL FIVE resources # simultaneously, they are NOT substitutes for each other. DACCS_METHOD = "High-temp DACCS" E_C = 423787.8 # Electrical energy, MWh/MtCO2 L_C = 0.000223 # Non-arable land, Mha/MtCO2 T_C = 1750000.0 # Thermal energy high grade, MWh/MtCO2 S_C = 1.0 # Geologic storage, Mt/MtCO2 W_C = 0.0047 # Other water, bcm/MtCO2 DACCS_COSTS = { DACCS_METHOD: { "Electrical energy": E_C, "Non-arable land": L_C, "Thermal energy high grade": T_C, "Geologic storage": S_C, "Other water": W_C, } } DACCS_CONSTRAINTS = {DACCS_METHOD: {"active": True, "cap_type": "percent", "cap_value": 100}} # All 5 standard caps set to "10 MtCO2-equivalent" → clean baseline of 10.0 TEN_MT_CAPS = { "Electrical energy": 10 * E_C, "Non-arable land": 10 * L_C, "Thermal energy high grade": 10 * T_C, "Geologic storage": 10 * S_C, "Other water": 10 * W_C, } energy_a = { "name": "Energy A", "group": "Electrical energy", "amount": 10 * E_C, "unit": "MWh", "methods": {DACCS_METHOD: {"min": E_C, "median": E_C, "max": E_C}}, } land_a = { "name": "Land A", "group": "Non-arable land", "amount": 10 * L_C, "unit": "Mha", "methods": {DACCS_METHOD: {"min": L_C, "median": L_C, "max": L_C}}, } def check_16_x(name, custom_resources, caps_override, expected_total): ok, res = run_optimisation( resource_caps={**TEN_MT_CAPS, **caps_override}, method_constraints=DACCS_CONSTRAINTS, method_costs=DACCS_COSTS, custom_resources=custom_resources, ) if not ok: _check(name, False, f"LP failed: {res.get('message', '')}") return None total = res["total_removed"] _check(name, abs(total - expected_total) < 1e-3, f"got {total:.4f}, expected {expected_total:.4f}") return res # 16.1 Baseline: standard only, no batches → all 5 resources bind at 10 MtCO2 check_16_x("16.1 Baseline — standard only, no batches", [], {}, 10.0) # 16.2 Energy A alone + all standards: Energy role doubles to 20-equiv, but # Land/Thermal/Storage/Water stay at 10-equiv → total stays 10 (Leontief holds: # extra electricity alone doesn't unlock more removal). check_16_x("16.2 Energy A alone + all standards — still bound by Land/Thermal/Water", [energy_a], {}, 10.0) # 16.3 Land A alone + all standards: symmetric case. check_16_x("16.3 Land A alone + all standards — still bound by Energy/Thermal/Water", [land_a], {}, 10.0) # 16.4 Energy A AND Land A together + unchanged Thermal/Storage/Water: both # substituted roles double capacity, but Thermal/Storage/Water (untouched) # remain the binding constraint at 10 → total should STILL be 10, not 20. res_16_4 = check_16_x( "16.4 Energy A + Land A together + standards — still bound by Thermal/Water", [energy_a, land_a], {}, 10.0) # 16.4b Regression guard: the standard resource must still show non-zero # usage (no D2 misfire zeroing it out). NOTE: batch usage (Energy A / Land A) # is NOT asserted here — this exact test is degenerate (standard alone already # covers the full 10, so the solver is free to leave the batches at 0; that's # a valid optimum, not a bug). See 16.4bis for a non-degenerate mixed-usage check. if res_16_4: usage = res_16_4["resource_usage"] _check("16.4b Electrical energy (standard) used > 0", usage.get("Electrical energy", {}).get(DACCS_METHOD, 0) > 1e-6, f"usage={usage.get('Electrical energy', {})}") _check("16.4d Non-arable land (standard) used > 0", usage.get("Non-arable land", {}).get(DACCS_METHOD, 0) > 1e-6, f"usage={usage.get('Non-arable land', {})}") # 16.4bis Same total (10), but standard Energy/Land caps are DELIBERATELY too # small (6-equiv) to cover it alone → the LP is forced (not just allowed) to # draw part of its Energy and Land needs from the batches. This removes the # degeneracy of 16.4 (where standard alone already covered the full 10, so the # solver was free to leave the batches at 0 — that's not a bug, just an # arbitrary vertex choice when nothing distinguishes the two sources). res_16_4bis = check_16_x( "16.4bis Standard Energy/Land capped below demand — forces real mixed usage", [energy_a, land_a], {"Electrical energy": 6 * E_C, "Non-arable land": 6 * L_C}, 10.0) if res_16_4bis: usage = res_16_4bis["resource_usage"] _check("16.4bis-b Electrical energy (standard) used ≈ 6 (its own cap, fully used)", abs(usage.get("Electrical energy", {}).get(DACCS_METHOD, 0) - 6 * E_C) < 1, f"usage={usage.get('Electrical energy', {})}") _check("16.4bis-c Energy A (batch) used ≈ 4 MtCO2-equiv (the missing part)", abs(usage.get("Energy A", {}).get(DACCS_METHOD, 0) - 4 * E_C) < 1, f"usage={usage.get('Energy A', {})}") _check("16.4bis-d Non-arable land (standard) used ≈ 6-equiv (its own cap, fully used)", abs(usage.get("Non-arable land", {}).get(DACCS_METHOD, 0) - 6 * L_C) < 1e-6, f"usage={usage.get('Non-arable land', {})}") _check("16.4bis-e Land A (batch) used ≈ 4 MtCO2-equiv (the missing part)", abs(usage.get("Land A", {}).get(DACCS_METHOD, 0) - 4 * L_C) < 1e-6, f"usage={usage.get('Land A', {})}") # 16.5 Energy A + Land A together, with Thermal/Storage/Water made generous # (non-binding) → now Energy (20-equiv) and Land (20-equiv) ARE the bottleneck # → total should rise to 20 (pooling genuinely unlocks extra capacity when it # actually is the limiting resource). check_16_x( "16.5 Energy A + Land A, Thermal/Storage/Water non-binding — total rises to 20", [energy_a, land_a], {"Thermal energy high grade": 1000 * T_C, "Geologic storage": 1000 * S_C, "Other water": 1000 * W_C}, 20.0) # 16.6 Standard Electrical energy AND Non-arable land forced to 0 (exclusivity # for BOTH roles at once), Thermal/Storage/Water generous → total should be # bound solely by Energy A's own 10-equiv amount and Land A's own 10-equiv # amount (both maxed, no cross-contamination between the two D2 branches). check_16_x( "16.6 Standard Energy=0 & Land=0, batches only — bound by batch amounts (10)", [energy_a, land_a], {"Electrical energy": 0.0, "Non-arable land": 0.0, "Thermal energy high grade": 1000 * T_C, "Geologic storage": 1000 * S_C, "Other water": 1000 * W_C}, 10.0) # ══════════════════════════════════════════════════════════════════════════════ # SECTION 17 — Family-resource pooling (OR semantics for same-group resources) # ══════════════════════════════════════════════════════════════════════════════ print("\n── SECTION 17 · Family-resource pooling (resource_to_group) ──────────────") FAM_GROUPS = {"Arable land": "Land", "Other land": "Land", "Non-arable land": "Land"} # ── 17.1 Baseline (no resource_to_group) — two same-family resources are # still required at once (AND): total bounded by whichever is more binding. ── def check_17_1(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 5.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Arable land": 1.0, "Other land": 1.0}}, custom_resources=[], ) _check("17.1 No resource_to_group → AND (bounded by the tighter resource, 5.0)", ok and abs(res["total_removed"] - 5.0) < 1e-6, f"total={res.get('total_removed') if ok else res}") check_17_1() # ── 17.2 Same setup, pooled — OR/blend: total can exceed either resource # alone, up to the sum (equal coefficients). ───────────────────────────────── def check_17_2(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 5.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Arable land": 1.0, "Other land": 1.0}}, custom_resources=[], resource_to_group=FAM_GROUPS, ) _check("17.2 Pooled → OR (total = sum of both caps, 15.0)", ok and abs(res["total_removed"] - 15.0) < 1e-6, f"total={res.get('total_removed') if ok else res}") if ok: usage = res["resource_usage"] arable_used = usage.get("Arable land", {}).get("Afforestation", 0) other_used = usage.get("Other land", {}).get("Afforestation", 0) _check("17.2b Both pool members fully used (10 + 5)", abs(arable_used - 10.0) < 1e-6 and abs(other_used - 5.0) < 1e-6, f"Arable land={arable_used}, Other land={other_used}") check_17_2() # ── 17.3 Unequal coefficients — pooled capacity = sum of each member's own # ceiling (cap / coefficient), regardless of which is cheaper. ────────────── def check_17_3(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 10.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Arable land": 0.5, "Other land": 2.0}}, custom_resources=[], resource_to_group=FAM_GROUPS, ) expected = 10.0 / 0.5 + 10.0 / 2.0 # 20 + 5 = 25 _check("17.3 Pooled, unequal coefficients → total = 25", ok and abs(res["total_removed"] - expected) < 1e-4, f"total={res.get('total_removed') if ok else res}, expected={expected}") check_17_3() # ── 17.4 A custom batch attached to one pooled standard resource joins the # same pool (batch + its reference resource + the sibling family resource all # complement each other). ──────────────────────────────────────────────────── def check_17_4(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 5.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Arable land": 1.0, "Other land": 1.0}}, custom_resources=[{ "name": "Reserve land", "group": "Arable land", "amount": 5.0, "unit": "Mha", "methods": {"Afforestation": {"min": 1.0, "median": 1.0, "max": 1.0}}, }], resource_to_group=FAM_GROUPS, ) _check("17.4 Pool + custom batch on a pool member → total = 20 (10+5+5)", ok and abs(res["total_removed"] - 20.0) < 1e-4, f"total={res.get('total_removed') if ok else res}") check_17_4() # ── 17.5 One pool member at zero cap — fully sourced from the other, no crash. ── def check_17_5(): ok, res = run_optimisation( resource_caps={"Arable land": 0.0, "Other land": 8.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Afforestation": {"Arable land": 1.0, "Other land": 1.0}}, custom_resources=[], resource_to_group=FAM_GROUPS, ) _check("17.5 One pool member at 0 → total = 8 (fully from the other)", ok and abs(res["total_removed"] - 8.0) < 1e-6, f"total={res.get('total_removed') if ok else res}") check_17_5() # ── 17.6 A method with only ONE resource from the family (no sibling) is # never pooled, even when resource_to_group is supplied — singleton case is # unaffected. ───────────────────────────────────────────────────────────────── def check_17_6(): ok, res = run_optimisation( resource_caps={"Arable land": 6.0}, method_constraints={"Agroforestry": {"active": True, "cap_type": "percent", "cap_value": 100}}, method_costs={"Agroforestry": {"Arable land": 2.0}}, custom_resources=[], resource_to_group=FAM_GROUPS, ) _check("17.6 Single family member (no sibling) → unaffected (total = 3.0)", ok and abs(res["total_removed"] - 3.0) < 1e-6, f"total={res.get('total_removed') if ok else res}") check_17_6() # ── 17.7 Percent cap on a pooled method is computed against its pooled (OR) # standalone potential, not the AND-bounded one. ──────────────────────────── def check_17_7(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 5.0}, method_constraints={"Afforestation": {"active": True, "cap_type": "percent", "cap_value": 50}}, method_costs={"Afforestation": {"Arable land": 1.0, "Other land": 1.0}}, custom_resources=[], resource_to_group=FAM_GROUPS, ) _check("17.7 50% cap of pooled potential (15) → total = 7.5", ok and abs(res["total_removed"] - 7.5) < 1e-4, f"total={res.get('total_removed') if ok else res}") check_17_7() # ── 17.8 Two methods sharing a pooled resource with a non-pooled method that # also draws on one of the family members — the non-pooled method's own usage # must not be swallowed by the pool. ──────────────────────────────────────── def check_17_8(): ok, res = run_optimisation( resource_caps={"Arable land": 10.0, "Other land": 5.0}, method_constraints={ "Afforestation": {"active": True, "cap_type": "percent", "cap_value": 100}, "Cropland management": {"active": True, "cap_type": "percent", "cap_value": 100}, }, method_costs={ "Afforestation": {"Arable land": 1.0, "Other land": 1.0}, "Cropland management": {"Arable land": 1.0}, }, custom_resources=[], resource_to_group=FAM_GROUPS, ) # Both methods compete for total CO2; Arable land (10) is shared between # Afforestation's pool and Cropland's direct use, Other land (5) only # available to Afforestation's pool. Total achievable = 15 either way # (Cropland can use Arable land directly, or Afforestation can via the pool) # — what matters here is that it solves without error and fully uses both land pools. total_arable = 0.0 total_other = 0.0 if ok: usage = res["resource_usage"] total_arable = sum(usage.get("Arable land", {}).values()) total_other = sum(usage.get("Other land", {}).values()) _check("17.8 Pool alongside a non-pooled method sharing a family member — solves, fully uses both", ok and abs(total_arable - 10.0) < 1e-4 and abs(total_other - 5.0) < 1e-4, f"ok={ok} arable_used={total_arable} other_used={total_other} total={res.get('total_removed') if ok else res}") check_17_8()