from pathlib import Path import re import pandas as pd import streamlit as st import plotly.graph_objs as go from streamlit_plotly_events import plotly_events from features.map import build_map_figure, find_centroid_by_slug from features.utils import slug_name import math import numpy as np # Main: only show map (no right-hand panel) st.title("UK Nowcast Model (2026 General Elections)") st.subheader("Simulating optimal Your Party strategies to avoid left vote splitting.") # Configuration DATA_DIR = Path("data") HEXJSON_PATH = DATA_DIR / "uk-constituencies-2024.hexjson" NOWCAST_PATH = DATA_DIR / "nowcast.csv" st.set_page_config(layout="wide", page_title="UK Nowcast Model (2026 General Elections)", initial_sidebar_state="expanded") # Load and validate data if not HEXJSON_PATH.exists(): st.error(f"Missing hexjson file at {HEXJSON_PATH}. Place `uk-constituencies-2024.hexjson` in data/.") st.stop() if not NOWCAST_PATH.exists(): st.error(f"Missing nowcast CSV at {NOWCAST_PATH}. Place `nowcast.csv` in data/.") st.stop() # read CSV nowcast_raw = pd.read_csv(NOWCAST_PATH, dtype=str) # read as str to sanitize reliably # sanitize electorate -> int; turnout -> fraction; party columns -> numeric shares (0..1) def parse_int_like(s): if pd.isna(s): return 0 # remove commas, spaces, non-digits s = re.sub(r"[^\d]", "", str(s)) return int(s) if s != "" else 0 def parse_pct_like(s): if pd.isna(s): return 0.0 s = str(s).strip() # handle values like '49.30%' or '0.493' or '-' if s == "-" or s == "": return 0.0 if s.endswith("%"): try: return float(s.strip("%")) / 100.0 except Exception: return 0.0 try: return float(s) except Exception: return 0.0 # Standard party list (ensure 'YOUR' is included) COMMON_PARTIES = ["YOUR", "LAB", "CON", "RFM", "LDM", "GRN", "MIN", "Oth", "SNP", "PLC"] # Convert columns # Find the constituency column name constituency_col = None for candidate in ["Constituency", "constituency", "ConstituencyName", "Constituency Name"]: if candidate in nowcast_raw.columns: constituency_col = candidate break if constituency_col is None: constituency_col = nowcast_raw.columns[0] # Parse electorate and turnout into numeric columns nowcast = nowcast_raw.copy() nowcast["Electorate_parsed"] = nowcast.get("Electorate", "").apply(parse_int_like) nowcast["Turnout_parsed"] = nowcast.get("Turnout", "").apply(parse_pct_like) # Parse / coerce party share columns to numeric (0..1). Some CSVs use '-' for missing. # We'll make sure all COMMON_PARTIES exist as numeric columns; missing ones are created as 0.0 for p in COMMON_PARTIES: if p in nowcast.columns: nowcast[p] = pd.to_numeric(nowcast[p].replace("-", pd.NA), errors="coerce").fillna(0.0) else: nowcast[p] = 0.0 # create slug column for lookups nowcast["__slug"] = nowcast[constituency_col].apply(slug_name) nowcast_index = nowcast.set_index("__slug", drop=False) # --- Simulation control in sidebar --- st.sidebar.header("Simulation controls") add_nonvoter_pct = st.sidebar.number_input( "% non-voters convinced to vote for Your Party", min_value=0, max_value=50, value=0, step=1, help="This converts a percentage of the non-voting electorate into votes for Your Party." ) # NEW: % left voters converted to Your Party add_left_pct = st.sidebar.number_input( "% left voters converted to Your Party", min_value=0, max_value=50, value=0, step=1, help="This takes a percentage of current left-party voters (LAB, LDM, GRN, SNP, PLC, MIN) and transfers them to Your Party. The taken voters are split across left parties proportional to their current left vote shares." ) # --- Helper functions for seat counts and rendering row (unchanged visual style) --- try: from features.utils import PARTY_COLOR_MAP except Exception: PARTY_COLOR_MAP = { "YOUR": "#800080", "LAB": "#DC241f", "CON": "#0087DC", "SNP": "#FFF100", "LDM": "#FDBB30", "GRN": "#6AB023", "RFM": "#3BB7D5", "PLC": "#6E86FF", "MIN": "#A0A0A0", "Oth": "#999999", } def compute_plurality_winners(df): """ Compute plurality winner from numeric party share columns in df. This intentionally ignores any pre-existing 'Winner' column so results reflect simulated adjustments (e.g. YOUR votes). """ # detect party columns (include COMMON_PARTIES if present) party_cols = [c for c in COMMON_PARTIES if c in df.columns] if not party_cols: exclude = {constituency_col, "Current", "Winner", "__slug", "Electorate", "Turnout", "Electorate_parsed", "Turnout_parsed"} party_cols = [c for c in df.columns if c not in exclude and pd.api.types.is_numeric_dtype(df[c])] if not party_cols: # nothing to compute return pd.Series("", index=df.index, dtype=str) # coerce to numeric and fillna with 0.0 numeric = df[party_cols].apply(pd.to_numeric, errors="coerce").fillna(0.0) # idxmax picks the column with largest share per row winners = numeric.idxmax(axis=1).fillna("").astype(str) # Rows where all zeros will pick the first column; make those empty instead row_max = numeric.max(axis=1).fillna(0.0) winners = winners.where(row_max > 0, other="") winners = winners.replace({"nan": ""}) return winners def render_seat_count_row(df, max_items_in_row=None): winners = compute_plurality_winners(df) counts = winners[winners != ""].value_counts().sort_values(ascending=False) parties = list(counts.index) if max_items_in_row is not None and len(parties) > max_items_in_row: top = parties[: max_items_in_row - 1] rest = parties[max_items_in_row - 1 :] top_counts = counts.loc[top].to_dict() other_count = int(counts.loc[rest].sum()) display = top[:] + ["Other"] counts_map = {**{k:int(v) for k,v in top_counts.items()}, "Other": other_count} else: display = parties counts_map = {k:int(v) for k,v in counts.astype(int).to_dict().items()} if not display: st.info("No plurality winners available to compute seat counts.") return cols = st.columns(len(display), gap="small") for col, party in zip(cols, display): with col: party_count = counts_map.get(party, 0) color = PARTY_COLOR_MAP.get(party, "#888888") if party != "Other" else "#999999" col.markdown( f"""
{party}
{party_count}
projected seats
""", unsafe_allow_html=True, ) # --- Build adjusted dataframe according to slider --- def apply_nonvoter_conversion(df, pct_nonvoter, pct_left=0): """ Defensive implementation that freezes original shares from the passed-in DataFrame and computes absolute votes for other parties from those frozen shares. Caller should pass the baseline (original) DataFrame to guarantee correct behaviour. New behaviour: pct_left is the percent of left-party voters (LAB, LDM, GRN, SNP, PLC, MIN) that are converted to 'YOUR'. The voters taken are removed proportionally from each left party's absolute votes and added to YOUR. This is applied per-constituency. """ # work on a copy so we never mutate caller's DataFrame base = df.copy(deep=True) input_frac = float(pct_nonvoter) / 100.0 left_frac = float(pct_left) / 100.0 # party columns present (in baseline) party_cols = [p for p in COMMON_PARTIES if p in base.columns] # ensure numeric types for baseline shares and electorate/turnout for p in party_cols: base[p] = pd.to_numeric(base[p], errors="coerce").fillna(0.0) base["_electorate"] = pd.to_numeric(base["Electorate_parsed"], errors="coerce").fillna(0).astype(int) base["_turnout"] = pd.to_numeric(base["Turnout_parsed"], errors="coerce").fillna(0.0) # 1) current votes and non-voters (integers) base["_current_votes"] = (base["_electorate"] * base["_turnout"]).round().astype(int) base["_non_voters"] = (base["_electorate"] - base["_current_votes"]).clip(lower=0).astype(int) # 2) freeze baseline shares explicitly (safe even if df was adjusted earlier) for p in party_cols: base[f"_baseline_share_{p}"] = base[p].astype(float) # 3) compute absolute votes for other parties from frozen baseline shares for p in party_cols: base[f"_votes_{p}"] = (base["_current_votes"] * base[f"_baseline_share_{p}"]).round().astype(int) # 4) desired and actual new YOUR votes (from non-voter pool) base["_desired_your"] = (base["_electorate"] * (1.0 - base["_turnout"]) * input_frac).round().astype(int) base["_new_your_votes"] = base[["_desired_your", "_non_voters"]].min(axis=1).astype(int) # ensure YOUR vote buckets exist and add new votes from non-voters if "YOUR" not in base.columns: base["YOUR"] = 0.0 if "_votes_YOUR" not in base.columns: base["_votes_YOUR"] = 0 base["_votes_YOUR"] = base["_votes_YOUR"].astype(int) + base["_new_your_votes"].astype(int) # 5) LEFT-VOTER TRANSFER: take left_frac of each left party's absolute votes and transfer to YOUR left_parties = ["LAB", "LDM", "GRN", "SNP", "PLC", "MIN"] # only consider left parties that actually exist in this dataset left_parties_present = [p for p in left_parties if p in base.columns and f"_votes_{p}" in base.columns] # compute per-party taken voters and subtract from left parties' absolute votes # sum all taken voters to _new_from_left and add to YOUR if left_frac > 0 and left_parties_present: # compute taken voters per left party taken_cols = [] for p in left_parties_present: taken_col = f"_left_take_{p}" # compute as rounded int of the absolute votes * left_frac base[taken_col] = (base[f"_votes_{p}"] * left_frac).round().astype(int) # subtract taken voters from the left-party vote absolute counts base[f"_votes_{p}"] = (base[f"_votes_{p}"] - base[taken_col]).clip(lower=0).astype(int) taken_cols.append(taken_col) # total new votes taken from left parties base["_new_from_left"] = base[taken_cols].sum(axis=1).astype(int) # add those to YOUR absolute votes base["_votes_YOUR"] = base["_votes_YOUR"].astype(int) + base["_new_from_left"].astype(int) else: # no left transfer base["_new_from_left"] = 0 # 6) new total votes and safe denominator # Note: transfers from left do not change total voter count; only non-voter conversion increases totals. base["_new_total_votes"] = base["_current_votes"] + base["_new_your_votes"] base["_new_total_votes_safe"] = base["_new_total_votes"].replace({0: 1}) # 7) recompute shares from absolute votes (other parties keep absolute votes, left parties were reduced above) for p in party_cols: base[p] = (base[f"_votes_{p}"] / base["_new_total_votes_safe"]).fillna(0.0) # YOUR share from its absolute votes base["YOUR"] = (base["_votes_YOUR"] / base["_new_total_votes_safe"]).fillna(0.0) # 8) return only expected downstream columns (plus internal debug cols if you need them) keep_cols = [constituency_col, "__slug", "Electorate_parsed", "Turnout_parsed"] keep_cols += [c for c in COMMON_PARTIES if c in base.columns] if "Winner" in base.columns: keep_cols.append("Winner") # include internal debug columns (absolute votes and intermediates) debug_cols = [c for c in base.columns if c.startswith("_votes_")] debug_cols += [c for c in ("_current_votes", "_non_voters", "_desired_your", "_new_your_votes", "_votes_YOUR", "_new_total_votes", "_new_from_left")] # include left_take columns if present debug_cols += [c for c in base.columns if c.startswith("_left_take_")] # keep only those that actually exist debug_cols = [c for c in debug_cols if c in base.columns] keep_cols += debug_cols return base[keep_cols].copy() # Apply simulation baseline_nowcast = nowcast.copy(deep=True) adjusted = apply_nonvoter_conversion(baseline_nowcast, add_nonvoter_pct, add_left_pct) # Render seat row from adjusted data render_seat_count_row(adjusted, max_items_in_row=8) # ---- projected seats bar plot ---- winners = compute_plurality_winners(adjusted) # tally and ensure YOUR is shown even if zero counts = winners[winners != ""].value_counts().astype(int) if "YOUR" in adjusted.columns and "YOUR" not in counts.index: counts = pd.concat([pd.Series({"YOUR": 0}), counts]).sort_values(ascending=False) # sort descending (largest first) counts = counts.sort_values(ascending=False) if counts.empty: st.info("No projected seats to display.") else: parties = counts.index.tolist() values = counts.values.tolist() colors = [PARTY_COLOR_MAP.get(p, "#999999") for p in parties] bar_fig = go.Figure( go.Bar( x=values, y=parties, orientation="h", marker=dict(color=colors), text=values, textposition="auto" ) ) bar_fig.update_layout( title="", xaxis_title="Seats", yaxis_title="Party", height=350, margin=dict(l=40, r=40, t=40, b=40), ) bar_fig.update_traces(textangle=0) st.plotly_chart( bar_fig, width='stretch', config={"displayModeBar": False} ) # ---- end bar plot ---- # Sidebar: searchable selectbox (uses original nowcast names) st.sidebar.header("Map controls") constituency_options = [""] + sorted(nowcast[constituency_col].astype(str).tolist()) selected_constituency = st.sidebar.selectbox( "Find constituency", options=constituency_options, index=0, help="Type to filter the list, then select a constituency." ) # Build figure (centroid markers include hover text = constituency name) using adjusted df fig, slug_to_centroid = build_map_figure(DATA_DIR, adjusted) # render the map via plotly_events so we can capture clicks FIG_HEIGHT = 900 events = plotly_events(fig, click_event=True, hover_event=False, override_height=FIG_HEIGHT) # utility: find slug from clicked coordinates (tolerance to avoid float mismatches) def find_slug_from_click(slug_map, x_clicked, y_clicked, tol=1e-6): for slug, (cx, cy) in slug_map.items(): if math.isclose(cx, x_clicked, rel_tol=0.0, abs_tol=tol) and math.isclose(cy, y_clicked, rel_tol=0.0, abs_tol=tol): return slug return None # Determine selected slug either from click or sidebar selectbox selected_slug = None # prefer last click event if available if events: # find last plotly_click for ev in reversed(events): if isinstance(ev, dict) and ev.get("event") == "plotly_click": pts = ev.get("points") or [] if pts: pt = pts[0] x_clicked = pt.get("x") y_clicked = pt.get("y") # try exact match first, otherwise try small tolerance search selected_slug = find_slug_from_click(slug_to_centroid, x_clicked, y_clicked, tol=1e-6) if selected_slug is None: selected_slug = find_slug_from_click(slug_to_centroid, x_clicked, y_clicked, tol=1e-3) break # fallback to sidebar selection if no click if selected_slug is None and selected_constituency: selected_slug = slug_name(selected_constituency) # show debug panel as an expander below the map with st.expander("Constituency details"): if not selected_slug: st.write("No constituency selected. Click a hex on the map or choose one from the sidebar.") else: if selected_slug not in adjusted.set_index("__slug").index: st.write(f"Selected slug '{selected_slug}' not found in adjusted data.") else: # fetch row for slug; normalize to a single Series sel = adjusted.set_index("__slug").loc[selected_slug] if isinstance(sel, pd.DataFrame): # deterministic: take the first matching row row = sel.iloc[0] else: row = sel # already a Series # parse inputs (must be done after row is selected) electorate = int(row.get("Electorate_parsed", 0)) turnout = float(row.get("Turnout_parsed", 0.0)) current_votes = int(round(electorate * turnout)) non_voters = electorate - current_votes input_frac = float(add_nonvoter_pct) / 100.0 desired_your = int(round(electorate * (1.0 - turnout) * input_frac)) new_your_votes = min(desired_your, non_voters) new_total_votes = current_votes + new_your_votes new_total_votes_safe = new_total_votes if new_total_votes > 0 else 1 # compute existing absolute votes for each party using authoritative _votes_ columns when present vote_rows = {} for p in [c for c in COMMON_PARTIES if c in adjusted.columns]: votes_col = f"_votes_{p}" if votes_col in row.index: val = row[votes_col] # safe conversion to int for various possible types try: votes_abs = int(val) except Exception: # try .item() for numpy/pandas scalars, else handle array-like try: votes_abs = int(val.item()) except Exception: if hasattr(val, "__len__") and len(val) > 0: votes_abs = int(val[0]) else: votes_abs = int(float(val)) else: # fallback: infer from current_votes * original_share (only used if _votes_ not present) votes_abs = int(round(current_votes * (row.get(p, 0.0) if p != "YOUR" else 0.0))) vote_rows[p] = votes_abs # Prefer any explicit internal fields for exactness if "_new_your_votes" in row.index: new_your_votes = int(row["_new_your_votes"]) if "_current_votes" in row.index: current_votes = int(row["_current_votes"]) # recompute dependent totals just in case non_voters = electorate - current_votes new_total_votes = current_votes + new_your_votes new_total_votes_safe = new_total_votes if new_total_votes > 0 else 1 # recompute final shares from authoritative absolute votes # small helper: coerce a scalar/array-like into an int safely def _safe_int_from_val(val, default=0): # handle None / NaN try: if val is None: return int(default) except Exception: pass # pandas NA / np.nan try: if pd.isna(val): return int(default) except Exception: pass # Try straightforward conversion first try: return int(val) except Exception: pass # Try .item() (numpy / pandas scalar) try: return int(val.item()) except Exception: pass # If array-like/Series, take the first element try: if hasattr(val, "__len__") and len(val) > 0: return int(val[0]) except Exception: pass # Last resort: float conversion try: return int(float(val)) except Exception: return int(default) # recompute final shares from authoritative absolute votes final_shares = {} for p, v in vote_rows.items(): if p == "YOUR": v_final = _safe_int_from_val(row.get("_votes_YOUR", new_your_votes), default=new_your_votes) else: v_final = int(v) # v comes from vote_rows and is already int final_shares[p] = v_final / new_total_votes_safe # prepare display table table_rows = [] for p in sorted(final_shares.keys(), key=lambda x: -final_shares[x]): votes_col = f"_votes_{p}" # Prefer authoritative _votes_ column when present, but coerce safely if votes_col in row.index: votes_abs = _safe_int_from_val(row.loc[votes_col], default=vote_rows.get(p, 0)) else: votes_abs = int(vote_rows.get(p, 0)) # Ensure YOUR uses the explicit _votes_YOUR field when available if p == "YOUR": votes_abs = _safe_int_from_val(row.get("_votes_YOUR", new_your_votes), default=new_your_votes) pct = final_shares[p] table_rows.append((p, votes_abs, f"{pct:.1%}")) df_dbg = pd.DataFrame(table_rows, columns=["Party", "Absolute votes", "Percent total votes"]) st.markdown(f"**Constituency:** {row.get(constituency_col, selected_slug)}") st.write(f"Electorate: {electorate:,}") st.write(f"Turnout: {turnout + input_frac:.2%} → Voters: {new_total_votes:,}") st.write(f"Non-voters: {electorate - new_total_votes:,}") st.write(f"Total Your Party votes: {desired_your:,}") st.dataframe(df_dbg, width='content') # -------------------------- # New: grid plot of winners for combinations of non-voter % (x) and left-voter % (y) # We'll compute a grid from 0..50 (inclusive) for both axes with step 1. # For each cell we simulate just this constituency using apply_nonvoter_conversion # on a one-row baseline and read the plurality winner. # -------------------------- # Prepare baseline one-row df for simulation (use baseline_nowcast original shares) # Ensure we produce a DataFrame with a __slug column (not just an index) if selected_slug not in baseline_nowcast.set_index("__slug").index: st.error("Selected constituency not found in baseline data.") # -------------------------- # New: grid plot of winners for combinations of non-voter % (x) and left-voter % (y) # Compute a 0..50 x 0..50 grid and display a heatmap inside the expander. # -------------------------- # Build a one-row baseline DataFrame for the selected constituency. baseline_row = baseline_nowcast[baseline_nowcast["__slug"] == selected_slug].copy() if baseline_row.shape[0] == 0: st.error("Selected constituency not found in baseline data.") else: # Ensure a proper DataFrame with columns (not index) baseline_row = baseline_row.reset_index(drop=True) # Ensure constituency_col and __slug exist as columns (apply_nonvoter_conversion expects them) if "__slug" not in baseline_row.columns: baseline_row["__slug"] = selected_slug if constituency_col not in baseline_row.columns: try: baseline_row[constituency_col] = nowcast_index.loc[selected_slug, constituency_col] except Exception: baseline_row[constituency_col] = selected_slug # Ensure numeric columns are present so apply_nonvoter_conversion won't drop them later for col in ("Electorate_parsed", "Turnout_parsed"): if col not in baseline_row.columns: baseline_row[col] = nowcast_index.loc[selected_slug, col] if col in nowcast_index.columns else 0 for p in COMMON_PARTIES: if p not in baseline_row.columns: baseline_row[p] = 0.0 # Grid axes x_vals = list(range(0, 51)) # % non-voters (x-axis) y_vals = list(range(0, 51)) # % left voters (y-axis) # Parties present for this constituency parties_present = [p for p in COMMON_PARTIES if p in baseline_row.columns] party_codes = {p: i for i, p in enumerate(parties_present)} code_to_party = {i: p for p, i in party_codes.items()} colors_for_codes = [PARTY_COLOR_MAP.get(code_to_party[i], "#888888") for i in range(len(party_codes))] # Pre-allocate z = np.full((len(y_vals), len(x_vals)), fill_value=-1, dtype=int) hovertext = [["" for _ in x_vals] for _ in y_vals] # Compute grid winners for iy, left_pct in enumerate(y_vals): for ix, nonv_pct in enumerate(x_vals): sim = apply_nonvoter_conversion(baseline_row, pct_nonvoter=nonv_pct, pct_left=left_pct) winner_series = compute_plurality_winners(sim) winner = "" if len(winner_series) > 0: val = winner_series.iloc[0] if pd.notna(val): winner = str(val) if winner == "" or winner not in party_codes: # fallback: prefer Oth if present, else pick first party if "Oth" in party_codes: code = party_codes["Oth"] winner_label = "Oth" else: code = 0 winner_label = code_to_party.get(code, "") else: code = party_codes[winner] winner_label = winner z[iy, ix] = int(code) hovertext[iy][ix] = f"{winner_label}
% non-voters: {nonv_pct}%
% left converted: {left_pct}%" # Build colorscale n_codes = len(colors_for_codes) if n_codes == 0: st.write("No party columns available to build winner grid.") else: # Build discrete (flat) colorscale so bands are solid, not blended if n_codes == 1: colorscale = [[0.0, colors_for_codes[0]], [1.0, colors_for_codes[0]]] else: colorscale = [] # create contiguous same-color intervals: [start,color], [end,color] for each band for idx, col in enumerate(colors_for_codes): start = idx / n_codes end = (idx + 1) / n_codes colorscale.append([start, col]) colorscale.append([end, col]) heatmap = go.Figure( data=go.Heatmap( z=z, x=x_vals, y=y_vals, colorscale=colorscale, colorbar=dict( title="Legend", tickmode="array", tickvals=list(party_codes.values()), ticktext=[code_to_party[i] for i in sorted(code_to_party.keys())], ), hoverinfo="text", text=hovertext, zmin=0, zmax=max(0, n_codes - 1) ) ) heatmap.update_layout( title="Winning party under different simulations", xaxis_title="% non-voters convinced to vote for Your Party", yaxis_title="% left voters converted to Your Party", height=480, margin=dict(l=60, r=40, t=60, b=60), ) heatmap.update_xaxes(tickmode="linear", dtick=5) heatmap.update_yaxes(tickmode="linear", dtick=5, autorange="reversed") # Force rendering inside the expander (use_container_width True so it fills the expander) st.plotly_chart(heatmap, width='content', config={"displayModeBar": False}) # If user selected a constituency from the sidebar, add an annotation on the map if selected_constituency: sel_slug = slug_name(selected_constituency) centroid = find_centroid_by_slug(slug_to_centroid, sel_slug) if centroid is not None: cx, cy = centroid fig.update_layout(annotations=[dict(x=cx, y=cy, text=f"{selected_constituency}", showarrow=True, arrowhead=2, ax=20, ay=-20)])