import pandas as pd import streamlit as st from config.config import CO2_UNIT, method_groups, method_order, method_tooltips, method_notes, secondary_resources, hidden_resource_names, section_intros from utils.ui_helpers import show_bar from utils.data_utils import strip_unit_suffix, get_other_method_variants, make_variant_name, default_constraint, get_resource_to_group def _make_slider_cb(m): """on_change for the percent-cap slider: write the new value into constraints.""" def _cb(): st.session_state.constraints[m]["cap_value"] = st.session_state[f"slider_{m}"] return _cb def _make_manual_cb(m): """on_change for the absolute-cap number_input: write the new value into constraints.""" def _cb(): v = st.session_state.get(f"manual_{m}") st.session_state.constraints[m]["cap_value"] = float(v) if v is not None else 0.0 return _cb def get_hidden_resources_for_method(m, method_to_primary_resources, cost_sliders_norm): """Return (required, optional) lists of hidden resource names for method m. Returns (None, None) if the method has at least one non-hidden primary resource or a custom batch already assigned. """ primary = method_to_primary_resources.get(m, []) if not primary or not all(r in hidden_resource_names for r in primary): return None, None custom_resources = st.session_state.get("custom_resources", []) if any(m in b.get("methods", {}) for b in custom_resources): return None, None required = [r for r in primary if r in hidden_resource_names] optional = [ r for r in cost_sliders_norm.get(m, {}) if r in hidden_resource_names and (m, r) in secondary_resources ] return required, optional def get_all_missing_required(m, cost_sliders_norm): """Return all required resources for method m that are not yet available. A resource (standard or hidden) is available if resource_caps.get(r, 0.0) > 0 OR if a custom batch with group == r is assigned to m — a custom batch substitutes for its reference resource regardless of whether that resource is a generic "Other X" (hidden) one or a specifically named standard one (e.g. a "Basalt A" batch referencing "Basalt mineral"). A resource the user has explicitly disabled for this method via the "Related CDR Methods" popover (Tab 1) is treated as missing too, unless a custom batch covers it — otherwise tab4 silently zeroes its coefficient (via enabled_methods) while this tab still reports the method as active, a discrepancy that can lead to "No active methods" once every resource of a method is disabled this way. Secondary resources and resources with median <= 0 are excluded. Family pooling: when two or more of the method's resources share the same resource family (e.g. "Arable land" and "Other land" are both "Land" — see build_family_pools in optimisation.py), they are OR-substitutable rather than all-required. A resource in such a family is only reported as missing if every member of that family is unavailable; a family member currently enabled as a secondary resource (its slider toggled on in Tab 3) also counts as available, since it becomes a real substitute in the LP. Returns: list[str]: missing required resource names (standard and hidden combined). """ resource_caps = st.session_state.get("resource_caps", {}) custom_resources = st.session_state.get("custom_resources", []) enabled_methods = st.session_state.get("enabled_methods", {}) method_costs = st.session_state.get("costs", {}).get(m, {}) covered_by_batch = { b["group"] for b in custom_resources if m in b.get("methods", {}) } resource_to_group = get_resource_to_group() def _available(r): if r in covered_by_batch: return True if resource_caps.get(r, 0.0) <= 0: return False return enabled_methods.get(r, {}).get(m, True) # Family membership includes this method's primary resources plus any # secondary resource currently switched on (cost > 0) — both act as real # substitutes for one another in the LP once active. family_members: dict[str, list[str]] = {} for r in cost_sliders_norm.get(m, {}): is_secondary = (m, r) in secondary_resources if is_secondary and method_costs.get(r, 0.0) <= 0: continue family_members.setdefault(resource_to_group.get(r, r), []).append(r) missing = [] for r, coeff_dict in cost_sliders_norm.get(m, {}).items(): if (m, r) in secondary_resources: continue if coeff_dict.get("median", 0.0) <= 0: continue siblings = family_members.get(resource_to_group.get(r, r), [r]) if len(siblings) > 1: if not any(_available(s) for s in siblings): missing.append(r) elif not _available(r): missing.append(r) return missing def get_uncovered_hidden_resources(m, cost_sliders_norm): """Return (required, optional) hidden resources for method m that leave it unconstrained. A hidden resource is "uncovered" only when the user has provided no resource at all from the same group — neither a standard cap (> 0 in tab1) nor a custom batch assigned to this method. If at least one standard resource in the group has a positive cap, the method is already constrained by it and the missing hidden resource does not inflate results. Returns: tuple[list[str], list[str]]: (required, optional) uncovered hidden resource names. required — must be added for the method to run (not in secondary_resources). optional — in secondary_resources, shown as informational only. Returns ([], []) when nothing is uncovered. """ custom_resources = st.session_state.get("custom_resources", []) # A hidden resource r (e.g. "Other mineral") is covered when any custom batch # whose group == r is assigned to this method — the batch name differs from # the group name (e.g. "Limestone" covers "Other mineral"). covered_groups = { b["group"] for b in custom_resources if m in b.get("methods", {}) } required, optional = [], [] for r, coeff_dict in cost_sliders_norm.get(m, {}).items(): if r not in hidden_resource_names: continue if coeff_dict.get("median", 0.0) <= 0: continue if r in covered_groups: continue if (m, r) in secondary_resources: optional.append(r) else: required.append(r) return required, optional def _render_resource_status(m, cost_sliders_norm): """Render an expander showing all resources of method m with their availability status. Standard resources: ✅ if cap > 0 and not disabled for this method via the "Related CDR Methods" popover (Tab 1), ❌ otherwise — unless a custom batch covers it. Hidden resources: ✅ if covered by a custom batch, ❌ if not. Secondary resources are shown in a separate section. Family pooling: when 2+ of the method's required resources share the same resource family (e.g. "Forest biomass" / "Other biomass" are both "Biomass" — see build_family_pools in optimisation.py), only one of them is actually needed. A family member that isn't set is shown as ✅ "optional" (not a blocking ❌) once a sibling already satisfies the family, since adding it is a bonus, not a requirement. """ resource_caps = st.session_state.get("resource_caps", {}) custom_resources = st.session_state.get("custom_resources", []) enabled_methods = st.session_state.get("enabled_methods", {}) resource_to_group = get_resource_to_group() # Map resource group → list of batch names covering it for method m (any # named standard resource or hidden "Other X" resource can be substituted # by a custom batch referencing it). covered_by_batch: dict[str, list[str]] = {} for b in custom_resources: if m in b.get("methods", {}): covered_by_batch.setdefault(b["group"], []).append(b["name"]) all_res = cost_sliders_norm.get(m, {}) def _is_available(r): batch_names = covered_by_batch.get(r) if r in hidden_resource_names: return bool(batch_names) disabled_for_method = not enabled_methods.get(r, {}).get(m, True) return (resource_caps.get(r, 0.0) > 0 and not disabled_for_method) or bool(batch_names) # Group this method's required (non-secondary) resources by family so # siblings can be recognised as OR-substitutable below. primary_by_family: dict[str, list[str]] = {} for r, coeff_dict in all_res.items(): if coeff_dict.get("median", 0.0) <= 0 or (m, r) in secondary_resources: continue primary_by_family.setdefault(resource_to_group.get(r, r), []).append(r) required_rows = [] optional_rows = [] for r, coeff_dict in all_res.items(): if coeff_dict.get("median", 0.0) <= 0: continue batch_names = covered_by_batch.get(r) available = _is_available(r) if r in hidden_resource_names: # Show actual batch name(s) instead of the generic "Other X" group display_r = ", ".join(batch_names) if batch_names else r source = "Add as new resource in Resource inputs" else: disabled_for_method = not enabled_methods.get(r, {}).get(m, True) display_r = ", ".join(batch_names) if (not resource_caps.get(r, 0.0) > 0 and batch_names) else r source = ( "Re-enable this method in the Related CDR Methods popover (Resource inputs tab)" if disabled_for_method and resource_caps.get(r, 0.0) > 0 else "Set quantity in Resource inputs" ) is_secondary = (m, r) in secondary_resources siblings = primary_by_family.get(resource_to_group.get(r, r), [r]) if not is_secondary else [r] available_siblings = [s for s in siblings if s != r and _is_available(s)] if available and len(siblings) > 1: icon, hint, is_blocking = "✅", "", False elif not available and available_siblings: # Family already satisfied by a sibling — not a blocker, just a bonus. icon, is_blocking = "✅", False hint = ( f"{', '.join(available_siblings)} already covers this " f"(only one of {', '.join(siblings)} is required). You can still add " f"{r} too." ) elif not available and len(siblings) > 1: icon, is_blocking = "❌", True hint = f"{source} (only one of {', '.join(siblings)} is required)" else: icon, is_blocking = ("✅", False) if available else ("❌", True) hint = "" if available else source row = (icon, display_r, hint, is_blocking) if is_secondary: optional_rows.append(row) else: required_rows.append(row) with st.expander("Why is this method disabled?", expanded=False): note = method_notes.get(m) if note: st.info(note, icon=":material/lightbulb:") if required_rows: st.markdown("**Required resources**") for _, r, hint, is_blocking in required_rows: color = "#FF5F4D" if is_blocking else "#55C95B" label = f'{r}' if hint: label += f' : {hint}' st.markdown(label, unsafe_allow_html=True) if optional_rows: st.markdown("**Optional resources**") for _, r, hint, is_blocking in optional_rows: color = "#FF5F4D" if is_blocking else "#55C95B" label = f'{r}' if hint: label += f' : {hint}' st.markdown(label, unsafe_allow_html=True) st.caption("Optional resources can be added in the **Resource coefficients** tab (tab 3).") def render_method(m, cols, i, cost_sliders_norm): """Render the constraint widget group (toggle + cap slider/input) for one method. Writes the result into st.session_state.constraints[m]. Args: m (str): CDR method name. cols (list): list of Streamlit column objects. i (int): index used to pick the column (i % 3). Returns: None """ with cols[i % 3]: tooltip = method_tooltips.get(m, None) if m not in st.session_state.constraints: st.session_state.constraints[m] = default_constraint() constraint = st.session_state.constraints[m] toggle_key = f"toggle_{m}" cap_type_key = f"cap_type_{m}" # Block the method if any required resource is missing (standard at 0 or # hidden resource not covered by a custom batch). Run this before the # regular st.toggle so we never touch toggle_{m} for blocked methods — # the pre-render sync already set it and Streamlit would raise # "cannot be modified after widget is instantiated" otherwise. missing_required = get_all_missing_required(m, cost_sliders_norm) if missing_required: st.session_state.constraints[m]["active"] = False st.session_state.pop(toggle_key, None) st.toggle(m, value=False, key=toggle_key, disabled=True, help=tooltip) _render_resource_status(m, cost_sliders_norm) return was_active = constraint["active"] st.session_state.setdefault(toggle_key, was_active) is_active = st.toggle(m, key=toggle_key, help=tooltip) st.session_state.constraints[m]["active"] = is_active if is_active and not was_active: st.session_state.constraints[m]["cap_type"] = "percent" st.session_state.constraints[m]["cap_value"] = 100 st.session_state[cap_type_key] = "percent" if not is_active: st.session_state.pop(cap_type_key, None) return # If cap_type is invalid (e.g. "nan" from a corrupted import), show a # warning and reset to "percent" so the radio and slider render correctly. # The flag is cleared once the user interacts with the radio. if constraint.get("_cap_type_invalid") or constraint.get("cap_type") not in ("percent", "absolute"): st.warning( f"Invalid **cap_type** value (`{constraint.get('cap_type')}`) loaded from file. " "Select a constraint type below to fix it.", icon=":material/warning:" ) constraint["cap_type"] = "percent" constraint["cap_value"] = 100 constraint.pop("_cap_type_invalid", None) st.session_state.pop(cap_type_key, None) st.session_state.pop(f"slider_{m}", None) st.session_state.pop(f"manual_{m}", None) # cap_type radio keeps a key so the user's choice persists across # reruns; slider and number_input are keyless so they always read # directly from the constraint — no stale session-state value possible. st.session_state.setdefault(cap_type_key, constraint["cap_type"]) # Capture the cap_type that was active in the previous render so we can # detect a mode switch (absolute → percent) below. old_cap_type = constraint.get("cap_type", "percent") cap_type = st.radio( "Cap value type", ["percent", "absolute"], horizontal=True, key=cap_type_key, format_func=lambda x: "% of its total potential" if x == "percent" else f"Absolute ({CO2_UNIT}/yr)", label_visibility="collapsed", ) st.session_state.constraints[m]["cap_type"] = cap_type # When the user switches from absolute → percent the stored cap_value is # an absolute number (MtCO₂/yr) which is meaningless as a percentage. # Reset both the constraint and the slider key to the default 100 % so # neither value nor widget is "stuck" at whatever number was left over. if cap_type == "percent" and old_cap_type == "absolute": constraint["cap_value"] = 100 st.session_state[f"slider_{m}"] = 100 st.session_state.pop(f"manual_{m}", None) if cap_type == "absolute" and old_cap_type == "percent": constraint["cap_value"] = 0.0 st.session_state.pop(f"manual_{m}", None) st.session_state.pop(f"slider_{m}", None) if cap_type == "percent": # Seed the slider key if absent or if it's spuriously 0 while the # constraint still records a non-zero intent. constraint["cap_value"] # is only written via on_change callbacks (user-initiated), so it # faithfully reflects the user's last explicit choice even across tab # navigations — making the discrepancy check reliable. _cur = st.session_state.get(f"slider_{m}") if _cur is None or (_cur == 0 and int(constraint["cap_value"]) > 0): st.session_state[f"slider_{m}"] = int(constraint["cap_value"]) value = st.slider( "Max share (%)", 0, 100, step=5, key=f"slider_{m}", on_change=_make_slider_cb(m), ) if value == 0: st.warning("Cap is 0% but method is active.", icon=":material/warning:") else: if f"manual_{m}" not in st.session_state and constraint.get("cap_value"): st.session_state[f"manual_{m}"] = float(constraint["cap_value"]) value = st.number_input( f"Max cap ({CO2_UNIT}/yr)", min_value=0.0, value=None, format="%.4f", key=f"manual_{m}", placeholder="0.0000", on_change=_make_manual_cb(m), ) if not value: st.warning(f"Cap is 0 {CO2_UNIT}/yr but method is active.", icon=":material/warning:") def reset_all_constraints(): """Reset all method constraints to their default values and rerun the app. Returns: None """ for m in method_order: st.session_state.constraints[m] = default_constraint() for key in [f"toggle_{m}", f"cap_type_{m}", f"slider_{m}", f"manual_{m}"]: st.session_state.pop(key, None) st.success("All method constraints reset to default.") st.rerun() def _iter_display_methods(custom_resources): """Yield (display_name, original_method_name) for all methods. 'Other' methods with custom batches are expanded into one entry per batch. All other methods yield (m, m). """ for m in method_order: variants = get_other_method_variants(m, custom_resources) if variants: for batch in variants: yield make_variant_name(m, batch["name"]), m else: yield m, m def build_summary_df(): """Build a summary DataFrame of currently active methods and their caps. Returns: pd.DataFrame | None: index = method name, column "Cap value" as a formatted string. Returns None if no method is active. """ custom_resources = st.session_state.get("custom_resources", []) rows = {} for display_m, _ in _iter_display_methods(custom_resources): c = st.session_state.constraints.get(display_m, default_constraint()) if not c["active"]: continue if c["cap_type"] == "percent": rows[display_m] = {"Cap value": f"{int(c['cap_value'])}%"} elif c["cap_type"] == "absolute": rows[display_m] = {"Cap value": f"{c['cap_value']:,g} {CO2_UNIT}/yr"} else: rows[display_m] = {"Cap value": f"⚠️ Invalid type ({c['cap_type']!r}) — fix in Tab 2"} return pd.DataFrame.from_dict(rows, orient="index") if rows else None def refresh_active_methods(cost_sliders_norm): """Auto-activate/deactivate methods based on real resource availability. A method is auto-deactivated when it's missing a required resource, and auto-reactivated (only if it was previously auto-deactivated, not manually turned off by the user) once that resource becomes available again. This must run on every app render — not just when Tab 2 itself draws — otherwise a user who goes straight from Tab 1 to Tab 3/4/5 without visiting Tab 2 first finds every method stuck at its initial "active: False" state (set once at app startup), and the optimisation fails with "No active methods" despite correctly configured resources. Call this from app.py's shared init in addition to Tab 2's own _sync_constraint_widgets (which calls it too, for its widget-key sync). Args: cost_sliders_norm (dict): {method: {resource: {min, median, max}}} with unit suffixes stripped. Returns: set[str]: method names that were just (re)activated by this call. """ auto_disabled = st.session_state.get("auto_disabled_methods", set()) just_activated = set() for m, c in st.session_state.constraints.items(): if get_all_missing_required(m, cost_sliders_norm): st.session_state.constraints[m]["active"] = False # Keep the method in auto_disabled so it gets re-enabled (and # fires just_activated) once the resource is added. auto_disabled.add(m) st.session_state.pop(f"toggle_{m}", None) st.session_state.pop(f"cap_type_{m}", None) st.session_state.pop(f"slider_{m}", None) st.session_state.pop(f"manual_{m}", None) continue if m in auto_disabled: st.session_state.constraints[m]["active"] = True auto_disabled.discard(m) just_activated.add(m) st.session_state["auto_disabled_methods"] = auto_disabled return just_activated def _sync_constraint_widgets(cost_sliders_norm): """Drive all constraint widget keys from the constraint dict before any widget is drawn. Two modes: - force mode: after a file/scenario load (constraints_just_loaded flag) or for freshly activated methods (just_activated_methods). All keys are set unconditionally from the constraint. - normal mode: setdefault only; never overrides a value the user already set. Args: cost_sliders_norm (dict): {method: {resource: {min, median, max}}} with unit suffixes stripped. Returns: None """ just_loaded = st.session_state.pop("constraints_just_loaded", False) just_activated = st.session_state.pop("just_activated_methods", set()) just_activated |= refresh_active_methods(cost_sliders_norm) for m, c in st.session_state.constraints.items(): force = just_loaded or (m in just_activated) if c["active"]: cap_type = c.get("cap_type", "percent") if force: st.session_state[f"toggle_{m}"] = True st.session_state[f"cap_type_{m}"] = cap_type if cap_type == "percent": st.session_state[f"slider_{m}"] = int(c.get("cap_value", 100)) st.session_state.pop(f"manual_{m}", None) else: cap_val = c.get("cap_value", 0.0) if cap_val: st.session_state[f"manual_{m}"] = float(cap_val) else: st.session_state.pop(f"manual_{m}", None) st.session_state.pop(f"slider_{m}", None) else: st.session_state.setdefault(f"toggle_{m}", True) st.session_state.setdefault(f"cap_type_{m}", cap_type) if cap_type == "percent": st.session_state.setdefault(f"slider_{m}", int(c.get("cap_value", 100))) st.session_state.pop(f"manual_{m}", None) else: cap_val = c.get("cap_value", 0.0) if cap_val and f"manual_{m}" not in st.session_state: st.session_state[f"manual_{m}"] = float(cap_val) st.session_state.pop(f"slider_{m}", None) else: if force: st.session_state[f"toggle_{m}"] = False else: st.session_state.setdefault(f"toggle_{m}", False) st.session_state.pop(f"cap_type_{m}", None) st.session_state.pop(f"slider_{m}", None) st.session_state.pop(f"manual_{m}", None) def render_constraints_summary(): """Render the summary dataframe of active methods and their caps. Returns: None """ st.markdown("""

