mirror of
https://github.com/harvard-edge/cs249r_book.git
synced 2026-11-09 10:39:41 -06:00
979 lines
43 KiB
Python
979 lines
43 KiB
Python
"""Compile chapter-owned learner graphs into maps and curriculum catalogs.
|
|
|
|
Book records live in chapter *_concepts.yml files. The independent field model
|
|
and book-link overlay live in taxonomy. Published JSON, DOT, SVG, chapter
|
|
viewers, and the catalog are deterministic build artifacts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import html
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import textwrap
|
|
import xml.etree.ElementTree as ET
|
|
from urllib.parse import quote as urlquote
|
|
|
|
import yaml
|
|
|
|
|
|
# The ML Systems Taxonomy page: its source folder and its public address.
|
|
SITE_URL = "https://mlsysbook.ai"
|
|
SITE_DIR = "taxonomy"
|
|
PAGE_URL = f"{SITE_URL}/{SITE_DIR}/"
|
|
|
|
|
|
class ConceptGraphError(ValueError):
|
|
"""An authored graph cannot be resolved against its chapter contracts."""
|
|
|
|
|
|
def digest(value: str) -> str:
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def sections(text: str) -> dict[str, str]:
|
|
"""Index anchored headings, excluding executable and literal fenced blocks."""
|
|
result = {}
|
|
heading = re.compile(r"^(#{1,6})\s+.*?\{#([\w-]+)[^}]*\}.*$", re.M)
|
|
# Preserve offsets but mask fences so code comments cannot become headings.
|
|
masked = re.sub(
|
|
r"^(`{3,}|~{3,})[^\n]*\n.*?^\1\s*$",
|
|
lambda m: " " * len(m[0]),
|
|
text,
|
|
flags=re.M | re.S,
|
|
)
|
|
matches = list(heading.finditer(masked))
|
|
for i, match in enumerate(matches):
|
|
level = len(match[1])
|
|
end = next(
|
|
(m.start() for m in matches[i + 1 :] if len(m[1]) <= level), len(text)
|
|
)
|
|
if match[2] in result:
|
|
raise ConceptGraphError(f"Duplicate section anchor: {match[2]}")
|
|
result[match[2]] = text[match.start() : end]
|
|
return result
|
|
|
|
|
|
def _required(record: dict, keys: tuple[str, ...], context: str) -> None:
|
|
if not isinstance(record, dict):
|
|
raise ConceptGraphError(f"{context}: expected a mapping")
|
|
for key in keys:
|
|
if not isinstance(record.get(key), str) or not record[key].strip():
|
|
raise ConceptGraphError(f"{context}: missing nonempty {key}")
|
|
|
|
|
|
def companion_resources(root: Path, chapter: dict, name: str) -> list[dict]:
|
|
"""Resolve explicitly authored activities; existence never implies readiness."""
|
|
records = chapter["contract"].get("explorer_graph", {}).get("resources", {})
|
|
if not isinstance(records, dict) or any(k not in chapter["owned"] for k in records):
|
|
raise ConceptGraphError(f"{chapter['id']}: resources must reference owned concepts")
|
|
items = records.get(name, [])
|
|
if not isinstance(items, list):
|
|
raise ConceptGraphError(f"{chapter['id']}: companion resources must be a list")
|
|
result = []
|
|
for record in items:
|
|
_required(record, ("title", "path", "kind", "status"), chapter["id"])
|
|
path = Path(record["path"])
|
|
resolved = (root / path).resolve()
|
|
if path.is_absolute() or ".." in path.parts or not resolved.is_relative_to(root.resolve()) or not resolved.is_file():
|
|
raise ConceptGraphError(f"{chapter['id']}: missing or unsafe companion source: {path}")
|
|
if record["kind"] not in ("implementation", "simulation", "lab", "slides", "hardware"):
|
|
raise ConceptGraphError(f"{chapter['id']}: unknown companion kind")
|
|
if record["status"] not in ("early-release", "ready", "skeleton"):
|
|
raise ConceptGraphError(f"{chapter['id']}: unknown companion status")
|
|
result.append({**record, "url": "https://github.com/harvard-edge/cs249r_book/blob/dev/" + urlquote(path.as_posix(), safe="/")})
|
|
return result
|
|
|
|
|
|
def load_graph(root: Path, volumes: tuple[str, ...] = ()) -> dict:
|
|
"""Resolve all nodes before relationships, allowing cross-chapter reuse."""
|
|
chapters, nodes, edges, maps = {}, {}, {}, {}
|
|
for number in range(1, 5):
|
|
volume = f"vol{number}"
|
|
if volumes and volume not in volumes:
|
|
continue
|
|
config = root / f"books/config/_quarto-pdf-{volume}.yml"
|
|
order = config.read_text() if config.exists() else ""
|
|
paths = list((root / "books" / volume).glob("*/*_concepts.yml"))
|
|
paths.sort(
|
|
key=lambda p: (
|
|
order.find(f"{volume}/{p.parent.name}/")
|
|
if f"{volume}/{p.parent.name}/" in order
|
|
else len(order),
|
|
p.parent.name,
|
|
)
|
|
)
|
|
for path in paths:
|
|
contract = yaml.safe_load(path.read_text())["concept_map"]
|
|
source = path.parent / contract["source"]
|
|
# Do not let authored paths escape their owning chapter.
|
|
if source.resolve().parent != path.parent.resolve():
|
|
raise ConceptGraphError(f"Source must be in owning chapter: {path}")
|
|
text = source.read_text()
|
|
indexed = sections(text)
|
|
title_match = re.search(r"^#\s+(.+?)(?:\s+\{.*)?$", text, re.M)
|
|
key = f"{volume}/{path.parent.name}"
|
|
owned = contract.get("primary_concepts", []) + contract.get(
|
|
"secondary_concepts", []
|
|
)
|
|
claims = contract.get("claims", {})
|
|
title = title_match[1] if title_match else path.parent.name
|
|
# Clean published form (.github/scripts/flatten-vol-urls.sh): the
|
|
# volume tree is flattened, so chapters sit directly under /volN/.
|
|
url = f"https://mlsysbook.ai/{volume}/{path.parent.name}/{source.stem}.html"
|
|
chapters[key] = dict(
|
|
id=key,
|
|
volume=volume,
|
|
title=title,
|
|
path=path,
|
|
source=source,
|
|
sections=indexed,
|
|
contract=contract,
|
|
owned=owned,
|
|
url=url,
|
|
source_hash=digest(text),
|
|
maps=[],
|
|
)
|
|
graph = contract.get("learner_graph")
|
|
if graph is None:
|
|
continue
|
|
if not isinstance(graph, dict) or graph.get("schema_version") != 1:
|
|
raise ConceptGraphError(f"{path}: unsupported learner_graph schema")
|
|
for node in graph.get("concepts", []):
|
|
_required(node, ("id", "label", "teaches", "summary", "section"), key)
|
|
aliases = node.get("aliases", [])
|
|
if not isinstance(aliases, list) or any(not isinstance(a, str) or not a.strip() for a in aliases):
|
|
raise ConceptGraphError(f"{key}: concept aliases must be nonempty strings")
|
|
ident = node["id"]
|
|
if not re.fullmatch(
|
|
r"vol[1-4]\.[a-z0-9.-]+", ident
|
|
) or not ident.startswith(volume + "."):
|
|
raise ConceptGraphError(
|
|
f"Invalid volume-scoped concept ID: {ident}"
|
|
)
|
|
if ident in nodes:
|
|
raise ConceptGraphError(f"Duplicate concept ID: {ident}")
|
|
if node["teaches"] not in owned or node["teaches"] not in claims:
|
|
raise ConceptGraphError(
|
|
f"{ident}: teaches must name an owned concept with a claim"
|
|
)
|
|
if node["section"] not in indexed:
|
|
raise ConceptGraphError(
|
|
f"{ident}: missing section {node['section']}"
|
|
)
|
|
nodes[ident] = {
|
|
**node,
|
|
"chapter": key,
|
|
"volume": number,
|
|
"resources": companion_resources(root, chapters[key], node["teaches"]),
|
|
"claim": claims[node["teaches"]],
|
|
"concept": node["teaches"],
|
|
"url": url + "#" + node["section"],
|
|
}
|
|
for relation in graph.get("relationships", []):
|
|
_required(
|
|
relation,
|
|
(
|
|
"id",
|
|
"source",
|
|
"target",
|
|
"predicate",
|
|
"explanation",
|
|
"conditions",
|
|
"status",
|
|
),
|
|
key,
|
|
)
|
|
if relation["id"] in edges:
|
|
raise ConceptGraphError(
|
|
f"Duplicate relationship ID: {relation['id']}"
|
|
)
|
|
if relation["status"] not in ("draft", "reviewed"):
|
|
raise ConceptGraphError(f"Unknown review status: {relation['id']}")
|
|
evidence = relation.get("evidence", {})
|
|
_required(evidence, ("section", "quote"), relation["id"])
|
|
section = indexed.get(evidence["section"])
|
|
if section is None:
|
|
raise ConceptGraphError(
|
|
f"{relation['id']}: missing evidence section"
|
|
)
|
|
normalized = lambda s: " ".join(s.split())
|
|
if normalized(evidence["quote"]) not in normalized(section):
|
|
raise ConceptGraphError(
|
|
f"{relation['id']}: evidence quote changed or is absent"
|
|
)
|
|
edges[relation["id"]] = {
|
|
**relation,
|
|
"label": relation["predicate"],
|
|
"chapter": key,
|
|
"evidence_url": url + "#" + evidence["section"],
|
|
"evidence_hash": digest(section),
|
|
}
|
|
for view in graph.get("maps", []):
|
|
_required(view, ("id", "focus_question", "description"), key)
|
|
if view["id"] in maps or not re.fullmatch(
|
|
r"vol[1-4]\.[a-z0-9]+(?:[.-][a-z0-9]+)*", view["id"]
|
|
):
|
|
raise ConceptGraphError(
|
|
f"Duplicate or invalid map ID: {view['id']}"
|
|
)
|
|
maps[view["id"]] = {
|
|
**view,
|
|
"chapter": key,
|
|
"title": view["focus_question"],
|
|
}
|
|
chapters[key]["maps"].append(view["id"])
|
|
for edge in edges.values():
|
|
if edge["source"] not in nodes or edge["target"] not in nodes:
|
|
raise ConceptGraphError(
|
|
f"{edge['id']}: relationship references missing concept"
|
|
)
|
|
for view in maps.values():
|
|
members = view.get("concepts", [])
|
|
relations = view.get("relationships", [])
|
|
if not members or len(members) != len(set(members)):
|
|
raise ConceptGraphError(
|
|
f"{view['id']}: empty or repeated concept membership"
|
|
)
|
|
if not relations or len(relations) != len(set(relations)):
|
|
raise ConceptGraphError(
|
|
f"{view['id']}: empty or repeated relationship membership"
|
|
)
|
|
for ident in members:
|
|
if ident not in nodes:
|
|
raise ConceptGraphError(f"{view['id']}: missing concept {ident}")
|
|
for ident in relations:
|
|
if ident not in edges:
|
|
raise ConceptGraphError(f"{view['id']}: missing relationship {ident}")
|
|
edge = edges[ident]
|
|
if edge["source"] not in members or edge["target"] not in members:
|
|
raise ConceptGraphError(
|
|
f"{view['id']}: relationship endpoint outside map"
|
|
)
|
|
for check in view.get("checks", []):
|
|
if (
|
|
check.get("concept") not in members
|
|
or not check.get("question")
|
|
or not check.get("explanation")
|
|
):
|
|
raise ConceptGraphError(f"{view['id']}: invalid learning check")
|
|
choices = check.get("choices", [])
|
|
if (
|
|
len(choices) < 2
|
|
or not isinstance(check.get("answer"), int)
|
|
or not 0 <= check["answer"] < len(choices)
|
|
):
|
|
raise ConceptGraphError(f"{view['id']}: invalid check answer")
|
|
volume_views = {}
|
|
for view in maps.values():
|
|
scope = view.get("scope", "focused")
|
|
if scope not in ("focused", "volume"):
|
|
raise ConceptGraphError(f"{view['id']}: unknown authored map scope")
|
|
volume = view["chapter"].split("/")[0]
|
|
if any(not nodes[i]["chapter"].startswith(volume + "/") for i in view["concepts"]):
|
|
raise ConceptGraphError(f"{view['id']}: map must stay inside its volume")
|
|
if scope == "volume":
|
|
if volume in volume_views:
|
|
raise ConceptGraphError(f"{volume}: only one primary volume map is allowed")
|
|
if view.get("volume") != volume:
|
|
raise ConceptGraphError(f"{view['id']}: volume scope must name its owning volume")
|
|
volume_views[volume] = view["id"]
|
|
for chapter in chapters.values():
|
|
if chapter["owned"] and chapter["volume"] in volume_views:
|
|
ident = volume_views[chapter["volume"]]
|
|
chapter["maps"] = [ident] + [m for m in chapter["maps"] if m != ident]
|
|
result = dict(chapters=chapters, nodes=nodes, edges=edges, maps=maps)
|
|
add_volume_maps(result, root)
|
|
return result
|
|
|
|
|
|
def add_volume_maps(graph: dict, root: Path) -> None:
|
|
"""Join chapter-authored concept propositions into one navigable map per book.
|
|
|
|
Evidence is the existing, editorially authored teaching claim. It establishes
|
|
provenance and freshness; draft edges still require semantic review.
|
|
"""
|
|
chapters, nodes, edges, maps = (
|
|
graph[k] for k in ("chapters", "nodes", "edges", "maps")
|
|
)
|
|
for volume in sorted({c["volume"] for c in chapters.values()}):
|
|
members, relations = set(), []
|
|
|
|
def resolve(chapter_id: str, name: str) -> str:
|
|
chapter = chapters.get(chapter_id)
|
|
if (
|
|
not chapter
|
|
or not chapter_id.startswith(volume + "/")
|
|
or name not in chapter["owned"]
|
|
):
|
|
raise ConceptGraphError(
|
|
f"Volume map: unresolved concept {chapter_id}: {name}"
|
|
)
|
|
ident = volume + ".goal." + digest(chapter_id + ":" + name)[:16]
|
|
alias = (
|
|
chapter["contract"]
|
|
.get("explorer_graph", {})
|
|
.get("labels", {})
|
|
.get(name, name)
|
|
)
|
|
nodes.setdefault(
|
|
ident,
|
|
dict(
|
|
id=ident,
|
|
label=alias,
|
|
summary=chapter["contract"]["claims"][name],
|
|
claim=chapter["contract"]["claims"][name],
|
|
concept=name,
|
|
chapter=chapter_id,
|
|
volume=int(volume[-1]),
|
|
url=chapter["url"],
|
|
kind="teaching-concept",
|
|
resources=companion_resources(root, chapter, name),
|
|
),
|
|
)
|
|
members.add(ident)
|
|
return ident
|
|
|
|
for chapter in chapters.values():
|
|
if chapter["volume"] != volume:
|
|
continue
|
|
authored = chapter["contract"].get("explorer_graph", {})
|
|
if not isinstance(authored.get("resources", {}), dict):
|
|
raise ConceptGraphError(f"{chapter['id']}: resources must reference owned concepts")
|
|
for goal in authored.get("resources", {}):
|
|
companion_resources(root, chapter, goal)
|
|
if authored and authored.get("schema_version") != 1:
|
|
raise ConceptGraphError(
|
|
f"{chapter['id']}: unsupported explorer_graph schema"
|
|
)
|
|
for relation in authored.get("relationships", []):
|
|
_required(
|
|
relation,
|
|
("source", "predicate", "explanation", "conditions", "status"),
|
|
chapter["id"],
|
|
)
|
|
if relation["status"] not in ("draft", "reviewed"):
|
|
raise ConceptGraphError(
|
|
"Volume relationship status must be draft or reviewed"
|
|
)
|
|
target = relation.get("target", {})
|
|
_required(target, ("chapter", "concept"), chapter["id"])
|
|
source_id = resolve(chapter["id"], relation["source"])
|
|
target_id = resolve(volume + "/" + target["chapter"], target["concept"])
|
|
evidence = relation.get("evidence", [])
|
|
if not evidence:
|
|
raise ConceptGraphError("Volume relationship needs claim evidence")
|
|
for item in evidence:
|
|
_required(item, ("chapter", "concept", "quote"), chapter["id"])
|
|
owner = chapters.get(volume + "/" + item["chapter"])
|
|
claim = (
|
|
owner["contract"].get("claims", {}).get(item["concept"], "")
|
|
if owner
|
|
else ""
|
|
)
|
|
if not claim or " ".join(item["quote"].split()) not in " ".join(
|
|
claim.split()
|
|
):
|
|
raise ConceptGraphError(
|
|
f"{chapter['id']}: volume relationship evidence claim changed"
|
|
)
|
|
ident = (
|
|
volume
|
|
+ ".connection."
|
|
+ digest(source_id + relation["predicate"] + target_id)[:16]
|
|
)
|
|
if ident in edges:
|
|
raise ConceptGraphError(f"Duplicate volume relationship: {ident}")
|
|
edges[ident] = {
|
|
**relation,
|
|
"id": ident,
|
|
"source": source_id,
|
|
"target": target_id,
|
|
"label": relation["predicate"],
|
|
"chapter": chapter["id"],
|
|
"evidence_url": chapter["url"],
|
|
"evidence_kind": "chapter-claim",
|
|
}
|
|
relations.append(ident)
|
|
if not relations:
|
|
continue
|
|
ident = volume + ".overview"
|
|
title = {
|
|
"vol1": "How do data, algorithms, and machines constrain one another?",
|
|
"vol2": "How does a fleet turn capacity into useful work?",
|
|
"vol3": "How does an agent turn proposals into accepted outcomes?",
|
|
"vol4": "How does a physical system turn proposals into safe actions?",
|
|
}[volume]
|
|
# Membership follows canonical chapter and concept order, not edge traversal.
|
|
ordered = [
|
|
i
|
|
for c in chapters.values()
|
|
for name in c["owned"]
|
|
if (i := volume + ".goal." + digest(c["id"] + ":" + name)[:16]) in members
|
|
]
|
|
owner = next(c for c in chapters.values() if c["volume"] == volume)
|
|
maps[ident] = dict(
|
|
id=ident,
|
|
title=title,
|
|
focus_question=title,
|
|
scope="argument" if any(m.get("scope") == "volume" and m["chapter"].startswith(volume + "/") for m in maps.values()) else "volume",
|
|
volume=volume,
|
|
chapter=owner["id"],
|
|
description="Follow concepts across chapter boundaries. Chapter selection highlights a region of this same map.",
|
|
concepts=ordered,
|
|
relationships=relations,
|
|
)
|
|
for chapter in chapters.values():
|
|
if chapter["volume"] == volume and chapter["owned"]:
|
|
if maps[ident]["scope"] == "volume":
|
|
chapter["maps"].insert(0, ident)
|
|
|
|
|
|
def map_data(graph: dict, ident: str) -> dict:
|
|
view = graph["maps"][ident]
|
|
nodes = []
|
|
for node_id in view["concepts"]:
|
|
node = graph["nodes"][node_id]
|
|
# Explicit teaches ownership joins overview arguments to finer questions.
|
|
# This is a refinement/navigation link, never inferred equivalence.
|
|
refinements = [
|
|
{"id": candidate["id"], "title": candidate["title"]}
|
|
for candidate in graph["maps"].values()
|
|
if candidate.get("scope", "focused") == "focused" and candidate["id"] != ident
|
|
and any(
|
|
graph["nodes"][i]["chapter"] == node["chapter"]
|
|
and graph["nodes"][i]["concept"] == node["concept"]
|
|
for i in candidate["concepts"]
|
|
)
|
|
]
|
|
nodes.append({
|
|
**node, "detail_maps": refinements,
|
|
"teaching_goal": node["chapter"] + ":" + digest(node["concept"])[:16],
|
|
})
|
|
return {
|
|
**view,
|
|
"nodes": nodes,
|
|
"coverage": map_coverage(graph, view),
|
|
"edges": [graph["edges"][i] for i in view["relationships"]],
|
|
}
|
|
|
|
|
|
def map_coverage(graph: dict, view: dict) -> dict:
|
|
"""Report editorial-goal representation, never claim field completeness."""
|
|
volume = view["chapter"].split("/")[0]
|
|
represented = {(graph["nodes"][i]["chapter"], graph["nodes"][i]["concept"]) for i in view["concepts"]}
|
|
chapters = []
|
|
for chapter in graph["chapters"].values():
|
|
if chapter["volume"] != volume or not chapter["owned"]:
|
|
continue
|
|
covered = [name for name in chapter["owned"] if (chapter["id"], name) in represented]
|
|
chapters.append(dict(
|
|
id=chapter["id"], title=chapter["title"], represented=covered,
|
|
outside_view=[name for name in chapter["owned"] if name not in covered],
|
|
primary_outside_view=[name for name in chapter["contract"].get("primary_concepts", []) if name not in covered],
|
|
))
|
|
return dict(
|
|
teaching_goals=sum(len(c["owned"]) for c in graph["chapters"].values() if c["volume"] == volume),
|
|
represented_goals=len(represented),
|
|
teaching_chapters=len(chapters),
|
|
represented_chapters=sum(bool(c["represented"]) for c in chapters),
|
|
chapters=chapters,
|
|
)
|
|
|
|
|
|
def svg_map(data: dict, *, pack_components: bool = False, emphasize_first: bool = True, pad_edge_labels: bool = False, horizontal: bool = False) -> tuple[str, bytes]:
|
|
"""Place labels and arrows together, retaining stable IDs for interaction."""
|
|
quote = lambda value: json.dumps(value, ensure_ascii=False)
|
|
lines = [
|
|
"digraph concepts {",
|
|
'graph [rankdir=TB, bgcolor="white", pad="0.35", nodesep="0.45", ranksep="0.85", splines=polyline, outputorder=edgesfirst];',
|
|
'node [shape=box, style="rounded,filled", fillcolor="#f7f7f7", color="#88949d", penwidth=1.15, fontname="Helvetica Neue", fontsize=16, margin="0.18,0.12"];',
|
|
'edge [fontname="Helvetica Neue", fontsize=12, color="#7a858e", fontcolor="#47535d", arrowsize=0.65];',
|
|
]
|
|
if horizontal:
|
|
lines.append("graph [rankdir=LR];")
|
|
if pack_components:
|
|
lines.append('graph [pack=true, packmode="array_u3"];')
|
|
for i, node in enumerate(data["nodes"]):
|
|
label = quote("\n".join(textwrap.wrap(node["label"], 23)))
|
|
extra = ', fillcolor="#fcebed", color="#a51c30", penwidth=1.8' if i == 0 and emphasize_first else ""
|
|
lines.append(
|
|
f"{quote(node['id'])} [id={quote('concept-' + node['id'])}, label={label}, tooltip={quote(node['summary'])}{extra}];"
|
|
)
|
|
for edge in data["edges"]:
|
|
if pad_edge_labels:
|
|
words = "<BR/>".join(html.escape(line) for line in textwrap.wrap(edge["label"], 28))
|
|
label = '<<TABLE BORDER="0" CELLBORDER="0" CELLSPACING="0" CELLPADDING="5"><TR><TD>' + words + '</TD></TR></TABLE>>'
|
|
else:
|
|
label = quote("\n".join(textwrap.wrap(edge["label"], 23)))
|
|
lines.append(
|
|
f"{quote(edge['source'])} -> {quote(edge['target'])} [id={quote('relation-' + edge['id'])}, label={label}];"
|
|
)
|
|
dot = "\n".join(lines + ["}"]) + "\n"
|
|
if not shutil.which("dot"):
|
|
raise ConceptGraphError("Graphviz 'dot' is required to build maps")
|
|
try:
|
|
raw = subprocess.check_output(
|
|
["dot", "-Tsvg"], input=dot.encode(), stderr=subprocess.PIPE
|
|
)
|
|
except subprocess.CalledProcessError as exc:
|
|
raise ConceptGraphError(exc.stderr.decode()) from exc
|
|
root = ET.fromstring(raw)
|
|
root.set("font-family", "Helvetica Neue, Helvetica, Arial, sans-serif")
|
|
for text in root.findall(
|
|
".//{http://www.w3.org/2000/svg}g[@class='edge']/{http://www.w3.org/2000/svg}text"
|
|
):
|
|
text.set("paint-order", "stroke")
|
|
text.set("stroke", "white")
|
|
text.set("stroke-width", "4")
|
|
text.set("stroke-linejoin", "round")
|
|
root.set("role", "group")
|
|
root.set("aria-label", data["title"])
|
|
ET.register_namespace("", "http://www.w3.org/2000/svg")
|
|
ET.register_namespace("xlink", "http://www.w3.org/1999/xlink")
|
|
return dot, ET.tostring(root, encoding="utf-8", xml_declaration=True)
|
|
|
|
|
|
def curriculum_catalog(graph: dict, navigation: dict | None = None) -> dict:
|
|
"""Expose ownership and declared reuse without inventing semantic edges.
|
|
|
|
Identities remain volume-scoped: similarly named concepts in independent
|
|
textbooks are not automatically merged. Navigation lenses group chapters.
|
|
"""
|
|
navigation = navigation or {}
|
|
chapters = {}
|
|
goals = {}
|
|
lookup = {}
|
|
for chapter in graph["chapters"].values():
|
|
contract = chapter["contract"]
|
|
concepts = []
|
|
for kind, field in (
|
|
("core", "primary_concepts"),
|
|
("supporting", "secondary_concepts"),
|
|
):
|
|
for name in contract.get(field, []):
|
|
ident = chapter["id"] + ":" + digest(name)[:16]
|
|
goal = dict(
|
|
id=ident,
|
|
label=name,
|
|
order=kind,
|
|
claim=contract.get("claims", {}).get(name, ""),
|
|
chapter=chapter["id"],
|
|
uses=[],
|
|
)
|
|
concepts.append(goal)
|
|
goals[ident] = goal
|
|
lookup[(chapter["id"], name)] = goal
|
|
chapters[chapter["id"]] = {
|
|
**{k: chapter[k] for k in ("id", "title", "url", "maps")},
|
|
"concepts": concepts,
|
|
"references": [],
|
|
}
|
|
for chapter in graph["chapters"].values():
|
|
for field, relation in (("touches", "revisits"), ("previews", "previews")):
|
|
for pointer in chapter["contract"].get(field, []):
|
|
owner = chapter["volume"] + "/" + pointer["owner"]
|
|
goal = lookup.get((owner, pointer["concept"]))
|
|
if goal is None:
|
|
raise ConceptGraphError(
|
|
f"{chapter['id']}: unresolved {field} pointer: {pointer['concept']}"
|
|
)
|
|
goal["uses"].append(dict(chapter=chapter["id"], relation=relation))
|
|
chapters[chapter["id"]]["references"].append(
|
|
dict(concept=goal["id"], relation=relation)
|
|
)
|
|
themes = navigation.get("themes", [])
|
|
seen = set()
|
|
for theme in themes:
|
|
if (
|
|
not re.fullmatch(r"[a-z][a-z0-9-]*", theme.get("id", ""))
|
|
or theme["id"] in seen
|
|
):
|
|
raise ConceptGraphError("Invalid or duplicate atlas theme ID")
|
|
seen.add(theme["id"])
|
|
_required(theme, ("label", "question"), theme["id"])
|
|
for ident in theme.get("chapters", []):
|
|
if ident not in chapters:
|
|
raise ConceptGraphError(
|
|
f"Atlas theme {theme['id']}: unknown chapter {ident}"
|
|
)
|
|
volumes = []
|
|
for volume in sorted({c["volume"] for c in graph["chapters"].values()}):
|
|
metadata = navigation.get("volumes", {}).get(volume, {})
|
|
volumes.append(
|
|
dict(
|
|
id=volume,
|
|
**metadata,
|
|
chapters=[
|
|
ch for key, ch in chapters.items() if key.startswith(volume + "/")
|
|
],
|
|
)
|
|
)
|
|
return dict(
|
|
schema_version=2,
|
|
scope="series",
|
|
volumes=volumes,
|
|
themes=themes,
|
|
counts=dict(
|
|
concepts=len(goals),
|
|
chapters=len(chapters),
|
|
references=sum(len(g["uses"]) for g in goals.values()),
|
|
),
|
|
maps=[
|
|
{k: view[k] for k in ("id", "title", "description", "chapter")}
|
|
| {"scope": view.get("scope", "focused")}
|
|
for view in sorted(
|
|
(m for m in graph["maps"].values() if m.get("scope") != "argument"), key=lambda m: m.get("scope") != "volume"
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _json(value: dict) -> bytes:
|
|
return (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode()
|
|
|
|
|
|
def build(root: Path, volumes: tuple[str, ...] = (), check: bool = False) -> dict:
|
|
"""Compile first, then publish artifacts; invalid authoring never overwrites outputs."""
|
|
# Shared catalogs/maps always cover the series; volume flags scope chapter viewers.
|
|
graph = load_graph(root)
|
|
site = root / SITE_DIR
|
|
navigation_path = site / "atlas.yml"
|
|
navigation = (
|
|
yaml.safe_load(navigation_path.read_text()) if navigation_path.exists() else {}
|
|
)
|
|
catalog = curriculum_catalog(graph, navigation)
|
|
artifacts = {site / "catalog.json": _json(catalog)}
|
|
# The same portable viewer is served by the explorer and embedded chapters.
|
|
template_path = site / "volume-maps.qmd"
|
|
if not template_path.exists():
|
|
template_path = site / "index.qmd"
|
|
template_source = template_path.read_text()
|
|
if "```{=html}\n" not in template_source:
|
|
raise ConceptGraphError("Chapter viewers require a raw HTML volume-map template")
|
|
template = template_source.split("```{=html}\n", 1)[1].split("\n```", 1)[0]
|
|
if 'id="study-root"' in template:
|
|
raise ConceptGraphError("Chapter viewers require the volume-map template, not the field explorer")
|
|
for ident in graph["maps"]:
|
|
data = map_data(graph, ident)
|
|
dot, svg = svg_map(data)
|
|
destination = site / "maps" / ident
|
|
artifacts[destination / "graph.json"] = _json(data)
|
|
artifacts[destination / "map.dot"] = dot.encode()
|
|
artifacts[destination / "map.svg"] = svg
|
|
for chapter in graph["chapters"].values():
|
|
if not chapter["maps"] and not chapter["owned"]:
|
|
continue
|
|
if volumes and chapter["volume"] not in volumes:
|
|
continue
|
|
destination = chapter["source"].parent / "concepts-generated"
|
|
# Chapter viewers start locally but retain the full ownership atlas.
|
|
available_maps = [
|
|
m["id"]
|
|
for m in catalog["maps"]
|
|
if m["id"] in graph["maps"]
|
|
]
|
|
local_catalog = {
|
|
**catalog,
|
|
"scope": "chapter",
|
|
"chapter": chapter["id"],
|
|
"maps": [m for m in catalog["maps"] if m["id"] in available_maps],
|
|
}
|
|
artifacts[destination / "catalog.json"] = _json(local_catalog)
|
|
for ident in available_maps:
|
|
source = site / "maps" / ident
|
|
for name in ("graph.json", "map.dot", "map.svg"):
|
|
artifacts[destination / "maps" / ident / name] = artifacts[
|
|
source / name
|
|
]
|
|
for name in (
|
|
"explorer.js",
|
|
"explorer.css",
|
|
"atlas.js",
|
|
"vendor/cytoscape.min.js",
|
|
"vendor/LICENSE.cytoscape",
|
|
):
|
|
artifacts[destination / name] = (site / name).read_bytes()
|
|
# Inline generated data lets a chapter viewer open directly from disk as well.
|
|
embedded = {
|
|
"catalog": local_catalog,
|
|
"maps": {
|
|
ident: {
|
|
"graph": json.loads(
|
|
artifacts[site / "maps" / ident / "graph.json"]
|
|
),
|
|
"svg": artifacts[site / "maps" / ident / "map.svg"].decode(),
|
|
}
|
|
for ident in available_maps
|
|
},
|
|
}
|
|
payload = (
|
|
json.dumps(embedded, ensure_ascii=False)
|
|
.replace("<", "\\u003c")
|
|
.replace(">", "\\u003e")
|
|
.replace("&", "\\u0026")
|
|
)
|
|
page = (
|
|
'<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>'
|
|
+ html.escape(chapter["title"])
|
|
+ ' concept maps</title><link rel="stylesheet" href="explorer.css"><style>body{margin:0;font-family:system-ui,sans-serif}.concept-explorer{padding:14px}.map-header h1{font-size:25px}.map-tools{gap:10px}</style></head><body>'
|
|
+ '<script type="application/json" id="concept-data">'
|
|
+ payload
|
|
+ "</script>"
|
|
+ template
|
|
+ "</body></html>"
|
|
)
|
|
artifacts[destination / "index.html"] = page.encode()
|
|
manifest = {}
|
|
for chapter in graph["chapters"].values():
|
|
if chapter["maps"] or chapter["owned"]:
|
|
manifest.setdefault(chapter["volume"], {})[
|
|
chapter["source"].parent.name
|
|
] = dict(
|
|
title=chapter["title"],
|
|
viewer="concepts-generated/index.html",
|
|
maps=chapter["maps"],
|
|
)
|
|
artifacts[root / "books/_build/concepts/chapters.json"] = _json(manifest)
|
|
for volume in {c["volume"] for c in graph["chapters"].values()}:
|
|
artifacts[site / "volumes" / f"{volume}.json"] = _json(
|
|
next(v for v in catalog["volumes"] if v["id"] == volume)
|
|
)
|
|
series_payload = json.dumps(
|
|
{
|
|
"catalog": catalog,
|
|
"maps": {
|
|
ident: {
|
|
"graph": json.loads(
|
|
artifacts[site / "maps" / ident / "graph.json"]
|
|
),
|
|
"svg": artifacts[site / "maps" / ident / "map.svg"].decode(),
|
|
}
|
|
for ident in graph["maps"] if any(m["id"] == ident for m in catalog["maps"])
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
series_payload = (
|
|
series_payload.replace("<", "\\u003c")
|
|
.replace(">", "\\u003e")
|
|
.replace("&", "\\u0026")
|
|
)
|
|
artifacts[site / "atlas.html"] = (
|
|
'<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">'
|
|
'<title>MLSysBook Concept Atlas</title><link rel="stylesheet" href="explorer.css">'
|
|
"<style>body{margin:0;font-family:system-ui,sans-serif;background:#fafbfc}</style></head><body>"
|
|
'<script type="application/json" id="concept-data">'
|
|
+ series_payload
|
|
+ "</script>"
|
|
+ template
|
|
+ "</body></html>"
|
|
).encode()
|
|
# The field reference model has independent identities; book links are an overlay.
|
|
from .field_graph import field_artifacts
|
|
artifacts.update(field_artifacts(root, graph))
|
|
orphaned: list[Path] = []
|
|
if not volumes:
|
|
for managed_dir in (site / "downloads", site / "field-focus", site / "maps", site / "volumes"):
|
|
if managed_dir.is_dir():
|
|
for existing in sorted(managed_dir.rglob("*")):
|
|
if existing.is_file() and existing not in artifacts:
|
|
orphaned.append(existing)
|
|
for pattern in ("field-*.svg", "field-*.dot"):
|
|
for existing in sorted(site.glob(pattern)):
|
|
if existing.is_file() and existing not in artifacts:
|
|
orphaned.append(existing)
|
|
stale = [
|
|
str(p.relative_to(root))
|
|
for p, data in artifacts.items()
|
|
if not p.exists() or p.read_bytes() != data
|
|
] + [f"{p.relative_to(root)} (orphaned)" for p in orphaned]
|
|
if not check:
|
|
for path, data in artifacts.items():
|
|
if path.exists() and path.read_bytes() == data:
|
|
continue
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
for orphan in orphaned:
|
|
orphan.unlink(missing_ok=True)
|
|
if orphan.parent != site and orphan.parent.is_dir() and not any(orphan.parent.iterdir()):
|
|
orphan.parent.rmdir()
|
|
field_payload = artifacts.get(site / "field-data.json")
|
|
return dict(
|
|
chapters=len(graph["chapters"]),
|
|
concepts=len(graph["nodes"]),
|
|
relationships=len(graph["edges"]),
|
|
maps=len(graph["maps"]),
|
|
stale=stale,
|
|
field=json.loads(field_payload)["summary"] if field_payload else None,
|
|
)
|
|
|
|
|
|
def refresh(root: Path, volumes: tuple[str, ...] = ()) -> dict:
|
|
"""Find definition candidates and changed evidence; never author accepted records.
|
|
|
|
This deterministic inventory is deliberately local: no model or embedding
|
|
service is invoked. Semantic relationship discovery remains future work.
|
|
"""
|
|
report = dict(schema_version=1, candidates=[], changed_evidence=[], invalid=[])
|
|
snapshot_path = root / "books/_build/concepts/evidence-snapshot.json"
|
|
previous = json.loads(snapshot_path.read_text()) if snapshot_path.exists() else {}
|
|
current = {}
|
|
contracts = {
|
|
(p.parent.parent.name, p.parent.name): yaml.safe_load(p.read_text())[
|
|
"concept_map"
|
|
]
|
|
for p in (root / "books").glob("vol*/*/*_concepts.yml")
|
|
}
|
|
for number in range(1, 5):
|
|
volume = f"vol{number}"
|
|
if volumes and volume not in volumes:
|
|
continue
|
|
for path in sorted((root / "books" / volume).glob("*/*_concepts.yml")):
|
|
contract = yaml.safe_load(path.read_text())["concept_map"]
|
|
source = path.parent / contract["source"]
|
|
text = source.read_text()
|
|
indexed = sections(text)
|
|
learner = contract.get("learner_graph", {})
|
|
labels = {
|
|
label.casefold()
|
|
for n in learner.get("concepts", [])
|
|
for label in [n["label"], *n.get("aliases", [])]
|
|
}
|
|
for match in re.finditer(
|
|
r':::\s*\{#([^\s}]+)[^\n]*\.callout-definition[^\n]*title="([^"]+)"',
|
|
text,
|
|
):
|
|
if match[2].casefold() not in labels:
|
|
report["candidates"].append(
|
|
dict(
|
|
label=match[2],
|
|
source=str(source.relative_to(root)),
|
|
anchor=match[1],
|
|
kind="definition",
|
|
status="candidate",
|
|
)
|
|
)
|
|
# Index definition tags expose vocabulary outside callout boxes.
|
|
known = labels | {
|
|
c["label"].casefold() for c in report["candidates"]
|
|
if c["source"] == str(source.relative_to(root))
|
|
}
|
|
for match in re.finditer(r"\\index\{([^{}\n]+)!definition\}", text):
|
|
label = match[1].split("!", 1)[0].strip()
|
|
if label.casefold() in known:
|
|
continue
|
|
anchored = [
|
|
(anchor, section) for anchor, section in indexed.items()
|
|
if match[0] in section
|
|
]
|
|
if not anchored:
|
|
continue
|
|
anchor = min(anchored, key=lambda item: len(item[1]))[0]
|
|
report["candidates"].append(dict(
|
|
label=label, source=str(source.relative_to(root)), anchor=anchor,
|
|
kind="index-definition", status="candidate",
|
|
))
|
|
known.add(label.casefold())
|
|
for edge in learner.get("relationships", []):
|
|
ident = edge["id"]
|
|
section = indexed.get(edge["evidence"]["section"])
|
|
if section is None or " ".join(
|
|
edge["evidence"]["quote"].split()
|
|
) not in " ".join((section or "").split()):
|
|
report["invalid"].append(
|
|
dict(id=ident, reason="Supporting section or quote changed")
|
|
)
|
|
now = dict(
|
|
section_hash=digest(section or ""),
|
|
authored_hash=digest(json.dumps(edge, sort_keys=True)),
|
|
)
|
|
old = previous.get(ident)
|
|
if isinstance(old, str):
|
|
old = dict(section_hash=old, authored_hash=now["authored_hash"])
|
|
if old and old["authored_hash"] == now["authored_hash"]:
|
|
current[ident] = old
|
|
if old["section_hash"] != now["section_hash"]:
|
|
report["changed_evidence"].append(ident)
|
|
else:
|
|
current[ident] = now
|
|
for relation in contract.get("explorer_graph", {}).get("relationships", []):
|
|
target = relation["target"]
|
|
source_id = (
|
|
volume
|
|
+ ".goal."
|
|
+ digest(
|
|
volume + "/" + path.parent.name + ":" + relation["source"]
|
|
)[:16]
|
|
)
|
|
target_id = (
|
|
volume
|
|
+ ".goal."
|
|
+ digest(
|
|
volume + "/" + target["chapter"] + ":" + target["concept"]
|
|
)[:16]
|
|
)
|
|
ident = (
|
|
volume
|
|
+ ".connection."
|
|
+ digest(source_id + relation["predicate"] + target_id)[:16]
|
|
)
|
|
claims = []
|
|
for item in relation.get("evidence", []):
|
|
claim = (
|
|
contracts.get((volume, item["chapter"]), {})
|
|
.get("claims", {})
|
|
.get(item["concept"], "")
|
|
)
|
|
claims.append(claim)
|
|
if not claim or " ".join(item["quote"].split()) not in " ".join(
|
|
claim.split()
|
|
):
|
|
report["invalid"].append(
|
|
dict(id=ident, reason="Supporting teaching claim changed")
|
|
)
|
|
now = dict(
|
|
section_hash=digest(json.dumps(claims)),
|
|
authored_hash=digest(json.dumps(relation, sort_keys=True)),
|
|
)
|
|
old = previous.get(ident)
|
|
if (
|
|
isinstance(old, dict)
|
|
and old["authored_hash"] == now["authored_hash"]
|
|
):
|
|
current[ident] = old
|
|
if old["section_hash"] != now["section_hash"]:
|
|
report["changed_evidence"].append(ident)
|
|
else:
|
|
current[ident] = now
|
|
for volume in (volumes or tuple(f"vol{n}" for n in range(1, 5))):
|
|
labels = {
|
|
label.casefold()
|
|
for (vol, _), contract in contracts.items() if vol == volume
|
|
for n in contract.get("learner_graph", {}).get("concepts", [])
|
|
for label in [n["label"], *n.get("aliases", [])]
|
|
}
|
|
for glossary in sorted((root / "books" / volume / "backmatter").glob("**/glossary.qmd")):
|
|
text = glossary.read_text()
|
|
for match in re.finditer(r"^\*\*([^*\n]+)\*\*\s*\n:\s+([^\n]+)", text, re.M):
|
|
if match[1].casefold() not in labels:
|
|
report["candidates"].append(dict(
|
|
label=match[1], source=str(glossary.relative_to(root)),
|
|
quote=match[2], kind="glossary", status="candidate", volume=volume,
|
|
))
|
|
output = root / "books/_build/concepts"
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
(output / "review.json").write_bytes(_json(report))
|
|
# A restricted refresh must retain the baseline for other volumes.
|
|
retained = {
|
|
key: value
|
|
for key, value in previous.items()
|
|
if volumes and not any(key.startswith(v + ".") for v in volumes)
|
|
}
|
|
snapshot_path.write_bytes(_json({**retained, **current}))
|
|
return report
|