#!/usr/bin/env python3
"""Draw the S1-S7 pipeline as two data-flow diagrams, from pipeline.json.
WHY THIS EXISTS
---------------
The pipeline is described in four places: the manuscript's locked step table,
Supplementary Table 4, manuscript Figure 13, and the platform. Four descriptions
of one thing is four opportunities to disagree, and they already had: the
platform served a figure-provenance record a month stale while the letter
reported a different count.
pipeline.json is the single record. It is derived from the locked table and the
deposited Supplementary Table 4, and it carries Table 4's SHA-256 so a reader
can tell whether the mapping drawn here is the mapping deposited. This script is
one consumer of it. Nothing here restates a fact about the pipeline.
FOLLOWING THE HOUSE DIAGRAM SYSTEM, DELIBERATELY THIS TIME
----------------------------------------------------------
Figure 11 departed from the diagram-design skill by omission. This does not.
It implements the `data-flow` type: role lanes, numbered step columns, node
cards with input/output chips, one focal step, one focal node, one focal arrow,
single-bend orthogonal routing, and the three-row legend. Geometry comes from
the type's own formulas rather than from taste.
The type's budget is six steps. There are seven, and the skill says split rather
than crowd. The split is the paper's own: Stage 1 is deterministic and ends when
the corpus freezes at G3; Stage 2 is LLM-assisted and every one of its outputs is
checked back against Stage 1. Drawing them separately is what the architecture
already claims.
The LLM lane in the Stage 1 diagram is empty across all three steps. That is not
a gap in the drawing. It is the strongest form of the sentence "no model
produced any reported value in Stage 1" — the row is there and nothing is in it.
Usage:
python3 deposit/build_pipeline_diagram.py
"""
from __future__ import annotations
import json
import os
import pathlib
HERE = pathlib.Path(__file__).resolve().parent
SSOT = HERE / "pipeline.json"
# The output directory is overridable so a checker can render into a scratch
# directory and compare byte for byte. The first version of that check
# copied this script into the scratch directory instead, which moved HERE
# with it and made the generator look for pipeline.json beside the copy.
OUT = pathlib.Path(os.environ.get("PIPELINE_FIGURES_OUT", HERE / "figures"))
# ── Skin ──────────────────────────────────────────────────────────────────────
# The Sumarhús editorial palette from the skill's style guide. This one is a
# platform surface, not a journal figure, so the house skin is right here in a
# way it was not for Figure 11.
PAPER, INK = "#F5F0E8", "#14110D"
MUTED, SOFT = "#6E6659", "#8A7D68"
ACCENT, LINK = "#B87333", "#1A6E78"
SANS = "Archivo, 'Helvetica Neue', Helvetica, Arial, sans-serif"
MONO = "'SF Mono', SFMono-Regular, ui-monospace, Menlo, monospace"
# Data-type chips (§8). Codes chosen for what actually moves in this pipeline.
CHIPS = {
"WB": ("#6e6479", "Public register"),
"DB": ("#5e7a9b", "Source dataset"),
"TB": ("#b8915a", "Frozen table"),
"FL": ("#9c6b50", "Prose / report"),
}
# ── Geometry, from the type reference §2 ──────────────────────────────────────
LABEL_COL_W, STEP_SLOT_W, RIGHT_PAD = 140, 112, 28
HEADER_H, LANE_H, LEGEND_H = 36, 80, 80
NODE_W, NODE_H = 100, 64
LANES = [
("SRC", ["EXTERNAL", "SOURCES"]),
("PIPE", ["DETERMINISTIC", "PIPELINE"]),
("AI", ["LLM", "TOOL"]),
("GATE", ["HUMAN", "GATE"]),
]
LANE_IX = {k: i for i, (k, _) in enumerate(LANES)}
# The badge on each node names its lane. It printed the lane KEY, so the key
# "AI" reached the figure after the paper stopped saying AI (6 Oct 2026). The
# key is a schema identifier read by pipeline.json and the app; what a reader
# sees is a separate decision and lives here.
BADGE = {"SRC": "SRC", "PIPE": "PIPE", "AI": "LLM", "GATE": "GATE"}
def fit(s: str, size: float, width: float = NODE_W - 8) -> str:
"""Trim to what fits the node box, on a word boundary, with an ellipsis.
The first render clipped "IUCN Red List assessme" and "Conservation-status
compil" mid-word. A box that cannot hold its label is a layout error, and
silently letting the glyphs run past the border is the version of it that
looks like a bug rather than a decision.
"""
per = size * 0.60 # Archivo advance, measured against the render
if len(s) * per <= width:
return s
keep = max(1, int(width / per) - 1)
cut = s[:keep]
if " " in cut[int(keep * 0.5):]:
cut = cut.rsplit(" ", 1)[0]
return cut.rstrip(" ,;·-") + "\u2026"
def esc(s: str) -> str:
return (s.replace("&", "&").replace("<", "<").replace(">", ">"))
def txt(x, y, s, size, fill, weight="400", anchor="middle", family=SANS, extra=""):
return (f'{esc(s)}\n')
def step_cx(j: int) -> float:
return LABEL_COL_W + j * STEP_SLOT_W + STEP_SLOT_W / 2
def lane_y_top(k: int) -> float:
return HEADER_H + k * LANE_H
def elbow(x1, y1, x2, y2, r=8) -> str:
"""Horizontal first, then vertical, with a quarter-arc corner. No diagonals."""
if abs(y1 - y2) < 0.5:
return f"M{x1},{y1} L{x2},{y2}"
if abs(x1 - x2) < 0.5:
return f"M{x1},{y1} L{x2},{y2}"
sy = 1 if y2 > y1 else -1
sx = 1 if x2 > x1 else -1
return (f"M{x1},{y1} L{x2 - sx * r},{y1} "
f"Q{x2},{y1} {x2},{y1 + sy * r} L{x2},{y2}")
def diagram(title, subtitle, steps, nodes, arrows, focal_step,
accent_label="Focal hand-off") -> str:
n_steps, n_lanes = len(steps), len(LANES)
W = LABEL_COL_W + n_steps * STEP_SLOT_W + RIGHT_PAD
legend_top = HEADER_H + n_lanes * LANE_H
H = legend_top + LEGEND_H
o = [f'\n")
return "".join(o)
def main() -> int:
ssot = json.loads(SSOT.read_text())
steps = {s["id"]: s for s in ssot["steps"]}
OUT.mkdir(exist_ok=True)
# ── Stage 1 ───────────────────────────────────────────────────────────────
s1_steps = [{"number": "01", "label": "IDENTIFY"},
{"number": "02", "label": "COMPILE"},
{"number": "03", "label": "INTEGRATE"}]
s1_nodes, s1_arrows = [], []
for j, sid in enumerate(("S1", "S2", "S3")):
st = steps[sid]
s1_nodes.append({"lane": "SRC", "step": j, "title": st["source_name"],
"sub": "external register", "tool": "custodian record",
"cout": "WB" if j == 0 else "DB"})
s1_nodes.append({"lane": "PIPE", "step": j, "title": f"{st['id']} {st['short_title']}",
"sub": ", ".join(st["audit_questions"]) or "no audit question",
"tool": "deterministic query",
"cin": "WB" if j == 0 else "DB",
"cout": "TB" if sid == "S3" else "DB",
"focal": sid == "S3"})
s1_nodes.append({"lane": "GATE", "step": j, "title": st["gate"],
"sub": "author-verified", "tool": "reconciled to source",
"cin": "TB" if sid == "S3" else "DB"})
s1_arrows.append({"frm": ("SRC", j), "to": ("PIPE", j), "style": "muted"})
s1_arrows.append({"frm": ("PIPE", j), "to": ("GATE", j),
"style": "accent" if sid == "S3" else "muted",
"label": "corpus frozen" if sid == "S3" else None})
if j:
s1_arrows.append({"frm": ("PIPE", j - 1), "to": ("PIPE", j), "style": "muted"})
(OUT / "pipeline-stage1.svg").write_text(diagram(
"Stage 1: deterministic", "No model produces any value in this stage",
s1_steps, s1_nodes, s1_arrows, focal_step=2,
accent_label="Corpus frozen, Stage 1 ends"))
# ── Stage 2 ───────────────────────────────────────────────────────────────
s2_steps = [{"number": "04", "label": "RETRIEVE"}, {"number": "05", "label": "SYNTHESISE"},
{"number": "06", "label": "CLASSIFY"}, {"number": "07", "label": "PUBLISH"}]
s2_nodes, s2_arrows = [], []
for j, sid in enumerate(("S4", "S5", "S6", "S7")):
st = steps[sid]
s2_nodes.append({"lane": "AI", "step": j, "title": st["tool_name"],
"sub": "model-assisted" if sid in ("S4", "S5") else "summary only",
"tool": "walled off" if sid == "S7" else "LLM tool",
"cin": "TB" if j == 0 else "FL", "cout": "FL"})
s2_nodes.append({"lane": "GATE", "step": j, "title": st["gate"],
"sub": ", ".join(st["audit_questions"]) or "no audit question",
"tool": "author-scored" if sid == "S6" else "author-checked",
"cin": "FL", "cout": "TB" if sid == "S6" else "FL",
"focal": sid == "S6"})
s2_arrows.append({"frm": ("AI", j), "to": ("GATE", j),
"style": "accent" if sid == "S6" else "muted",
"label": "scored by authors" if sid == "S6" else None})
if j:
s2_arrows.append({"frm": ("AI", j - 1), "to": ("AI", j), "style": "muted"})
s2_nodes.append({"lane": "PIPE", "step": 0, "title": "Frozen corpus",
"sub": "Stage 1 output", "tool": "corpus frozen at G3",
"cout": "TB"})
s2_arrows.append({"frm": ("PIPE", 0), "to": ("AI", 0), "style": "muted"})
s2_arrows.append({"frm": ("GATE", 2), "to": ("GATE", 3), "style": "link"})
(OUT / "pipeline-stage2.svg").write_text(diagram(
"Stage 2: LLM-assisted", "Every output is checked back against Stage 1",
s2_steps, s2_nodes, s2_arrows, focal_step=2,
accent_label="Scored by the authors, not the model"))
# PNG beside each SVG, because a diagram that has to be pasted into a
# document or a slide is pasted as a raster and somebody will otherwise
# export one by hand, which is a copy that drifts. Rendered at 3x so it
# survives a retina screen and a printed page.
#
# rsvg-convert is not assumed: a reader cloning this repository has the SVGs
# either way, and a missing rasteriser is reported rather than failing the
# build. The SVGs are the artefact of record; the PNGs are a convenience.
import shutil
import subprocess
raster = shutil.which("rsvg-convert")
for f in ("pipeline-stage1.svg", "pipeline-stage2.svg"):
print(f" wrote figures/{f} ({(OUT / f).stat().st_size:,} bytes)")
if raster:
png = OUT / f.replace(".svg", ".png")
subprocess.run([raster, "-z", "3", "-o", str(png), str(OUT / f)],
check=True)
print(f" wrote figures/{png.name} ({png.stat().st_size:,} bytes)")
if not raster:
print(" rsvg-convert not found: PNGs not refreshed. The SVGs are the "
"record; install librsvg to regenerate the rasters.")
return 0
if __name__ == "__main__":
raise SystemExit(main())