| """Real design apps behind the design-canvas GUI interface (skin="penpot", ...). |
| |
| Each backend sets an episode up inside the app (a blank board at the task's size with its assets loaded), hands |
| the GUI driver a URL and cookies to open, and on submit reads the app's own document back as a DesignDoc so the |
| same reward scores it. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import os |
| import threading |
| from pathlib import Path |
|
|
| _BACKENDS: dict = {} |
| _LOCK = threading.Lock() |
|
|
| |
| READY = {"penpot": "[class*=viewport]"} |
|
|
|
|
| def get_backend(skin: str, fonts_dir: Path | None = None): |
| """The shared backend for a real-app skin. |
| |
| With DESIGN_CANVAS_APPS=daytona the server runs the apps in Daytona sandboxes it manages (realapps/daytona.py; |
| DESIGN_CANVAS_DAYTONA_DESKTOPS desktops, default 2). Otherwise they are reached at fixed addresses: |
| DESIGN_CANVAS_PENPOT_URL (default http://localhost:9011) with DESIGN_CANVAS_PENPOT_CREDS (where the test |
| profile's credentials are kept), and DESIGN_CANVAS_DESKTOP_URLS (desktop agents, one episode each at a time). |
| The desktop apps (impress, inkscape-desktop, gimp) share one pool of desktops.""" |
| with _LOCK: |
| if skin not in _BACKENDS: |
| daytona = os.environ.get("DESIGN_CANVAS_APPS", "").lower() == "daytona" |
| if skin == "penpot": |
| if daytona: |
| from .daytona import DaytonaPenpotBackend |
|
|
| _BACKENDS[skin] = DaytonaPenpotBackend(_apps()) |
| else: |
| from .penpot.backend import PenpotBackend |
|
|
| creds = os.environ.get("DESIGN_CANVAS_PENPOT_CREDS", |
| str(Path.home() / ".cache" / "design_canvas" / "penpot-test-user.json")) |
| _BACKENDS[skin] = PenpotBackend( |
| os.environ.get("DESIGN_CANVAS_PENPOT_URL", "http://localhost:9011"), creds) |
| elif skin in ("impress", "inkscape-desktop", "gimp"): |
| desk = next((b for k, b in _BACKENDS.items() if k != "penpot"), None) |
| if desk is None and daytona: |
| from .daytona import DaytonaDesktopBackend |
|
|
| desk = DaytonaDesktopBackend(_apps(), int(os.environ.get("DESIGN_CANVAS_DAYTONA_DESKTOPS", "2")), |
| fonts_dir) |
| elif desk is None: |
| from .desktop.client import DesktopBackend |
|
|
| desk = DesktopBackend() |
| _BACKENDS[skin] = desk |
| else: |
| raise ValueError(f"no real-app backend for skin {skin!r}") |
| return _BACKENDS[skin] |
|
|
|
|
| _APPS = None |
|
|
|
|
| def _apps(): |
| global _APPS |
| if _APPS is None: |
| from .daytona import DaytonaApps |
|
|
| _APPS = DaytonaApps() |
| return _APPS |
|
|
|
|
| def status(check: bool = False) -> dict: |
| """Which real apps this server offers and where they run, for the playground and GET /api/apps. |
| |
| With check, each app's endpoint (or the Daytona pool) is also asked whether it is up; without it nothing is |
| contacted, so the call is free.""" |
| try: |
| from ..server.episodes import REAL_SKINS |
| except ImportError: |
| from server.episodes import REAL_SKINS |
| daytona = os.environ.get("DESIGN_CANVAS_APPS", "").lower() == "daytona" |
| out = {"runs_on": "daytona" if daytona else "local", |
| "apps": [{"skin": k, "name": n, "about": d, "surface": "gui"} for k, (n, d) in REAL_SKINS.items()]} |
| if daytona: |
| out["desktops"] = int(os.environ.get("DESIGN_CANVAS_DAYTONA_DESKTOPS", "2")) |
| if check: |
| apps = _apps() |
| out["sandboxes"] = [{"role": sb.labels.get("role"), "id": sb.id[:8], |
| "state": str(sb.state).split(".")[-1].lower(), "cpu": sb.cpu, |
| "memory_gb": sb.memory, "episodes": apps.busy.get(sb.id, 0)} |
| for sb in apps.sandboxes() if sb.labels.get("role")] |
| else: |
| penpot = os.environ.get("DESIGN_CANVAS_PENPOT_URL", "http://localhost:9011") |
| desktops = [u.strip() for u in os.environ.get("DESIGN_CANVAS_DESKTOP_URLS", "http://localhost:5005").split(",") |
| if u.strip()] |
| out["endpoints"] = [{"role": "penpot", "url": penpot}] + [{"role": "desktop", "url": u} for u in desktops] |
| if check: |
| for e in out["endpoints"]: |
| e["up"] = _up(e["url"] + ("/api/rpc/command/get-profile" if e["role"] == "penpot" else "/health")) |
| return out |
|
|
|
|
| def _up(url: str) -> bool: |
| import urllib.error |
| import urllib.request |
|
|
| try: |
| with urllib.request.urlopen(url, timeout=3): |
| return True |
| except urllib.error.HTTPError as e: |
| return e.code < 500 |
| except Exception: |
| return False |
|
|
|
|
| _AVAILABLE: dict = {} |
|
|
|
|
| def available() -> tuple[bool, str]: |
| """Whether this server can run real apps, and if not, why (cached for 30 s; cheap enough for every page load). |
| |
| On Daytona that needs the SDK and a key; locally, the Penpot and desktop endpoints must answer.""" |
| import time |
|
|
| if _AVAILABLE and time.time() - _AVAILABLE["at"] < 30: |
| return _AVAILABLE["ok"], _AVAILABLE["why"] |
| if os.environ.get("DESIGN_CANVAS_APPS", "").lower() == "daytona": |
| try: |
| import daytona |
| except ImportError: |
| ok, why = False, "the daytona package is not installed (install the `daytona` extra)" |
| else: |
| ok, why = (bool(os.environ.get("DAYTONA_API_KEY")), "DAYTONA_API_KEY is not set") |
| else: |
| st = status(check=True) |
| down = [e["url"] for e in st["endpoints"] if not e.get("up")] |
| ok, why = not down, ("real apps run locally here and nothing answers at " + ", ".join(down) + |
| "; run them (realapps/) or set DESIGN_CANVAS_APPS=daytona") |
| _AVAILABLE.update(at=time.time(), ok=ok, why="" if ok else why) |
| return ok, _AVAILABLE["why"] |
|
|
