File size: 6,118 Bytes
c63cf78 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """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)
|