CRRA_Optimization_Tool_V2 / tabs /tab2_constraints.py
CarlyneB's picture
[UX] British spelling: rename optimization → optimisation throughout code (conflict resolve)
7a80c0b
Raw
History Blame Contribute Delete
29.7 kB
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'<span style="color:{color};font-weight:600"><strong>{r}</strong></span>'
if hint:
label += f' <span style="color:#888;font-style:italic;font-weight:400">: {hint}</span>'
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'<span style="color:{color};font-weight:600"><strong>{r}</strong></span>'
if hint:
label += f' <span style="color:#888;font-style:italic;font-weight:400">: {hint}</span>'
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("""
<p class="crra-sub-heading">Summary of Active methods and caps</p>""",
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('<h1 class="crra-heading">Method constraints</h1>', unsafe_allow_html=True)
show_bar()
st.markdown(
"""
<p class="objective_title">OBJECTIVE</p>
""",
unsafe_allow_html=True,
)
st.markdown(
f"""
<div class="text_with_border">
<p>{section_intros.get("tab2_constraints", "").format(CO2_UNIT=CO2_UNIT)}</p>
</div>
<br>
""",
unsafe_allow_html=True,
)
st.markdown('<p class="crra-sub-heading">Controls</p>', 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"""
<p class="crra-group-name">{group}</p>""",
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()