Download app.py from neovalle/SpatialSemanticAnalysis: direct link, hf CLI and curl.
- Browser
- Download file 26.5 kB
-
https://huggingface.co/spaces/neovalle/SpatialSemanticAnalysis/resolve/main/app.py
- Command line
-
hf download hf://spaces/neovalle/SpatialSemanticAnalysis/app.py
-
curl -L -o app.py https://huggingface.co/spaces/neovalle/SpatialSemanticAnalysis/resolve/main/app.py
26.5 kB
| """ | |
| SpatialDiscourseAnalysis v2 | |
| H4rmony Project · Jorge Vallego · https://theh4rmonyproject.org | |
| Places the sentences of up to four texts on an evaluative axis defined by two sets of | |
| example sentences (poles), in the semantic space of all-mpnet-base-v2. | |
| New in v2 | |
| - Relevance filter: keeps only sentences that match an editable lexicon, and logs why | |
| - Pole check: leave-one-out test that flags pole sentences sitting on the wrong side | |
| - Spread: median, SD, IQR, share of sentences leaning to each pole, tail direction | |
| - Principal-axis leaning: how each text's own main dimension of variation relates to the axis | |
| - Pairwise differences between texts with effect size and permutation test | |
| - Most aligned / most misaligned sentences for each text | |
| - Optional neutral baseline text | |
| - Per-sentence scores as a downloadable CSV | |
| Scale: 0% = centroid of pole A, 100% = centroid of pole B, 50% = midpoint. | |
| """ | |
| import re | |
| import json | |
| import tempfile | |
| import numpy as np | |
| import pandas as pd | |
| try: | |
| import gradio as gr | |
| except ImportError: # allows the analysis core to be tested without gradio | |
| gr = None | |
| try: | |
| import spaces # Hugging Face ZeroGPU | |
| GPU = spaces.GPU | |
| except ImportError: | |
| def GPU(fn=None, **kwargs): | |
| return fn if fn is not None else (lambda f: f) | |
| MODEL_NAME = "sentence-transformers/all-mpnet-base-v2" | |
| _model = None | |
| def get_model(): | |
| global _model | |
| if _model is None: | |
| from sentence_transformers import SentenceTransformer | |
| _model = SentenceTransformer(MODEL_NAME) | |
| return _model | |
| def _embed_gpu(sentences): | |
| import torch | |
| m = get_model() | |
| if torch.cuda.is_available(): | |
| m = m.to("cuda") | |
| return m.encode(sentences, batch_size=64, normalize_embeddings=True, | |
| convert_to_numpy=True, show_progress_bar=False) | |
| def embed(sentences): | |
| """Replaceable in tests.""" | |
| if not sentences: | |
| return np.zeros((0, 768)) | |
| return np.asarray(_embed_gpu(list(sentences)), dtype=np.float64) | |
| # ---------------------------------------------------------------------------------------- | |
| # Text handling | |
| # ---------------------------------------------------------------------------------------- | |
| SENT_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-Z“"‘(\[])') | |
| def split_sentences(text, min_words=4): | |
| """Paragraphs or lines -> sentences. One-sentence-per-line input is preserved.""" | |
| out = [] | |
| for block in re.split(r'\n+', text or ""): | |
| block = " ".join(block.split()) | |
| if not block: | |
| continue | |
| for s in SENT_SPLIT.split(block): | |
| s = s.strip() | |
| if len(s.split()) >= min_words: | |
| out.append(s) | |
| # de-duplicate, keep order | |
| seen, uniq = set(), [] | |
| for s in out: | |
| if s not in seen: | |
| seen.add(s) | |
| uniq.append(s) | |
| return uniq | |
| def pole_sentences(text): | |
| return [" ".join(l.split()) for l in (text or "").splitlines() if l.strip()] | |
| # ---------------------------------------------------------------------------------------- | |
| # Relevance filter | |
| # One category per line: Category name: term, term*, multi word term | |
| # Optional co-occurrence: Category name: term, term | with: term, term | |
| # '*' = any word ending. Matching is case-insensitive and on whole words. | |
| # ---------------------------------------------------------------------------------------- | |
| DEFAULT_LEXICON = """\ | |
| Key nodes: nature, natural, sustainab*, smart, green | |
| Appraisal of nature: charming, crystal clear, crystal-clear, stunning, unspoil*, beautiful, relaxing, pristine, untouched, wild, wilder, wilderness, breathtaking, enchanting, paradise, oasis, oases, scenery, landscape, landscapes, panoramic, picturesque, evocative, untamed, unbridled | |
| Sustainable activities: bike, bikes, bicycle, bicycles, cycling, cycle, cyclist, cyclists, itinerary, itineraries, horseback, birdwatch*, trekking, hiking, nordic walking, canoe*, trail, trails, slow | |
| Slow-travel framing: authentic*, hidden gem, under the radar, under-the-radar, natural wonders, traveller, travellers, traveler, travelers, responsible, eco-conscious, eco-friendly, best-kept secret, best kept secret, lesser-known, lesser known, off the beaten, meaningful | |
| Ecological care: preserv*, protect*, conserv*, ecosystem, ecosystems, biodiversity, wildlife, species, habitat, habitats, safeguard*, coexist*, balance, fragile, reintroduc*, climate change | |
| Commercial modifiers: luxury, investment, invest, value, property, properties, exclusive, privacy, price, deal, deals, upscale, villa, villas, buyer, buyers, income, secluded, asset | |
| Offer framing: offer, offers, offered, offering | with: nature, park, coast, coastline, area, maremma, grosseto, beach, beaches, landscape, sea | |
| """ | |
| def _term_regex(term): | |
| t = re.escape(term.strip().lower()).replace(r"\*", r"\w*") | |
| t = t.replace(r"\ ", r"[\s-]+").replace(r"\-", r"[\s-]+") | |
| return t | |
| def parse_lexicon(text): | |
| cats = [] | |
| for line in (text or "").splitlines(): | |
| if ":" not in line or not line.strip(): | |
| continue | |
| name, rest = line.split(":", 1) | |
| with_part = None | |
| if "| with:" in rest: | |
| rest, with_part = rest.split("| with:", 1) | |
| terms = [t for t in rest.split(",") if t.strip()] | |
| if not terms: | |
| continue | |
| rx = re.compile(r"\b(" + "|".join(_term_regex(t) for t in terms) + r")\b", re.I) | |
| wrx = None | |
| if with_part: | |
| wt = [t for t in with_part.split(",") if t.strip()] | |
| wrx = re.compile(r"\b(" + "|".join(_term_regex(t) for t in wt) + r")\b", re.I) | |
| cats.append((name.strip(), rx, wrx)) | |
| return cats | |
| def match_categories(sentence, cats): | |
| hit = [] | |
| for name, rx, wrx in cats: | |
| if rx.search(sentence) and (wrx is None or wrx.search(sentence)): | |
| hit.append(name) | |
| return hit | |
| # ---------------------------------------------------------------------------------------- | |
| # Geometry | |
| # ---------------------------------------------------------------------------------------- | |
| def axis_scores(E, cA, cB): | |
| d = cB - cA | |
| return 100.0 * ((E - cA) @ d) / (d @ d) | |
| def pole_check(EA, EB, sents_a, sents_b): | |
| """Leave-one-out: re-score each pole sentence against centroids computed without it.""" | |
| cA, cB = EA.mean(0), EB.mean(0) | |
| sA, sB = axis_scores(EA, cA, cB), axis_scores(EB, cA, cB) | |
| pooled = np.sqrt((sA.var(ddof=1) + sB.var(ddof=1)) / 2) | |
| dprime = (sB.mean() - sA.mean()) / pooled if pooled > 0 else float("inf") | |
| stray = [] | |
| for i in range(len(EA)): | |
| cA_i = np.delete(EA, i, 0).mean(0) | |
| s = axis_scores(EA[i:i + 1], cA_i, cB)[0] | |
| if s >= 50: | |
| stray.append(("A", sents_a[i], s)) | |
| for i in range(len(EB)): | |
| cB_i = np.delete(EB, i, 0).mean(0) | |
| s = axis_scores(EB[i:i + 1], cA, cB_i)[0] | |
| if s <= 50: | |
| stray.append(("B", sents_b[i], s)) | |
| cos = float(cA @ cB / (np.linalg.norm(cA) * np.linalg.norm(cB))) | |
| if dprime >= 3: | |
| quality = "clearly defined" | |
| elif dprime >= 1.5: | |
| quality = "adequately defined" | |
| else: | |
| quality = "weakly defined: poles overlap; consider rewriting the stray sentences" | |
| return dict(cA=cA, cB=cB, sA=sA, sB=sB, dprime=float(dprime), stray=stray, | |
| centroid_cosine=cos, quality=quality) | |
| def spread(scores, low, high): | |
| q1, med, q3 = np.percentile(scores, [25, 50, 75]) | |
| sd = scores.std(ddof=1) if len(scores) > 1 else 0.0 | |
| skew = float((((scores - scores.mean()) / sd) ** 3).mean()) if sd > 0 else 0.0 | |
| return dict(n=int(len(scores)), mean=float(scores.mean()), median=float(med), sd=float(sd), | |
| iqr=float(q3 - q1), p10=float(np.percentile(scores, 10)), | |
| p90=float(np.percentile(scores, 90)), | |
| share_a=float((scores < low).mean()), | |
| share_neutral=float(((scores >= low) & (scores <= high)).mean()), | |
| share_b=float((scores > high).mean()), skew=skew) | |
| def principal_axis(E, cA, cB, k=3): | |
| d_hat = (cB - cA) / np.linalg.norm(cB - cA) | |
| X = E - E.mean(0) | |
| _, S, Vt = np.linalg.svd(X, full_matrices=False) | |
| var = S ** 2 / max(len(E) - 1, 1) | |
| total = float(var.sum()) if var.sum() > 0 else 1.0 | |
| comps = [] | |
| for i in range(min(k, len(var))): | |
| pc, proj = Vt[i], X @ Vt[i] | |
| if abs(proj.min()) > abs(proj.max()): # orient towards the longer tail | |
| pc = -pc | |
| cos = float(pc @ d_hat) | |
| comps.append(dict(pc=i + 1, variance_share=float(var[i] / total), | |
| angle=float(np.degrees(np.arccos(min(1.0, abs(cos))))), | |
| leaning="B" if cos > 0 else "A")) | |
| axis_var = float(((X @ d_hat) ** 2).sum() / max(len(E) - 1, 1)) | |
| return dict(components=comps, axis_share=axis_var / total, baseline=1.0 / E.shape[1]) | |
| def compare(s1, s2, n_perm=5000, seed=0): | |
| diff = float(s2.mean() - s1.mean()) | |
| pooled = np.sqrt((s1.var(ddof=1) + s2.var(ddof=1)) / 2) | |
| d = diff / pooled if pooled > 0 else 0.0 | |
| rng = np.random.default_rng(seed) | |
| allv, n1 = np.concatenate([s1, s2]), len(s1) | |
| count = 0 | |
| for _ in range(n_perm): | |
| rng.shuffle(allv) | |
| if abs(allv[n1:].mean() - allv[:n1].mean()) >= abs(diff): | |
| count += 1 | |
| p = (count + 1) / (n_perm + 1) | |
| return dict(diff=diff, cohen_d=float(d), p=float(p)) | |
| def size_label(d): | |
| d = abs(d) | |
| return "negligible" if d < 0.2 else "small" if d < 0.5 else "medium" if d < 0.8 else "large" | |
| # ---------------------------------------------------------------------------------------- | |
| # Analysis pipeline | |
| # ---------------------------------------------------------------------------------------- | |
| def analyse(name_a, poles_a, name_b, poles_b, texts, baseline_text, use_filter, lexicon, | |
| low, high, n_extremes=10): | |
| """texts: list of (name, raw_text). Returns a dict of results (no plotting).""" | |
| name_a = name_a.strip() or "Pole A" | |
| name_b = name_b.strip() or "Pole B" | |
| PA, PB = pole_sentences(poles_a), pole_sentences(poles_b) | |
| if len(PA) < 3 or len(PB) < 3: | |
| raise ValueError("Each pole needs at least 3 sentences (one per line).") | |
| EA, EB = embed(PA), embed(PB) | |
| pc = pole_check(EA, EB, PA, PB) | |
| cA, cB = pc["cA"], pc["cB"] | |
| cats = parse_lexicon(lexicon) if use_filter else [] | |
| items, filter_log = [], [] | |
| if baseline_text and baseline_text.strip(): | |
| texts = list(texts) + [("Neutral baseline", baseline_text)] | |
| for name, raw in texts: | |
| if not raw or not raw.strip(): | |
| continue | |
| name = name.strip() or f"Text {len(items) + 1}" | |
| sents = split_sentences(raw) | |
| kept, kept_cats = [], [] | |
| for s in sents: | |
| c = match_categories(s, cats) if (cats and name != "Neutral baseline") else [] | |
| if cats and name != "Neutral baseline" and not c: | |
| continue | |
| kept.append(s) | |
| kept_cats.append(c) | |
| filter_log.append(dict(text=name, total=len(sents), kept=len(kept), | |
| by_category={n: sum(n in c for c in kept_cats) for n, _, _ in cats})) | |
| if len(kept) < 3: | |
| continue | |
| E = embed(kept) | |
| s = axis_scores(E, cA, cB) | |
| order = np.argsort(s) | |
| items.append(dict(name=name, sentences=kept, categories=kept_cats, E=E, scores=s, | |
| spread=spread(s, low, high), | |
| pa=principal_axis(E, cA, cB) if len(kept) > 3 else None, | |
| most_a=[(kept[i], float(s[i])) for i in order[:n_extremes]], | |
| most_b=[(kept[i], float(s[i])) for i in order[::-1][:n_extremes]])) | |
| if not items: | |
| raise ValueError("No text has at least 3 sentences after filtering.") | |
| pairs = [] | |
| main = [it for it in items if it["name"] != "Neutral baseline"] | |
| for i in range(len(main)): | |
| for j in range(i + 1, len(main)): | |
| pairs.append((main[i]["name"], main[j]["name"], | |
| compare(main[i]["scores"], main[j]["scores"]))) | |
| base = next((it for it in items if it["name"] == "Neutral baseline"), None) | |
| return dict(name_a=name_a, name_b=name_b, PA=PA, PB=PB, EA=EA, EB=EB, pole=pc, | |
| items=items, pairs=pairs, baseline=base, filter_log=filter_log, | |
| use_filter=use_filter, low=low, high=high) | |
| # ---------------------------------------------------------------------------------------- | |
| # Reporting | |
| # ---------------------------------------------------------------------------------------- | |
| def z(x): | |
| """Round for display without printing -0.""" | |
| r = round(float(x)) | |
| return 0 if r == 0 else r | |
| def bar(pct, width=41): | |
| pos = int(round(min(max(pct, 0), 100) / 100 * (width - 1))) | |
| return "░" * pos + "●" + "░" * (width - 1 - pos) | |
| def format_report(R): | |
| A, B, pc = R["name_a"], R["name_b"], R["pole"] | |
| L = ["═" * 66, " SPATIAL DISCOURSE ANALYSIS v2 — Results", "═" * 66, "", | |
| f" AXIS: {A} ←────────→ {B}", | |
| f" 0% = {A} centroid · 100% = {B} centroid · 50% = midpoint", | |
| f" ({len(R['PA'])} sentences at {A} · {len(R['PB'])} at {B})", "", | |
| "─" * 66, " 1. POLE CHECK", "─" * 66, | |
| f" Axis quality: {pc['quality']} (separation d′ = {pc['dprime']:.2f})", | |
| f" Leave-one-out: {len(pc['stray'])} of {len(R['PA']) + len(R['PB'])} pole sentences fall on the wrong side"] | |
| for side, s, sc in pc["stray"]: | |
| L.append(f" [{A if side == 'A' else B}] {sc:.0f}% {s}") | |
| L += ["", "─" * 66, " 2. SENTENCE SELECTION", "─" * 66] | |
| if R["use_filter"]: | |
| L.append(" Relevance filter ON: sentences kept only if they match at least one lexicon category.") | |
| else: | |
| L.append(" Relevance filter OFF: all sentences used.") | |
| for f in R["filter_log"]: | |
| L.append(f" {f['text']}: {f['kept']} of {f['total']} sentences kept") | |
| if R["use_filter"] and f["text"] != "Neutral baseline": | |
| L.append(" " + " · ".join(f"{k}: {v}" for k, v in f["by_category"].items())) | |
| L += ["", "─" * 66, " 3. ORIENTATION AND SPREAD", "─" * 66, | |
| f" Leaning bands: {A} < {R['low']:.0f}% ≤ neutral ≤ {R['high']:.0f}% < {B}", ""] | |
| for it in R["items"]: | |
| sp = it["spread"] | |
| L += [f" {it['name']} ({sp['n']} sentences)", | |
| f" {bar(sp['mean'])} mean {z(sp['mean'])}% · median {z(sp['median'])}%", | |
| f" spread: SD {sp['sd']:.0f} · IQR {sp['iqr']:.0f} · middle 80% between {z(sp['p10'])}% and {z(sp['p90'])}%", | |
| f" leaning {A}: {sp['share_a']:.0%} · neutral: {sp['share_neutral']:.0%} · leaning {B}: {sp['share_b']:.0%}", | |
| " tail: " + (f"longer towards {B}" if sp['skew'] > 0.2 else f"longer towards {A}" if sp['skew'] < -0.2 else "symmetric") | |
| + f" (skew {sp['skew']:+.2f})", ""] | |
| if R["baseline"] is not None: | |
| b = R["baseline"]["spread"]["mean"] | |
| L.append(f" Relative to the neutral baseline ({b:.0f}%):") | |
| for it in R["items"]: | |
| if it["name"] != "Neutral baseline": | |
| dlt = it["spread"]["mean"] - b | |
| L.append(f" {it['name']}: {dlt:+.0f} points ({'towards ' + B if dlt > 0 else 'towards ' + A})") | |
| L.append("") | |
| L += ["─" * 66, " 4. PRINCIPAL-AXIS LEANING", "─" * 66, | |
| " Does each text vary mainly along this axis, or along other dimensions (topics, registers)?", ""] | |
| for it in R["items"]: | |
| pa = it["pa"] | |
| if pa is None: | |
| continue | |
| c1 = pa["components"][0] | |
| L += [f" {it['name']}", | |
| f" main dimension of variation (PC1, {c1['variance_share']:.0%} of variance): " | |
| f"{c1['angle']:.0f}° from the axis, far end leaning {A if c1['leaning'] == 'A' else B}", | |
| f" variance along the axis: {pa['axis_share']:.1%} of total " | |
| f"({pa['axis_share'] / pa['baseline']:.0f}× a random direction)"] | |
| for c in pa["components"][1:]: | |
| L.append(f" PC{c['pc']} ({c['variance_share']:.0%}): {c['angle']:.0f}° from the axis") | |
| L.append("") | |
| if R["pairs"]: | |
| L += ["─" * 66, " 5. DIFFERENCES BETWEEN TEXTS", "─" * 66] | |
| for a, b, c in R["pairs"]: | |
| L.append(f" {b} vs {a}: {c['diff']:+.1f} points ({'towards ' + B if c['diff'] > 0 else 'towards ' + A}) · " | |
| f"Cohen's d = {c['cohen_d']:+.2f} ({size_label(c['cohen_d'])}) · permutation p = {c['p']:.4f}") | |
| L.append("") | |
| L += ["═" * 66, | |
| " Scores computed in the full 768-dimensional space of all-mpnet-base-v2", | |
| " (normalised embeddings). The 3D map is a reduced view for orientation only.", | |
| "═" * 66] | |
| return "\n".join(L) | |
| def format_extremes(R): | |
| A, B = R["name_a"], R["name_b"] | |
| out = [] | |
| for it in R["items"]: | |
| if it["name"] == "Neutral baseline": | |
| continue | |
| out.append(f"### {it['name']}\n\n**Closest to {A}**\n") | |
| out += [f"{k}. ({sc:.0f}%) {s}" for k, (s, sc) in enumerate(it["most_a"], 1)] | |
| out.append(f"\n**Closest to {B}**\n") | |
| out += [f"{k}. ({sc:.0f}%) {s}" for k, (s, sc) in enumerate(it["most_b"], 1)] | |
| out.append("") | |
| return "\n".join(out) | |
| def results_table(R): | |
| rows = [] | |
| for it in R["items"]: | |
| sp = it["spread"] | |
| pa = it["pa"] or {"components": [{"angle": np.nan}], "axis_share": np.nan} | |
| rows.append({"Text": it["name"], "Sentences": sp["n"], "Mean %": round(sp["mean"], 1), | |
| "Median %": round(sp["median"], 1), "SD": round(sp["sd"], 1), | |
| "IQR": round(sp["iqr"], 1), | |
| f"Leaning {R['name_a']}": f"{sp['share_a']:.0%}", | |
| "Neutral": f"{sp['share_neutral']:.0%}", | |
| f"Leaning {R['name_b']}": f"{sp['share_b']:.0%}", | |
| "Skew": round(sp["skew"], 2), | |
| "PC1 angle °": round(pa["components"][0]["angle"], 0), | |
| "Axis variance %": round(100 * pa["axis_share"], 1)}) | |
| return pd.DataFrame(rows) | |
| def sentences_csv(R): | |
| rows = [] | |
| for side, sents, E in [(R["name_a"], R["PA"], R["EA"]), (R["name_b"], R["PB"], R["EB"])]: | |
| s = axis_scores(E, R["pole"]["cA"], R["pole"]["cB"]) | |
| rows += [{"text": f"POLE: {side}", "sentence": x, "score_pct": round(float(v), 2), "categories": ""} | |
| for x, v in zip(sents, s)] | |
| for it in R["items"]: | |
| rows += [{"text": it["name"], "sentence": x, "score_pct": round(float(v), 2), "categories": "; ".join(c)} | |
| for x, v, c in zip(it["sentences"], it["scores"], it["categories"])] | |
| f = tempfile.NamedTemporaryFile(delete=False, suffix="_sentence_scores.csv", mode="w", encoding="utf-8") | |
| pd.DataFrame(rows).to_csv(f, index=False) | |
| f.close() | |
| return f.name | |
| # ---------------------------------------------------------------------------------------- | |
| # Plots | |
| # ---------------------------------------------------------------------------------------- | |
| PALETTE = ["#2a7f62", "#c0563f", "#3b6fb6", "#b08a1e", "#7a7a7a"] | |
| def wrap(s, n=70): | |
| words, lines, cur = s.split(), [], "" | |
| for w in words: | |
| if len(cur) + len(w) + 1 > n: | |
| lines.append(cur) | |
| cur = w | |
| else: | |
| cur = (cur + " " + w).strip() | |
| lines.append(cur) | |
| return "<br>".join(lines) | |
| def distribution_plot(R): | |
| import plotly.graph_objects as go | |
| fig = go.Figure() | |
| for k, it in enumerate(R["items"]): | |
| fig.add_trace(go.Violin(x=it["scores"], name=f"{it['name']} ({len(it['scores'])})", | |
| orientation="h", side="positive", points="all", pointpos=-0.4, | |
| jitter=0.3, marker=dict(size=4, opacity=0.55), | |
| line_color=PALETTE[k % len(PALETTE)], meanline_visible=True, | |
| text=[wrap(s) for s in it["sentences"]], | |
| hovertemplate="%{x:.0f}%<br>%{text}<extra></extra>")) | |
| for x, label in [(0, R["name_a"]), (50, "midpoint"), (100, R["name_b"])]: | |
| fig.add_vline(x=x, line_dash="dot" if x == 50 else "solid", line_color="#999", | |
| annotation_text=label, annotation_position="top") | |
| for x in (R["low"], R["high"]): | |
| fig.add_vline(x=x, line_dash="dash", line_color="#ccc") | |
| fig.update_layout(title="Distribution of sentences along the axis", | |
| xaxis_title=f"← {R['name_a']} · position (%) · {R['name_b']} →", | |
| height=180 + 120 * len(R["items"]), showlegend=False, | |
| margin=dict(l=10, r=10, t=60, b=40)) | |
| return fig | |
| def map_3d(R): | |
| """x = position on the axis (true scale); y, z = main dimensions of variation orthogonal to it.""" | |
| import plotly.graph_objects as go | |
| cA, cB = R["pole"]["cA"], R["pole"]["cB"] | |
| d_hat = (cB - cA) / np.linalg.norm(cB - cA) | |
| blocks = [(f"Pole: {R['name_a']}", R["PA"], R["EA"]), (f"Pole: {R['name_b']}", R["PB"], R["EB"])] | |
| blocks += [(it["name"], it["sentences"], it["E"]) for it in R["items"]] | |
| allE = np.vstack([b[2] for b in blocks]) | |
| resid = allE - np.outer(allE @ d_hat, d_hat) | |
| resid -= resid.mean(0) | |
| _, _, Vt = np.linalg.svd(resid, full_matrices=False) | |
| fig = go.Figure() | |
| start = 0 | |
| for k, (name, sents, E) in enumerate(blocks): | |
| n = len(E) | |
| sc = axis_scores(E, cA, cB) | |
| yz = (resid[start:start + n]) @ Vt[:2].T | |
| start += n | |
| is_pole = name.startswith("Pole:") | |
| colour = ("#2a7f62" if k == 0 else "#c0563f") if is_pole else PALETTE[(k - 2) % len(PALETTE)] | |
| fig.add_trace(go.Scatter3d(x=sc, y=yz[:, 0], z=yz[:, 1], mode="markers", name=name, | |
| marker=dict(size=6 if is_pole else 3.5, color=colour, | |
| symbol="diamond" if is_pole else "circle", | |
| opacity=0.9 if is_pole else 0.6), | |
| text=[wrap(s) for s in sents], | |
| hovertemplate="%{x:.0f}%<br>%{text}<extra>" + name + "</extra>")) | |
| fig.update_layout(title="Semantic map: x = position on the axis; y, z = other main dimensions", | |
| scene=dict(xaxis_title=f"{R['name_a']} ← axis % → {R['name_b']}", | |
| yaxis_title="other dim. 1", zaxis_title="other dim. 2"), | |
| height=640, margin=dict(l=0, r=0, t=50, b=0)) | |
| return fig | |
| # ---------------------------------------------------------------------------------------- | |
| # UI | |
| # ---------------------------------------------------------------------------------------- | |
| INTRO = """ | |
| # SpatialDiscourseAnalysis v2 | |
| Place the sentences of up to four texts on an evaluative axis defined by two sets of example sentences. | |
| **Poles**: one sentence per line; avoid negations and keep both poles on the same topics and register, | |
| so the axis captures stance rather than topic. **Texts**: paste running text or one sentence per line. | |
| The optional **relevance filter** keeps only sentences matching an editable lexicon and reports what it kept. | |
| Part of the [H4rmony Project](https://theh4rmonyproject.org). | |
| """ | |
| def run(name_a, poles_a, name_b, poles_b, | |
| n1, t1, n2, t2, n3, t3, n4, t4, baseline, use_filter, lexicon, low, high): | |
| if low >= high: | |
| raise gr.Error("The lower band edge must be below the upper band edge.") | |
| try: | |
| R = analyse(name_a, poles_a, name_b, poles_b, | |
| [(n1, t1), (n2, t2), (n3, t3), (n4, t4)], baseline, | |
| use_filter, lexicon, float(low), float(high)) | |
| except ValueError as e: | |
| raise gr.Error(str(e)) | |
| return ("```\n" + format_report(R) + "\n```", distribution_plot(R), map_3d(R), | |
| results_table(R), format_extremes(R), sentences_csv(R)) | |
| def build_ui(): | |
| with gr.Blocks(title="SpatialDiscourseAnalysis v2") as demo: | |
| gr.Markdown(INTRO) | |
| with gr.Tab("1 · Poles"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| name_a = gr.Textbox(label="Pole A name", value="Ecologically aligned") | |
| poles_a = gr.Textbox(label="Pole A sentences (one per line)", lines=14) | |
| with gr.Column(): | |
| name_b = gr.Textbox(label="Pole B name", value="Ecologically misaligned") | |
| poles_b = gr.Textbox(label="Pole B sentences (one per line)", lines=14) | |
| with gr.Tab("2 · Texts"): | |
| texts = [] | |
| for i in range(4): | |
| with gr.Row(): | |
| n = gr.Textbox(label=f"Text {i + 1} name", value=f"Text {i + 1}", scale=1) | |
| t = gr.Textbox(label=f"Text {i + 1}" + (" (optional)" if i >= 1 else ""), lines=6, scale=4) | |
| texts += [n, t] | |
| baseline = gr.Textbox(label="Neutral baseline (optional): factual, stance-free text about the same subject", | |
| lines=4) | |
| with gr.Tab("3 · Settings"): | |
| use_filter = gr.Checkbox(label="Apply relevance filter", value=False) | |
| lexicon = gr.Textbox(label="Lexicon: 'Category: term, term*, multi word' · optional '| with: term, term' " | |
| "requires a second term in the same sentence", | |
| value=DEFAULT_LEXICON, lines=10) | |
| with gr.Row(): | |
| low = gr.Slider(5, 50, value=40, step=1, label=f"Leaning-A band edge (%)") | |
| high = gr.Slider(50, 95, value=60, step=1, label=f"Leaning-B band edge (%)") | |
| go_btn = gr.Button("Run analysis", variant="primary") | |
| report = gr.Markdown() | |
| dist = gr.Plot(label="Distribution along the axis") | |
| map3d = gr.Plot(label="3D semantic map") | |
| table = gr.Dataframe(label="Summary table") | |
| extremes = gr.Markdown() | |
| csv = gr.File(label="Per-sentence scores (CSV)") | |
| go_btn.click(run, inputs=[name_a, poles_a, name_b, poles_b] + texts + [baseline, use_filter, lexicon, low, high], | |
| outputs=[report, dist, map3d, table, extremes, csv]) | |
| return demo | |
| if gr is not None: | |
| try: | |
| get_model() # load once at start-up (recommended for ZeroGPU) | |
| except Exception as e: # the app still starts; the error resurfaces on first run | |
| print("Model not preloaded:", e) | |
| demo = build_ui() | |
| if __name__ == "__main__" and gr is not None: | |
| demo.launch() | |