[REFACTOR] utils: centralize helpers — add state_utils, move render_resource_input to ui_helpers, consolidate data_utils

utils/data_utils.py

@@ -1,11 +1,13 @@
-import io
 import json
 import re
+from io import BytesIO
+import numpy as np
 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, resource_groups, method_groups
+from config.config import (
+    resource_unit_map, resource_order, method_order,
+    resource_groups, method_groups, secondary_resources, CO2_UNIT,
+)
 
 def get_resource_to_group() -> dict[str, str]:
     """Map each resource name to its resource group.
@@ -45,36 +47,14 @@ def load_slider_data():
         data = json.load(f)
         return data
     
-def render_resource_input(name_clean, r, tooltip=None):
-    """Render a number_input widget for a single resource availability cap.
-
-    Writes the entered value into st.session_state.resource_caps[r].
-
-    Args:
-        name_clean (str): display label (resource name without unit suffix).
-        r (str): canonical resource name used as the session state key.
-        tooltip (str | None): optional help text shown on hover.
+def default_constraint() -> dict:
+    """Return the default constraint dict for a CDR method (active, 100% cap).
 
     Returns:
-        None
+        dict: {"active": True, "cap_type": "percent", "cap_value": 100}.
     """
-    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
-    )
+    return {"active": True, "cap_type": "percent", "cap_value": int(100)}
 
-    st.session_state.resource_caps[r] = cap if cap is not None else 0.0
 
 def is_valid_scenario(scenario):
     """Check that a scenario dict has the minimum required keys to be loaded.
@@ -237,7 +217,6 @@ def neutralize_zero_coefficient_methods(method_costs: dict, method_constraints:
 
 # ---------------------------------------------------------------------------
 # "Other" method expansion helpers
-# ---------------------------------------------------------------------------
 
 def is_other_method(method_name: str) -> bool:
     """True when the method name contains 'other' (case-insensitive).
@@ -261,4 +240,214 @@ def get_other_method_variants(method_name: str, custom_resources: list) -> list:
 
 def make_variant_name(method_name: str, batch_name: str) -> str:
     """Build the display/LP name for a method variant: '<Method>: <Batch>'."""
