""" core/layout_generate.py — `layout.generate`: the mesh makes the document. The annotator at `layout.docbank` solves the hard problem: given a document somebody else wrote, recover exactly where every element sits. It does it by compiling twice — once clean, once with each semantic element recoloured — and reading glyph colours out of the coloured twin. It is careful because it has to be: it does not know what the template contains, so it infers columns, merges lines into regions, and guesses where one block ends. Generation does not have that problem. When the mesh writes the LaTeX, it knows what every block is *before* the page exists. So this uses the same proven mechanism — two compiles, colour is dimensionless, prove it — and throws away every heuristic on top of it: the annotator this ------------------ ------------------------------------------ colour per label colour per BLOCK INSTANCE merge lines to guess the region is the union of that block's own a region words; there is nothing to guess infer columns the spec chose the column count unprofiled classes there is no foreign class to profile A unique colour per block is the whole difference. Block 12's words are the only words in the document drawing in block 12's colour, so its bounding box is exact by construction and a paragraph that breaks across a column or a page yields one region per page without a single tolerance constant. What is kept, unchanged, is the fidelity gate. Colour being layout-neutral is an assumption, and an assumption that is wrong by two points produces a dataset that trains a model to be confidently, uniformly wrong with nothing in the pipeline complaining. Every run compares the two PDFs glyph by glyph and a run that drifts is not a result, whatever the boxes look like. The other thing this refuses to be is a volume machine. The excerpt this comes from is right: nobody needs another million perfect documents. `weakness` takes a per-label score from an evaluation run and biases sampling toward what the model is currently bad at, so a generation run is an answer to a measurement rather than a number someone picked. Runs hosted. The mesh image carries latexmk; nobody installs TeX to try this. Where it cannot compile it says so and hands back the source, because "here is the document, you need a runtime" is a usable answer and a traceback is not. """ from __future__ import annotations import json import random import re import shutil import subprocess from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple SCHEMA = "signalmesh.layout_generate/v1" COORD = "layout.generate" # Same tolerance the annotator uses. A coloured run that moves a glyph further # than this did not measure the clean page. DRIFT_TOL_PT = 0.05 # Label names are deliberately the annotator's, so a generated dataset and an # annotated one merge without a translation table. LABELS = ("headline", "byline", "date", "abstract", "subhead", "paragraph", "list", "table", "figure", "caption", "equation", "quote", "code", "reference") # Blocks that carry a body of prose and can therefore be sampled repeatedly. # `headline`/`byline`/`date`/`abstract` are structural — one per document — so # they are not in the sampling pool however weak the model is on them. SAMPLED = ("paragraph", "list", "table", "figure", "equation", "quote", "code", "reference") _WORDS = ( "signal mesh coordinate parcel antenna frequency broadcast lattice region " "boundary gradient threshold aperture manifold corridor junction terminal " "cadence drift anchor survey beacon harbor district tower archive registry " "transit ledger foundry quarry orchard viaduct cistern annex bastion" ).split() _TITLES = ("SignalMesh: Coordinate-Addressable Cognition", "Query the Mesh, Do Not Crawl the Repository", "Ground Truth Before the Page Exists", "Every Capability Has an Address", "Zero Egress by Construction", "A Map for Systems You Already Run") # Filler is not filler. Every unsourced document generated here is a sample # somebody screenshots, publishes to a dataset card, or opens in the world, so # the sentences on it say what the mesh does instead of saying nothing at all. # # Every claim below is one this repository actually implements — the fidelity # gate, the exporters, the runtime split, drivemesh charting, the R4 rules. A # generated page is the wrong place to promise a feature that does not exist, # because it is exactly the artefact that outlives the conversation. # # Lengths run from six words to forty on purpose: sentence length is a layout # property, and a corpus of one length produces a corpus of one line-breaking # behaviour. _BYLINES = ("SignalMesh Field Notes", "The Mesh Operations Desk", "Coordinate Registry", "SignalMesh Research", "Atlas Survey Office", "The Inspection Department") _HEADS = ( "Coordinate Addressing", "The Fidelity Gate", "Charting a Folder", "Where a Capability Runs", "Frequencies as Places", "Exporting the Run", "Hosted or Edge", "Steering by Measurement", "Ground Truth by Construction", "The Thought Log", "Observed Provenance", "Zero Egress", "Wrapping a Tool as a Coordinate", "Visibility Is Asked", "Districts, Not Rivals", ) _PITCH = ( "SignalMesh is not an orchestrator. It is the topology that orchestration " "occurs inside of.", "Every file, symbol, capability and thought gets a stable address.", "Sub-millisecond lookups against a local index. No API, no model, no egress.", "The mesh is charted once and stays current incrementally, so a coordinate " "resolved today still resolves after the refactor.", "Ask the mesh where something is instead of searching for where it might be.", "A capability is reachable by name, and where it runs is derived rather than " "assumed.", "Something runs on your own machine for exactly one of two reasons: the " "authority only exists there, or nobody has hosted the runtime yet.", "The first of those is permanent and correct. The second is just work not " "done, and it is tracked as work.", "Secrets are masked at chart time. A coordinate never carries a credential.", "This document was generated by the mesh, and every region on it was known " "before the page existed.", "The generator writes the LaTeX, so the ground truth is exact by " "construction rather than recovered afterwards.", "Each block draws in its own colour. Its bounding box is the union of its " "own words, and there is nothing left to infer.", "The document is compiled twice, once clean and once instrumented, and a run " "whose glyphs moved produces no labels at all.", "A silent three percent of misaligned boxes is worse than a failed run, " "because it trains a model to be confidently wrong.", "Annotations export as COCO, YOLO, Pascal VOC, Hugging Face datasets or " "JSONL without a translation table.", "Points are not pixels, and the origin is not where you assume. Both are " "asserted rather than left to the caller.", "Generation is steered by measurement. Per-class scores from an evaluation " "run bias sampling toward whatever the model is currently worst at.", "Nobody needs another million perfect documents. They need the documents " "their model fails on.", "Point the mesh at a folder and every file in it has a coordinate before " "anything has been downloaded.", "Bytes stay out of the mesh. A signal carries the identity of a file, never " "the file.", "Private means private. Visibility is asked at intake and never guessed.", "Frequencies are places. A broadcast is somewhere you can go and stand.", "Roads in the atlas are observed provenance, not decoration.", "The lens asks the mesh, so a fuzzy trail still routes. Nothing is matched " "by substring in the browser.", "A place that does not match your search dims. It never disappears.", "Drop a file onto a card and you have handed it to that place.", "Controls are generated from a repository's own argument parser, so the " "console cannot offer a flag the tool does not have.", "Any tool with a declared surface can be wrapped as a coordinate and called " "like anything else in the mesh.", "R4 watches for the mistakes already made once, and reports them rather " "than quietly rewriting them.", "A detector that silently patches a bug makes it invisible, and it still " "ships to everyone not running this mesh.", "Governance is per neighbourhood. Quarantine is a place, not a status flag.", "Devices pair with the mesh, and a paired device keeps its own credentials.", "A hosted browser session would be everyone's cookies in one place, so " "there is not one.", "The thought log grades an intent as new, partial, existing or specified, " "and hands back the coordinates to act from.", "Reading a two thousand line file to answer one question is the thing this " "replaces.", "Districts are the tools you already run. The mesh gives them an address, " "it does not replace them.", "Maps do not replace buildings. They help you navigate them.", "Every stage of the pipeline declares itself, including the stages nobody " "has built, because a pipeline that hides its gaps is a wish.", "An unprofiled document class is a request the mesh can satisfy, not a " "crash and not a guess.", "The whole index is a single file you own, and it never leaves the machine " "it was built on.", ) # ── colour: one per block instance ─────────────────────────────────────────── # 0-255 per channel, kept away from the extremes so no channel rounds into # another block's, and never r == g == b: PDF encodes greys in DeviceGray, a # 1-tuple, which cannot be told apart from unlabelled body text. def colour_of(n: int) -> Tuple[int, int, int]: """Block index to an RGB triple that survives a PDF round trip.""" if n < 0: raise ValueError("block index is not negative") r = 20 + (n % 200) g = 20 + ((n // 200) % 200) b = 20 + ((n // 40000) % 200) if r == g == b: b += 1 return r, g, b def _key(rgb: Sequence[float]) -> Tuple[float, float, float]: return tuple(round(float(v), 3) for v in rgb) def resolve(colour: Any, table: Dict[Tuple[float, float, float], int]) -> Optional[int]: """One pdfplumber colour value back to the block that drew it. Colour space varies per operator — DeviceGray is a 1-tuple, DeviceCMYK a 4-tuple — and only a 3-tuple can carry a block. """ if colour is None or isinstance(colour, (int, float)): return None try: if len(colour) != 3: return None return table.get(_key(colour)) except (TypeError, ValueError): return None # ── the spec: what document to write ───────────────────────────────────────── def weights(weakness: Optional[Dict[str, float]] = None) -> Dict[str, float]: """Sampling weight per block label, from an evaluation run. `weakness` maps a label to how well the model currently does on it — an AP in 0..1 — so the weight is what is left over. A label the model has never seen scores 0 and dominates; a label at AP 0.95 all but disappears. Absent scores keep the flat weight, because "unmeasured" is not "solved". """ w = {k: 1.0 for k in SAMPLED} for label, ap in (weakness or {}).items(): if label in w: w[label] = max(0.05, 1.0 - min(max(float(ap), 0.0), 1.0)) return w def _ascii(text: str) -> str: """Real prose carries curly quotes, en dashes and whatever else a word processor left in it. pdflatex fails on most of that, and a document that will not compile is not a training example — so sourced text is folded to ASCII before it can reach the page.""" for a, b in (("‘", "'"), ("’", "'"), ("“", '"'), ("”", '"'), ("–", "-"), ("—", "---"), ("…", "..."), (" ", " ")): text = text.replace(a, b) return text.encode("ascii", "ignore").decode("ascii") def _bank(lines: Sequence[str]) -> Tuple[str, ...]: """Single words for the short fills — a subhead, a table cell, a byline. Taken from the same corpus as the sentences, so a page never mixes real prose with invented vocabulary. """ return tuple(dict.fromkeys( w.strip(".,;:()[]\"'") for s in lines for w in _ascii(s).split() if w.strip(".,;:()[]\"'").isalpha() and len(w) > 3)) def _headline(rng: random.Random, bank: Sequence[str], say: Sequence[str]) -> str: """A title. Where there is real prose the words come from it, so a sourced document has no invented vocabulary anywhere on the page.""" if not say: return rng.choice(_TITLES) return " ".join(rng.choice(bank).title() for _ in range(rng.randint(3, 7))) def _sentence(rng: random.Random, lo: int = 6, hi: int = 16, pool: Sequence[str] = ()) -> str: """One sentence — a real one where the mesh was pointed at real documents. A sourced sentence is used as it stands rather than trimmed to a word count. Its length is the thing worth having: real corpora have real sentence lengths, and that is a layout property, not a text property. """ if pool: return _ascii(rng.choice(pool)) or _sentence(rng, lo, hi) n = rng.randint(lo, hi) s = " ".join(rng.choice(_WORDS) for _ in range(n)) return s[0].upper() + s[1:] + "." def _para(rng: random.Random, lo: int = 2, hi: int = 6, pool: Sequence[str] = ()) -> str: return " ".join(_sentence(rng, pool=pool) for _ in range(rng.randint(lo, hi))) def spec(seed: int, *, sections: int = 3, blocks_per_section: int = 4, weakness: Optional[Dict[str, float]] = None, twocolumn: Optional[bool] = None, source: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """A whole document as a flat list of labelled blocks, from one seed. The same seed gives the same document, which is what makes a dataset reproducible and a regression bisectable. `source` is what a charted folder supplied — real sentences, and the per-class scores from an eval run. Given one, the pages carry that corpus and the sampling answers a measurement instead of a guess. Passing `weakness` explicitly still wins: an argument beats a file somebody dropped in a folder. Without one the pages carry `_PITCH`, which is the mesh describing itself. A generated document is a sample somebody screenshots or publishes, so it says what this does rather than saying nothing in six invented nouns. """ rng = random.Random(seed) src = source or {} supplied = tuple(src.get("prose") or ()) say = supplied or _PITCH # Short fills — a subhead, a table cell, a byline — are not sentences, so # they take single words. Where there is real prose they take them from it, # and the invented vocabulary disappears from the document entirely. bank = _bank(supplied) or _bank(_PITCH) or tuple(_WORDS) weakness = weakness if weakness is not None else (src.get("weakness") or None) w = weights(weakness) pool, wts = list(w), [w[k] for k in w] blocks: List[Dict[str, Any]] = [ {"label": "headline", "kind": "title", "text": _headline(rng, bank, supplied)}, {"label": "byline", "kind": "byline", "text": rng.choice(_BYLINES) if not supplied else rng.choice(bank).title() + " " + rng.choice(bank).title()}, {"label": "date", "kind": "date", "text": "{:04d}-{:02d}-{:02d}".format(rng.randint(2019, 2026), rng.randint(1, 12), rng.randint(1, 28))}, {"label": "abstract", "kind": "abstract", "text": _para(rng, 2, 4, say)}, ] for _ in range(sections): blocks.append({"label": "subhead", "kind": "subhead", "text": rng.choice(_HEADS) if not supplied else " ".join(rng.choice(bank).title() for _ in range(rng.randint(2, 4)))}) for _ in range(blocks_per_section): label = rng.choices(pool, weights=wts, k=1)[0] blocks.append(_block(label, rng, say, bank)) for i, b in enumerate(blocks): b["id"] = i b["rgb"] = colour_of(i) return {"schema": SCHEMA, "seed": seed, "blocks": blocks, "twocolumn": rng.random() < 0.35 if twocolumn is None else bool(twocolumn), "weakness": dict(weakness or {}), "sourced": bool(supplied), "source": src.get("from") or None, "label_counts": _counts(blocks)} def _block(label: str, rng: random.Random, say: Sequence[str] = (), bank: Sequence[str] = _WORDS) -> Dict[str, Any]: if label == "list": return {"label": label, "kind": "list", "items": [_sentence(rng, 3, 9, say) for _ in range(rng.randint(3, 6))]} if label == "table": cols = rng.randint(2, 4) return {"label": label, "kind": "table", "cols": cols, "rows": [[rng.choice(bank) for _ in range(cols)] for _ in range(rng.randint(2, 5))]} if label == "figure": return {"label": label, "kind": "figure", "width_pt": rng.randint(120, 300), "height_pt": rng.randint(60, 180), "caption": _sentence(rng, 4, 12, say)} if label == "equation": return {"label": label, "kind": "equation", "text": rng.choice([r"\sum_{i=1}^{n} x_i^2 = \lambda \int_0^1 f(t)\,dt", r"\frac{\partial u}{\partial t} = \alpha \nabla^2 u", r"E = mc^2 + \sum_{k} \beta_k \log p_k"])} if label == "code": return {"label": label, "kind": "code", "lines": ["def " + rng.choice(bank).lower() + "(x):", " return x * " + str(rng.randint(2, 99)), "# " + rng.choice(bank).lower()]} if label == "reference": return {"label": label, "kind": "reference", "items": [rng.choice(bank).title() + ", " + rng.choice(bank).title() + ". " + _sentence(rng, 3, 7, say) for _ in range(rng.randint(2, 4))]} if label == "quote": return {"label": label, "kind": "quote", "text": _sentence(rng, 8, 20, say)} return {"label": "paragraph", "kind": "paragraph", "text": _para(rng, pool=say)} def _counts(blocks: Sequence[Dict[str, Any]]) -> Dict[str, int]: out: Dict[str, int] = {} for b in blocks: out[b["label"]] = out.get(b["label"], 0) + 1 return dict(sorted(out.items(), key=lambda kv: -kv[1])) # ── the LaTeX: one body, two definitions of the same two macros ────────────── def _esc(s: str) -> str: out = [] for ch in str(s): if ch in "#$%&_{}": out.append("\\" + ch) elif ch == "~": out.append(r"\textasciitilde{}") elif ch == "^": out.append(r"\textasciicircum{}") elif ch == "\\": out.append(r"\textbackslash{}") else: out.append(ch) return "".join(out) def _c(b: Dict[str, Any]) -> str: return "SM" + str(b["id"]) def _wrap(b: Dict[str, Any], body: str) -> str: r"""Colour one span — never as the thing that starts the paragraph. `\SMb` becomes `\textcolor`, and `\textcolor` does `\leavevmode`. So in the annotated render a paragraph beginning with `\SMb` starts on an empty hbox, while the clean render starts on its first character and TeX inserts a `\parindent` box. Different first box, different line width, reflow — and the gate rightly rejected the whole run. The fix is not to place the colour more cleverly. It is to put the token that begins the paragraph in the body the two renders SHARE, so they cannot disagree about it: `\leavevmode` here is a no-op wherever we are already in horizontal mode, and where we are not, both files do it identically. """ return r"\leavevmode\SMb{" + _c(b) + "}{" + body + "}" def _body_of(b: Dict[str, Any]) -> str: k, w = b["kind"], _wrap if k == "title": return r"\begin{center}{\LARGE\bfseries " + w(b, _esc(b["text"])) + "}\\par\\vspace{6pt}\\end{center}" if k == "byline": return r"\begin{center}{\large " + w(b, _esc(b["text"])) + "}\\par\\end{center}" if k == "date": return r"\begin{center}{\small " + w(b, _esc(b["text"])) + "}\\par\\vspace{10pt}\\end{center}" if k == "abstract": return r"\noindent " + w(b, r"\textbf{Abstract.} " + _esc(b["text"])) + r"\par\vspace{8pt}" if k == "subhead": return r"\vspace{8pt}\noindent{\large\bfseries " + w(b, _esc(b["text"])) + r"}\par\vspace{4pt}" if k == "paragraph": # \indent, not the \leavevmode _wrap supplies, so body paragraphs keep # the indentation a real article has. Both renders see it. return r"\indent " + w(b, _esc(b["text"])) + r"\par" if k == "quote": return r"\begin{quote}" + w(b, r"\itshape " + _esc(b["text"])) + r"\end{quote}" if k == "list": # The marker is given explicitly, and coloured, because \item draws its # default label in black — so the bullets extracted as unlabelled and # every list region started at its first word, clipping its own markers # off. \textbullet IS labelitemi, so the label box is unchanged. items = "".join(r"\item[" + w(b, r"\textbullet") + "] " + w(b, _esc(it)) + "\n" for it in b["items"]) return r"\begin{itemize}" + "\n" + items + r"\end{itemize}" if k == "reference": items = ("\\par\n" + r"\noindent ").join( w(b, r"\small " + _esc(it)) for it in b["items"]) return r"\vspace{6pt}\noindent " + items + r"\par\vspace{4pt}" if k == "code": # \mbox{} first, because a code line's leading indentation is `~` glue # and TeX discards glue at the start of a line. The clean render dropped # it; the annotated one kept it, because the colour whatsit in front is # not discardable and shields the glue behind it. An empty hbox in the # shared body shields it identically in both. lines = (r"\\" + "\n" + r"\mbox{}").join( w(b, r"\texttt{" + _esc(ln).replace(" ", "~") + "}") for ln in b["lines"]) return r"\vspace{4pt}\noindent \mbox{}" + lines + r"\par\vspace{4pt}" if k == "equation": # \color inside math mode is a whatsit in the math list, not the # vertical list — the placement the annotator measured as costing a # whole \baselineskip. The gate proves it either way. return (r"\begin{equation}\color{" + _c(b) + "}" + b["text"] + r"\end{equation}") if k == "table": spec_ = "l" * int(b["cols"]) rows = ("\\\\\n").join(" & ".join(w(b, _esc(c)) for c in row) for row in b["rows"]) return (r"\vspace{4pt}\begin{center}\begin{tabular}{" + spec_ + r"}\hline" + "\n" + rows + r"\\\hline\end{tabular}\end{center}\vspace{4pt}") if k == "figure": rect = (r"\SMr{" + _c(b) + "}{" + str(b["width_pt"]) + "pt}{" + str(b["height_pt"]) + "pt}") cap = w(b, r"\small Figure: " + _esc(b["caption"])) return (r"\begin{center}" + rect + r"\par\vspace{3pt}\noindent " + cap + r"\par\end{center}\vspace{4pt}") raise ValueError("no emitter for block kind " + repr(k)) _PREAMBLE = r"""\documentclass[__OPTS__11pt]{article} \usepackage{xcolor} \usepackage{amsmath} \pagestyle{empty} \definecolor{SMgrey}{RGB}{150,150,150} __COLOURS__ """ # `{#2}`, not `#2`. \textcolor opens a group around its argument, so a font # change inside a span — \small, \itshape — is confined in the annotated render # and would leak into the rest of the paragraph in the clean one. Two braces; # without them the two pages are set at different sizes and the gate says so. _CLEAN_MACROS = r"""\newcommand{\SMb}[2]{{#2}} \newcommand{\SMr}[3]{{\fboxsep=0pt\colorbox{SMgrey}{\phantom{\rule{#2}{#3}}}}} """ _ANNOT_MACROS = r"""\newcommand{\SMb}[2]{\textcolor{#1}{#2}} \newcommand{\SMr}[3]{{\fboxsep=0pt\colorbox{#1}{\phantom{\rule{#2}{#3}}}}} """ def latex(sp: Dict[str, Any]) -> Dict[str, str]: """The two .tex files. Identical body; the macros differ, nothing else.""" colours = "\n".join( r"\definecolor{" + _c(b) + "}{RGB}{" + ",".join(str(v) for v in b["rgb"]) + "}" for b in sp["blocks"]) pre = (_PREAMBLE.replace("__OPTS__", "twocolumn," if sp.get("twocolumn") else "") .replace("__COLOURS__", colours)) body = "\n".join(_body_of(b) for b in sp["blocks"]) doc = "\n" + r"\begin{document}" + "\n" + body + "\n" + r"\end{document}" + "\n" return {"clean.tex": pre + _CLEAN_MACROS + doc, "annot.tex": pre + _ANNOT_MACROS + doc} # ── running it ─────────────────────────────────────────────────────────────── def available() -> Dict[str, Any]: """Whether this coordinate can execute here, proved rather than declared.""" try: import pdfplumber # noqa: F401 plumber = True except Exception: plumber = False tex = bool(shutil.which("latexmk")) return {"latexmk": tex, "pdfplumber": plumber, "ready": tex and plumber, "missing": [n for n, ok in (("latexmk", tex), ("pdfplumber", plumber)) if not ok]} def _compile(work: Path, name: str) -> Path: proc = subprocess.run( ["latexmk", "-pdf", "-interaction=nonstopmode", "-halt-on-error", "-file-line-error", name], cwd=work, capture_output=True, text=True) pdf = work / (Path(name).stem + ".pdf") if proc.returncode != 0 or not pdf.exists(): errs = [l for l in (proc.stdout or "").splitlines() if re.search(r":\d+:|^!", l)] raise RuntimeError("latexmk failed on " + name + "\n " + "\n ".join(errs[:8] or (proc.stdout or "").splitlines()[-8:])) return pdf def _fidelity(clean_pdf: Path, annot_pdf: Path) -> Dict[str, Any]: """Colour is supposed to be dimensionless. Check, do not assume.""" import pdfplumber rep: Dict[str, Any] = {"ok": True, "max_drift_pt": 0.0, "pages": []} with pdfplumber.open(clean_pdf) as a, pdfplumber.open(annot_pdf) as b: if len(a.pages) != len(b.pages): return {"ok": False, "max_drift_pt": None, "pages": [], "error": "page count {} vs {}".format(len(a.pages), len(b.pages))} for i, (pa, pb) in enumerate(zip(a.pages, b.pages), 1): ca, cb = pa.chars, pb.chars e: Dict[str, Any] = {"page": i, "glyphs": len(ca)} if len(ca) != len(cb): e["status"] = "GLYPH COUNT {} vs {}".format(len(ca), len(cb)) rep["ok"] = False else: d = max((max(abs(x["x0"] - y["x0"]), abs(x["top"] - y["top"])) for x, y in zip(ca, cb)), default=0.0) e["drift_pt"] = round(d, 4) e["status"] = "ok" if d <= DRIFT_TOL_PT else "DRIFT" rep["max_drift_pt"] = max(rep["max_drift_pt"], d) rep["ok"] = rep["ok"] and d <= DRIFT_TOL_PT rep["pages"].append(e) return rep def _extract(annot_pdf: Path, sp: Dict[str, Any]) -> List[Dict[str, Any]]: """Words, and one exact region per block per page. The region needs no merging rule: every word drawing in block n's colour belongs to block n, so the region is the union of that block's own boxes on that page. A block that breaks across a column or a page therefore yields one region per page, correctly, with no tolerance to tune. """ import pdfplumber table = {_key(b["rgb"] and [v / 255.0 for v in b["rgb"]]): b["id"] for b in sp["blocks"]} label_of = {b["id"]: b["label"] for b in sp["blocks"]} pages: List[Dict[str, Any]] = [] with pdfplumber.open(annot_pdf) as pdf: for n, page in enumerate(pdf.pages, 1): els: List[Dict[str, Any]] = [] spans: Dict[int, List[float]] = {} def note(bid: int, box: Sequence[float]) -> None: cur = spans.get(bid) if cur is None: spans[bid] = list(box) else: cur[0] = min(cur[0], box[0]); cur[1] = min(cur[1], box[1]) cur[2] = max(cur[2], box[2]); cur[3] = max(cur[3], box[3]) for w in page.extract_words(extra_attrs=["non_stroking_color"]): bid = resolve(w.get("non_stroking_color"), table) box = [round(w["x0"], 2), round(w["top"], 2), round(w["x1"], 2), round(w["bottom"], 2)] els.append({"label": label_of.get(bid, "unlabelled"), "text": w["text"], "kind": "word", "bbox": box, **({"block": bid} if bid is not None else {})}) if bid is not None: note(bid, box) for r in page.rects: bid = resolve(r.get("non_stroking_color"), table) if bid is not None: note(bid, [round(r["x0"], 2), round(r["top"], 2), round(r["x1"], 2), round(r["bottom"], 2)]) regions = [{"label": label_of[bid], "kind": "region", "block": bid, "bbox": [round(v, 2) for v in box], "text": None} for bid, box in sorted(spans.items())] els.extend(regions) pages.append({"page": n, "width": round(page.width, 2), "height": round(page.height, 2), "elements": els, "regions": regions}) return pages def generate(seed: int, out_dir: Optional[Path] = None, *, keep: bool = False, **kw: Any) -> Dict[str, Any]: """One synthetic document with exact ground truth, or an honest refusal. The returned shape is the annotator's, so `/ui/layout/load`, the viewer and every exporter take it without knowing which coordinate produced it. """ sp = spec(seed, **kw) src = latex(sp) name = "gen_{}".format(seed) have = available() base = {"schema": SCHEMA, "coordinate": COORD, "template": name, "documentclass": "article", "profile": "generated", "seed": seed, "spec": sp, "source": src, "labels": sorted(set(LABELS) | {"unlabelled"})} if not have["ready"]: return {**base, "pages": [], "runs": "nowhere", "runtime": have, "why": "this document is written; " + " and ".join(have["missing"]) + (" are" if len(have["missing"]) > 1 else " is") + " not here to render it", "action": "run this where the mesh image is — it carries latexmk"} import tempfile work = Path(out_dir) if out_dir else Path(tempfile.mkdtemp(prefix="smlayout_")) work.mkdir(parents=True, exist_ok=True) try: for fn, text in src.items(): (work / fn).write_text(text, encoding="utf-8") clean_pdf = _compile(work, "clean.tex") annot_pdf = _compile(work, "annot.tex") fid = _fidelity(clean_pdf, annot_pdf) pages = _extract(annot_pdf, sp) if fid["ok"] else [] doc = {**base, "runs": "hosted", "fidelity": fid, "pages": pages, "pdf": str(clean_pdf) if keep else None, "counts": _counts([e for p in pages for e in p["regions"]])} if not fid["ok"]: doc["why"] = ("the coloured twin does not match the clean page, so its " "boxes do not describe this document — no labels emitted") return doc finally: if not keep and not out_dir: shutil.rmtree(work, ignore_errors=True) def batch(seeds: Sequence[int], **kw: Any) -> Dict[str, Any]: """A run. Documents that fail the gate are reported, never quietly kept. `rejected` means the fidelity gate refused it. Not running at all is a different answer and gets its own key, because "your labels are wrong" and "there is no TeX here" call for opposite actions. """ seeds = list(seeds) docs = [generate(s, **kw) for s in seeds] ok = [d for d in docs if (d.get("fidelity") or {}).get("ok")] return {"schema": SCHEMA, "coordinate": COORD, "requested": len(seeds), "passed": len(ok), "documents": docs, "counts": _counts([e for d in ok for p in d["pages"] for e in p["regions"]]), "rejected": [d["template"] for d in docs if d.get("fidelity") and not d["fidelity"]["ok"]], "not_run": [d["template"] for d in docs if d["runs"] == "nowhere"]} if __name__ == "__main__": # colour round trip: every block distinguishable after a PDF's rounding seen = {_key([v / 255.0 for v in colour_of(i)]) for i in range(4000)} assert len(seen) == 4000, len(seen) assert all(len(set(colour_of(i))) > 1 for i in range(4000)), "a grey is unreadable" tbl = {_key([v / 255.0 for v in colour_of(i)]): i for i in range(50)} assert resolve([c / 255.0 for c in colour_of(7)], tbl) == 7 assert resolve(0.0, tbl) is None and resolve(None, tbl) is None assert resolve((0, 0, 0, 1), tbl) is None # CMYK carries no block # the same seed is the same document; a different seed is not a, b = spec(11), spec(11) assert json.dumps(a, sort_keys=True) == json.dumps(b, sort_keys=True) assert json.dumps(spec(12), sort_keys=True) != json.dumps(a, sort_keys=True) assert [x["id"] for x in a["blocks"]] == list(range(len(a["blocks"]))) # weakness biases what gets generated, and does it in the right direction strong = spec(5, sections=8, blocks_per_section=8, weakness={k: 0.99 for k in SAMPLED if k != "table"}) assert strong["label_counts"]["table"] > sum( v for k, v in strong["label_counts"].items() if k in SAMPLED and k != "table") flat = weights(None) assert set(flat) == set(SAMPLED) and len(set(flat.values())) == 1 assert weights({"table": 0.0})["table"] > weights({"table": 1.0})["table"] assert weights({"headline": 0.0}) == flat, "structural labels are not sampled" # a charted folder's prose reaches the page, and its scores steer sampling real = ["The corridor was measured twice before anyone signed for it.", "A second inspection found the same drift in the opposite direction.", "Nobody could say which of the two surveys had been resurveyed."] sourced = spec(4, sections=4, blocks_per_section=6, source={"from": "drive.corpus", "prose": real, "weakness": {"table": 0.0, "paragraph": 0.99}}) assert sourced["sourced"] and sourced["source"] == "drive.corpus" page = latex(sourced)["clean.tex"] assert "corridor was measured twice" in page, "sourced prose never reached the page" assert not any(w in page for w in ("viaduct", "bastion")), \ "the word bank must not leak in once there is real prose" assert sourced["label_counts"].get("table", 0) > \ sourced["label_counts"].get("paragraph", 0), sourced["label_counts"] # an argument beats a file somebody dropped in a folder both = spec(4, weakness={"table": 0.99}, source={"weakness": {"table": 0.0}}) assert both["weakness"] == {"table": 0.99}, both["weakness"] # a word processor's punctuation cannot reach pdflatex assert _ascii("it’s — “here”…") == 'it\'s --- "here"...' assert _ascii("你好") == "", "unrenderable glyphs are dropped, not passed on" # both files carry the identical body and differ only in the two macros src = latex(a) assert set(src) == {"clean.tex", "annot.tex"} for fn in src: assert src[fn].count(r"\begin{document}") == 1 body_of = lambda t: t.split(r"\begin{document}", 1)[1] assert body_of(src["clean.tex"]) == body_of(src["annot.tex"]), "the pages must match" assert r"\textcolor" in src["annot.tex"] and r"\textcolor" not in src["clean.tex"] for blk in a["blocks"]: assert (r"\definecolor{SM" + str(blk["id"]) + "}") in src["annot.tex"] assert "twocolumn" in latex({**a, "twocolumn": True})["clean.tex"] # every block kind emits, and nothing escapes unescaped into the source every = {"paragraph", "list", "table", "figure", "equation", "quote", "code", "reference", "title", "byline", "date", "abstract", "subhead"} rng = random.Random(0) for kind in every - {"title", "byline", "date", "abstract", "subhead"}: blk = {**_block(kind, rng), "id": 3} src_ = _body_of(blk) assert any(m + "{SM3}" in src_ for m in (r"\SMb", r"\color")), kind assert _esc("100% & {x}_y") == r"100\% \& \{x\}\_y" try: _body_of({"kind": "hologram", "id": 0, "label": "x"}) raise AssertionError("an unknown block kind must not silently vanish") except ValueError as expected: assert "hologram" in str(expected), expected # with no runtime it hands back the document and says where it runs have = available() d = generate(3) if not have["ready"]: assert d["runs"] == "nowhere" and d["pages"] == [] assert d["source"]["annot.tex"], "the source is the deliverable when TeX is not" assert have["missing"] and have["missing"][0] in d["why"] print("layout_generate self-check OK (source only: missing " + ", ".join(have["missing"]) + ")") else: assert d["runs"] == "hosted" and d["fidelity"]["ok"], d.get("fidelity") assert d["pages"] and d["counts"], "a passing run has regions" blocks = {e["block"] for p in d["pages"] for e in p["regions"]} assert blocks <= {b["id"] for b in d["spec"]["blocks"]} for p in d["pages"]: for e in p["regions"]: x0, top, x1, bottom = e["bbox"] assert x1 > x0 and bottom > top, e assert 0 <= x0 and x1 <= p["width"] + 1, e # Every block kind, in one document, on real pages. Each of these has # cost a run before: a colour whatsit shielding glue TeX would have # discarded, a font change grouped on one side only, a paragraph # started by \textcolor's \leavevmode instead of by \parindent. They # all present as drift, so one composite document catches the lot. full = generate(99, sections=6, blocks_per_section=6) assert full["fidelity"]["ok"], full["fidelity"] seen = {e["label"] for p in full["pages"] for e in p["regions"]} assert set(SAMPLED) <= seen, sorted(set(SAMPLED) - seen) stray = [e for p in full["pages"] for e in p["elements"] if e["kind"] == "word" and e["label"] == "unlabelled"] assert not stray, ("every glyph belongs to a block, or a region clips " "its own content: " + str([e["text"] for e in stray[:8]])) r = batch([1, 2]) assert r["passed"] == 2 and not r["rejected"], r["rejected"] print("layout_generate self-check OK (compiled: " + str(len(d["pages"])) + " pages, drift " + str(d["fidelity"]["max_drift_pt"]) + "pt, " + str(d["counts"]) + ")")