Pegah
Claude Opus 4.7 (1M context)
chore: initial commit — SAVANT corpus, KG, agentic pipeline, 9-page UI
c63cf78 | """End-to-end Track A pipeline: KG → figures → enumeration → ranking. | |
| Single-command rebuild of every artifact that depends on the 24 candidate | |
| extraction JSONs. Re-run after David's verification updates, after schema | |
| bumps, or after adding new papers. | |
| What it does (in order): | |
| A3 — load 24 extractions → build NetworkX KG → save .json + .graphml | |
| A4 — render 3 static figures (lineage / coverage / dataset overlay) | |
| + 1 interactive HTML (pyvis) | |
| A5 — enumerate coverage matrices → save enumeration.json | |
| A6 — rank empty 3-D cells → save ranked.json | |
| Out of scope: | |
| A7 (top5_rationale.md) is hand-curated and NOT regenerated by this script. | |
| A2 (relationships in extraction JSONs) requires hand-curation too — see | |
| scripts/populate_relationships.py. | |
| Usage: | |
| PYTHONPATH=src python scripts/run_gap_analysis.py | |
| PYTHONPATH=src python scripts/run_gap_analysis.py --skip-interactive | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import sys | |
| import time | |
| from pathlib import Path | |
| from typing import Any | |
| # Allow `python scripts/run_gap_analysis.py` even when PYTHONPATH=src is not set. | |
| REPO = Path(__file__).resolve().parents[1] | |
| if str(REPO / "src") not in sys.path: | |
| sys.path.insert(0, str(REPO / "src")) | |
| from savant.gap_analysis import ( # noqa: E402 | |
| enumerate_cells, | |
| rank_gaps, | |
| save_enumeration, | |
| save_ranking, | |
| ) | |
| from savant.knowledge_graph import ( # noqa: E402 | |
| build_kg, | |
| load_all_extractions, | |
| save_kg, | |
| ) | |
| from savant.visualization import render_all, render_interactive # noqa: E402 | |
| def _log(stage: str, msg: str, elapsed: float | None = None) -> None: | |
| suffix = f" [{elapsed:5.2f}s]" if elapsed is not None else "" | |
| print(f" {stage} {msg}{suffix}") | |
| def run(out_kg_dir: Path | None = None, | |
| out_figures_dir: Path | None = None, | |
| out_gaps_dir: Path | None = None, | |
| skip_figures: bool = False, | |
| skip_interactive: bool = False) -> dict[str, Any]: | |
| """Rebuild every Track-A artifact. | |
| Parameters | |
| ---------- | |
| out_kg_dir, out_figures_dir, out_gaps_dir | |
| Override default output locations. Useful for smoke tests. | |
| skip_figures | |
| If True, skip the static matplotlib figures (still run interactive). | |
| skip_interactive | |
| If True, skip the pyvis interactive HTML (still run static figures). | |
| Returns | |
| ------- | |
| Summary dict of generated artifacts: {label: Path}. | |
| """ | |
| artifacts: dict[str, Path] = {} | |
| total_start = time.perf_counter() | |
| print("\n=== SAVANT Track A — Knowledge Graph + Gap Analysis ===\n") | |
| # ---- A3: load extractions + build KG ---- | |
| t0 = time.perf_counter() | |
| records = load_all_extractions() | |
| g = build_kg(records) | |
| kg_paths = save_kg(g, out_dir=out_kg_dir) | |
| artifacts["kg_json"] = kg_paths["json"] | |
| artifacts["kg_graphml"] = kg_paths["graphml"] | |
| _log("A3", f"built KG: {g.number_of_nodes()} nodes / {g.number_of_edges()} edges " | |
| f"from {len(records)} records", time.perf_counter() - t0) | |
| # ---- A4: static figures ---- | |
| if not skip_figures: | |
| t0 = time.perf_counter() | |
| fig_paths = render_all(g, out_dir=out_figures_dir) | |
| artifacts["fig_lineage"] = fig_paths["lineage"] | |
| artifacts["fig_coverage"] = fig_paths["coverage"] | |
| artifacts["fig_dataset"] = fig_paths["dataset"] | |
| _log("A4", "rendered 3 static figures", time.perf_counter() - t0) | |
| else: | |
| _log("A4", "skipped (static figures)") | |
| # ---- A4: interactive HTML ---- | |
| if not skip_interactive: | |
| t0 = time.perf_counter() | |
| # Interactive HTML is anchored to the default location (lives next to | |
| # the kg artifacts). Out-of-tree paths aren't supported by pyvis cleanly. | |
| html_path = render_interactive(g) | |
| artifacts["interactive_html"] = html_path | |
| _log("A4", "rendered interactive pyvis HTML", time.perf_counter() - t0) | |
| else: | |
| _log("A4", "skipped (interactive HTML)") | |
| # ---- A5: enumerate coverage matrices ---- | |
| t0 = time.perf_counter() | |
| enumeration = enumerate_cells(g) | |
| enum_path = save_enumeration( | |
| enumeration, | |
| out_path=(out_gaps_dir / "enumeration.json") if out_gaps_dir else None, | |
| ) | |
| artifacts["enumeration"] = enum_path | |
| cells_3d = enumeration["matrices"]["framework_x_valuation_unit_x_action_type"] | |
| nonempty = sum(1 for c in cells_3d if c["count"] > 0) | |
| _log("A5", f"enumerated coverage: {nonempty}/{len(cells_3d)} 3-D cells occupied", | |
| time.perf_counter() - t0) | |
| # ---- A6: rank empty 3-D cells ---- | |
| t0 = time.perf_counter() | |
| ranking = rank_gaps(enumeration) | |
| rank_path = save_ranking( | |
| ranking, | |
| out_path=(out_gaps_dir / "ranked.json") if out_gaps_dir else None, | |
| ) | |
| artifacts["ranking"] = rank_path | |
| _log("A6", f"ranked {ranking['n_candidate_cells']} empty cells " | |
| f"({ranking['n_with_positive_methodological_score']} with non-zero " | |
| f"methodological score)", time.perf_counter() - t0) | |
| elapsed = time.perf_counter() - total_start | |
| print(f"\n=== Done in {elapsed:.2f}s ===\n") | |
| print("Artifacts written:") | |
| for label, path in artifacts.items(): | |
| try: | |
| rel = path.relative_to(REPO) if path.is_absolute() else path | |
| except ValueError: | |
| rel = path | |
| size_kb = path.stat().st_size // 1024 | |
| print(f" {label:20s} {rel} ({size_kb} KB)") | |
| print() | |
| return artifacts | |
| def _parse_args() -> argparse.Namespace: | |
| p = argparse.ArgumentParser( | |
| description=__doc__, | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| ) | |
| p.add_argument("--skip-figures", action="store_true", | |
| help="skip the 3 static matplotlib figures (still run interactive)") | |
| p.add_argument("--skip-interactive", action="store_true", | |
| help="skip the interactive pyvis HTML (still run static figures)") | |
| return p.parse_args() | |
| if __name__ == "__main__": | |
| args = _parse_args() | |
| run(skip_figures=args.skip_figures, skip_interactive=args.skip_interactive) | |