import copy
import pandas as pd
import streamlit as st
import plotly.express as px
from config.config import CO2_UNIT, hidden_resource_names, method_order, resource_order, resource_unit_map, section_intros
from utils.data_utils import (get_resource_to_unit, get_resource_to_group, run_expanded_optimisation,
is_other_method, get_other_method_variants, make_variant_name,
default_constraint)
from utils.ui_helpers import register_chart_data, show_bar
def compute_baseline():
"""Run the baseline optimisation and build combined resource metadata.
Returns:
tuple[float, list[str], dict[str, str]]:
baseline_removed – total CO₂ removed at current inputs (0 if failed).
all_resources – standard resources + custom batch names.
resource_to_unit – {resource: unit} including custom batches.
"""
baseline_success, baseline_result = run_expanded_optimisation(
resource_caps=st.session_state.resource_caps,
method_constraints=st.session_state.constraints,
costs=st.session_state.costs,
custom_resources=st.session_state.get("custom_resources", []),
enabled_methods=st.session_state.get("enabled_methods", {}),
)
baseline_removed = baseline_result["total_removed"] if baseline_success else 0
# Only offer resources that actually have a quantity set — sweeping a
# resource currently at 0 always gives 0 regardless of the % slider
# (any percentage of a 0 base cap is still 0), so it's never useful to
# compare.
custom_resource_names = [
b["name"] for b in st.session_state.get("custom_resources", [])
if float(b.get("amount", 0)) > 0
]
all_resources = [
r for r in resource_order
if r not in hidden_resource_names and st.session_state.resource_caps.get(r, 0) > 0
] + custom_resource_names
resource_to_unit = get_resource_to_unit()
for b in st.session_state.get("custom_resources", []):
resource_to_unit[b["name"]] = b.get("unit", "")
return baseline_removed, all_resources, resource_to_unit
def panel_resource_sensitivity(baseline_removed, all_resources, resource_to_unit):
"""Render the resource-availability sensitivity sub-panel.
Sweeps each selected resource from 0 % to 200 % of its current cap in
10 % steps, re-solves the LP at each point, and plots total CO₂ removed.
Args:
baseline_removed (float): total CO₂ removed at current inputs.
all_resources (list[str]): standard + custom resource names to select from.
resource_to_unit (dict[str, str]): {resource: unit} for axis labels.
Returns:
None
"""
st.markdown(
'
Impact of resource availability
',
unsafe_allow_html=True,
)
st.markdown(
'Each line shows how total CO₂ removal changes as one resource\'s availability varies from 0% to 200% of its current cap. '
'
The range goes up to 200% so you can see both the impact of a reduction and the potential gains from doubling availability. '
'
A steep curve means that resource strongly drives portfolio performance: small increases can yield large gains. '
'
A flat line means the portfolio is limited by something else (another resource or a method cap).
',
unsafe_allow_html=True,
)
selected_resources = st.multiselect(
"Select one or more resources:", all_resources, key="crra-select-resource"
)
if not selected_resources:
st.warning("Please select at least one resource to run the sensitivity analysis.", icon=":material/warning:")
return
percentages = list(range(0, 210, 10))
custom_amount_by_name = {
b["name"]: float(b.get("amount", 0))
for b in st.session_state.get("custom_resources", [])
}
all_data = []
for resource in selected_resources:
unit = resource_to_unit.get(resource, "")
base_cap = (
st.session_state.resource_caps.get(resource)
or custom_amount_by_name.get(resource, 0)
)
for pct in percentages:
perturbed_caps = st.session_state.resource_caps.copy()
actual_amount = base_cap * pct / 100
perturbed_caps[resource] = actual_amount
success, result = run_expanded_optimisation(
resource_caps=perturbed_caps,
method_constraints=st.session_state.constraints,
costs=st.session_state.costs,
custom_resources=st.session_state.get("custom_resources", []),
enabled_methods=st.session_state.get("enabled_methods", {}),
)
all_data.append({
"Input": resource,
"% of cap": pct,
"Actual amount": round(actual_amount, 4),
"Unit": unit,
f"CO₂ removed ({CO2_UNIT})": result["total_removed"] if success else 0,
})
df = pd.DataFrame(all_data)
st.markdown("#### Sensitivity to resource availability")
fig = px.line(
df, x="% of cap", y=f"CO₂ removed ({CO2_UNIT})", color="Input", markers=True,
hover_data={"Actual amount": ":.4f", "Unit": True},
)
fig.update_layout(
xaxis_title="Resource availability (% of current cap)",
yaxis_title=f"Total portfolio CO₂ removal ({CO2_UNIT})",
)
fig.add_hline(
y=baseline_removed, line_dash="dash", line_color="gray",
annotation_text=f"Baseline: {baseline_removed:,} {CO2_UNIT}",
)
st.plotly_chart(fig, width='stretch')
with st.expander("📊 View / download chart data"):
st.dataframe(df, width='stretch', hide_index=True)
register_chart_data("Sensitivity to resource availability", df)
def panel_method_sensitivity(baseline_removed):
"""Render the method-cap sensitivity sub-panel.
Sweeps each selected method's cap from 0 % to 100 % of its current value
in 10 % steps, re-solves the LP at each point, and plots total CO₂ removed.
Methods whose current cap is already 0 are excluded from the selector
entirely (sweeping "X% of 0" is always 0, never useful to compare).
Args:
baseline_removed (float): total CO₂ removed at current inputs.
Returns:
None
"""
st.markdown(
'Impact of max allocation per method
',
unsafe_allow_html=True,
)
st.markdown(
'Each line shows how total CO₂ removal changes as one method\'s maximum deployment cap varies from 0% to 100% of its current setting: '
'whether that cap was defined as a percentage of its potential or as an absolute value in the Method constraints tab. '
'
The scale stops at 100% because the cap represents a ceiling you set: going beyond it would mean overriding your own constraint. '
'
A steep curve means the cap is a binding constraint: relaxing it would allow more removal. '
'
A flat line means the method is already limited by resource availability, not by its cap.
',
unsafe_allow_html=True,
)
# "Other"-tagged methods with custom batches attached are expanded into
# independent variants (e.g. "Enhanced weathering - Other mineral: Ganite")
# by run_expanded_optimisation — offer those variants as selectable
# entries too, alongside the standard method names, so their own cap can
# be swept just like any other method. An "Other"-tagged method with NO
# batch attached has no variant AND can never produce any result on its
# own (its only resource is the hidden "Other X" pool, which is never
# directly settable) — skip it entirely rather than offer a sweep that
# would always show a flat 0 line.
#
# Same reasoning as the resource panel: a method whose current cap is
# already 0 would sweep to "X% of 0", always 0 regardless of the slider —
# never useful to compare — so it's excluded from the list up front
# instead of being selectable and only warned about afterwards. Same for
# a method that is simply inactive (missing resources, manually toggled
# off in Tab 2...): sweeping its cap_value doesn't touch "active", so the
# perturbed run stays forced to 0 regardless — just as pointless to offer.
def _current_cap(method_name):
constraint = st.session_state.constraints.get(method_name) or default_constraint()
cap_type = constraint.get("cap_type", "percent")
return constraint.get("cap_value", 100 if cap_type == "percent" else 0.0)
def _is_sweepable(method_name):
constraint = st.session_state.constraints.get(method_name) or default_constraint()
return constraint.get("active", True) and _current_cap(method_name) > 0
custom_resources = st.session_state.get("custom_resources", [])
method_options = []
for m in method_order:
variants = get_other_method_variants(m, custom_resources)
if variants:
method_options.extend(
make_variant_name(m, b["name"]) for b in variants
if _is_sweepable(make_variant_name(m, b["name"]))
)
elif not is_other_method(m) and _is_sweepable(m):
method_options.append(m)
selected_methods = st.multiselect(
"Select one or more methods:", method_options, key="crra-select-method"
)
if not selected_methods:
st.warning("Please select at least one method to run the sensitivity analysis.", icon=":material/warning:")
return
percentages = list(range(0, 110, 10))
all_data = []
for method in selected_methods:
constraint = st.session_state.constraints.get(method) or default_constraint()
cap_type = constraint.get("cap_type", "percent")
current_cap = _current_cap(method)
for pct in percentages:
perturbed_constraints = copy.deepcopy(st.session_state.constraints)
perturbed_constraints.setdefault(method, default_constraint())
if cap_type == "percent":
actual_cap = min(100, current_cap * pct / 100)
perturbed_constraints[method]["cap_value"] = actual_cap
cap_label = f"{actual_cap:.0f}% of its potential"
else:
actual_cap = current_cap * pct / 100
perturbed_constraints[method]["cap_value"] = actual_cap
cap_label = f"{actual_cap:.2f} {CO2_UNIT}/yr"
success, result = run_expanded_optimisation(
resource_caps=st.session_state.resource_caps,
method_constraints=perturbed_constraints,
costs=st.session_state.costs,
custom_resources=custom_resources,
enabled_methods=st.session_state.get("enabled_methods", {}),
)
all_data.append({
"Method": method,
"Cap type": cap_type,
"% of current cap": pct,
"Actual cap": cap_label,
f"CO₂ removed ({CO2_UNIT})": result["total_removed"] if success else 0,
})
if not all_data:
return
df = pd.DataFrame(all_data)
fig = px.line(
df, x="% of current cap", y=f"CO₂ removed ({CO2_UNIT})",
color="Method", markers=True,
hover_data={"Actual cap": True, "Cap type": True},
)
fig.add_hline(
y=baseline_removed, line_dash="dash", line_color="lightgray",
annotation_text=f"Baseline: {baseline_removed:,} {CO2_UNIT}",
)
fig.update_layout(
xaxis_title="% of current cap setting",
yaxis_title=f"Total portfolio CO₂ removal ({CO2_UNIT})",
)
st.plotly_chart(fig, width='stretch')
with st.expander("📊 View / download chart data"):
st.dataframe(df, width='stretch', hide_index=True)
register_chart_data("Sensitivity to method cap", df)
def show_bottleneck_analysis():
"""Render the resource sensitivity table (CDR gain per +1 unit of each resource).
Reads the latest optimisation result from session state. Shows nothing if no
result is available yet.
Returns:
None
"""
latest = st.session_state.get("latest_result")
if not latest:
st.info("Run the optimisation first (Portfolio generation tab) to see the bottleneck analysis.", icon=":material/info:")
return
st.markdown('Resource sensitivity
', unsafe_allow_html=True)
st.markdown(
''
'For each resource: the additional CO₂ that could be removed by adding exactly +1 unit, '
'recomputed with all other constraints held fixed. '
'
Resources at the top of the list are the most binding bottlenecks in your current portfolio: '
'prioritise increasing their availability to unlock the largest gains.'
'
',
unsafe_allow_html=True,
)
resource_to_group = get_resource_to_group()
custom_batch_lookup = {b["name"]: b for b in latest.get("custom_resources", [])}
actual_gains = latest.get("resource_actual_gain", {})
if actual_gains:
sp_rows = []
for r, gain in sorted(actual_gains.items(), key=lambda x: -x[1]):
if r in custom_batch_lookup:
unit = resource_unit_map.get(resource_to_group.get(custom_batch_lookup[r]["group"]), "?")
else:
unit = resource_unit_map.get(resource_to_group.get(r), "?")
sp_rows.append({
"Resource": r,
"Unit": unit,
f"CDR gain for +1 unit ({CO2_UNIT})": round(gain, 4),
})
df_sp = pd.DataFrame(sp_rows)
st.dataframe(df_sp, width='stretch', hide_index=True)
register_chart_data("Resource sensitivity (CDR gain per +1 unit)", df_sp)
else:
st.info("No resources to analyse.", icon=":material/info:")
def render_tab():
"""Render the Sensitivity Analysis tab.
Computes a baseline optimisation from session state, then lets the user
sweep one or more resources (or method caps) over a range and plots how
total CO₂ removed changes. resource_caps, constraints, and costs are
guaranteed present by app.py's shared init, which runs before any tab.
Returns:
None
"""
st.markdown('Sensitivity analysis
', unsafe_allow_html=True)
show_bar()
st.markdown('OBJECTIVE
', unsafe_allow_html=True)
st.markdown(
f'{section_intros.get("tab5_sensitivity", "")}
',
unsafe_allow_html=True,
)
baseline_removed, all_resources, resource_to_unit = compute_baseline()
tab1, tab2, tab3 = st.tabs(["🔁 Resource sensitivity", "🔁 Method sensitivity", "🔍 Bottleneck analysis"])
with tab1:
panel_resource_sensitivity(baseline_removed, all_resources, resource_to_unit)
with tab2:
panel_method_sensitivity(baseline_removed)
with tab3:
show_bottleneck_analysis()
if st.button("Scenario comparison →", key="btn-nav-next-sensitivity", type="primary", help="Compare multiple saved scenarios side by side."):
st.session_state["_navigate_to"] = "Scenario comparison"
st.rerun()