Merge pull request #53 from penpot/feature/penpot-client

Add client for penpot backend (supporting shape retrieval in particular)

.github/workflows/lint_and_docs.yaml

@@ -61,6 +61,7 @@ jobs:
         env:
           GOOGLE_STORAGE_KEY: ${{ secrets.GOOGLE_STORAGE_KEY }}
           GOOGLE_STORAGE_SECRET: ${{ secrets.GOOGLE_STORAGE_SECRET }}
+          PP_BACKEND_PASSWORD: ${{ secrets.PP_BACKEND_PASSWORD }}
         run: poetry run poe doc-build
 #      - name: Upload artifact
 #        uses: actions/upload-pages-artifact@v2

.github/workflows/pytest.yml

@@ -44,6 +44,7 @@ jobs:
         env:
           GOOGLE_STORAGE_KEY: ${{ secrets.GOOGLE_STORAGE_KEY }}
           GOOGLE_STORAGE_SECRET: ${{ secrets.GOOGLE_STORAGE_SECRET }}
+          PP_BACKEND_PASSWORD: ${{ secrets.PP_BACKEND_PASSWORD }}
         run: poetry run poe test
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v1

.github/workflows/windows_mac_test.yml

@@ -42,4 +42,5 @@ jobs:
         env:
           GOOGLE_STORAGE_KEY: ${{ secrets.GOOGLE_STORAGE_KEY }}
           GOOGLE_STORAGE_SECRET: ${{ secrets.GOOGLE_STORAGE_SECRET }}
+          PP_BACKEND_PASSWORD: ${{ secrets.PP_BACKEND_PASSWORD }}
         run: poetry run poe test-subset

.gitignore

@@ -3,6 +3,7 @@
 config_local.json
 temp
 test/log
