File size: 6,131 Bytes
df84e45 1d12f00 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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()
# CSS selector that appears once each app's editor has loaded (the GUI driver waits for it)
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: # Penpot answers get-profile without a session too
return e.code < 500
except Exception: # noqa: BLE001
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 # noqa: F401
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"]
|