"""Concrete Compressive Strength — Predict / Design (Streamlit).
Two tools:
* Predict strength — enter a mix design (any field may be left blank), get the
predicted compressive strength.
* Design a mix — enter a target strength, get detailed mix designs that
reach it (drawn from real data, optionally refined). All
suggestions are reported at 28-day strength.
Run locally: streamlit run app/streamlit_app.py
"""
from __future__ import annotations
import sys
from pathlib import Path
import pandas as pd
import streamlit as st
# Make sibling modules importable no matter how the script is launched
# (`streamlit run`, AppTest, or a Hugging Face Space at repo root).
sys.path.insert(0, str(Path(__file__).resolve().parent))
from inference import (
AGE_COL,
CURING_CARBON_FACTORS,
DESIGN_COLS,
EMBODIED_CARBON_FACTORS,
SCM_COLS,
Predictor,
embodied_carbon,
_default_ckpt_dir,
_default_index,
)
st.set_page_config(
page_title="Concrete Compressive Strength",
page_icon="◧",
layout="wide",
initial_sidebar_state="collapsed",
)
CKPT_DIR = _default_ckpt_dir()
INDEX_PATH = _default_index()
DESIGN_AGE = 28.0 # all design suggestions are reported at 28-day strength
# ---------------------------------------------------------------------------
# Styling — editorial: cream canvas, black frame + top bar, big grotesque type.
# ---------------------------------------------------------------------------
INK = "#1a1a1a" # near-black text / accent (monochrome)
SUBTLE = "#6b6b63" # warm gray secondary text
CREAM = "#ffffff" # page background / light text on dark elements
CARD = "#f6f6f7" # faint card surface (separates groups on white)
HAIRLINE = "rgba(0,0,0,0.10)"
FONT = ('"Inter", -apple-system, BlinkMacSystemFont, "Segoe UI", '
'Roboto, Helvetica, Arial, sans-serif')
CSS = f"""
"""
FRAME = """
", unsafe_allow_html=True)
def _category_select(label, options, mix, key):
choice = st.selectbox(label, list(options.keys()), index=0)
if options[choice] is not None:
mix[key] = options[choice]
def carbon_inventory_editor(key_prefix: str) -> tuple[dict, dict]:
"""Editable embodied-carbon inventory.
Returns ``(material_factors, curing_factors)`` — material kg CO₂e/kg and
curing-process kg CO₂e/m³.
"""
factors, curing = {}, {}
with st.expander("Embodied-carbon inventory (emission factors)"):
st.caption("Cradle-to-gate kg CO₂e per kg of material. Indicative "
"defaults (ICE database / EPD literature) — adjust to your "
"suppliers' EPDs.")
cols = st.columns(3)
for i, (col, default) in enumerate(EMBODIED_CARBON_FACTORS.items()):
with cols[i % 3]:
factors[col] = st.number_input(
CARBON_LABELS.get(col, col),
value=float(default), min_value=0.0, step=0.001,
format="%.4f", key=f"{key_prefix}_cf_{col}",
)
st.caption("Curing-process energy, kg CO₂e per m³ (heated regimes only).")
ccols = st.columns(3)
# Standard/ambient curing adds no process energy, so it isn't editable.
editable = [(r, d) for r, d in CURING_CARBON_FACTORS.items() if r != "standard"]
for i, (regime, default) in enumerate(editable):
with ccols[i % 3]:
curing[regime] = st.number_input(
f"Curing: {regime}",
value=float(default), min_value=0.0, step=1.0,
format="%.1f", key=f"{key_prefix}_cure_{regime}",
)
return factors, curing
def show_carbon_breakdown(mix: dict, factors: dict, curing: dict,
strength: float | None = None) -> float:
"""Render the embodied-carbon and intensity metrics only. Returns total."""
total, _ = embodied_carbon(mix, factors, curing)
cols = st.columns(2)
cols[0].metric("Embodied carbon", f"{total:,.0f} kg CO₂e / m³")
if strength and strength > 0:
cols[1].metric("Carbon intensity", f"{total / strength:.1f} kg CO₂e / m³ / MPa")
return total
def ingredient_lines(row: pd.Series) -> list[tuple[str, float]]:
"""(label, amount) for each non-trivial design ingredient in a suggested mix."""
always = {"cement_kg_m3", "water_kg_m3"}
items = []
for col in DESIGN_COLS:
if col == AGE_COL or col not in row:
continue
val = pd.to_numeric(row[col], errors="coerce")
if pd.isna(val) or (val == 0 and col not in always):
continue
items.append((COLUMN_LABELS.get(col, col), float(val)))
return items
def format_suggestions(out: pd.DataFrame) -> pd.DataFrame:
"""Clean, clearly-named subset of the suggested mixes."""
always = {"cement_kg_m3", "water_kg_m3", "age_days"}
ing = [c for c in DESIGN_COLS if c in out.columns]
keep_ing = [c for c in ing if c in always
or pd.to_numeric(out[c], errors="coerce").fillna(0).abs().sum() > 0]
head = [c for c in ["pred_gnn", "water_binder_ratio"]
if c in out.columns]
disp = out[head + keep_ing].copy()
return disp.rename(columns=COLUMN_LABELS)
# ---------------------------------------------------------------------------
# Tab 1 — forward
# ---------------------------------------------------------------------------
def forward_tab(pred: Predictor) -> None:
_step("Enter your mix design")
st.caption("Leave any field blank if you don't have it. Amounts are per m³ of concrete.")
mix: dict = {}
c1, c2, c3 = st.columns(3, gap="large")
with c1:
with st.container(border=True):
_sec("Binder")
med = pred.bounds.get("cement_kg_m3", {}).get("p50", 350.0)
for col, label, unit in BINDER:
default = med if col == "cement_kg_m3" else 0.0
mix[col] = _num(pred, col, label, unit, default=default)
with c2:
with st.container(border=True):
_sec("Water & admixture")
mix["water_kg_m3"] = _num(
pred, "water_kg_m3", "Water", "kg/m³",
default=pred.bounds.get("water_kg_m3", {}).get("p50", 175.0), lo=120, hi=600,
)
mix["superplasticizer_kg_m3"] = _num(
pred, "superplasticizer_kg_m3", "Superplasticizer", "kg/m³",
default=pred.bounds.get("superplasticizer_kg_m3", {}).get("p50", 0.0),
)
with st.container(border=True):
_sec("Aggregate")
for col, label, unit in AGG:
d = pred.bounds.get(col, {}).get("p50", 0.0)
mix[col] = _num(pred, col, label, unit, default=d)
mix["max_coarse_aggregate_size_mm"] = _num(
pred, "max_coarse_aggregate_size_mm", "Max coarse agg. size", "mm",
optional=True, lo=10, hi=40, min_value=10, max_value=40,
)
mix["max_fine_aggregate_size_mm"] = _num(
pred, "max_fine_aggregate_size_mm", "Max fine agg. size", "mm", optional=True,
)
with c3:
with st.container(border=True):
_sec("Curing & age")
mix[AGE_COL] = st.number_input("Curing age (days)", value=28.0, min_value=1.0, step=1.0)
mix["curing_temperature_c"] = _num(
pred, "curing_temperature_c", "Curing temperature", "°C", optional=True)
mix["curing_humidity_pct"] = _num(
pred, "curing_humidity_pct", "Curing humidity", "%", optional=True)
with st.expander("Fibres (optional)"):
fc = st.columns(len(FIBRE))
for (col, label, unit), c in zip(FIBRE, fc):
with c:
mix[col] = _num(pred, col, label, unit, optional=True)
_category_select("Fibre type", FIBRE_TYPE_OPTIONS, mix, "fibre_type_norm")
with st.expander("Cement / SCM chemistry & type (advanced, optional)"):
gcols = st.columns(4)
for i, (col, label, unit) in enumerate(CHEM):
with gcols[i % 4]:
mix[col] = _num(pred, col, label, unit, optional=True)
s1, s2 = st.columns(2)
with s1:
_category_select("Cement type", CEMENT_TYPE_OPTIONS, mix, "cement_type_norm")
with s2:
_category_select("Curing regime", CURING_OPTIONS, mix, "curing_regime_norm")
carbon_factors, curing_factors = carbon_inventory_editor("fwd")
show_curve = st.checkbox("Show strength gain with curing age", value=True)
st.write("")
go = st.button("Predict strength", type="primary")
if go:
clean = {k: v for k, v in mix.items() if v not in (None, "")}
with st.spinner("Predicting…"):
res = pred.predict_strength(clean)
st.write("")
st.metric("Predicted compressive strength", f"{res['gnn']:.1f} MPa")
rmse = pred.config.get("test_metrics", {}).get("rmse")
if rmse:
st.caption(f"Typical accuracy ≈ ±{rmse:.0f} MPa on held-out test data.")
flags = pred.out_of_range(clean)
if flags:
pretty = ", ".join(COLUMN_LABELS.get(f, f) for f in flags)
st.warning(f"Some inputs are outside the usual data range ({pretty}); "
"the prediction there is an extrapolation.")
st.markdown("##### Embodied carbon")
show_carbon_breakdown(clean, carbon_factors, curing_factors, strength=res["gnn"])
if show_curve:
curve = pred.age_curve(clean)[["age_days", "gnn_mpa"]].rename(
columns={"gnn_mpa": "Predicted (MPa)"}
)
st.markdown("##### Predicted strength gain with curing age")
chart_col, _ = st.columns([3, 2])
with chart_col:
st.line_chart(curve.set_index("age_days"))
# ---------------------------------------------------------------------------
# Tab 2 — inverse
# ---------------------------------------------------------------------------
def inverse_tab(pred: Predictor, index: pd.DataFrame) -> None:
_step("Choose a target strength")
if index.empty:
st.info("The mix database (`inverse_index.csv`) isn't built yet. "
"Run `python app/build_inverse_index.py` first.")
return
smin = float(pred.config.get("strength_min", 10))
smax = float(pred.config.get("strength_max", 200))
c1, c2 = st.columns([2, 1])
with c1:
target = st.slider("Target compressive strength (MPa)", smin, smax,
min(80.0, smax), step=1.0)
with c2:
k = st.slider("Number of mix options", 1, 4, 1)
o1, o2, o3 = st.columns(3)
allow_fibre = o1.checkbox("Allow fibres", value=True)
require_coarse = o2.checkbox(
"Require coarse aggregate", value=True,
help="Exclude binder/sand-only mixes (UHPC, mortars). Switch off only if "
"you're designing a UHPC-style mix.")
refine = o3.checkbox("Refine to target", value=False,
help="Fine-tune each candidate so its predicted strength matches the target.")
e1, e2 = st.columns(2)
with e1:
excluded_labels = st.multiselect(
"SCMs you can't use", list(SCM_LABELS.values()),
help="Mixes containing any selected supplementary cementitious material are excluded.")
with e2:
preferred_labels = st.multiselect(
"Preferred curing methods", list(CURING_PREF_OPTIONS.keys()),
help="Leave empty to allow any curing method. Records with no recorded "
"regime are treated as standard; steam/heat/autoclave data comes "
"only from the UHPC dataset.")
exclude_scms = [c for c, lab in SCM_LABELS.items() if lab in excluded_labels]
curing_regimes = [CURING_PREF_OPTIONS[lab] for lab in preferred_labels]
carbon_factors, curing_factors = carbon_inventory_editor("inv")
st.caption("All suggestions are reported at 28-day strength.")
st.write("")
if st.button("Design mixes", type="primary"):
with st.spinner("Searching mix designs…"):
out = pred.suggest_mixes(
target, index, k=k, allow_fibre=allow_fibre,
require_coarse_aggregate=require_coarse,
exclude_scms=exclude_scms, curing_regimes=curing_regimes,
domain="any", age=DESIGN_AGE, refine=refine,
)
if out.empty:
st.warning("No mixes match those constraints — try widening them.")
return
st.success(f"{len(out)} mix design(s) predicted to reach ≈ {target:.0f} MPa at 28 days")
out = out.reset_index(drop=True)
out["embodied_carbon_kgco2e_m3"] = [
round(embodied_carbon(r.to_dict(), carbon_factors, curing_factors)[0], 1)
for _, r in out.iterrows()
]
# One clearly-itemised card per mix option (no wide table).
for i, row in out.iterrows():
with st.container(border=True):
st.markdown(f"#### Mix option {i + 1}")
m = st.columns(3)
m[0].metric("Predicted strength", f"{row['pred_gnn']:.1f} MPa")
wb = pd.to_numeric(row.get("water_binder_ratio"), errors="coerce")
m[1].metric("Water / binder", f"{wb:.2f}" if pd.notna(wb) else "—")
m[2].metric("Embodied carbon",
f"{row['embodied_carbon_kgco2e_m3']:,.0f} kg CO₂e/m³")
st.markdown("**Ingredients (per m³ of concrete)**")
items = ingredient_lines(row)
lc = st.columns(2)
for j, (label, val) in enumerate(items):
lc[j % 2].markdown(f"- {label}: **{val:g}**")
# Full machine-readable export still available.
export = format_suggestions(out)
if "embodied_carbon_kgco2e_m3" in out.columns:
export["Embodied carbon (kg CO₂e/m³)"] = out["embodied_carbon_kgco2e_m3"].values
st.write("")
st.download_button("Download these mixes (CSV)", export.to_csv(index=False),
file_name=f"mix_designs_{int(target)}MPa.csv", mime="text/csv")
def main() -> None:
st.markdown(CSS, unsafe_allow_html=True)
st.markdown(FRAME, unsafe_allow_html=True)
st.markdown(HERO, unsafe_allow_html=True)
if not (Path(CKPT_DIR) / "hierarchical.pt").exists():
st.error(f"No trained model found in `{CKPT_DIR}`.")
st.stop()
pred = get_predictor()
index = get_index()
tab1, tab2 = st.tabs(["Predict strength", "Design a mix"])
with tab1:
forward_tab(pred)
with tab2:
inverse_tab(pred, index)
if __name__ == "__main__":
main()
else:
main()