"""Desktop agent: one X display with a real design app, driven over HTTP (runs inside the desktop container). GET /health GET /screenshot -> PNG of the 1280x800 screen POST /input {"type": "click"|"double"|"triple"|"right"|"move"|"drag"|"type"|"key"|"scroll", ...} POST /reset {"app": "impress"|"inkscape"|"gimp", "width", "height", "background", "assets": {id: base64}} POST /read -> {"width", "height", "background", "elements": [DesignDoc element dicts], "skipped": n} POST /expert {"doc": DesignDoc dict} -> builds the design in the app (for the score-1.0 check) The env turns /read into a DesignDoc and scores it; everything app-specific lives in the App classes below. Coordinates in documents are canvas pixels; each app's own unit is converted at this boundary. """ from __future__ import annotations import argparse import base64 import io import json import math import os import re import shutil import socket import subprocess import threading import time import xml.etree.ElementTree as ET from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path W, H = 1280, 800 DISPLAY = os.environ.get("DISPLAY", ":1") HOME = Path.home() PICS = HOME / "Pictures" WORK = Path("/work") LOCK = threading.Lock() def run(cmd: list[str], check=False, timeout=30, **kw) -> subprocess.CompletedProcess: return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=check, env={**os.environ, "DISPLAY": DISPLAY}, **kw) # ---------------------------------------------------------------------------------------------- display def start_display(): n = DISPLAY.lstrip(":") for stale in (Path(f"/tmp/.X{n}-lock"), Path(f"/tmp/.X11-unix/X{n}")): # left by a restarted container stale.unlink(missing_ok=True) subprocess.Popen(["Xvfb", DISPLAY, "-screen", "0", f"{W}x{H}x24", "-nolisten", "tcp"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(100): if run(["xdpyinfo"]).returncode == 0: break time.sleep(0.1) cfg = HOME / ".config/openbox" cfg.mkdir(parents=True, exist_ok=True) # every application window opens maximised and undecorated, like a full-screen app on a laptop (cfg / "rc.xml").write_text(""" yesno """) subprocess.Popen(["openbox"], env={**os.environ, "DISPLAY": DISPLAY}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) fonts = Path(os.environ.get("DESIGN_CANVAS_FONTS", "/fonts")) # mounted locally, uploaded in a sandbox if fonts.is_dir(): # the task pack's fonts, so the apps offer the families the designs use dst = HOME / ".local/share/fonts/pack" if not dst.exists(): dst.parent.mkdir(parents=True, exist_ok=True) dst.symlink_to(fonts) run(["fc-cache", "-f"], timeout=300) def screenshot() -> bytes: from PIL import ImageGrab img = ImageGrab.grab(xdisplay=DISPLAY) buf = io.BytesIO() img.convert("RGB").save(buf, "PNG") return buf.getvalue() XKEYS = {"ControlOrMeta": "ctrl", "Control": "ctrl", "Meta": "super", "Alt": "alt", "Shift": "shift", "Enter": "Return", "Escape": "Escape", "Backspace": "BackSpace", "Delete": "Delete", "Tab": "Tab", " ": "space", "ArrowUp": "Up", "ArrowDown": "Down", "ArrowLeft": "Left", "ArrowRight": "Right", "Home": "Home", "End": "End", "PageUp": "Prior", "PageDown": "Next", "+": "plus", "-": "minus"} def xkey(combo: str) -> str: """Playwright spelling ('ControlOrMeta+Shift+z') -> xdotool ('ctrl+shift+z').""" return "+".join(XKEYS.get(p, p) for p in combo.split("+") if p) def do_input(a: dict) -> dict: t = a["type"] xd = ["xdotool"] if t in ("click", "double", "triple", "right"): mods = [xkey(m) for m in a.get("modifiers") or []] for m in mods: run(xd + ["keydown", m]) n = {"click": 1, "double": 2, "triple": 3, "right": 1}[t] run(xd + ["mousemove", str(int(a["x"])), str(int(a["y"])), "click", "--repeat", str(n), "--delay", "60", "3" if t == "right" else "1"]) for m in reversed(mods): run(xd + ["keyup", m]) elif t == "move": run(xd + ["mousemove", str(int(a["x"])), str(int(a["y"]))]) elif t == "drag": (x0, y0), (x1, y1) = a["from"], a["to"] run(xd + ["mousemove", str(int(x0)), str(int(y0)), "mousedown", "1"]) for k in range(1, 11): # a real drag passes through intermediate points run(xd + ["mousemove", str(int(x0 + (x1 - x0) * k / 10)), str(int(y0 + (y1 - y0) * k / 10))]) time.sleep(0.02) run(xd + ["mouseup", "1"]) elif t == "type": run(xd + ["type", "--delay", "12", "--", a["text"]], timeout=120) elif t == "key": run(xd + ["key", "--delay", "40"] + [xkey(k) for k in a["keys"]]) elif t == "scroll": run(xd + ["mousemove", str(int(a["x"])), str(int(a["y"]))]) dy, dx = int(a.get("dy", 0)), int(a.get("dx", 0)) for button, n in ((5 if dy > 0 else 4, abs(dy)), (7 if dx > 0 else 6, abs(dx))): if n: run(xd + ["click", "--repeat", str(n), "--delay", "30", str(button)]) else: raise ValueError(f"unknown input type {t!r}") time.sleep(0.25) return {"ok": True} def wait_window(pattern: str, timeout: float = 90) -> bool: t0 = time.time() while time.time() - t0 < timeout: r = run(["xdotool", "search", "--onlyvisible", "--name", pattern]) if r.returncode == 0 and r.stdout.strip(): return True time.sleep(0.5) return False def fill_screen(pattern: str) -> None: """Give the app's main window (its largest visible one) the whole screen: an app restores its own last window size whatever the window manager says, so the agent would otherwise see a different layout per desktop.""" best = None for wid in run(["xdotool", "search", "--onlyvisible", "--name", pattern]).stdout.split(): m = re.search(r"Geometry: (\d+)x(\d+)", run(["xdotool", "getwindowgeometry", wid]).stdout) if m and (best is None or int(m[1]) * int(m[2]) > best[1]): best = (wid, int(m[1]) * int(m[2])) if best and best[1] < W * H * 0.95: run(["xdotool", "windowmove", best[0], "0", "0"]) run(["xdotool", "windowsize", best[0], str(W), str(H)]) time.sleep(0.5) def dismiss(titles: tuple[str, ...]) -> int: """Close windows by title (startup dialogs an app shows no matter its settings); returns how many.""" n = 0 for title in titles: for wid in run(["xdotool", "search", "--name", title]).stdout.split(): run(["xdotool", "windowactivate", "--sync", wid]) run(["xdotool", "key", "--window", wid, "Escape"]) run(["xdotool", "windowclose", wid]) n += 1 if n: time.sleep(0.5) return n GIMP_DIALOGS = ("Welcome to GIMP", "GIMP Message") def kill_apps(): for name in ("soffice.bin", "inkscape", "gimp", "gimp-3.0"): run(["pkill", "-9", "-x", name]) # GIMP's plug-ins (the Script-Fu server among them) are separate processes that can outlive it run(["pkill", "-9", "-f", "/gimp/3.0/plug-ins/"]) for _ in range(40): # and the server's port must be free before the next GIMP binds it try: socket.create_connection(("127.0.0.1", Gimp.port), timeout=0.3).close() time.sleep(0.25) except OSError: break time.sleep(0.5) # ---------------------------------------------------------------------------------------------- assets def write_assets(assets: dict[str, str]) -> dict[str, Path]: """Task images as PNG files named after their asset ids, in ~/Pictures (where file dialogs open).""" from PIL import Image shutil.rmtree(PICS, ignore_errors=True) PICS.mkdir(parents=True) out = {} for aid, b64 in assets.items(): img = Image.open(io.BytesIO(base64.b64decode(b64))) p = PICS / f"{aid}.png" img.save(p, "PNG") out[aid] = p return out class Matcher: """Map an image an app embedded (no filename left) back to the task asset it came from, by pixels.""" def __init__(self, paths: dict[str, Path]): from PIL import Image self.sig = {aid: self._sig(Image.open(p)) for aid, p in paths.items()} @staticmethod def _sig(img): import numpy as np return np.asarray(img.convert("RGBA").resize((24, 24)), dtype="float32") / 255.0 def best(self, img) -> str | None: import numpy as np s = self._sig(img) scores = {aid: float(np.abs(s - t).mean()) for aid, t in self.sig.items()} aid = min(scores, key=scores.get) if scores else None return aid if aid is not None and scores[aid] < 0.08 else None # ---------------------------------------------------------------------------------------------- Impress (UNO) MM100_PER_PX = 2540 / 96 # 1/100 mm per CSS pixel class Impress: """LibreOffice Impress, one blank slide the size of the canvas; read back through the UNO API.""" def __init__(self): self.ctx = self.doc = None def _connect(self): import uno local = uno.getComponentContext() resolver = local.ServiceManager.createInstanceWithContext("com.sun.star.bridge.UnoUrlResolver", local) for _ in range(120): try: self.ctx = resolver.resolve("uno:socket,host=127.0.0.1,port=2002;urp;StarOffice.ComponentContext") return except Exception: # noqa: BLE001 - soffice still starting time.sleep(0.5) raise RuntimeError("LibreOffice did not open its UNO socket") def _desktop(self): return self.ctx.ServiceManager.createInstanceWithContext("com.sun.star.frame.Desktop", self.ctx) PROFILE = Path("/tmp/lo-profile") # a fresh profile per episode: no stale lock from a killed instance, no first-run dialogs, the same UI every time SETTINGS = """ false false true false """ def reset(self, width, height, background, paths): shutil.rmtree(self.PROFILE, ignore_errors=True) (self.PROFILE / "user").mkdir(parents=True) (self.PROFILE / "user" / "registrymodifications.xcu").write_text(self.SETTINGS) subprocess.Popen(["soffice", f"-env:UserInstallation={self.PROFILE.as_uri()}", "--impress", "--norestore", "--nologo", "--nodefault", "--accept=socket,host=127.0.0.1,port=2002;urp;"], env={**os.environ, "DISPLAY": DISPLAY, "SAL_USE_VCLPLUGIN": "gtk3"}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) self._connect() from com.sun.star.beans import PropertyValue # type: ignore[import-not-found] self.doc = self._desktop().loadComponentFromURL("private:factory/simpress", "_blank", 0, ()) page = self.doc.DrawPages.getByIndex(0) page.Layout = 20 # blank layout: no title / content placeholders while page.Count: page.remove(page.getByIndex(0)) page.Width, page.Height = round(width * MM100_PER_PX), round(height * MM100_PER_PX) page.BorderLeft = page.BorderRight = page.BorderTop = page.BorderBottom = 0 self._set_background(page, background) self.paths = paths self.matcher = Matcher(paths) self.size = (width, height) del PropertyValue wait_window("Impress", 60) time.sleep(1.5) fill_screen("Impress") def _set_background(self, page, color): bg = self.doc.createInstance("com.sun.star.drawing.Background") bg.FillStyle = uno_enum("com.sun.star.drawing.FillStyle", "SOLID") bg.FillColor = int(color.lstrip("#"), 16) page.Background = bg def _graphic_asset(self, graphic) -> str | None: """Which task asset a placed graphic is: LibreOffice keeps the file it was loaded from (OriginURL); otherwise export it to a file and match it by pixels.""" from PIL import Image origin = "" try: origin = graphic.OriginURL or "" except Exception: # noqa: BLE001 pass stem = Path(origin.replace("file://", "")).stem if origin else "" if stem in self.paths: return stem gp = self.ctx.ServiceManager.createInstanceWithContext("com.sun.star.graphic.GraphicProvider", self.ctx) tmp = WORK / "graphic-export.png" tmp.unlink(missing_ok=True) gp.storeGraphic(graphic, (pv("URL", tmp.as_uri()), pv("MimeType", "image/png"))) return self.matcher.best(Image.open(tmp)) if tmp.exists() and tmp.stat().st_size else None def read(self): page = self.doc.DrawPages.getByIndex(0) w, h = self.size bg = "#ffffff" try: b = page.Background if b is not None and int(b.FillColor) >= 0: bg = f"#{int(b.FillColor) & 0xFFFFFF:06x}" except Exception: # noqa: BLE001 pass els, skipped = [], 0 def flat(container): # groups are flattened; their members keep page coordinates for i in range(container.Count): s = container.getByIndex(i) if s.ShapeType == "com.sun.star.drawing.GroupShape": yield from flat(s) else: yield s for s in flat(page): kind = s.ShapeType if kind == "com.sun.star.drawing.CustomShape": # what the Basic Shapes toolbar draws geo = {p.Name: p.Value for p in s.CustomShapeGeometry}.get("Type", "") kind = ("com.sun.star.drawing.EllipseShape" if geo in ("ellipse", "circle") else "com.sun.star.drawing.RectangleShape" if geo in ("rectangle", "round-rectangle", "square", "round-square", "quadrat") else kind) pos, size = s.Position, s.Size common = {"x": pos.X / MM100_PER_PX, "y": pos.Y / MM100_PER_PX, "width": max(size.Width / MM100_PER_PX, 0.5), "height": max(size.Height / MM100_PER_PX, 0.5), "angle": -(getattr(s, "RotateAngle", 0) or 0) / 100.0} try: common["opacity"] = 1 - (getattr(s, "FillTransparence", 0) or 0) / 100.0 except Exception: # noqa: BLE001 pass if kind == "com.sun.star.drawing.GraphicObjectShape": aid = self._graphic_asset(s.Graphic) if aid is None: skipped += 1 continue els.append({"kind": "image", "asset_id": aid, **common}) elif kind in ("com.sun.star.drawing.TextShape", "com.sun.star.presentation.TitleTextShape", "com.sun.star.presentation.SubtitleShape", "com.sun.star.presentation.OutlinerShape"): text = s.getString().replace("\r\n", "\n").replace("\r", "\n") if not text.strip(): skipped += 1 continue els.append({"kind": "text", **_read_text(s), **common}) elif kind in ("com.sun.star.drawing.RectangleShape", "com.sun.star.drawing.EllipseShape"): fill = f"#{int(s.FillColor) & 0xFFFFFF:06x}" els.append({"kind": "shape", "shape": "ellipse" if "Ellipse" in kind else "rect", "fill": fill, **common}) else: skipped += 1 return {"width": w, "height": h, "background": bg, "elements": els, "skipped": skipped} def expert(self, doc): page = self.doc.DrawPages.getByIndex(0) self._set_background(page, doc.get("background", "#ffffff")) for e in doc["elements"]: if e["kind"] == "image": s = self.doc.createInstance("com.sun.star.drawing.GraphicObjectShape") page.add(s) gp = self.ctx.ServiceManager.createInstanceWithContext("com.sun.star.graphic.GraphicProvider", self.ctx) s.Graphic = gp.queryGraphic((pv("URL", self.paths[e["asset_id"]].as_uri()),)) elif e["kind"] == "shape": s = self.doc.createInstance("com.sun.star.drawing." + ("EllipseShape" if e.get("shape") == "ellipse" else "RectangleShape")) page.add(s) s.FillColor = int((e.get("fill") or "#000000").lstrip("#"), 16) s.LineStyle = uno_enum("com.sun.star.drawing.LineStyle", "NONE") else: s = self.doc.createInstance("com.sun.star.drawing.TextShape") page.add(s) s.TextAutoGrowWidth = True s.TextAutoGrowHeight = True _write_text(s, e) set_geom(s, e) time.sleep(0.5) def _hexcolor(v) -> str: return f"#{int(v) & 0xFFFFFF:06x}" if int(v) >= 0 else "#000000" def _style_of(r) -> dict: return {"color": _hexcolor(r.CharColor), "bold": float(r.CharWeight) >= 150, "italic": _ev(r.CharPosture) not in (0, "NONE")} def _read_text(shape) -> dict: """A text shape's string, typography and styled runs (one run per portion that differs in colour/weight).""" paras = shape.Text.createEnumeration() runs, first_para = [], None while paras.hasMoreElements(): para = paras.nextElement() if first_para is None: first_para = para elif runs: runs.append({"text": "\n", **{k: runs[-1][k] for k in ("color", "bold", "italic")}}) portions = para.createEnumeration() while portions.hasMoreElements(): por = portions.nextElement() t = por.getString() if t: runs.append({"text": t, **_style_of(por)}) text = "".join(r["text"] for r in runs) cur = shape.Text.createTextCursor() cur.gotoStart(False) font_px = float(cur.CharHeight) * 96 / 72 ls = getattr(first_para, "ParaLineSpacing", None) if first_para is not None else None mode = _ev(ls.Mode) if ls is not None else 0 if mode in (3, "FIX") and ls.Height: # a fixed line height in 1/100 mm lh = ls.Height / MM100_PER_PX / font_px elif mode in (0, "PROP") and ls is not None: # proportional: 100% is the font's own line height (~1.17 em) lh = ls.Height / 100.0 * 1.17 else: lh = 1.17 adjust = {0: "left", 1: "right", 3: "center", "LEFT": "left", "RIGHT": "right", "CENTER": "center"}.get( _ev(getattr(first_para, "ParaAdjust", 0)), "left") styles = {(r["color"], r["bold"], r["italic"]) for r in runs if r["text"].strip()} return {"text": text, "font_family": cur.CharFontName, "font_size": font_px, **(_style_of(cur)), "underline": int(_ev(cur.CharUnderline) or 0) != 0, "align": adjust, "line_height": lh, "letter_spacing": float(getattr(cur, "CharKerning", 0) or 0) / MM100_PER_PX, "uppercase": _ev(getattr(cur, "CharCaseMap", 0)) in (1, "UPPERCASE"), "runs": runs if len(styles) > 1 else None} def _write_text(shape, e: dict) -> None: import uno shape.setString(e["text"]) font_px = float(e.get("font_size", 48)) cur = shape.Text.createTextCursor() cur.gotoStart(False) cur.gotoEnd(True) cur.CharFontName = e.get("font_family") or "DejaVu Sans" cur.CharHeight = font_px * 72 / 96 cur.CharColor = int((e.get("color") or "#000000").lstrip("#"), 16) cur.CharWeight = 150.0 if e.get("bold") else 100.0 cur.CharPosture = uno.Enum("com.sun.star.awt.FontSlant", "ITALIC" if e.get("italic") else "NONE") cur.CharKerning = round(float(e.get("letter_spacing", 0)) * MM100_PER_PX) cur.CharCaseMap = 1 if e.get("uppercase") else 0 cur.ParaAdjust = {"left": 0, "right": 1, "center": 3}[e.get("align", "left")] spacing = uno.createUnoStruct("com.sun.star.style.LineSpacing") spacing.Mode, spacing.Height = 3, round(float(e.get("line_height", 1.0)) * font_px * MM100_PER_PX) cur.ParaLineSpacing = spacing pos = 0 for run in e.get("runs") or []: # colour / weight per styled stretch n = len(run["text"]) c = shape.Text.createTextCursor() c.gotoStart(False) c.goRight(pos, False) c.goRight(n, True) if run.get("color"): c.CharColor = int(run["color"].lstrip("#"), 16) if run.get("bold") is not None: c.CharWeight = 150.0 if run["bold"] else 100.0 if run.get("italic") is not None: c.CharPosture = uno.Enum("com.sun.star.awt.FontSlant", "ITALIC" if run["italic"] else "NONE") pos += n def _ev(v): """A UNO enum's name, or the plain value (UNO hands back some properties as enums, some as ints).""" return v.value if hasattr(v, "value") and not isinstance(v, (int, float, str)) else v def pv(name, value): import uno p = uno.createUnoStruct("com.sun.star.beans.PropertyValue") p.Name, p.Value = name, value return p def uno_enum(typ, value): import uno return uno.Enum(typ, value) def set_geom(shape, e): import uno pos = uno.createUnoStruct("com.sun.star.awt.Point") size = uno.createUnoStruct("com.sun.star.awt.Size") pos.X, pos.Y = round(e["x"] * MM100_PER_PX), round(e["y"] * MM100_PER_PX) size.Width, size.Height = round(e["width"] * MM100_PER_PX), round(e["height"] * MM100_PER_PX) shape.Size = size shape.Position = pos if e.get("angle"): shape.RotateAngle = round(-e["angle"] * 100) % 36000 # ---------------------------------------------------------------------------------------------- Inkscape (SVG) SVG = "{http://www.w3.org/2000/svg}" XLINK = "{http://www.w3.org/1999/xlink}" class Inkscape: """Inkscape with an SVG page the size of the canvas; read back by saving (Ctrl+S) and parsing the SVG.""" path = WORK / "design.svg" def reset(self, width, height, background, paths): self.size, self.paths, self.matcher = (width, height), paths, Matcher(paths) self.path.write_text(_svg(width, height, background, [])) subprocess.Popen(["inkscape", str(self.path)], env={**os.environ, "DISPLAY": DISPLAY}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) wait_window("Inkscape", 120) time.sleep(4) fill_screen("Inkscape") def save(self): before = self.path.stat().st_mtime win = run(["xdotool", "search", "--onlyvisible", "--name", "Inkscape"]).stdout.split() if win: run(["xdotool", "windowactivate", "--sync", win[-1]]) run(["xdotool", "key", "ctrl+s"]) for _ in range(40): time.sleep(0.25) if self.path.stat().st_mtime != before: break time.sleep(0.5) def read(self): self.save() return parse_svg(self.path.read_text(), self.size, self.matcher) def expert(self, doc): # write the design as SVG and reopen it in Inkscape (then a read saves through the app and re-parses) kill_apps() self.path.write_text(_svg(*self.size, doc.get("background", "#ffffff"), doc["elements"], self.paths)) subprocess.Popen(["inkscape", str(self.path)], env={**os.environ, "DISPLAY": DISPLAY}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) wait_window("Inkscape", 120) time.sleep(4) fill_screen("Inkscape") def _svg(width, height, background, elements, paths=None) -> str: parts = [] for e in elements: tr = "" if e.get("angle"): cx, cy = e["x"] + e["width"] / 2, e["y"] + e["height"] / 2 tr = f' transform="rotate({e["angle"]} {cx} {cy})"' op = f' opacity="{e.get("opacity", 1)}"' if e.get("opacity", 1) != 1 else "" if e["kind"] == "image": parts.append(f'') elif e["kind"] == "shape": tag = "ellipse" if e.get("shape") == "ellipse" else "rect" geo = (f'cx="{e["x"] + e["width"] / 2}" cy="{e["y"] + e["height"] / 2}" rx="{e["width"] / 2}" ' f'ry="{e["height"] / 2}"' if tag == "ellipse" else f'x="{e["x"]}" y="{e["y"]}" width="{e["width"]}" height="{e["height"]}"') parts.append(f'<{tag} {geo} fill="{e.get("fill") or "#000000"}"{tr}{op}/>') else: fs = float(e.get("font_size", 48)) step = float(e.get("_step") or float(e.get("line_height", 1.0)) * fs) base = float(e.get("_baseline") or e["y"] + fs * 0.8) align = e.get("align", "left") anchor = {"left": "start", "center": "middle", "right": "end"}[align] ax = float(e.get("_anchor_x") or {"left": e["x"], "center": e["x"] + e["width"] / 2, "right": e["x"] + e["width"]}[align]) style = (f'font-family:{e.get("font_family") or "DejaVu Sans"};font-size:{fs}px;' f'font-weight:{"bold" if e.get("bold") else "normal"};' f'font-style:{"italic" if e.get("italic") else "normal"};fill:{e.get("color") or "#000000"};' f'text-anchor:{anchor};letter-spacing:{float(e.get("letter_spacing", 0))}px;' f'line-height:{step / fs if fs else 1.0};' f'text-transform:{"uppercase" if e.get("uppercase") else "none"}') spans = "".join(f'{_line_runs(line)}' for k, line in enumerate(_lines_of(e))) parts.append(f'{spans}') return (f'\n' f'' + "".join(parts) + "") def _lines_of(e: dict) -> list[list[dict]]: """A text element as lines of styled pieces (its runs, or one piece per line).""" runs = e.get("runs") or [{"text": e["text"]}] lines: list[list[dict]] = [[]] for r in runs: for k, part in enumerate(r["text"].split("\n")): if k: lines.append([]) if part: lines[-1].append({**r, "text": part}) return lines def _line_runs(pieces: list[dict]) -> str: out = [] for p in pieces: st = [] if p.get("color"): st.append(f"fill:{p['color']}") if p.get("bold") is not None: st.append(f"font-weight:{'bold' if p['bold'] else 'normal'}") if p.get("italic") is not None: st.append(f"font-style:{'italic' if p['italic'] else 'normal'}") out.append(f'{_esc(p["text"])}' if st else _esc(p["text"])) return "".join(out) def _esc(s: str) -> str: return s.replace("&", "&").replace("<", "<").replace(">", ">") def _matrix(tr: str | None): """An SVG transform attribute as a 2x3 affine (a, b, c, d, e, f).""" m = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0) for name, args in re.findall(r"(\w+)\s*\(([^)]*)\)", tr or ""): v = [float(x) for x in re.split(r"[\s,]+", args.strip()) if x] if name == "matrix": t = tuple(v) elif name == "translate": t = (1, 0, 0, 1, v[0], v[1] if len(v) > 1 else 0) elif name == "scale": t = (v[0], 0, 0, v[1] if len(v) > 1 else v[0], 0, 0) elif name == "rotate": r = math.radians(v[0]) c, s = math.cos(r), math.sin(r) t = (c, s, -s, c, 0, 0) if len(v) == 3: t = _mul(_mul((1, 0, 0, 1, v[1], v[2]), t), (1, 0, 0, 1, -v[1], -v[2])) else: continue m = _mul(m, t) return m def _mul(m, n): a, b, c, d, e, f = m a2, b2, c2, d2, e2, f2 = n return (a * a2 + c * b2, b * a2 + d * b2, a * c2 + c * d2, b * c2 + d * d2, a * e2 + c * f2 + e, b * e2 + d * f2 + f) def _place(m, x, y, w, h): """Box (x, y, w, h) under affine m -> DesignDoc geometry (unrotated box + angle about its centre).""" a, b, c, d, e, f = m sx, sy = math.hypot(a, b), math.hypot(c, d) angle = math.degrees(math.atan2(b, a)) cx, cy = x + w / 2, y + h / 2 ccx, ccy = a * cx + c * cy + e, b * cx + d * cy + f W2, H2 = w * sx, h * sy return {"x": ccx - W2 / 2, "y": ccy - H2 / 2, "width": max(W2, 0.5), "height": max(H2, 0.5), "angle": angle} def _style(el) -> dict: st = dict(kv.split(":", 1) for kv in (el.get("style") or "").split(";") if ":" in kv) for k in ("fill", "font-family", "font-size", "font-weight", "font-style", "text-anchor", "opacity"): if el.get(k) is not None and k not in st: st[k] = el.get(k) return {k.strip(): v.strip() for k, v in st.items()} def _num(v, default=0.0) -> float: m = re.match(r"-?[\d.]+", str(v or "")) return float(m.group()) if m else default def parse_svg(text: str, size, matcher: Matcher) -> dict: from PIL import Image root = ET.fromstring(text) w, h = size els, skipped = [], 0 bg = "#ffffff" nv = root.find("{http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd}namedview") if nv is not None and nv.get("pagecolor"): bg = nv.get("pagecolor") def walk(node, m, inherited): nonlocal skipped, bg for el in node: tag = el.tag.replace(SVG, "") mm = _mul(m, _matrix(el.get("transform"))) st = {**inherited, **_style(el)} op = float(st.get("opacity", 1) or 1) if tag == "g": walk(el, mm, st) elif el.get("id") == "canvas-background": bg = st.get("fill", bg) elif tag == "image": href = el.get(XLINK + "href") or el.get("href") or "" aid = None if href.startswith("data:"): aid = matcher.best(Image.open(io.BytesIO(base64.b64decode(href.split(",", 1)[1])))) else: stem = Path(href.replace("file://", "")).stem aid = stem if stem in matcher.sig else None if aid is None: skipped += 1 continue els.append({"kind": "image", "asset_id": aid, "opacity": op, **_place(mm, _num(el.get("x")), _num(el.get("y")), _num(el.get("width")), _num(el.get("height")))}) elif tag in ("rect", "ellipse", "circle"): if tag == "rect": geo = (_num(el.get("x")), _num(el.get("y")), _num(el.get("width")), _num(el.get("height"))) else: rx = _num(el.get("rx") or el.get("r")) ry = _num(el.get("ry") or el.get("r")) geo = (_num(el.get("cx")) - rx, _num(el.get("cy")) - ry, 2 * rx, 2 * ry) fill = st.get("fill", "#000000") if not fill.startswith("#"): skipped += 1 continue els.append({"kind": "shape", "shape": "rect" if tag == "rect" else "ellipse", "fill": fill, "opacity": op, **_place(mm, *geo)}) elif tag == "text": spans = [t for t in el if t.tag.replace(SVG, "") == "tspan"] or [el] s0 = {**st, **_style(spans[0])} fs = _num(s0.get("font-size"), 16) runs = [] for k, line in enumerate(spans): if k and runs: runs.append({**{q: runs[-1][q] for q in ("color", "bold", "italic")}, "text": "\n"}) lst = {**s0, **_style(line)} pieces = [(line.text or "", lst)] + [ p for c in line for p in (("".join(c.itertext()), {**lst, **_style(c)}), (c.tail or "", lst))] for txt, pst in pieces: if txt: runs.append({"text": txt, "color": pst.get("fill", "#000000"), "bold": pst.get("font-weight", "normal") in ("bold", "600", "700", "800", "900"), "italic": pst.get("font-style") == "italic"}) text = "".join(r["text"] for r in runs) if not text.strip(): continue a, b, c, d, e2, f2 = mm xs, ys = _num(spans[0].get("x") or el.get("x")), _num(spans[0].get("y") or el.get("y")) lh_style = s0.get("line-height", "") lh_style = (_num(lh_style) / 100 if lh_style.endswith("%") else _num(lh_style) / fs if lh_style.endswith("px") else _num(lh_style, 1.25)) if lh_style else 1.25 step = (_num(spans[1].get("y")) - ys) if len(spans) > 1 and spans[1].get("y") else fs * lh_style anchor = s0.get("text-anchor", "start") styles = {(r["color"], r["bold"], r["italic"]) for r in runs if r["text"].strip()} els.append({"kind": "text", "text": text, "font_family": s0.get("font-family", "").strip("'\""), "font_size": fs * math.hypot(a, b), "color": runs[0]["color"] if runs else "#000000", "bold": runs[0]["bold"] if runs else False, "italic": runs[0]["italic"] if runs else False, "align": {"start": "left", "middle": "center", "end": "right"}.get(anchor, "left"), "line_height": step / fs if fs else 1.0, "opacity": op, "letter_spacing": _num(s0.get("letter-spacing"), 0.0), "uppercase": s0.get("text-transform") == "uppercase", "runs": runs if len(styles) > 1 else None, "angle": math.degrees(math.atan2(b, a)), "_anchor_x": a * xs + c * ys + e2, "_baseline": b * xs + d * ys + f2}) elif tag in ("defs", "namedview", "metadata", "title", "desc", "style"): continue elif tag.startswith("{"): continue else: skipped += 1 walk(root, (1, 0, 0, 1, 0, 0), {}) return {"width": w, "height": h, "background": bg, "elements": els, "skipped": skipped} # ---------------------------------------------------------------------------------------------- GIMP (Script-Fu) class Gimp: """GIMP with one white image the size of the canvas; read back through a Script-Fu server.""" port = 10008 def reset(self, width, height, background, paths): from PIL import Image self.size, self.paths = (width, height), paths base = WORK / "canvas.png" Image.new("RGB", (width, height), background).save(base) # no welcome dialog, and messages to stderr rather than modal boxes that would block the app cfg = HOME / ".config/GIMP/3.0" cfg.mkdir(parents=True, exist_ok=True) (cfg / "gimprc").write_text('(show-welcome-dialog no)\n(last-run-version "3.0.4")\n(message-handler stderr)\n' '(save-session-info no)\n') # GIMP 3: batch commands need an interpreter; the server keeps listening while the GUI is used subprocess.Popen(["gimp", "--no-splash", "--batch-interpreter=plug-in-script-fu-eval", "-b", f'(plug-in-script-fu-server #:run-mode RUN-NONINTERACTIVE #:ip "127.0.0.1" ' f'#:port {self.port} #:logfile "/tmp/sf.log")', str(base)], env={**os.environ, "DISPLAY": DISPLAY}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) wait_window("GIMP", 180) for _ in range(10): # the welcome window and the batch call's message box appear a moment after time.sleep(1) dismiss(GIMP_DIALOGS) fill_screen("GIMP") for _ in range(120): try: socket.create_connection(("127.0.0.1", self.port), timeout=1).close() break except OSError: time.sleep(0.5) time.sleep(2) def _sf(self, expr: str) -> str: """Evaluate Scheme in the running GIMP (Script-Fu server protocol: 'G', len, expr -> 'G', err, len, out).""" dismiss(("GIMP Message",)) # a modal message box would hold the call data = expr.encode() with socket.create_connection(("127.0.0.1", self.port), timeout=60) as s: s.sendall(b"G" + len(data).to_bytes(2, "big") + data) head = b"" while len(head) < 4: head += s.recv(4 - len(head)) n = int.from_bytes(head[2:4], "big") out = b"" while len(out) < n: out += s.recv(n - len(out)) if head[1] != 0: raise RuntimeError(f"script-fu error: {out.decode(errors='replace')[:300]}") return out.decode(errors="replace") def _val(self, expr: str) -> str: """The printed result of one Script-Fu call, outer list stripped: '(151 245)' -> '151 245'.""" out = self._sf(expr).strip() return out[1:-1] if out.startswith("(") and out.endswith(")") else out def _str(self, expr: str) -> str: v = self._val(expr) m = re.match(r'^"(.*)"$', v, re.S) return (m.group(1).replace('\\"', '"').replace("\\n", "\n").replace("\\\\", "\\") if m else v) def read(self): w, h = self.size img = "(vector-ref (car (gimp-get-images)) 0)" def layers(expr): # top first; layer groups are flattened in place, skipping hidden groups out = [] for lid in (int(x) for x in re.findall(r"\d+", self._val(expr))): if self._val(f"(gimp-item-is-group {lid})") in ("1", "#t"): if self._val(f"(gimp-item-get-visible {lid})") in ("1", "#t"): out += layers(f"(gimp-item-get-children {lid})") else: out.append(lid) return out ids = layers(f"(gimp-image-get-layers {img})") els, skipped, bg = [], 0, "#ffffff" for lid in reversed(ids): # GIMP lists top first; DesignDoc is back to front name = self._str(f"(gimp-item-get-name {lid})") if self._val(f"(gimp-item-get-visible {lid})") not in ("1", "#t"): continue x, y = (float(v) for v in self._val(f"(gimp-drawable-get-offsets {lid})").split()[:2]) geom = {"x": x, "y": y, "width": max(float(self._val(f"(gimp-drawable-get-width {lid})")), 0.5), "height": max(float(self._val(f"(gimp-drawable-get-height {lid})")), 0.5), "opacity": float(self._val(f"(gimp-layer-get-opacity {lid})")) / 100} if name == "canvas.png": # the base layer: its colour is the background (read live, it may be filled) px = [float(v) for v in re.findall(r"[\d.]+", self._val(f"(gimp-drawable-get-pixel {lid} 1 1)"))] k = 255.0 if (len(px) > 3 and px[3] > 1.0) or max(px[:3] or [0]) > 1.0 else 1.0 bg = "#" + "".join(f"{round(c / k * 255):02x}" for c in (px + [255, 255, 255])[:3]) continue if self._val(f"(gimp-item-id-is-text-layer {lid})") in ("1", "#t"): rgba = [float(v) for v in re.findall(r"[\d.]+", self._val(f"(gimp-text-layer-get-color {lid})"))] k = 255.0 if (len(rgba) > 3 and rgba[3] > 1.0) or max(rgba[:3] or [0]) > 1.0 else 1.0 color = "#" + "".join(f"{round(c / k * 255):02x}" for c in (rgba + [0, 0, 0])[:3]) font = self._str(f"(gimp-resource-get-name (car (gimp-text-layer-get-font {lid})))") size = float(self._val(f"(gimp-text-layer-get-font-size {lid})").split()[0]) just = self._val(f"(gimp-text-layer-get-justification {lid})") align = {"1": "right", "2": "center"}.get(just.strip(), "left") family = re.sub(r"(\s+(Bold|Italic|Regular|Medium|Light|SemiBold|Black|Oblique))+$", "", font) els.append({"kind": "text", "text": self._str(f"(gimp-text-layer-get-text {lid})"), "font_family": family, "bold": "Bold" in font, "italic": "Italic" in font, "font_size": size, "color": color, "align": align, "letter_spacing": float(self._val(f"(gimp-text-layer-get-letter-spacing {lid})") or 0), "_line_extra": float(self._val(f"(gimp-text-layer-get-line-spacing {lid})") or 0), **geom, "_top": geom["y"]}) continue stem = Path(name).stem.split(" ")[0].split("#")[0] if stem in self.paths: els.append({"kind": "image", "asset_id": stem, **geom}) else: skipped += 1 return {"width": w, "height": h, "background": bg, "elements": els, "skipped": skipped} def expert(self, doc): r, g, b = (int(doc.get("background", "#ffffff")[i:i + 2], 16) for i in (1, 3, 5)) self._sf(f""" (let* ((img (vector-ref (car (gimp-get-images)) 0)) (layers (car (gimp-image-get-layers img))) (base (vector-ref layers (- (vector-length layers) 1)))) (gimp-context-set-foreground (list {r} {g} {b})) (gimp-drawable-fill base FILL-FOREGROUND) (gimp-displays-flush))""") for e in doc["elements"]: if e["kind"] == "image": p = self.paths[e["asset_id"]] self._sf(f""" (let* ((img (vector-ref (car (gimp-get-images)) 0)) (l (car (gimp-file-load-layer RUN-NONINTERACTIVE img "{p}")))) (gimp-image-insert-layer img l 0 -1) (gimp-layer-scale l {max(1, round(e['width']))} {max(1, round(e['height']))} FALSE) (gimp-layer-set-offsets l {round(e['x'])} {round(e['y'])}) (gimp-displays-flush))""") elif e["kind"] == "text": text = e["text"].replace("\\", "\\\\").replace('"', '\\"') font = self._font(e.get("font_family") or "Sans", bool(e.get("bold")), bool(e.get("italic"))) fs = float(e.get("font_size", 48)) r, g, b = (int((e.get("color") or "#000000")[i:i + 2], 16) for i in (1, 3, 5)) just = {"left": "TEXT-JUSTIFY-LEFT", "center": "TEXT-JUSTIFY-CENTER", "right": "TEXT-JUSTIFY-RIGHT"}[ e.get("align", "left")] extra = float(e.get("_step", fs)) - float(e.get("_natural", fs)) # pixels between lines beyond natural self._sf(f""" (let* ((img (vector-ref (car (gimp-get-images)) 0)) (l (car (gimp-text-layer-new img "{text}" (car (gimp-font-get-by-name "{font}")) {fs} UNIT-PIXEL)))) (gimp-image-insert-layer img l 0 -1) (gimp-text-layer-set-color l (list {r} {g} {b})) (gimp-text-layer-set-justification l {just}) (gimp-text-layer-set-line-spacing l {extra}) (gimp-text-layer-set-letter-spacing l {float(e.get('letter_spacing', 0))}) (gimp-text-layer-resize l {max(1.0, float(e['width']))} (car (gimp-drawable-get-height l))) (gimp-layer-set-offsets l {round(e['x'])} {round(float(e.get('_top', e['y'])))}) (gimp-displays-flush))""") def _font(self, family: str, bold: bool, italic: bool) -> str: """The installed font for a family and style ('Montserrat' bold -> 'Montserrat Bold'), exact names first.""" names = re.findall(r'"([^"]+)"', self._sf( f'(map (lambda (f) (car (gimp-resource-get-name f))) (vector->list (car (gimp-fonts-get-list "{family}"))))')) style = "Bold Italic" if bold and italic else "Bold" if bold else "Italic" if italic else "" for want in ([f"{family} {style}"] if style else []) + [family, f"{family} Regular"]: for n in names: if n.lower() == want.lower(): return n return names[0] if names else "Sans-serif" APPS = {"impress": Impress, "inkscape": Inkscape, "gimp": Gimp} STATE: dict = {"app": None} # ---------------------------------------------------------------------------------------------- http class Handler(BaseHTTPRequestHandler): def log_message(self, *a): pass def _send(self, code, body, ctype="application/json"): data = body if isinstance(body, bytes) else json.dumps(body).encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): try: if self.path == "/health": self._send(200, {"ok": True, "app": type(STATE["app"]).__name__ if STATE["app"] else None}) elif self.path == "/screenshot": self._send(200, screenshot(), "image/png") else: self._send(404, {"error": "not found"}) except Exception as exc: # noqa: BLE001 self._send(500, {"error": f"{type(exc).__name__}: {exc}"}) def do_POST(self): n = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(n) or b"{}") try: with LOCK: if self.path == "/input": out = do_input(body) elif self.path == "/reset": kill_apps() paths = write_assets(body.get("assets") or {}) app = APPS[body["app"]]() app.reset(int(body["width"]), int(body["height"]), body.get("background", "#ffffff"), paths) STATE["app"] = app out = {"ok": True} elif self.path == "/read": out = STATE["app"].read() elif self.path == "/expert": STATE["app"].expert(body["doc"]) out = {"ok": True} else: self._send(404, {"error": "not found"}) return self._send(200, out) except Exception as exc: # noqa: BLE001 import traceback self._send(500, {"error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc()[-1500:]}) def main(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=5005) a = ap.parse_args() WORK.mkdir(parents=True, exist_ok=True) start_display() ThreadingHTTPServer(("0.0.0.0", a.port), Handler).serve_forever() if __name__ == "__main__": main()