Spaces:
Runtime error
Runtime error
File size: 6,026 Bytes
cfeb018 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | # features/map.py
import json
from pathlib import Path
from typing import Dict, Tuple
import pandas as pd
import plotly.graph_objs as go
from .utils import hex_to_pixel, polygon_hex, slug_name, PARTY_COLOR_MAP
# keep YOUR first so it gets priority where present
DEFAULT_PARTY_ORDER = ["YOUR", "LAB", "CON", "RFM", "LDM", "GRN", "MIN", "Oth", "SNP", "PLC"]
def _safe_float(val):
"""
Convert val to float safely. Return 0.0 for '-', '', None or any non-convertible value.
Handles numeric types and string forms.
"""
try:
if val is None:
return 0.0
# if it's already numeric, cast to float directly
if isinstance(val, (int, float,)):
return float(val)
v = str(val).strip()
if v == "" or v == "-":
return 0.0
# remove commas (just in case)
v = v.replace(",", "")
return float(v)
except Exception:
return 0.0
def build_map_figure(data_dir: Path, nowcast: pd.DataFrame) -> Tuple[go.Figure, Dict[str, Tuple[float, float]]]:
"""
Build a map figure using numeric party share columns from the provided nowcast DataFrame.
Important behavior:
- Party columns detected from DEFAULT_PARTY_ORDER are coerced to numeric here.
- The "winner" for a hex is selected from numeric shares (max share).
The original 'Winner' column is used only as a fallback when numeric shares are all zero/missing.
"""
hexjson_path = data_dir / "uk-constituencies-2024.hexjson"
with open(hexjson_path, "r", encoding="utf-8") as f:
hexjson = json.load(f)
# detect party columns in the incoming DataFrame (priority from DEFAULT_PARTY_ORDER)
parties = [p for p in DEFAULT_PARTY_ORDER if p in nowcast.columns]
if not parties:
exclude = {"Constituency", "constituency", "ConstituencyName", "Constituency Name", "Current", "Winner", "__slug"}
parties = [c for c in nowcast.columns if c not in exclude][:9]
# Work on a local copy to avoid mutating caller's DataFrame
df = nowcast.copy()
# Ensure the party columns we care about are numeric (coerce strings -> floats, replace NaN with 0.0)
numeric_party_cols = [p for p in parties if p in df.columns]
if numeric_party_cols:
df[numeric_party_cols] = df[numeric_party_cols].apply(pd.to_numeric, errors="coerce").fillna(0.0)
# index by slug for fast lookup; assume caller provided __slug
if "__slug" not in df.columns:
df["__slug"] = df.get("Constituency", "").apply(slug_name)
nowcast_index = df.set_index("__slug", drop=False)
hex_entries = hexjson.get("hexes", {})
trace_polygons = []
trace_centroids = []
slug_to_centroid = {}
for hid, meta in hex_entries.items():
name = meta.get("n") or meta.get("name") or hid
q = meta.get("q")
r = meta.get("r")
hex_colour = meta.get("colour")
slug = slug_name(name)
now_row = nowcast_index.loc[slug] if slug in nowcast_index.index else None
# Determine winner using numeric shares first
winner = None
if now_row is not None:
# collect numeric shares in the same order as numeric_party_cols
vals = [_safe_float(now_row.get(p, 0.0)) for p in numeric_party_cols]
if any(v > 0.0 for v in vals):
# choose index of the maximum value safely (no pandas.np)
max_ix = max(range(len(vals)), key=lambda i: vals[i])
winner = numeric_party_cols[max_ix]
else:
# fallback to the original Winner column if present and informative
w = now_row.get("Winner", None) if "Winner" in now_row else None
if w is not None and str(w).strip() not in ("", "-", "nan"):
winner = w
else:
winner = None
# pick colour from computed winner; fallback to hexjson colour or neutral grey
color = PARTY_COLOR_MAP.get(str(winner), hex_colour or "#CCCCCC")
cx, cy = hex_to_pixel(q, r)
verts = polygon_hex(cx, cy)
xs, ys = zip(*verts)
# polygon trace for the hex
poly = go.Scatter(
x=xs,
y=ys,
fill="toself",
mode="lines",
line=dict(width=0.6, color="white"),
fillcolor=color,
hoverinfo="none",
name=name,
showlegend=False,
)
trace_polygons.append(poly)
# centroid invisible marker with hover text = name
centroid = go.Scatter(
x=[cx],
y=[cy],
mode="markers",
marker=dict(size=44, color="rgba(0,0,0,0)"),
hoverinfo="text",
text=[name],
hovertemplate="%{text}<extra></extra>",
name=f"{name}-hit",
showlegend=False,
)
trace_centroids.append(centroid)
slug_to_centroid[slug] = (cx, cy)
all_traces = trace_polygons + trace_centroids
fig = go.Figure(data=all_traces)
fig.update_layout(
margin=dict(l=8, r=8, t=8, b=8),
xaxis=dict(visible=False, showgrid=False, zeroline=False),
yaxis=dict(visible=False, showgrid=False, zeroline=False, scaleanchor="x"),
hovermode="closest",
plot_bgcolor="rgba(0,0,0,0)",
paper_bgcolor="rgba(0,0,0,0)",
)
# tighten axis ranges
all_x = [x for tr in all_traces for x in tr.x if x is not None]
all_y = [y for tr in all_traces for y in tr.y if y is not None]
if all_x and all_y:
minx, maxx = min(all_x), max(all_x)
miny, maxy = min(all_y), max(all_y)
dx = (maxx - minx) * 0.005
dy = (maxy - miny) * 0.005
fig.update_xaxes(range=[minx - dx, maxx + dx])
fig.update_yaxes(range=[miny - dy, maxy + dy], autorange=False)
return fig, slug_to_centroid
def find_centroid_by_slug(slug_to_centroid: Dict[str, Tuple[float, float]], slug: str):
return slug_to_centroid.get(slug)
|