Summary of Active methods and caps

""", unsafe_allow_html=True, ) df = build_summary_df() if df is not None: df_display = df.reset_index().rename(columns={"index": "CDR Method"}) df_display = df_display[["CDR Method", "Cap value"]] num_rows = len(df_display) st.dataframe( df_display, width='stretch', hide_index=True, height=(num_rows * 36 + 36), column_config={ "CDR Method": st.column_config.TextColumn("CDR Method", width="large"), "Cap value": st.column_config.TextColumn("Cap value"), }, row_height=36, ) else: st.warning("No active methods selected.", icon=":material/warning:") def render_tab(cost_sliders): """Render the Method Constraints tab. Args: cost_sliders (dict): full cost slider config {method: {resource: {min, median, max}}}. Returns: None """ cost_sliders_norm = { m: {strip_unit_suffix(name): conf for name, conf in resources.items()} for m, resources in cost_sliders.items() } _sync_constraint_widgets(cost_sliders_norm) st.markdown('

Method constraints

', unsafe_allow_html=True) show_bar() st.markdown( """

OBJECTIVE

""", unsafe_allow_html=True, ) st.markdown( f"""

{section_intros.get("tab2_constraints", "").format(CO2_UNIT=CO2_UNIT)}


""", unsafe_allow_html=True, ) st.markdown('

Controls

', unsafe_allow_html=True) if st.button("RESET ALL", key="btn-reset", icon=":material/reset_settings:"): reset_all_constraints() custom_resources = st.session_state.get("custom_resources", []) # Build the full list of display names (variants replace "Other" methods with batches) all_display = list(_iter_display_methods(custom_resources)) any_active = any( st.session_state.constraints.get(display_m, default_constraint()).get("active", True) for display_m, _ in all_display ) # Activatable: variants are always activatable; standard methods only if they # A method is activatable when all its required resources are available. activatable = [] for display_m, orig_m in all_display: if not get_all_missing_required(orig_m, cost_sliders_norm): activatable.append(display_m) if any_active: if st.button("Deactivate all", icon=":material/remove_done:", key="btn-deactivate-all"): for display_m, _ in all_display: st.session_state.constraints.setdefault(display_m, default_constraint())["active"] = False st.session_state[f"toggle_{display_m}"] = False st.rerun() else: if st.button("Activate all", icon=":material/done_all:", key="btn-activate-all"): for m in activatable: st.session_state.constraints.setdefault(m, default_constraint())["active"] = True st.session_state[f"toggle_{m}"] = True st.rerun() with st.expander("Manage methods", expanded=True, icon=":material/settings:"): for group, method_list in method_groups.items(): st.markdown(f"""

{group}

""", unsafe_allow_html=True, ) cols = st.columns(3) col_idx = 0 for m in method_list: variants = get_other_method_variants(m, custom_resources) if variants: for batch in variants: vname = make_variant_name(m, batch["name"]) render_method(vname, cols, col_idx, cost_sliders_norm) col_idx += 1 else: render_method(m, cols, col_idx, cost_sliders_norm) col_idx += 1 render_constraints_summary() if st.button("Resource coefficients →", key="btn-nav-next-constraints", type="primary", help="Optional: adjust resource efficiency coefficients per method."): st.session_state["_navigate_to"] = "Resource coefficients" st.rerun()