| """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) |
|
|
|
|
| |
|
|
|
|
| def start_display(): |
| n = DISPLAY.lstrip(":") |
| for stale in (Path(f"/tmp/.X{n}-lock"), Path(f"/tmp/.X11-unix/X{n}")): |
| 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) |
| |
| (cfg / "rc.xml").write_text("""<?xml version="1.0"?><openbox_config xmlns="http://openbox.org/3.4/rc"> |
| <applications><application class="*" type="normal"><maximized>yes</maximized><decor>no</decor></application> |
| </applications></openbox_config>""") |
| subprocess.Popen(["openbox"], env={**os.environ, "DISPLAY": DISPLAY}, |
| stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| fonts = Path(os.environ.get("DESIGN_CANVAS_FONTS", "/fonts")) |
| if fonts.is_dir(): |
| 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): |
| 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]) |
| |
| run(["pkill", "-9", "-f", "/gimp/3.0/plug-ins/"]) |
| for _ in range(40): |
| 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) |
|
|
|
|
| |
|
|
|
|
| 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 |
|
|
|
|
| |
|
|
|
|
| MM100_PER_PX = 2540 / 96 |
|
|
|
|
| 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: |
| 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") |
| |
| SETTINGS = """<?xml version="1.0" encoding="UTF-8"?> |
| <oor:items xmlns:oor="http://openoffice.org/2001/registry" xmlns:xs="http://www.w3.org/2001/XMLSchema"> |
| <item oor:path="/org.openoffice.Office.Common/Misc"><prop oor:name="ShowTipOfTheDay" oor:op="fuse"><value>false</value></prop></item> |
| <item oor:path="/org.openoffice.Office.Common/Misc"><prop oor:name="FirstRun" oor:op="fuse"><value>false</value></prop></item> |
| <item oor:path="/org.openoffice.Setup/Office"><prop oor:name="ooSetupInstCompleted" oor:op="fuse"><value>true</value></prop></item> |
| <item oor:path="/org.openoffice.Office.Impress/Misc"><prop oor:name="StartWithTemplate" oor:op="fuse"><value>false</value></prop></item> |
| </oor:items>""" |
|
|
| 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 |
|
|
| self.doc = self._desktop().loadComponentFromURL("private:factory/simpress", "_blank", 0, ()) |
| page = self.doc.DrawPages.getByIndex(0) |
| page.Layout = 20 |
| 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: |
| 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: |
| pass |
| els, skipped = [], 0 |
|
|
| def flat(container): |
| 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": |
| 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: |
| 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: |
| lh = ls.Height / MM100_PER_PX / font_px |
| elif mode in (0, "PROP") and ls is not None: |
| 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 []: |
| 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 |
|
|
|
|
| |
|
|
|
|
| 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): |
| |
| 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'<image x="{e["x"]}" y="{e["y"]}" width="{e["width"]}" height="{e["height"]}" ' |
| f'preserveAspectRatio="none" xlink:href="{paths[e["asset_id"]].as_uri()}"{tr}{op}/>') |
| 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'<tspan x="{ax}" y="{base + k * step}" sodipodi:role="line">{_line_runs(line)}</tspan>' |
| for k, line in enumerate(_lines_of(e))) |
| parts.append(f'<text xml:space="preserve" style="{style}"{tr}{op}>{spans}</text>') |
| return (f'<?xml version="1.0" encoding="UTF-8"?>\n<svg xmlns="http://www.w3.org/2000/svg" ' |
| f'xmlns:xlink="http://www.w3.org/1999/xlink" ' |
| f'xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd" ' |
| f'xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape" width="{width}px" height="{height}px" ' |
| f'viewBox="0 0 {width} {height}"><sodipodi:namedview pagecolor="{background}" inkscape:pageopacity="1"/>' |
| f'<rect id="canvas-background" x="0" y="0" width="{width}" height="{height}" fill="{background}" ' |
| f'sodipodi:insensitive="true"/><g inkscape:groupmode="layer" inkscape:label="Design" id="design">' |
| + "".join(parts) + "</g></svg>") |
|
|
|
|
| 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'<tspan style="{";".join(st)}">{_esc(p["text"])}</tspan>' 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} |
|
|
|
|
| |
|
|
|
|
| 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) |
| |
| 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') |
| |
| 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): |
| 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",)) |
| 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): |
| 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): |
| 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": |
| 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)) |
| 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} |
|
|
|
|
| |
|
|
|
|
| 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: |
| 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: |
| 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() |
|
|
