Spaces:
Running
Running
| import io | |
| import json | |
| import re | |
| import pandas as pd | |
| import streamlit as st | |
| import openpyxl | |
| from openpyxl.styles import PatternFill, Font | |
| from config.config import resource_unit_map, resource_order, method_order | |
| def load_slider_data(): | |
| """Load slider data from a JSON file and cache it for performance.""" | |
| with open("data/cost_sliders.json") as f: | |
| data = json.load(f) | |
| return data | |
| def render_resource_input(name_clean, r, tooltip=None): | |
| widget_key = f"cap_{r}" | |
| if widget_key not in st.session_state: | |
| existing = st.session_state.resource_caps.get(r, 0.0) | |
| st.session_state[widget_key] = existing if existing else None | |
| cap = st.number_input( | |
| label=name_clean, | |
| min_value=0.0, | |
| format="%.2f", | |
| value=None, | |
| placeholder="0.00", | |
| key=widget_key, | |
| help=tooltip | |
| ) | |
| st.session_state.resource_caps[r] = cap if cap is not None else 0.0 | |
| def is_valid_scenario(scenario): | |
| base_ok = ( | |
| isinstance(scenario, dict) | |
| and "allocations" in scenario | |
| and "resource_caps" in scenario | |
| and "method_costs" in scenario | |
| and "method_constraints" in scenario | |
| and "summary" in scenario | |
| ) | |
| if not base_ok: | |
| return False | |
| return True | |
| def is_complete_scenario(scenario): | |
| """A scenario that also carries the detailed optimisation outputs and the | |
| per-resource manual allocation state, so reloading it reproduces the | |
| exact same optimisation (not just the same inputs).""" | |
| required_extra = { | |
| "resource_usage", "resource_used_total", "resource_remaining","enabled_methods", | |
| } | |
| return is_valid_scenario(scenario) and required_extra.issubset(scenario.keys()) | |
| def load_inputs_from_excel(file): | |
| xl = pd.read_excel(file, sheet_name=None) | |
| resource_caps = {} | |
| if "resource_caps" in xl: | |
| for _, row in xl["resource_caps"].iterrows(): | |
| resource_caps[row["Resource"]] = float(row["Available Amount"]) | |
| constraints = {} | |
| if "constraints" in xl: | |
| for _, row in xl["constraints"].iterrows(): | |
| constraints[row["method"]] = { | |
| "active": bool(row["active"]), | |
| "cap_type": str(row["cap_type (percent or absolute)"]), | |
| "cap_value": float(row["cap_value"]) | |
| } | |
| costs = {} | |
| if "coefficients" in xl: | |
| for _, row in xl["coefficients"].iterrows(): | |
| m, r, v = row["method"], row["resource"], float(row["value"]) | |
| costs.setdefault(m, {})[r] = v | |
| custom_resources = [] | |
| warnings = [] | |
| valid_resources = set(resource_order) | |
| valid_methods = set(method_order) | |
| if "custom_resources" in xl: | |
| df_custom = xl["custom_resources"] | |
| required_custom = {"name", "group", "amount", "unit", "method", "min", "median", "max"} | |
| if required_custom.issubset(df_custom.columns): | |
| for bname in df_custom["name"].dropna().unique(): | |
| df_b = df_custom[df_custom["name"] == bname] | |
| group = str(df_b["group"].iloc[0]) | |
| if group not in valid_resources: | |
| warnings.append(f"New resource \"{bname}\" - group \"{group}\" not found: check spelling.") | |
| continue | |
| methods = {} | |
| for _, row in df_b.iterrows(): | |
| m = str(row["method"]) | |
| if m not in valid_methods: | |
| warnings.append(f"New resource \"{bname}\" - method \"{m}\" not found: check spelling.") | |
| continue | |
| methods[m] = { | |
| "min": float(row["min"]), | |
| "median": float(row["median"]), | |
| "max": float(row["max"]), | |
| } | |
| custom_resources.append({ | |
| "name": str(bname), | |
| "group": group, | |
| "amount": float(df_b["amount"].iloc[0]), | |
| "unit": str(df_b["unit"].iloc[0]), | |
| "methods": methods, | |
| }) | |
| # Per-resource manual method allocation (which CDR methods are allowed to | |
| # use a given resource). Optional sheet for backward compatibility with | |
| # older templates that predate this feature. | |
| enabled_methods = {} | |
| if "enabled_methods" in xl: | |
| df_enabled = xl["enabled_methods"] | |
| required_enabled = {"resource", "method", "enabled"} | |
| if required_enabled.issubset(df_enabled.columns): | |
| for _, row in df_enabled.iterrows(): | |
| r, m = row["resource"], row["method"] | |
| enabled_methods.setdefault(r, {})[m] = bool(row["enabled"]) | |
| return resource_caps, constraints, costs, custom_resources, enabled_methods, warnings | |
| def strip_unit_suffix(resource_name: str) -> str: | |
| """Remove legacy unit suffix from resource name, e.g. 'Unspecified land (ha/tCO2)' -> 'Unspecified land'""" | |
| return re.sub(r"\s*\(.*?\)\s*$", "", resource_name).strip() | |
| def neutralize_zero_coefficient_methods(method_costs: dict, method_constraints: dict) -> dict: | |
| safe_constraints = {m: c.copy() for m, c in method_constraints.items()} | |
| for method, constraint in safe_constraints.items(): | |
| if not constraint.get("active", True): | |
| continue | |
| coeffs = method_costs.get(method, {}) | |
| all_zero = all(v == 0 for v in coeffs.values()) if coeffs else True | |
| if all_zero: | |
| constraint["active"] = False | |
| return safe_constraints |