Spaces:
Running
Running
| """Export analysis results in formats that are easy to process downstream. | |
| Three files per download: | |
| - summary.csv one row per text (tidy, wide). Rows from different downloads can be | |
| stacked, e.g. to follow a source over time or compare groups of texts. | |
| - sentences.csv one row per sentence, with its position, relevance and whether it was analysed. | |
| - results.json everything above, nested, plus the instrument (model, poles, axis overlap). | |
| Every row carries the information needed to judge whether two results are comparable: | |
| the encoder model, a fingerprint of the pole exemplars, the topic-filter settings and the | |
| app version. Results are only directly comparable when model and poles fingerprint match. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import re | |
| import tempfile | |
| import uuid | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| def fingerprint(path: Path) -> str: | |
| """Short SHA-256 of a file (e.g. poles.tsv), to check that two results used the same poles.""" | |
| return hashlib.sha256(Path(path).read_bytes()).hexdigest()[:12] | |
| def parse_tags(tags: str) -> dict: | |
| """'source=Guardian; year=2024; draft' -> {'source': 'Guardian', 'year': '2024', 'draft': 'yes'}""" | |
| out = {} | |
| for item in re.split(r"[;\n]", tags or ""): | |
| item = item.strip() | |
| if not item: | |
| continue | |
| if "=" in item: | |
| k, v = item.split("=", 1) | |
| out[k.strip()] = v.strip() | |
| else: | |
| out[item] = "yes" | |
| return out | |
| def _slug(s: str) -> str: | |
| return re.sub(r"[^0-9A-Za-z]+", "_", s).strip("_").lower() | |
| def record(res: dict, sp, lm: pd.DataFrame, meta: dict) -> tuple[dict, list[dict], dict]: | |
| """Build the summary row, the sentence rows and the JSON record for one analysed text. | |
| meta: run_id, label, date, source_file, tags (dict), model, model_label, poles_sha, | |
| app_version, topic_filter, threshold, filter_model, topic_definition, topic_sha. | |
| """ | |
| ids = [ax["id"] for ax in sp.axes] | |
| xyz = res["text_xyz"] | |
| sx = res["sent_xyz"] | |
| c = res["cloud"] | |
| raw = res["table"]["Before orthogonalisation"].to_numpy(dtype=float) | |
| W = lm[ids].to_numpy(dtype=float) | |
| dist = np.linalg.norm(xyz - W, axis=1) | |
| zone = int(np.argmin(np.linalg.norm(xyz[:2] - W[:, :2], axis=1))) # eco × change map zone | |
| nearest = int(np.argmin(dist)) | |
| row = { | |
| "run_id": meta["run_id"], | |
| "label": meta["label"], | |
| "date": meta["date"], | |
| "source_file": meta["source_file"], | |
| } | |
| for k, v in meta["tags"].items(): | |
| row[f"tag_{_slug(k)}"] = v | |
| row.update({ | |
| "n_sentences_total": res["n_total"], | |
| "n_sentences_analysed": len(res["sentences"]), | |
| "filter_fallback": res["fallback"], | |
| "n_negation_analysed": sum(bool(n) for n in res["sent_neg"]), | |
| }) | |
| for k, i in enumerate(ids): | |
| row[f"{i}_pct"] = round(float(xyz[k]), 2) | |
| for k, i in enumerate(ids): | |
| row[f"{i}_sd"] = round(float(sx[:, k].std()), 2) if len(sx) > 1 else None | |
| for k, i in enumerate(ids): | |
| row[f"{i}_raw_pct"] = round(float(raw[k]), 2) | |
| row["spread_frobenius"] = round(c["frobenius"], 2) if "frobenius" in c else None | |
| row["variation_captured"] = round(c["captured"], 4) if "captured" in c else None | |
| if "eigvals" in c: | |
| tot = float(c["eigvals"].sum()) or float("nan") | |
| for j in range(3): | |
| row[f"eig{j + 1}_share"] = round(float(c["eigvals"][j] / tot), 4) | |
| row["main_direction_axis"] = ids[c["main_axis"]] | |
| row["main_direction_angle_deg"] = round(float(c["angles"][c["main_axis"]]), 1) | |
| for k, i in enumerate(ids): | |
| row[f"main_direction_{i}"] = round(float(c["eigvecs"][k, 0]), 4) | |
| else: | |
| for j in range(3): | |
| row[f"eig{j + 1}_share"] = None | |
| row["main_direction_axis"] = None | |
| row["main_direction_angle_deg"] = None | |
| for i in ids: | |
| row[f"main_direction_{i}"] = None | |
| row["nearest_worldview"] = lm["name"].iloc[nearest] | |
| row["nearest_worldview_distance"] = round(float(dist[nearest]), 2) | |
| row["map_zone"] = lm["name"].iloc[zone] | |
| for name, d in zip(lm["name"], dist): | |
| row[f"dist_{_slug(name)}"] = round(float(d), 2) | |
| row.update({ | |
| "model": meta["model"], | |
| "poles_sha": meta["poles_sha"], | |
| "topic_filter": meta["topic_filter"], | |
| "threshold": meta["threshold"] if meta["topic_filter"] else None, | |
| "filter_model": meta["filter_model"], | |
| "topic_definition": meta["topic_definition"], | |
| "topic_sha": meta["topic_sha"], | |
| "app_version": meta["app_version"], | |
| "analysed_utc": meta["analysed_utc"], | |
| }) | |
| # Sentences: analysed and filtered-out, in their original order | |
| sent_rows = [] | |
| for n, (s, used, r, p, neg) in enumerate(zip(res["all_sentences"], res["keep"], res["all_rel"], | |
| res["all_xyz"], res["all_neg"]), start=1): | |
| sr = {"run_id": meta["run_id"], "label": meta["label"], "sentence_no": n, | |
| "analysed": bool(used), "relevance": round(float(r), 4), "negation_cues": neg} | |
| for k, i in enumerate(ids): | |
| sr[f"{i}_pct"] = round(float(p[k]), 2) | |
| sr["sentence"] = s | |
| sent_rows.append(sr) | |
| js = { | |
| "run_id": meta["run_id"], | |
| "metadata": {"label": meta["label"], "date": meta["date"], | |
| "source_file": meta["source_file"], "tags": meta["tags"]}, | |
| "instrument": { | |
| "app_version": meta["app_version"], | |
| "model": meta["model"], | |
| "model_label": meta["model_label"], | |
| "poles_sha": meta["poles_sha"], | |
| "orthogonalisation": "Loewdin (symmetric)", | |
| "axes": sp.axes, | |
| "axis_overlap_cosine": sp.overlap.to_numpy().round(4).tolist(), | |
| "topic_filter": meta["topic_filter"], | |
| "threshold": meta["threshold"] if meta["topic_filter"] else None, | |
| "filter_model": meta["filter_model"], | |
| "topic_definition": meta["topic_definition"], | |
| "topic_sha": meta["topic_sha"], | |
| }, | |
| "analysed_utc": meta["analysed_utc"], | |
| "summary": row, | |
| "sentences": sent_rows, | |
| } | |
| return row, sent_rows, js | |
| def write_files(rows: list[dict], sent_rows: list[dict], records: list[dict], stem: str) -> list[str]: | |
| """Write summary.csv, sentences.csv and results.json into a fresh temp folder.""" | |
| folder = Path(tempfile.mkdtemp(prefix="discoursegeometry_")) | |
| stem = _slug(stem) or "discoursegeometry" | |
| paths = [folder / f"{stem}_summary.csv", folder / f"{stem}_sentences.csv", folder / f"{stem}_results.json"] | |
| pd.DataFrame(rows).to_csv(paths[0], index=False, encoding="utf-8") | |
| pd.DataFrame(sent_rows).to_csv(paths[1], index=False, encoding="utf-8") | |
| payload = records[0] if len(records) == 1 else {"runs": records} | |
| paths[2].write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=_jsonable), encoding="utf-8") | |
| return [str(p) for p in paths] | |
| def _jsonable(o): | |
| if isinstance(o, (np.integer,)): | |
| return int(o) | |
| if isinstance(o, (np.floating,)): | |
| return float(o) | |
| if isinstance(o, (np.bool_,)): | |
| return bool(o) | |
| if isinstance(o, np.ndarray): | |
| return o.tolist() | |
| return str(o) | |
| def new_meta(**kw) -> dict: | |
| kw.setdefault("run_id", uuid.uuid4().hex[:8]) | |
| kw.setdefault("analysed_utc", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")) | |
| return kw | |