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 @st.cache_data 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