diff --git a/scripts/probe_iw_themes.py b/scripts/probe_iw_themes.py new file mode 100644 index 0000000..900ee6c --- /dev/null +++ b/scripts/probe_iw_themes.py @@ -0,0 +1,53 @@ +"""Dump FULL (untruncated) text + options + coverage for WVS questions matching each IW theme, so +the precise per-item selector and axis-positive pole can be locked by eye. -- authored by Claude + + uv run python scripts/probe_iw_themes.py +""" +from __future__ import annotations + +import ast +import re + +from datasets import load_dataset + +from tinymfv.zones import zone_of + +SKIP = re.compile(r"don'?t know|no answer|refus|decline|none of|not applicable|^other|missing|inap", + re.I) + +THEMES = { + "Y_religion": r"how important it is in your life|believe in any|importance of god|attend religious", + "Y_abortion": r"\babortion\b", + "Y_childaut": r"encouraged to learn at home", + "X_homosex": r"homosexual", + "X_trust": r"most people can be trusted", + "X_politaction": r"forms of political action", +} + + +def main() -> None: + ds = load_dataset("Anthropic/llm_global_opinions", split="train") + recs = [] + for r in ds: + if r["source"] != "WVS" or not r["question"]: + continue + opts = ast.literal_eval(r["options"]) if isinstance(r["options"], str) else r["options"] + keep = [i for i, o in enumerate(opts) if not SKIP.search(o)] + sel = ast.literal_eval(re.search(r"\{.*\}", r["selections"], re.S).group(0)) + ncov = sum(1 for c in sel if zone_of(c) and sum(sel[c][i] for i in keep) > 0) + recs.append({"q": r["question"], "opts": [opts[i] for i in keep], "ncov": ncov}) + + for theme, pat in THEMES.items(): + rx = re.compile(pat, re.I) + hits = sorted([r for r in recs if rx.search(r["q"])], key=lambda r: -r["ncov"]) + print(f"\n{'='*90}\n{theme}: {len(hits)} matches\n{'='*90}") + for r in hits: + if r["ncov"] < 40: + continue + tail = r["q"][-90:].replace("\n", " / ") + print(f"[cov {r['ncov']:3d}] n_opts={len(r['opts'])} ...TAIL: {tail!r}") + print(f" opts: {r['opts']}") + + +if __name__ == "__main__": + main() diff --git a/scripts/wvs_map.py b/scripts/wvs_map.py index a5126fd..8105877 100644 --- a/scripts/wvs_map.py +++ b/scripts/wvs_map.py @@ -1,11 +1,16 @@ -"""WVS / Inglehart-Welzel style culture map: place LLMs among human societies (the Economist chart). +"""WVS Inglehart-Welzel culture map with LABELED axes: place LLMs among human societies (the +Economist chart), on the two named IW dimensions instead of a blind PCA. -Runs tinymfv on the WVS subset of Anthropic/llm_global_opinions "like the others": each 4-option WVS -question is one ordinal item; a country x question matrix of expected-score E is ipsative-PCA'd (the -same maps.ipsative_pca the instrument maps use) into 2 axes, with IW zone ellipses. Each model is -administered the SAME questions -- open models via the logprob reader (read_items), logprob-less API -models via the sampling reader (read_items_sampled) -- reduced to an E-vector and projected onto the -country axes as a labelled dot. + X = Survival <-> Self-expression (homosexuality tolerance, interpersonal trust, political action) + Y = Traditional <-> Secular-Rational (religion importance + belief, abortion, child autonomy) + +Each axis is a small hand-picked battery of GlobalOpinionQA WVS items (tinymfv.iw_axes), every item +oriented to its axis-positive pole by reading the option order. A country's coordinate is the mean +`positiveness` (0-1) over that axis's items from the human WVS distribution; a model's coordinate is +the SAME items administered through the answer-token reader (open models: read_items; logprob-less +API models: read_items_sampled), reduced identically. This is an APPROXIMATE IW (3 themes/axis, not +the canonical 5 -- national pride/authority/materialism are absent from GlobalOpinionQA), not a +verbatim reproduction; the caveat is printed on the figure. uv run python scripts/wvs_map.py --local-model Qwen/Qwen3-0.6B \ --api-models meta-llama/llama-3.1-8b-instruct openai/gpt-4o-mini @@ -16,7 +21,6 @@ import argparse import ast import json import re -from collections import Counter from pathlib import Path import dotenv @@ -32,31 +36,36 @@ from datasets import load_dataset from transformers import AutoModelForCausalLM, AutoTokenizer from tinymfv import maps -from tinymfv.zones import zones_for, zone_of, ECONOMIST_OUTLIERS +from tinymfv.zones import zones_for, zone_of from tinymfv.instrument import Instrument, InstrItem from tinymfv.read import read_items, resolve_answer_ids from tinymfv.read_api import read_items_sampled -from tinymfv.readouts import expected_score +from tinymfv.iw_axes import AXIS_ITEMS, X_AXIS, Y_AXIS, SKIP, resolve_items, positiveness -SKIP = re.compile(r"don'?t know|no answer|refus|decline|none of|not applicable|^other", re.I) MODEL_COLORS = ["#c0392b", "#8e44ad", "#16a085", "#d35400", "#2980b9", "#c2185b"] +# option labels are single digits 0..n-1 -- single-token (unlike '10' on the justifiable scale) and +# the format the answer-token reader is tuned for (a bare digit, not a letter the model ignores in +# favour of the option word). +DIGITS = "0123456789" -def load_wvs(n_opts: int) -> list[dict]: - """WVS questions with exactly `n_opts` substantive options (DK/No-answer dropped), each carrying - its per-country human distribution renormalized over the substantive options.""" +def load_wvs_all() -> list[dict]: + """Every WVS question with its substantive options (DK/refusal/Missing/INAP dropped) and each + zone-mapped country's distribution renormalized over those options.""" ds = load_dataset("Anthropic/llm_global_opinions", split="train") out = [] for r in ds: - if r["source"] != "WVS": + if r["source"] != "WVS" or not r["question"]: continue opts = ast.literal_eval(r["options"]) if isinstance(r["options"], str) else r["options"] keep = [i for i, o in enumerate(opts) if not SKIP.search(o)] - if len(keep) != n_opts: + if len(keep) < 2: continue sel = ast.literal_eval(re.search(r"\{.*\}", r["selections"], re.S).group(0)) dist = {} for c, ps in sel.items(): + if not zone_of(c): + continue v = np.array([ps[i] for i in keep], float) if v.sum() > 0: dist[c] = v / v.sum() @@ -64,86 +73,94 @@ def load_wvs(n_opts: int) -> list[dict]: return out -def dense_block(recs: list[dict]) -> tuple[list[str], list[dict]]: - """Largest complete country x question block (no imputation): start from every zone-mapped - country x every question, then greedily drop whichever row or column has the most missing cells - until no cell is missing. WVS coverage is patchy per-question, so a complete block needs this - trim rather than a fixed coverage threshold.""" - countries = sorted({c for r in recs for c in r["dist"] if zone_of(c)}) - A = np.array([[c in r["dist"] for r in recs] for c in countries]) # C x Q presence - rows, cols = list(range(len(countries))), list(range(len(recs))) - while True: - sub = A[np.ix_(rows, cols)] - if sub.all(): - break - frac_r, frac_c = (~sub).mean(1), (~sub).mean(0) # missing FRACTION per country / per question - if frac_r.max() >= frac_c.max(): - rows.pop(int(np.argmax(frac_r))) - else: - cols.pop(int(np.argmax(frac_c))) - return [countries[i] for i in rows], [recs[j] for j in cols] +def human_axis_scores(resolved: dict[str, list[dict]]) -> tuple[list[str], np.ndarray]: + """Per-country (X, Y). A country is kept if it covers at least half of each axis's items; its + axis value is the mean positiveness over the items it does cover.""" + countries = sorted({c for items in resolved.values() for it in items for c in it["rec"]["dist"]}) + rows, keep = [], [] + for c in countries: + xy, ok = [], True + for axis in (X_AXIS, Y_AXIS): + vals = [positiveness(it["rec"]["dist"][c], it["pole_idx"], it["n"]) + for it in resolved[axis] if c in it["rec"]["dist"]] + if len(vals) < (len(resolved[axis]) + 1) // 2: + ok = False + break + xy.append(float(np.mean(vals))) + if ok: + keep.append(c) + rows.append(xy) + return keep, np.array(rows) -def human_matrix(countries: list[str], block: list[dict], n_opts: int) -> np.ndarray: - """countries x questions matrix of expected-score E as a 0-1 fraction (E-1)/(n_opts-1).""" - w = np.arange(1, n_opts + 1) - M = np.array([[float((block[q]["dist"][c] * w).sum()) for q in range(len(block))] for c in countries]) - return (M - 1) / (n_opts - 1) +def build_instruments(resolved: dict[str, list[dict]]) -> tuple[list[Instrument], dict[str, dict]]: + """One nominal Instrument per distinct option-count (answer_space = single letters), covering the + union of both axes' items. Returns the instruments + a {suffix: {pole_idx, n, axis}} index.""" + items_by_n: dict[int, list[InstrItem]] = {} + meta: dict[str, dict] = {} + seen: set[str] = set() + for axis, items in resolved.items(): + for it in items: + s = it["suffix"] + meta[s] = {"pole_idx": it["pole_idx"], "n": it["n"], "axis": axis} + if s in seen: + continue + seen.add(s) + n, opts = it["n"], it["rec"]["opts"] + legend = "; ".join(f"{DIGITS[k]}) {o}" for k, o in enumerate(opts)) + task = f"Answer options: {legend}. Respond with only the number." + items_by_n.setdefault(n, []).append( + InstrItem(id=s, prompt=it["rec"]["q"], dimension="iw", sign=1, + frame="forward", meta={"task": task})) + instrs = [Instrument(name=f"wvs_iw_n{n}", construct="opinion", kind="nominal", + answer_space=list(DIGITS[:n]), dimensions=["iw"], items=its, + prefill="(", display="WVS-IW") + for n, its in sorted(items_by_n.items())] + return instrs, meta -def build_instrument(block: list[dict], n_opts: int) -> Instrument: - """One ordinal Instrument over the WVS questions: answer_space digits 1..n_opts, each item the - question + its numbered options legend.""" - items = [] - for i, r in enumerate(block): - legend = "; ".join(f"{k+1}) {o}" for k, o in enumerate(r["opts"])) - task = f"Answer options: {legend}. Respond with only the number." - items.append(InstrItem(id=str(i), prompt=r["q"], dimension="wvs", sign=1, - frame="forward", meta={"task": task})) - return Instrument(name="wvs", construct="opinion", kind="ordinal", - answer_space=[str(i) for i in range(1, n_opts + 1)], - dimensions=["wvs"], items=items, prefill="(", scale_max=n_opts, - human_scale_max=n_opts, display="WVS") +def model_axis_scores(vecs: dict[str, np.ndarray], meta: dict[str, dict], + resolved: dict[str, list[dict]]) -> tuple[float, float]: + """(X, Y) for one model from its per-item p vectors (suffix -> p over options).""" + xy = [] + for axis in (X_AXIS, Y_AXIS): + vals = [positiveness(vecs[it["suffix"]], it["pole_idx"], it["n"]) for it in resolved[axis]] + xy.append(float(np.mean(vals))) + return xy[0], xy[1] -def model_vector(rows: list[dict], n: int, n_opts: int) -> np.ndarray: - """Per-question E as a 0-1 fraction, in item order. NaN if a question's read collapsed.""" - by_id = {int(r["id"]): r for r in rows} - out = np.full(n, np.nan) - for i in range(n): - p = np.asarray(by_id[i]["p"], float) - if np.isfinite(p).all() and abs(p.sum() - 1) < 1e-6: - out[i] = (expected_score(p, n_opts) - 1) / (n_opts - 1) - return out +def read_model(rows: list[dict], meta: dict[str, dict]) -> dict[str, np.ndarray]: + """rows from read_items / read_items_sampled -> {suffix: p over that item's options}. NaN p (read + collapse) fails loud later via positiveness rather than being imputed.""" + return {r["id"]: np.asarray(r["p"], float)[: meta[r["id"]]["n"]] for r in rows} def main() -> None: ap = argparse.ArgumentParser() - ap.add_argument("--n-opts", type=int, default=4) ap.add_argument("--local-model", default="Qwen/Qwen3-0.6B") ap.add_argument("--api-models", nargs="*", default=[]) ap.add_argument("--api-samples", type=int, default=20) ap.add_argument("--max-think-tokens", type=int, default=64) ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") - ap.add_argument("--out", default="/tmp/claude-1000/wvs_map.png") - ap.add_argument("--cache", default="/tmp/claude-1000/wvs_model_vectors.json", - help="cache model E-vectors so re-plotting (new zone style) skips the API calls") + ap.add_argument("--out", default="/tmp/claude-1000/wvs_map_iw.png") + ap.add_argument("--cache", default="/tmp/claude-1000/wvs_iw_vectors.json", + help="cache model per-item p vectors so re-styling skips the API/model calls") args = ap.parse_args() - recs = load_wvs(args.n_opts) - countries, block = dense_block(recs) - logger.info(f"{len(recs)} {args.n_opts}-option WVS questions -> dense block " - f"{len(countries)} countries x {len(block)} questions") - Hfrac = human_matrix(countries, block, args.n_opts) - P, Vt, var, mu, Pc = maps.ipsative_pca(Hfrac) - instr = build_instrument(block, args.n_opts) + recs = load_wvs_all() + resolved = resolve_items(recs) + for axis, items in resolved.items(): + logger.info(f"{axis}: " + ", ".join(f"{it['suffix']}[n{it['n']},pole{it['pole_idx']}]" + for it in items)) + countries, P = human_axis_scores(resolved) + logger.info(f"{len(recs)} WVS questions -> {len(countries)} countries on 2 IW axes") - # cache keyed by the exact question block: same block -> reuse E-vectors, only re-projecting + - # re-drawing (so iterating on the zone style costs no API calls). - sig = f"{args.n_opts}:{len(block)}:{hash(tuple(r['q'] for r in block)) & 0xffffffff}" + instrs, meta = build_instruments(resolved) + sig = str(hash(tuple(sorted((s, m["n"]) for s, m in meta.items()))) & 0xffffffff) cpath = Path(args.cache) cache = json.loads(cpath.read_text()).get(sig, {}) if cpath.exists() else {} - vecs: dict[str, np.ndarray] = {k: np.array(v) for k, v in cache.items()} + vecs: dict[str, dict[str, np.ndarray]] = {k: {s: np.array(p) for s, p in v.items()} + for k, v in cache.items()} if args.local_model: key = args.local_model.split("/")[-1] + " (lp)" @@ -153,41 +170,58 @@ def main() -> None: tok.pad_token = tok.eos_token tok.padding_side = "left" lm = AutoModelForCausalLM.from_pretrained(args.local_model, dtype=torch.bfloat16).to(args.device).eval() - rows = read_items(lm, tok, instr, instr.items, resolve_answer_ids(tok, instr.answer_space), - max_think_tokens=args.max_think_tokens, batch_size=16, verbose_first=True) - vecs[key] = model_vector(rows, len(block), args.n_opts) + rows = [] + for k, instr in enumerate(instrs): + rows += read_items(lm, tok, instr, instr.items, + resolve_answer_ids(tok, instr.answer_space), + max_think_tokens=args.max_think_tokens, batch_size=16, + verbose_first=(k == 0)) + vecs[key] = read_model(rows, meta) for m in args.api_models: key = m.split("/")[-1] + " (sampled)" if key not in vecs: - rows = read_items_sampled(m, instr, instr.items, n_samples=args.api_samples, verbose_first=True) - vecs[key] = model_vector(rows, len(block), args.n_opts) + rows = [] + for k, instr in enumerate(instrs): + rows += read_items_sampled(m, instr, instr.items, n_samples=args.api_samples, + verbose_first=(k == 0)) + vecs[key] = read_model(rows, meta) cpath.parent.mkdir(parents=True, exist_ok=True) allc = json.loads(cpath.read_text()) if cpath.exists() else {} - allc[sig] = {k: v.tolist() for k, v in vecs.items()} + allc[sig] = {k: {s: p.tolist() for s, p in v.items()} for k, v in vecs.items()} cpath.write_text(json.dumps(allc)) - hmean = Hfrac.mean(0) # a question the model didn't answer coherently -> the human average there - models = {k: ((np.where(np.isnan(v), hmean, v) @ Pc) - mu) @ Vt[:2].T for k, v in vecs.items()} + models = {k: model_axis_scores(v, meta, resolved) for k, v in vecs.items()} zones, emph = zones_for(countries) - fig, ax = plt.subplots(figsize=(10, 8)) + zone_of_c = {c: z for z, members in zones.items() for c in members} + dot_cols = [maps.ZONE_COLORS.get(zone_of_c[c], "#888888") for c in countries] + fig, ax = plt.subplots(figsize=(11, 9)) ax.set_facecolor("#faf8f2") ax.grid(True, color="#eceadf", lw=0.3, zorder=0) - maps.draw_zone_regions(ax, P, countries, zones) - ax.scatter(P[:, 0], P[:, 1], s=22, c=maps.C_HUM, alpha=0.7, edgecolors="white", linewidths=0.4, zorder=3) + ax.axhline(0.5, color="#d9d5c6", lw=0.8, zorder=1) + ax.axvline(0.5, color="#d9d5c6", lw=0.8, zorder=1) + maps.draw_zone_hulls(ax, P, countries, zones) + # Dots coloured by zone (region shown by colour, like the Economist) + only the named outliers + # labelled -- 90 country labels is the clutter the user flagged; the region name (drawn by + # draw_zone_regions at each zone centroid) carries the rest. + ax.scatter(P[:, 0], P[:, 1], s=28, c=dot_cols, alpha=0.85, edgecolors="white", linewidths=0.5, zorder=3) for i, c in enumerate(countries): - is_e = c in emph - ax.annotate(c, (P[i, 0], P[i, 1]), fontsize=7.5 if is_e else 6, xytext=(3, 2), - textcoords="offset points", color="#111" if is_e else "#666", - fontweight="bold" if is_e else "normal", zorder=6) + if c in emph: + ax.annotate(c, (P[i, 0], P[i, 1]), fontsize=9, xytext=(4, 3), + textcoords="offset points", color="#111", fontweight="bold", zorder=6) for (name, pt), col in zip(models.items(), MODEL_COLORS): - ax.scatter(*pt, s=120, marker="*", c=col, edgecolors="white", linewidths=1.0, zorder=8) - ax.annotate(name, pt, xytext=(6, 4), textcoords="offset points", fontsize=9, + ax.scatter(*pt, s=150, marker="*", c=col, edgecolors="white", linewidths=1.0, zorder=8) + ax.annotate(name, pt, xytext=(7, 4), textcoords="offset points", fontsize=9, fontweight="bold", color=col, zorder=9) - ax.set_xlabel(f"PC1 ({var[0]*100:.0f}% var)") - ax.set_ylabel(f"PC2 ({var[1]*100:.0f}% var)") - ax.set_title(f"WVS values map: LLMs among {len(countries)} human societies " - f"({len(block)} WVS questions, ipsative PCA)", fontsize=11) + ax.set_xlabel(f"{X_AXIS} (right = self-expression)") + ax.set_ylabel(f"{Y_AXIS} (up = secular-rational)") + ax.set_title(f"WVS Inglehart-Welzel map: LLMs among {len(countries)} human societies " + f"(approximate IW axes)", fontsize=12) + ax.text(0.01, 0.01, + "Approximate IW: axes built from GlobalOpinionQA WVS items (3 themes/axis, not the\n" + "canonical 5; national pride / authority / materialism absent). Not a verbatim WVS " + "factor score.", + transform=ax.transAxes, fontsize=6.5, color="#888", va="bottom", ha="left", zorder=10) fig.tight_layout() fig.savefig(args.out, dpi=200, bbox_inches="tight") logger.info(f"wrote {args.out}") diff --git a/src/tinymfv/iw_axes.py b/src/tinymfv/iw_axes.py new file mode 100644 index 0000000..20d82f3 --- /dev/null +++ b/src/tinymfv/iw_axes.py @@ -0,0 +1,88 @@ +"""Approximate Inglehart-Welzel axes over the GlobalOpinionQA WVS items. -- authored by Claude + +NOT the verbatim WVS factor scores. The canonical IW battery is 10 items (5 per axis); national +pride, respect-for-authority, materialist/postmaterialist priorities and happiness are absent or too +sparse in GlobalOpinionQA, so this is a keyword-selected APPROXIMATION: two axes, a few items each, +each item oriented to its axis-positive pole by reading the OPTION text -- so a reversed option order +(one child-quality row is stored ['Not mentioned','Important']) cannot silently flip a country. + + X = Survival (0) <-> Self-expression (1) + Y = Traditional (0) <-> Secular-Rational (1) + +An entity's axis value = mean over that axis's covered items of `positiveness` in [0,1], where +positiveness is the answer's expected position toward the axis-positive pole (0 = opposite pole, +1 = the pole). Each item is keyed by a unique trailing substring of the full WVS question (the +sub-item name GlobalOpinionQA appends after the shared stem). +""" +from __future__ import annotations + +import re + +import numpy as np + +# drop non-substantive options before renormalizing (DK / refusals / the "Missing"/"INAP" fillers +# GlobalOpinionQA leaves in some option lists). +SKIP = re.compile(r"don'?t know|no answer|refus|decline|none of|not applicable|^other|missing|inap", + re.I) + +X_AXIS = "Survival <-> Self-expression" +Y_AXIS = "Traditional <-> Secular-Rational" + +# axis -> [(question-text suffix uniquely naming the WVS sub-item, option substring naming the +# axis-POSITIVE pole)]. Suffix is matched by str.endswith on the full question; pole must be an +# endpoint option (both asserted in resolve_items). +AXIS_ITEMS = { + Y_AXIS: [ + ("Religion", "Not at all important"), # importance of religion (secular = not important) + ("God", "No"), # believe in God (secular = No) + ("Abortion", "Always justifiable"), # abortion (secular = justifiable) + ("Obedience", "Not mentioned"), # child quality: obedience (secular = don't teach) + ("Independence", "Important"), # child quality: independence (secular = teach) + ("Determination, perseverance", "Important"), # child quality: determination (secular = teach) + ("Imagination", "Important"), # child quality: imagination (secular = teach) + ], + X_AXIS: [ + ("Homosexuality", "Always justifiable"), # homosexuality (self-expr = justifiable) + ("dealing with people?", "Most people can be trusted"), # interpersonal trust (self-expr = trust) + ("Signing a petition", "Have done"), # political action (self-expr = have done) + ("Attending peaceful demonstrations", "Have done"), # political action + ("Joining in boycotts", "Have done"), # political action + ], +} + + +def e_frac(dist: np.ndarray, n: int) -> float: + """Expected normalized option position in [0,1]: sum_k (k/(n-1)) * p_k, k = 0..n-1.""" + k = np.arange(n, dtype=float) / (n - 1) + return float((np.asarray(dist, dtype=float) * k).sum()) + + +def positiveness(dist: np.ndarray, pole_idx: int, n: int) -> float: + """Position toward the axis-positive pole in [0,1]. pole at the LAST option -> e_frac; at the + FIRST option -> 1 - e_frac. Only endpoint poles are meaningful for a signed axis.""" + e = e_frac(dist, n) + if pole_idx == n - 1: + return e + if pole_idx == 0: + return 1.0 - e + raise ValueError(f"pole must be an endpoint option, got idx {pole_idx} of {n}") + + +def resolve_items(recs: list[dict]) -> dict[str, list[dict]]: + """recs: [{q, opts, dist}]. Resolve each AXIS_ITEMS entry to exactly one question; fail loud on + 0 or >1 selector matches, an ambiguous pole, or a non-endpoint pole. Returns axis -> list of + {suffix, rec, pole_idx, n}.""" + resolved: dict[str, list[dict]] = {} + for axis, items in AXIS_ITEMS.items(): + rl = [] + for suffix, pole_sub in items: + hits = [r for r in recs if r["q"].strip().endswith(suffix)] + assert len(hits) == 1, f"{axis}: suffix {suffix!r} matched {len(hits)} questions (want 1)" + r = hits[0] + n = len(r["opts"]) + pole = [i for i, o in enumerate(r["opts"]) if pole_sub.lower() in o.lower()] + assert len(pole) == 1, f"{suffix!r}: pole {pole_sub!r} matched options {pole} of {r['opts']}" + assert pole[0] in (0, n - 1), f"{suffix!r}: pole not an endpoint (idx {pole[0]} of {n})" + rl.append({"suffix": suffix, "rec": r, "pole_idx": pole[0], "n": n}) + resolved[axis] = rl + return resolved diff --git a/src/tinymfv/maps.py b/src/tinymfv/maps.py index 2432cee..370f440 100644 --- a/src/tinymfv/maps.py +++ b/src/tinymfv/maps.py @@ -112,6 +112,33 @@ def _country_region(cen: np.ndarray, pts: np.ndarray | None, sigma: float, r_fix return Point(*cen).buffer(r_fixed, quad_segs=24) +def draw_zone_hulls(ax, P: np.ndarray, countries: list[str], zones: dict[str, list[str]], + pad: float = 0.022, alpha: float = 0.13) -> None: + """Economist-style zone outline: the tight CONVEX HULL of a zone's country-mean points, rounded + and slightly inflated (shapely buffer), lightly filled with the zone colour, thin outline, and a + white-haloed italic label at the centroid. A 1- or 2-country zone degenerates to a rounded + disc/capsule via the same buffer. Far cleaner than a union of per-country discs when the axes + already separate the countries (WVS IW map); the disc-union `draw_zone_regions` stays for the + instrument maps that overlay real within-country respondent spread.""" + import matplotlib.patheffects as pe + from shapely.geometry import MultiPoint + from matplotlib.patches import Polygon as MplPolygon + cidx = {c: i for i, c in enumerate(countries)} + buf = pad * float(np.hypot(*(P.max(0) - P.min(0)))) + for zname, members in zones.items(): + pts = np.array([P[cidx[c]] for c in members if c in cidx]) + if not len(pts): + continue + geom = MultiPoint([tuple(p) for p in pts]).convex_hull.buffer(buf, quad_segs=16) + zcol = ZONE_COLORS.get(zname, "#888888") + ax.add_patch(MplPolygon(np.asarray(geom.exterior.coords), closed=True, facecolor=zcol, + edgecolor=zcol, alpha=alpha, lw=1.1, zorder=1.5)) + cen = pts.mean(0) + ax.text(cen[0], cen[1], zname, fontsize=9.5, color=zcol, ha="center", va="center", + style="italic", fontweight="bold", zorder=5, + path_effects=[pe.withStroke(linewidth=3.0, foreground="white")]) + + def draw_zone_regions(ax, P: np.ndarray, countries: list[str], zones: dict[str, list[str]], cloud_P: np.ndarray | None = None, cloud_countries: list[str] | None = None, sigma: float = 1.0) -> None: @@ -123,7 +150,7 @@ def draw_zone_regions(ax, P: np.ndarray, countries: list[str], zones: dict[str, from shapely.ops import unary_union from matplotlib.patches import Polygon as MplPolygon cidx = {c: i for i, c in enumerate(countries)} - r_fixed = 0.06 * float(np.hypot(*(P.max(0) - P.min(0)))) + r_fixed = 0.09 * float(np.hypot(*(P.max(0) - P.min(0)))) # ~50% bigger than the first pass by_country: dict[str, np.ndarray] = {} if cloud_P is not None and cloud_countries is not None: cc = np.asarray(cloud_countries) @@ -148,8 +175,12 @@ def draw_zone_regions(ax, P: np.ndarray, countries: list[str], zones: dict[str, ax.add_patch(MplPolygon(np.asarray(g.exterior.coords), closed=True, facecolor=zcol, edgecolor=zcol, alpha=0.15, lw=1.1, zorder=1.5)) cen = P[idxs].mean(0) - ax.text(cen[0], cen[1], zname, fontsize=8.5, color=zcol, ha="center", va="center", - style="italic", fontweight="bold", zorder=2, alpha=0.9) + # white halo so the zone-coloured label reads on top of the same-coloured blob (no contrast + # otherwise). Darkened text + heavier stroke over the pale fill. + import matplotlib.patheffects as pe + ax.text(cen[0], cen[1], zname, fontsize=9.5, color=zcol, ha="center", va="center", + style="italic", fontweight="bold", zorder=5, + path_effects=[pe.withStroke(linewidth=3.0, foreground="white")])