Spaces:
Sleeping
Sleeping
| """ | |
| 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() | |