-    return f"{method_name}: {batch_name}"
+    return f"{method_name}: {batch_name}"
+
+
+# ---------------------------------------------------------------------------
+# Resource / Excel helpers (moved from tab1)
+
+def standard_resource_names() -> set:
+    """Return the flat set of all standard resource names defined in resource_groups."""
+    return {r for rlist in resource_groups.values() for r in rlist}
+
+
+def to_excel_bytes(batches: list) -> bytes:
+    """Serialise all custom resource batches to an Excel file (one row per batch × method).
+
+    Args:
+        batches (list): list of custom batch dicts with keys name, group, amount, unit, methods.
+
+    Returns:
+        bytes: raw bytes of the .xlsx workbook.
+    """
+    rows = []
+    for batch in batches:
+        for m, coefs in batch["methods"].items():
+            rows.append({
+                "name":   batch["name"],
+                "group":  batch["group"],
+                "amount": float(batch["amount"]),
+                "unit":   batch["unit"],
+                "method": m,
+                "min":    float(coefs["min"]),
+                "median": float(coefs["median"]),
+                "max":    float(coefs["max"]),
+            })
+    excel_buffer = BytesIO()
+    with pd.ExcelWriter(excel_buffer, engine="openpyxl") as writer:
+        pd.DataFrame(rows).to_excel(writer, index=False, sheet_name="custom_resources")
+    return excel_buffer.getvalue()
+
+
+def parse_custom_excel(file) -> list:
+    """Parse a custom resource Excel file into a list of batch dicts.
+
+    Expects columns: name, group, amount, unit, method, min, median, max.
+
+    Args:
+        file: file-like object pointing to an .xlsx workbook.
+
+    Returns:
+        list[dict]: one dict per unique batch name with keys name, group, amount, unit, methods.
+
+    Raises:
+        ValueError: if required columns are missing.
+    """
+    df = pd.read_excel(file, sheet_name=0)
+    required = {"name", "group", "amount", "unit", "method", "min", "median", "max"}
+    if not required.issubset(df.columns):
+        raise ValueError(f"Missing columns: {required - set(df.columns)}")
+    batches = []
+    for batch_name in df["name"].unique():
+        df_b = df[df["name"] == batch_name]
+        methods = {
+            row["method"]: {
+                "min":    float(row["min"]),
+                "median": float(row["median"]),
+                "max":    float(row["max"]),
+            }
+            for _, row in df_b.iterrows()
+        }
+        group_name = str(df_b["group"].iloc[0])
+        resource_to_group = get_resource_to_group()
+        batches.append({
+            "name":    str(batch_name),
+            "group":   group_name,
+            "amount":  float(df_b["amount"].iloc[0]),
+            "unit":    resource_unit_map.get(resource_to_group.get(group_name), "?"),
+            "methods": methods,
+        })
+    return batches
+
+
+# ---------------------------------------------------------------------------
+# Coefficient / slider helpers (moved from tab3)
+
+def normalize_costs(costs_dict: dict) -> dict:
+    """Strip unit suffixes from resource keys for legacy scenario compatibility."""
+    return {strip_unit_suffix(r): v for r, v in costs_dict.items()}
+
+
+def get_slider_params(conf: dict) -> tuple:
+    """Derive slider step size and display format from a coefficient config dict.
+
+    Args:
+        conf (dict): slider config with keys "min", "median", "max".
+
+    Returns:
+        tuple[float, str]: (step, fmt).
+    """
+    min_coeff = conf["min"]
+    coeff_range = conf["max"] - min_coeff
+
+    if min_coeff < 1e-10:
+        fmt = "%.11f"
+    elif min_coeff < 1e-08:
+        fmt = "%.9f"
+    elif min_coeff < 1e-05:
+        fmt = "%.6f"
+    elif min_coeff < 0.01:
+        fmt = "%.4f"
+    elif min_coeff < 1:
+        fmt = "%.3f"
+    elif min_coeff < 100:
+        fmt = "%.2f"
+    else:
+        fmt = "%.0f"
+
+    if coeff_range < 0.001:
+        step = 1e-06
+    elif coeff_range < 0.1:
+        step = 0.0001
+    elif coeff_range < 1:
+        step = 0.001
+    elif coeff_range < 10:
+        step = 0.01
+    elif coeff_range < 100:
+        step = 0.1
+    else:
+        step = 1.0
+
+    return step, fmt
+
+
+def is_method_enabled(method: str, resource: str, enabled_methods: dict) -> bool:
+    """Check whether a method is enabled for a given resource.
+
+    Args:
+        method (str): CDR method name.
+        resource (str): resource name.
+        enabled_methods (dict): {resource: {method: bool}} from session state.
+
+    Returns:
+        bool: True if enabled (defaults to True when not explicitly set).
+    """
+    return enabled_methods.get(resource, {}).get(method, True)
+
+
+def is_custom_secondary(method_name: str, custom_r: str, custom_resource_to_sub_group: dict) -> bool:
+    """Return True if the reference resource for custom_r is secondary for method_name.
+
+    Args:
+        method_name (str): CDR method name.
+        custom_r (str): custom batch resource name.
+        custom_resource_to_sub_group (dict): {custom_r: reference_resource}.
+
+    Returns:
+        bool
+    """
+    sub_group = custom_resource_to_sub_group.get(custom_r)
+    return sub_group is not None and (method_name, sub_group) in secondary_resources
+
+
+# ---------------------------------------------------------------------------
+# JSON / results helpers (moved from tab4)
+
+def to_json_safe(obj):
+    """Recursively convert numpy/pandas types to plain Python so json.dumps works."""
+    if isinstance(obj, dict):
+        return {k: to_json_safe(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [to_json_safe(v) for v in obj]
+    if isinstance(obj, pd.Series):
+        return obj.iloc[0] if len(obj) == 1 else obj.tolist()
+    if isinstance(obj, pd.DataFrame):
+        return obj.to_dict(orient="records")
+    if isinstance(obj, np.integer):
+        return int(obj)
+    if isinstance(obj, np.floating):
+        return float(obj)
+    if isinstance(obj, np.ndarray):
+        return obj.tolist()
+    return obj
+
+
+def build_results_df(allocations: dict, method_constraints: dict | None = None) -> pd.DataFrame:
+    """Build a DataFrame of per-method CO₂ allocations with share and cap columns.
+
+    Args:
+        allocations (dict): {method: CO₂ removed (float)}.
+        method_constraints (dict | None): {method: {cap_type, cap_value}} for the Cap column.
+
+    Returns:
+        pd.DataFrame: indexed by method, sorted descending by CO₂ removed.
+    """
+    df = pd.DataFrame.from_dict(allocations, orient="index", columns=[f"{CO2_UNIT} removed"])
+    df[f"{CO2_UNIT} removed"] = pd.to_numeric(df[f"{CO2_UNIT} removed"], errors="coerce")
+    total = df[f"{CO2_UNIT} removed"].sum()
+    if total > 0:
+        df["Total share (%)"] = (df[f"{CO2_UNIT} removed"] / total * 100).round(2)
+
+    if method_constraints:
+        caps = {}
+        for m in df.index:
+            c = method_constraints.get(m, {})
+            if c.get("cap_type") == "absolute":
+                caps[m] = f"{c['cap_value']:,.2f} {CO2_UNIT}/yr"
+            elif c.get("cap_type") == "percent":
+                caps[m] = f"{int(c.get('cap_value', 100))}% of potential"
+            else:
+                caps[m] = f"⚠️ Invalid type ({c.get('cap_type')!r}) — fix in Tab 2"
+        df["Cap"] = pd.Series(caps)
+
+    return df.sort_values(f"{CO2_UNIT} removed", ascending=False)

utils/state_utils.py

@@ -0,0 +1,96 @@
+import streamlit as st
+
+
+def init_custom_resources():
+    """Initialise custom resource session state keys and coerce amounts/coefficients to float.
+
+    Ensures custom_resources, enabled_methods, and auto_disabled_methods exist in
+    session state, then sanitises every batch so numeric fields are float (not str).
+    """
+    if "custom_resources" not in st.session_state:
+        st.session_state.custom_resources = []
+    if "enabled_methods" not in st.session_state:
+        st.session_state.enabled_methods = {}
+    if "auto_disabled_methods" not in st.session_state:
+        st.session_state.auto_disabled_methods = set()
+    for batch in st.session_state.custom_resources:
+        try:
+            batch["amount"] = float(batch["amount"])
+        except (ValueError, TypeError):
+            batch["amount"] = 0.0
+        for m, coefs in batch.get("methods", {}).items():
+            for key in ("min", "median", "max"):
+                try:
+                    coefs[key] = float(coefs[key])
+                except (ValueError, TypeError):
+                    coefs[key] = 0.0
+
+
+def clear_widget_keys():
+    """Delete all Streamlit widget state keys for sliders, manual inputs, and toggles.
+
+    Called after loading a new file or scenario so widgets re-initialise from
+    fresh session-state values instead of showing stale cached values.
+    """
+    prefixes = ("toggle_", "cap_type_", "slider_", "manual_")
+    for k in [k for k in st.session_state if k.startswith(prefixes)]:
+        del st.session_state[k]
+
+
+def apply_auto_disable_logic(method_to_primary_resources, secondary_resources):
+    """Synchronise auto-disabled method state based on current resource availability.
+
+    Runs three steps:
+    1. Count active primary resources per method (custom batches also count).
+    2. Re-enable any method previously auto-disabled that now has resources.
+    3. Disable any method whose primary resource count just dropped to zero.
+
+    Args:
+        method_to_primary_resources (dict[str, list[str]]): method → list of
+            primary resource names (excludes secondary resources).
+        secondary_resources (set[tuple[str, str]]): set of (method, resource)
+            pairs where the resource is optional for that method.
+
+    Returns:
+        None
+    """
+    method_active_resource_count = {}
+    for r, methods in st.session_state.enabled_methods.items():
+        for m, ok in methods.items():
+            if (m, r) in secondary_resources:
+                continue
+            method_active_resource_count.setdefault(m, 0)
+            if ok:
+                method_active_resource_count[m] += 1
+
+    for batch in st.session_state.get("custom_resources", []):
+        if float(batch.get("amount") or 0) <= 0:
+            continue
+        for m_name in batch.get("methods", {}).keys():
+            method_active_resource_count.setdefault(m_name, 0)
+            method_active_resource_count[m_name] += 1
+
+    methods_with_no_active_resource = [
+        m for m in method_to_primary_resources
+        if method_active_resource_count.get(m, 0) == 0
+    ]
+
+    for m in list(st.session_state.auto_disabled_methods):
+        if method_active_resource_count.get(m, 0) > 0:
+            if m in st.session_state.constraints:
+                st.session_state.constraints[m]["active"] = True
+                st.session_state.constraints[m]["cap_type"] = "percent"
+                st.session_state.constraints[m]["cap_value"] = 100
+            st.session_state[f"toggle_{m}"] = True
+            st.session_state[f"slider_{m}"] = 100
+            st.session_state.auto_disabled_methods.discard(m)
+            st.session_state.setdefault("just_activated_methods", set()).add(m)
+
+    for m in methods_with_no_active_resource:
+        st.session_state.auto_disabled_methods.add(m)
+        if st.session_state.constraints.get(m, {}).get("active", True):
+            st.session_state.constraints.setdefault(m, {"active": True, "cap_type": "percent", "cap_value": 100})
+            st.session_state.constraints[m]["active"] = False
+            toggle_key = f"toggle_{m}"
+            if toggle_key in st.session_state:
+                st.session_state[toggle_key] = False

utils/ui_helpers.py

@@ -3,6 +3,38 @@ import streamlit as st
 from config.config import CRRA_COLORS, CRRA_FONT, PLOTLY_TEMPLATE
 
 
+def render_resource_input(name_clean, r, tooltip=None):
+    """Render a number_input widget for a single resource availability cap.
+
+    Writes the entered value into st.session_state.resource_caps[r].
+
+    Args:
+        name_clean (str): display label (resource name without unit suffix).
+        r (str): canonical resource name used as the session state key.
+        tooltip (str | None): optional help text shown on hover.
+
+    Returns:
+        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 apply_chart_style(fig, height=None):
     """Apply the standard CRRA theme (font, colours, margins) to a Plotly figure.