| """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 {}) |
|
|
| 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 |
|
|
|
|
| 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: |
| 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] = {} |
| 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: |
| 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) |
| 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) |
|
|
