"""The env's side of the desktop agent (realapps/desktop/agent.py): screen, input, and the app's document. `DesktopBackend` gives each desktop-app episode one desktop (one X display = one episode at a time), from the pool in DESIGN_CANVAS_DESKTOP_URLS (comma list; default http://localhost:5005), or from Daytona sandboxes the env starts (realapps/daytona.py). The GUI driver forwards that episode's screenshots and input to the desktop instead of a browser page, so the GUI tools are the same. """ from __future__ import annotations import base64 import fcntl import hashlib import json import os import threading import time import urllib.request from dataclasses import dataclass from pathlib import Path from ...core.doc import DesignDoc, Element from ...core.text import layout LOCKS = Path(os.environ.get("DESIGN_CANVAS_LOCK_DIR", "/tmp/design-canvas-locks")) class DesktopClient: def __init__(self, base_url: str, headers: dict[str, str] | None = None): self.base = base_url.rstrip("/") self.headers = dict(headers or {}) # e.g. a Daytona sandbox's preview token def _post(self, path: str, body: dict, timeout: float = 300): req = urllib.request.Request(self.base + path, data=json.dumps(body).encode(), method="POST", headers={"Content-Type": "application/json", **self.headers}) try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read()) except urllib.error.HTTPError as e: raise RuntimeError(f"desktop {path}: {e.read()[:1500].decode(errors='replace')}") from None def screenshot(self) -> bytes: req = urllib.request.Request(self.base + "/screenshot", headers=self.headers) with urllib.request.urlopen(req, timeout=60) as r: return r.read() def input(self, **action) -> None: self._post("/input", action, timeout=120) def reset(self, app: str, width: int, height: int, background: str, assets: dict[str, Path]) -> None: self._post("/reset", {"app": app, "width": width, "height": height, "background": background, "assets": {a: base64.b64encode(Path(p).read_bytes()).decode() for a, p in assets.items()}}, timeout=120) def read(self) -> dict: return self._post("/read", {}, timeout=300) def expert(self, doc: DesignDoc) -> None: self._post("/expert", {"doc": doc.model_dump(mode="json")}, timeout=120) @dataclass class DesktopSession: app: str client: DesktopClient url: str skipped: int = 0 fonts: object = None # the pack's FontRegistry: SVG text is placed by baseline, converted with our own layout def text_anchor(e: Element, fonts) -> dict: """Where an app that places text by baseline (SVG) should put this element: first baseline, anchor x, line step.""" lay = layout(e, fonts) return {"_baseline": e.y + lay.lines[0].baseline, "_step": lay.line_box, "_anchor_x": e.x + {"left": 0.0, "center": e.width / 2, "right": e.width}[e.align]} def text_top(e: Element, fonts) -> dict: """For apps whose text layers start at the first line's top (GIMP): our box top plus the half-leading.""" lay = layout(e, fonts) ascent, descent = fonts.metrics(e.font_family or "", e.font_size, e.bold, e.italic) return {"_top": e.y + (lay.line_box - (ascent + descent)) / 2, "_step": lay.line_box, "_natural": ascent + descent} def from_top(raw: dict, fonts) -> dict: top = raw.pop("_top") extra = raw.pop("_line_extra", None) if extra is not None: # GIMP's spacing is pixels added to the font's own line height a, d = fonts.metrics(raw.get("font_family") or "", raw["font_size"], raw.get("bold", False), raw.get("italic", False)) raw["line_height"] = max(0.1, (a + d + extra) / raw["font_size"]) el = Element.model_validate({"id": "t", "x": raw["x"], "y": 0, "width": raw["width"], "height": raw["height"], **{k: v for k, v in raw.items() if k not in ("x", "y", "width", "height")}}) lay = layout(el, fonts) ascent, descent = fonts.metrics(el.font_family or "", el.font_size, el.bold, el.italic) return {**raw, "y": top - (lay.line_box - (ascent + descent)) / 2, "height": max(1.0, lay.height)} def from_anchor(raw: dict, fonts) -> dict: """The inverse: a text element read back by baseline and anchor -> DesignDoc box, through the same layout.""" base, ax = raw.pop("_baseline"), raw.pop("_anchor_x") raw.pop("_step", None) el = Element.model_validate({"id": "t", "x": 0, "y": 0, "width": 10, "height": 10, "auto_width": True, **raw}) lay = layout(el, fonts) w = max(1.0, lay.width + 1.0) return {**raw, "auto_width": True, "width": w, "height": max(1.0, lay.height), "x": ax - {"left": 0.0, "center": w / 2, "right": w}[el.align], "y": base - lay.lines[0].baseline} class DesktopBackend: """Real desktop apps (LibreOffice Impress, Inkscape, GIMP) behind the GUI interface.""" APPS = {"impress": "impress", "inkscape-desktop": "inkscape", "gimp": "gimp"} def __init__(self, urls: list[str] | None = None): urls = urls or os.environ.get("DESIGN_CANVAS_DESKTOP_URLS", "http://localhost:5005").split(",") self.free = [u.strip() for u in urls if u.strip()] self.held: dict[str, int] = {} # desktop -> its lock file, held while an episode runs on it self.cv = threading.Condition() def _client(self, url: str) -> DesktopClient: return DesktopClient(url) @staticmethod def _lock(key: str) -> int | None: """An exclusive lock on one desktop shared by every process on this machine (the multi-worker server's workers all see the same desktops), or None if another process holds it.""" LOCKS.mkdir(parents=True, exist_ok=True) fd = os.open(LOCKS / f"desktop-{hashlib.sha1(key.encode()).hexdigest()[:16]}.lock", os.O_CREAT | os.O_RDWR, 0o600) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) return fd except BlockingIOError: os.close(fd) return None def new_episode(self, skin: str, width: int, height: int, assets: dict[str, Path], background: str = "#ffffff", fonts=None) -> DesktopSession: deadline = time.time() + 1800 with self.cv: # one episode per desktop: wait for one no process is using while True: url = next((u for u in self.free if (fd := self._lock(u)) is not None), None) if url is not None: self.free.remove(url) self.held[url] = fd break if time.time() > deadline: raise TimeoutError("no desktop became free in 30 minutes") self.cv.wait(timeout=1.0) # a release here wakes it; a release in another process is polled try: client = self._client(url) client.reset(self.APPS[skin], width, height, background, assets) except Exception: self.release(url) raise return DesktopSession(self.APPS[skin], client, url, fonts=fonts) def release(self, url: str) -> None: with self.cv: fd = self.held.pop(url, None) if fd is not None: fcntl.flock(fd, fcntl.LOCK_UN) os.close(fd) if url not in self.free: self.free.append(url) self.cv.notify() def read_doc(self, s: DesktopSession) -> DesignDoc: raw = s.client.read() doc = DesignDoc(width=raw["width"], height=raw["height"], background=raw.get("background") or "#ffffff") s.skipped = int(raw.get("skipped", 0)) for n, e in enumerate(raw["elements"], 1): e = {k: v for k, v in e.items() if v is not None} if "_baseline" in e or "_top" in e: if s.fonts is None: raise RuntimeError("this app places text by its own metrics: pass the pack's fonts to new_episode") e = from_anchor(e, s.fonts) if "_baseline" in e else from_top(e, s.fonts) e["opacity"] = min(1.0, max(0.0, float(e.get("opacity", 1.0)))) try: doc.elements.append(Element.model_validate({"id": f"e{n}", **e})) except ValueError: s.skipped += 1 doc.next_id = len(doc.elements) + 1 return doc def write_doc(self, s: DesktopSession, doc: DesignDoc) -> None: d = doc.model_dump(mode="json") if s.fonts is not None: for e, el in zip(d["elements"], doc.elements, strict=True): if el.kind == "text": e.update(text_anchor(el, s.fonts) if s.app == "inkscape" else text_top(el, s.fonts)) s.client._post("/expert", {"doc": d}, timeout=120)