Spaces:
Running
Running
| """ | |
| 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"]) + ")") | |