+/*.iml
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

config.json

@@ -19,5 +19,9 @@
     "openai": "env:OPENAI_API_KEY",
     "anthropic": "env:ANTHROPIC_API_KEY",
     "gemini": "env:GEMINI_API_KEY"
+  },
+  "penpot_backend": {
+    "user": "[email protected]",
+    "password": "env:PP_BACKEND_PASSWORD"
   }
 }

pyproject.toml

@@ -29,6 +29,7 @@ requests = "^2.32.2"
 resvg-py = "^0.1.5"
 selenium = "^4.21.0"
 tqdm = "^4.66.4"
+transit-python2 = "^0.8.321"
 webdriver-manager = "^4.0.1"
 
 [tool.poetry.group.dev]

scripts/api.py

@@ -1,3 +1,4 @@
+# type: ignore
 import requests
 import gzip 
 import json

scripts/penpot_client_demo.py

@@ -0,0 +1,20 @@
+from pprint import pprint
+
+from penai.client import PenpotClient, transit_to_py
+from penai.registries.projects import SavedPenpotProject
+
+if __name__ == '__main__':
+    saved_penpot_project = SavedPenpotProject.INTERACTIVE_MUSIC_APP
+    penpot_project = saved_penpot_project.load(pull=True)
+    main_file = penpot_project.get_main_file()
+    page = main_file.get_page_by_name("Interactive music app")
+    shape = page.svg.get_shape_by_name("ic_equalizer_48px-1")
+
+    client = PenpotClient.create_default()
+    result = client.get_shape_recursive_py(
+        project_id=saved_penpot_project.get_project_id(),
+        file_id=main_file.id,
+        page_id=page.id,
+        shape_id=shape.id
+    )
+    pprint(result)

src/penai/client.py

@@ -0,0 +1,137 @@
+import io
+from typing import Any, Self
+from uuid import UUID
+
+import requests
+from transit.reader import Reader
+from transit.transit_types import Keyword, TaggedValue, frozendict
+
+from penai.config import get_config
+
+SERVER_URL_DEFAULT = "https://design.penpot.app"
+
+
+class PenpotClient:
+    """Client for interaction with the Penpot backend."""
+
+    def __init__(self, user_email: str, user_password: str, server_url: str = SERVER_URL_DEFAULT):
+        self.server_url = server_url
+        self.base_url = server_url + "/api/rpc/command"
+        self.session = requests.Session()
+        login_response = self._login(user_email, user_password)
+        self.session.cookies.update(login_response.cookies)
+
+    @classmethod
+    def create_default(cls) -> Self:
+        cfg = get_config()
+        return cls(cfg.penpot_user, cfg.penpot_password)
+
+    def _login(self, email: str, password: str) -> requests.Response:
+        url = f"{self.base_url}/login-with-password"
+        json = {
+            "~:email": email,
+            "~:password": password,
+        }
+        headers = {
+            "Content-Type": "application/transit+json",
+        }
+        return self.session.post(url=url, headers=headers, json=json)
+
+    def _read_transit_dict(self, response: requests.Response) -> dict:
+        reader = Reader("json")
+        return reader.read(io.StringIO(response.text))
+
+    def get_file(self, project_id: str, file_id: str) -> dict:
+        url = f"{self.base_url}/get-file"
+        params = {
+            "id": file_id,
+            "project-id": project_id,
+            "features": [
+                "layout/grid",
+                "styles/v2",
+                "fdata/pointer-map",
+                "fdata/objects-map",
+                "components/v2",
+                "fdata/shape-data-type",
+            ],
+        }
+        resp = self.session.get(url=url, params=params)
+        return self._read_transit_dict(resp)
+
+    def _get_file_fragment(self, file_id: str, fragment_id: str) -> dict:
+        url = f"{self.base_url}/get-file-fragment"
+        params = {
+            "file-id": file_id,
+            "fragment-id": fragment_id,
+        }
+        resp = self.session.get(url=url, params=params)
+        return self._read_transit_dict(resp)
+
+    def get_page(self, project_id: str, file_id: str, page_id: str) -> dict:
+        data = self.get_file(project_id, file_id)
+        pages_index = data[Keyword("data")][Keyword("pages-index")]
+        page = pages_index[UUID(page_id)]
+        if Keyword("objects") not in page:
+            raise NotImplementedError("Retrieval of missing page fragments not implemented")
+            # TODO implement retrieval if necessary
+            # Code to be adapted for this:
+            # fragment_id = v["~#penpot/pointer"][0]
+            # fragment = self._get_file_fragment(file_id, fragment_id[2:])
+            # data["~:data"]["~:pages-index"][k] = fragment["~:content"]
+        return page
+
+    def get_shape(self, project_id: str, file_id: str, page_id: str, shape_id: str) -> TaggedValue:
+        page = self.get_page(project_id, file_id, page_id)
+        objects = page[Keyword("objects")]
+        return objects[UUID(shape_id)]
+
+    def get_shape_recursive_py(
+        self,
+        project_id: str,
+        file_id: str,
+        page_id: str,
+        shape_id: str,
+    ) -> dict:
+        """Gets the representation for the requested shape.
+
+        :param project_id: the project's UUID
+        :param file_id: the file's UUID
+        :param page_id: the page's UUID
+        :param shape_id: the shape's UUID
+        :return: a dictionary representation (containing mostly primitive types)
+            of the shape with all sub-shapes recursively expanded
+        """
+        page = self.get_page(project_id, file_id, page_id)
+        objects = page[Keyword("objects")]
+
+        def py_shape(uuid: str) -> dict:
+            shape = objects[UUID(uuid)]
+            shape_dict = transit_to_py(shape)["shape"]
+            if "shapes" in shape_dict:
+                subshapes = {}
+                for subshape_id in shape_dict["shapes"]:
+                    subshapes[subshape_id] = py_shape(subshape_id)
+                shape_dict["shapes"] = subshapes
+            return shape_dict
+
+        return py_shape(shape_id)
+
+
+def transit_to_py(obj: Any) -> Any:
+    """Recursively converts the given transit representation to more primitive Python types.
+
+    :param obj: the object the convert
+    :return: the simplified representation
+    """
+    if isinstance(obj, TaggedValue):
+        return {obj.tag: transit_to_py(obj.rep)}
+    elif isinstance(obj, frozendict):
+        return {transit_to_py(k): transit_to_py(v) for k, v in obj.items()}
+    elif isinstance(obj, Keyword):
+        return obj.name
+    elif isinstance(obj, tuple):
+        return tuple(transit_to_py(x) for x in obj)
+    elif isinstance(obj, UUID):
+        return obj.hex
+    else:
+        return obj

src/penai/config.py

@@ -53,6 +53,14 @@ def gemini_api_key(self) -> str:
     def get_openai_client(self, timeout: int = 100) -> OpenAI:
         return OpenAI(api_key=self.openai_api_key, timeout=timeout)
 
+    @property
+    def penpot_user(self) -> str:
+        return cast(str, self._get_non_empty_entry(["penpot_backend", "user"]))
+
+    @property
+    def penpot_password(self) -> str:
+        return cast(str, self._get_non_empty_entry(["penpot_backend", "password"]))
+
 
 class ConfigProvider(ConfigProviderBase[__Configuration]):
     pass

src/penai/registries/projects.py

@@ -23,6 +23,14 @@ class SavedPenpotProject(Enum):
     def get_project_name(self) -> str:
         return self.value
 
+    def get_project_id(self) -> str:
+        """:return: the project's UUID on the default server (design.penpot.app)"""
+        match self:
+            case SavedPenpotProject.INTERACTIVE_MUSIC_APP:
+                return "15586d98-a20a-8145-8004-69dd979da070"
+            case _:
+                raise NotImplementedError
+
     def get_path(self, pull: bool = False, force: bool = False) -> str:
         result = os.path.join(
             get_config().penpot_designs_basedir(),

src/penai/svg.py

@@ -449,6 +449,14 @@ def shape_id(self) -> str:
         # We actually have to ask its parent very kindly.
         return self.get_containing_g_element().get("id")
 
+    @property
+    def id(self) -> str:
+        """:return: the shape's UUID"""
+        shape_id = self.shape_id
+        prefix = "shape-"
+        assert shape_id.startswith(prefix)
+        return shape_id[len(prefix) :]
+
     @property
     def depth_in_svg(self) -> int:
         return self._depth_in_svg

test/penai/test_server_interaction.py

@@ -0,0 +1,6 @@
+from penai.client import PenpotClient
+
+
+def test_authentication_successful() -> None:
+    client = PenpotClient.create_default()
+    assert client.session.cookies, "Authentication to penpot server failed, check your config!"