CRRA_Optimization_Tool_V2 / tests /test_optimization_stress.py
CarlyneB's picture
[TEST] Add stress tests and edge case coverage for optimisation engine
3c03f4c
Raw
History Blame Contribute Delete
76.8 kB
"""
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()