sgeboers/motief
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: implement exclusive SVD motion assignment with label review report

Browse Source 
- Each motion now assigned to exactly one component (highest absolute score)
- Added --exclusive flag (default: True) for backward compatibility
- Added markdown report generation with motion details for label review
- Added --report-top-n for report size (default: 20 per component)
- Updated JSON output with 'exclusive' flag for transparency
This commit is contained in:
sgeboers
2026-04-04 17:55:28 +02:00
parent ee8ffea6e2
commit 33edb334c4
3 changed files with 1127 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_svd_json.py
+345 -32
View File
@@ -4,9 +4,14 @@ For each SVD component, finds the top N motions by absolute score (split
equally between positive and negative pole), joins with the motions table,
and writes the result to the output JSON file.
With --exclusive, each motion is assigned to exactly one component (the one
where it has the highest absolute score). This ensures cleaner axis labels.
Usage:
uv run python3 scripts/generate_svd_json.py --db data/motions.db --window current_parliament
uv run python3 scripts/generate_svd_json.py --db data/motions.db --window 2025
uv run python3 scripts/generate_svd_json.py --db data/motions.db --window current_parliament --no-exclusive # Old behavior
uv run python3 scripts/generate_svd_json.py --db data/motions.db --window current_parliament --report-top-n 20 # Detailed report
"""
from __future__ import annotations
@@ -16,6 +21,7 @@ import json
import logging
import os
import sys
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -26,9 +32,133 @@ logger = logging.getLogger("generate_svd_json")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
def find_best_component(vec: List[float], max_components: int) -> Tuple[int, float]:
"""Find component with highest absolute score within valid range.
Args:
vec: SVD vector for the motion
max_components: Maximum component index to consider
Returns: (component_index, score)
"""
if not vec:
return 0, 0.0
best_idx = 0
best_abs = abs(vec[0]) if len(vec) > 0 else 0.0
best_score = vec[0] if len(vec) > 0 else 0.0
# Only consider components within range
for i in range(min(len(vec), max_components)):
v = vec[i]
if abs(v) > best_abs:
best_abs = abs(v)
best_idx = i
best_score = v
return best_idx, best_score
def generate_markdown_report(
per_component: List[List[Tuple[int, float]]],
details_map: Dict[int, tuple],
window: str,
exclusive: bool,
report_top_n: int,
theme_labels: Optional[Dict[int, str]] = None,
) -> str:
"""Generate markdown report for label review."""
lines = [
"# SVD Motion Report",
f"",
f"**Window**: {window}",
f"**Generated**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"**Exclusive Assignment**: {'Yes' if exclusive else 'No'}",
f"**Motions per component**: {report_top_n} ({(report_top_n // 2)} per pole)",
f"",
f"---",
f"",
]
for comp_idx, top_motions in enumerate(per_component):
comp_num = comp_idx + 1
theme = (
theme_labels.get(comp_num, "TBD")
if theme_labels
else f"Component {comp_num}"
)
lines.append(f"## Component {comp_num}: {theme}")
lines.append(f"")
# Separate positive and negative
positive = [(mid, score) for mid, score in top_motions if score >= 0]
negative = [(mid, score) for mid, score in top_motions if score < 0]
# Sort: positive by score descending, negative by score ascending (most negative first)
positive.sort(key=lambda x: x[1], reverse=True)
negative.sort(key=lambda x: x[1])
lines.append(f"### Positive Pole ({len(positive)} motions)")
lines.append(f"")
lines.append(f"| Score | Motion ID | Title |")
lines.append(f"|-------|-----------|-------|")
for mid, score in positive:
detail = details_map.get(mid)
title = detail[1] if detail else f"Motion #{mid}"
# Truncate long titles
if title and len(title) > 80:
title = title[:77] + "..."
lines.append(f"| {score:+.3f} | {mid} | {title} |")
lines.append(f"")
lines.append(f"### Negative Pole ({len(negative)} motions)")
lines.append(f"")
lines.append(f"| Score | Motion ID | Title |")
lines.append(f"|-------|-----------|-------|")
for mid, score in negative:
detail = details_map.get(mid)
title = detail[1] if detail else f"Motion #{mid}"
# Truncate long titles
if title and len(title) > 80:
title = title[:77] + "..."
lines.append(f"| {score:+.3f} | {mid} | {title} |")
lines.append(f"")
lines.append(f"### Motion Details")
lines.append(f"")
# Show top 3 from each pole with full details
for pole_name, motions in [
("Positive", positive[:3]),
("Negative", negative[:3]),
]:
lines.append(f"#### {pole_name} Pole (top 3)")
lines.append(f"")
for mid, score in motions:
detail = details_map.get(mid)
if detail:
lines.append(f"**Motion #{mid}** (score: {score:+.3f})")
lines.append(f"- **Title**: {detail[1]}")
lines.append(f"- **Date**: {detail[3]}")
lines.append(f"- **Policy Area**: {detail[4] or 'N/A'}")
body = detail[2]
if body:
# Truncate body text
if len(body) > 500:
body = body[:497] + "..."
lines.append(f"- **Body**: {body}")
lines.append(f"")
lines.append(f"---")
lines.append(f"")
return "\n".join(lines)
def main(argv: Optional[List[str]] = None) -> int:
p = argparse.ArgumentParser(
description="Generate SVD top-motions JSON for a window."
description="Generate SVD top-motions JSON and report for a window."
)
p.add_argument("--db", default="data/motions.db", help="Path to motions.db")
p.add_argument(
@@ -38,7 +168,7 @@ def main(argv: Optional[List[str]] = None) -> int:
"--top-n",
type=int,
default=10,
help="Top N motions per component (split pos/neg)",
help="Top N motions per component for JSON output (split pos/neg)",
)
p.add_argument(
"--components", type=int, default=10, help="Number of SVD components to include"
@@ -48,8 +178,38 @@ def main(argv: Optional[List[str]] = None) -> int:
default="thoughts/explorer/top_svd_top_motions.json",
help="Output JSON file path",
)
p.add_argument(
"--no-exclusive",
action="store_true",
help="Disable exclusive assignment (each motion can appear on multiple components)",
)
p.add_argument(
"--report",
action="store_true",
default=True,
help="Generate markdown report (default: True)",
)
p.add_argument(
"--no-report",
action="store_true",
help="Disable markdown report generation",
)
p.add_argument(
"--report-top-n",
type=int,
default=20,
help="Number of motions per component to show in report (default: 20)",
)
p.add_argument(
"--report-out",
default=None,
help="Output path for markdown report (default: same dir as JSON, .md extension)",
)
args = p.parse_args(argv)
exclusive = not args.no_exclusive
generate_report = args.report and not args.no_report
try:
import duckdb
except ImportError:
@@ -98,23 +258,124 @@ def main(argv: Optional[List[str]] = None) -> int:
n_positive = args.top_n // 2
n_negative = args.top_n - n_positive
report_n_positive = args.report_top_n // 2
report_n_negative = args.report_top_n - report_n_positive
output_rows: List[Dict[str, Any]] = []
all_motion_ids: List[int] = []
# Collect top motions per component
per_component: List[List[Tuple[int, float]]] = []
for comp_idx in range(args.components):
scored: List[Tuple[int, float]] = []
for mid, vec in motion_scores.items():
if comp_idx < len(vec):
scored.append((mid, vec[comp_idx]))
scored.sort(key=lambda x: x[1], reverse=True)
top_positive = scored[:n_positive]
top_negative = scored[-n_negative:]
combined = top_positive + list(reversed(top_negative))
per_component.append(combined)
all_motion_ids.extend(mid for mid, _ in combined)
if exclusive:
# EXCLUSIVE ASSIGNMENT: each motion assigned to exactly one component
logger.info("Using exclusive assignment (each motion to its best component)")
# Step 1: For each motion, find its best component
motion_best: Dict[
int, Tuple[int, float]
] = {} # motion_id -> (component, score)
for mid, vec in motion_scores.items():
best_comp, best_score = find_best_component(vec, args.components)
motion_best[mid] = (best_comp, best_score)
# Step 2: Collect top motions per component
comp_scores: Dict[int, List[Tuple[int, float]]] = {
i: [] for i in range(args.components)
}
for mid, (best_comp, best_score) in motion_best.items():
comp_scores[best_comp].append((mid, best_score))
# Step 3: Sort and take top N per component for JSON output
for comp_idx in range(args.components):
scored = comp_scores[comp_idx]
scored.sort(key=lambda x: x[1], reverse=True)
# Get unique motions for positive and negative poles
# Positive: top N by score
# Negative: bottom N by score (excluding already used)
json_positive = []
json_negative = []
used_ids = set()
# Sort by score descending for positive
for mid, score in scored:
if len(json_positive) < n_positive and mid not in used_ids:
json_positive.append((mid, score))
used_ids.add(mid)
# Sort by score ascending for negative (most negative first)
for mid, score in sorted(scored, key=lambda x: x[1]):
if len(json_negative) < n_negative and mid not in used_ids:
json_negative.append((mid, score))
used_ids.add(mid)
json_combined = json_positive + list(reversed(json_negative))
per_component.append(json_combined)
# Track all IDs for fetching
all_motion_ids.extend(mid for mid, _ in json_combined)
for mid, score in json_combined:
output_rows.append(
{
"component": comp_idx + 1,
"motion_id": mid,
"score": score,
}
)
# Also track IDs for report (may need more motions)
report_all_ids: Dict[int, List[int]] = {i: [] for i in range(args.components)}
for comp_idx in range(args.components):
scored = comp_scores[comp_idx]
scored.sort(key=lambda x: x[1], reverse=True)
# Get more for report
report_ids = [mid for mid, _ in scored[: args.report_top_n]]
report_all_ids[comp_idx] = report_ids
report_motion_ids = []
for comp_idx in range(args.components):
report_motion_ids.extend(report_all_ids[comp_idx])
# Build per_component for report (with more motions)
report_per_component: List[List[Tuple[int, float]]] = []
for comp_idx in range(args.components):
scored = comp_scores[comp_idx]
scored.sort(key=lambda x: x[1], reverse=True)
report_top = scored[: args.report_top_n]
report_per_component.append(report_top)
else:
# NON-EXCLUSIVE: each component selects its own top motions (original behavior)
logger.info(
"Using non-exclusive assignment (motions can appear on multiple components)"
)
for comp_idx in range(args.components):
scored: List[Tuple[int, float]] = []
for mid, vec in motion_scores.items():
if comp_idx < len(vec):
scored.append((mid, vec[comp_idx]))
scored.sort(key=lambda x: x[1], reverse=True)
top_positive = scored[:n_positive]
top_negative = scored[-n_negative:]
combined = top_positive + list(reversed(top_negative))
per_component.append(combined)
all_motion_ids.extend(mid for mid, _ in combined)
# For non-exclusive, each motion in per_component goes to JSON
for comp_idx, top_motions in enumerate(per_component):
for mid, score in top_motions:
output_rows.append(
{
"component": comp_idx + 1,
"motion_id": mid,
"score": score,
}
)
# For report, use same per_component
report_per_component = per_component
report_motion_ids = all_motion_ids
# Batch-fetch motion details
unique_ids = list(set(all_motion_ids))
@@ -134,24 +395,22 @@ def main(argv: Optional[List[str]] = None) -> int:
details_map: Dict[int, tuple] = {row[0]: row for row in detail_rows}
logger.info("Fetched details for %d motions", len(details_map))
# Build output rows
for comp_idx, top_motions in enumerate(per_component):
comp_num = comp_idx + 1
for mid, score in top_motions:
detail = details_map.get(mid)
output_rows.append(
{
"component": comp_num,
"motion_id": mid,
"score": score,
"title": detail[1] if detail else None,
"body_text": detail[2] if detail else None,
"date": str(detail[3])[:10] if detail and detail[3] else None,
"policy_area": detail[4] if detail else None,
}
)
# Enrich output_rows with details
for row in output_rows:
mid = row["motion_id"]
detail = details_map.get(mid)
if detail:
row["title"] = detail[1]
row["body_text"] = detail[2]
row["date"] = str(detail[3])[:10] if detail[3] else None
row["policy_area"] = detail[4]
output: Dict[str, Any] = {"window": args.window, "rows": output_rows}
# Write JSON output
output: Dict[str, Any] = {
"window": args.window,
"exclusive": exclusive,
"rows": output_rows,
}
out_dir = os.path.dirname(args.out)
if out_dir:
@@ -166,6 +425,60 @@ def main(argv: Optional[List[str]] = None) -> int:
args.components,
args.out,
)
# Generate markdown report
if generate_report:
report_path = args.report_out
if report_path is None:
# Default: same directory, .md extension
base = args.out.rsplit(".", 1)[0]
report_path = base + "_report.md"
report_dir = os.path.dirname(report_path)
if report_dir:
os.makedirs(report_dir, exist_ok=True)
# Get theme labels from SVD_THEMES if available
theme_labels = None
try:
# Try to import theme labels
sys.path.insert(0, ROOT)
from explorer import SVD_THEMES
theme_labels = {
k: v.get("label", f"Component {k}") for k, v in SVD_THEMES.items()
}
except Exception:
logger.debug("Could not load theme labels, using defaults")
# For report, fetch details for all report motions
report_unique_ids = list(set(report_motion_ids))
if report_unique_ids and report_unique_ids != unique_ids:
con = duckdb.connect(database=args.db, read_only=True)
placeholders = ", ".join("?" for _ in report_unique_ids)
report_detail_rows = con.execute(
f"SELECT id, title, body_text, date, policy_area FROM motions WHERE id IN ({placeholders})",
report_unique_ids,
).fetchall()
con.close()
# Merge with existing details
for row in report_detail_rows:
details_map[row[0]] = row
markdown = generate_markdown_report(
report_per_component,
details_map,
args.window,
exclusive,
args.report_top_n,
theme_labels,
)
with open(report_path, "w", encoding="utf-8") as f:
f.write(markdown)
logger.info("Written markdown report to %s", report_path)
return 0