"""High-level project and library management for PhotonForge PDA.
This module provides the ``Project`` runtime object and async helpers to
create, list, load, retire, and restore PDA-backed projects.
"""
from __future__ import annotations
import asyncio
import importlib
import io
import keyword
import mimetypes
import os
import re
import shutil
import string
import sys
import zipfile
import zlib
from collections import defaultdict, namedtuple
from collections.abc import Awaitable, Iterable, Iterator, Mapping, Sequence
from datetime import datetime
from filecmp import dircmp
from json import dumps as json_dumps
from json import loads as json_loads
from pathlib import Path, PurePosixPath
from types import ModuleType
from typing import Literal
from warnings import warn
import httpx
from pydantic import ValidationError
from ..extension import (
Component,
Technology,
_component_registry,
_export,
_import,
_model_registry,
_technology_registry,
)
from . import _timing
from ._client import (
_add_tag,
_add_version,
_check_access,
_create_document,
_delete_attachment,
_delete_document,
_delete_tag,
_find_repo_document,
_get_attachment_content,
_get_document,
_get_version,
_grant_permission,
_list_attachments,
_list_documents,
_list_latest_versions,
_list_libs,
_list_permissions,
_list_tags,
_list_versions,
_project_type,
# _restore_document,
_revoke_permission,
_technology_type,
_transfer_ownership,
_ui_component_type,
_update_attachment,
_update_permission,
_upload_attachment,
_user_info,
user_info,
)
from ._component import _PREVIEW_FILENAME, _PREVIEW_MIME_TYPE, component_from_pf
from ._crdt.document import Document, _to_automerge_value
from ._crdt.refs import AutomergeRef
from ._parametric_schema import _select_schema
from ._sync_runtime import run as _run
from ._types import DocumentId, ReadOnlyURL, UiComponent
from ._types import ProjectModel as _ProjectModel
_DEFAULT_MODULE_PATH = "./pflibs"
_PDA_BATCH_CONCURRENCY = 32
_PDA_DOC_CREATE_CONCURRENCY = 4
_PF_NATIVE_MIME_TYPE = "application/vnd.photonforge.native+json+zlib"
_PF_NATIVE_FILENAME = "pf-native.json.zlib"
_PYTHON_MODULE_FILENAME = "python-module.zip"
_PYTHON_MODULE_MIME_TYPE = "application/vnd.photonforge.module+zip"
_LEGACY_PYTHON_MODULE_MIME_TYPE = "application/zip"
_table: defaultdict[int, int | None] = defaultdict(lambda: None)
for c in string.ascii_lowercase + string.digits:
n = ord(c)
_table[n] = n
for up, lo in zip(string.ascii_uppercase, string.ascii_lowercase, strict=True):
_table[ord(up)] = ord(lo)
_o = ord("_")
_table[_o] = _o
_table[ord(" ")] = _o
def _derive_module_name(name: str, doc_id: DocumentId) -> str:
module_name = name.translate(_table)
if len(module_name) == 0:
return f"m_{doc_id}"
if module_name[0].isdigit():
module_name = "m_" + module_name
while keyword.iskeyword(module_name) or keyword.issoftkeyword(module_name):
module_name = module_name + "_"
return module_name
def _validate_module_name(module_name: str) -> str:
if (
not module_name.isidentifier()
or keyword.iskeyword(module_name)
or keyword.issoftkeyword(module_name)
):
raise RuntimeError(f"Invalid python module name {module_name!r}.")
return module_name
def _python_module(
value: object,
) -> tuple[str | None, str | None, bytes | None]:
if isinstance(value, str):
return None, value or None, None
if isinstance(value, (bytes, bytearray)):
return None, None, bytes(value)
if (
isinstance(value, Sequence)
and not isinstance(value, (str, bytes, bytearray))
and len(value) == 2
and all(isinstance(item, str) for item in value)
):
if len(value[1]) == 0:
raise RuntimeError("Invalid empty python module attachment ID.")
return _validate_module_name(value[0]), value[1], None
return None, None, None
def _module_archive_name(archive: bytes) -> str | None:
with zipfile.ZipFile(io.BytesIO(archive), "r") as zf:
roots = set()
for info in zf.infolist():
member = PurePosixPath(info.filename)
if (
member.is_absolute()
or "\\" in info.filename
or any(part.endswith(":") for part in member.parts)
or any(part in ("", ".", "..") for part in member.parts)
):
raise RuntimeError("Invalid path in python module archive.")
if len(member.parts) == 0:
continue
if not info.is_dir() and len(member.parts) < 2:
raise RuntimeError("Python module archive must contain one rooted package.")
roots.add(member.parts[0])
if len(roots) > 1:
raise RuntimeError("Python module archive must contain one rooted package.")
return _validate_module_name(next(iter(roots))) if roots else None
def _extract_module_archive(archive: bytes, path: Path, module_name: str) -> None:
"""Extract rooted module ZIP data into ``path``."""
archive_name = _module_archive_name(archive)
if archive_name is not None and archive_name != module_name:
raise RuntimeError(
f"Python module archive contains {archive_name!r}, expected {module_name!r}."
)
with zipfile.ZipFile(io.BytesIO(archive), "r") as zf:
for info in zf.infolist():
member = PurePosixPath(info.filename)
relative_parts = member.parts[1:]
if len(relative_parts) == 0:
continue
target = path.joinpath(*relative_parts).resolve()
if path.resolve() not in target.parents:
raise RuntimeError("Invalid path in python module archive.")
if info.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with zf.open(info) as src, target.open("wb") as dst:
shutil.copyfileobj(src, dst)
def _check_module_name_available(module_name: str, *module_paths: str | Path | None) -> None:
if not any(key == module_name or key.startswith(f"{module_name}.") for key in sys.modules):
return
module = sys.modules.get(module_name)
expected_paths = {
Path(module_path).resolve() / module_name
for module_path in module_paths
if module_path is not None
}
if any(Path(path).resolve() in expected_paths for path in getattr(module, "__path__", ())):
return
raise RuntimeError(f"Python module name {module_name!r} is already in use.")
def _purge_module(module_name: str, module_path: str | Path | None) -> None:
_check_module_name_available(module_name, module_path)
for key in tuple(sys.modules):
if key == module_name or key.startswith(f"{module_name}."):
del sys.modules[key]
for registry in (_component_registry, _technology_registry):
for key in tuple(registry):
if key.startswith(f"{module_name}."):
del registry[key]
for key, model in tuple(_model_registry.items()):
model_module = getattr(model, "__module__", "")
if model_module == module_name or model_module.startswith(f"{module_name}."):
del _model_registry[key]
def _check_unique_names(data: dict[DocumentId, Technology | Component], check_for: str) -> None:
"""Validate that object names are unique within one collection.
Args:
data: Objects keyed by document ID.
check_for: Collection label used in error messages.
"""
if len({obj.name for obj in data.values()}) == len(data):
return
existing = set()
for obj in data.values():
if obj.name in existing:
raise RuntimeError(
f"{obj.name!r} will not be unique within project {check_for} after update."
)
existing.add(obj.name)
raise RuntimeError(f"Update would result in duplicate name within project {check_for}.")
async def _gather(*aws: Awaitable[object]) -> list[object]:
results = await asyncio.gather(*aws, return_exceptions=True)
for result in results:
if isinstance(result, BaseException):
raise result
return results
async def _limited_gather(*aws: Awaitable[object]) -> list[object]:
semaphore = asyncio.Semaphore(_PDA_BATCH_CONCURRENCY)
async def run(aw: Awaitable[object]) -> object:
async with semaphore:
return await aw
return await _gather(*(run(aw) for aw in aws))
async def _find_component_native_attachment(ui_doc_id: DocumentId, content: bytes) -> str | None:
with _timing.time_block("component_native.find_existing"):
attachments = await _list_attachments(document_id=ui_doc_id)
for attachment in attachments:
if (
attachment.get("filename") != _PF_NATIVE_FILENAME
or attachment.get("mimeType") != _PF_NATIVE_MIME_TYPE
):
continue
attachment_id = attachment["id"]
if await _get_attachment_content(attachment_id) == content:
_timing.record("component_native.reused")
return attachment_id
return None
async def _upload_component_native_data(ui_doc_id: DocumentId, data: dict[str, object]) -> str:
with _timing.time_block("component_native.encode"):
content = zlib.compress(json_dumps(data, ensure_ascii=False).encode("utf-8"))
_timing.record("component_native.bytes", bytes_count=len(content))
with _timing.time_block("component_native.upload"):
try:
record = await _upload_attachment(
content, _PF_NATIVE_FILENAME, _PF_NATIVE_MIME_TYPE, document_id=ui_doc_id
)
except httpx.HTTPStatusError as err:
if err.response.status_code != 409:
raise
attachment_id = await _find_component_native_attachment(ui_doc_id, content)
if attachment_id is not None:
return attachment_id
raise
if record.get("documentId") != ui_doc_id:
raise RuntimeError(
"Native data attachment upload returned documentId "
f"{record.get('documentId')!r} instead of {ui_doc_id!r}."
)
return record["id"]
async def _load_component_native_data(ui_data: Mapping[str, object]) -> dict[str, object]:
attachment_id = ui_data.get("pfNative")
if isinstance(attachment_id, str) and len(attachment_id) > 0:
try:
with _timing.time_block("component_native.download"):
encoded = await _get_attachment_content(attachment_id)
_timing.record("component_native.bytes", bytes_count=len(encoded))
with _timing.time_block("component_native.decode"):
return json_loads(zlib.decompress(encoded).decode("utf-8"))
except Exception as err:
raise RuntimeError(
f"Unable to load native component attachment {attachment_id!r}."
) from err
pf_ref = ui_data.get("pfRef")
if isinstance(pf_ref, str) and pf_ref.startswith("automerge:"):
with _timing.time_block("component_native.legacy_find"):
doc = await _find_repo_document(pf_ref)
if doc is not None:
with _timing.time_block("component_native.legacy_read"):
return doc.read()
raise RuntimeError("UI component is missing native PhotonForge data.")
async def _add_version_and_tag_info(info: dict[str, object], doc: Document) -> dict[str, object]:
"""Add version and tag metadata for one document reference.
Args:
info: Base metadata payload to enrich.
doc: Document whose current reference is used for filtering.
Returns:
Enriched metadata payload including matching tags and versions.
The returned payload includes tags from matching references and either
``version`` (single match) or ``versions`` (multiple matches).
"""
# TODO: endpoint to get version/tags by document reference
version_infos, tag_infos = await _gather(
_list_versions(doc.id),
_list_tags(None, doc.id),
)
versions = [
version_info["version"] for version_info in version_infos if version_info["ref"] == doc.ref
]
if len(versions) == 1:
info["version"] = versions[0]
elif len(versions) > 1:
info["versions"] = versions
info["tags"] = [tag_info["tag"] for tag_info in tag_infos if tag_info["ref"] == doc.ref]
return info
_TechnologyData = namedtuple("_TechnologyData", ["doc", "obj", "parent_id"])
_ComponentData = namedtuple(
"_ComponentData", ["native_data", "ui_doc", "obj", "parent_id", "external"]
)
def _attachment_id(attachment: Mapping[str, object] | str) -> str:
if isinstance(attachment, str):
return attachment
attachment_id = attachment.get("attachmentId", attachment.get("id"))
if isinstance(attachment_id, str) and len(attachment_id) > 0:
return attachment_id
raise ValueError(
"Expected an attachment ID or attachment metadata with 'attachmentId' or 'id'."
)
class _ProjectObjectMap(Mapping[str, Component | Technology]):
"""Read-only live view of one project's named objects."""
def __init__(
self, project: Project, kind: Literal["components", "technologies"], parent_id: DocumentId
) -> None:
self._project = project
self._kind = kind
self._parent_id = parent_id
def _items(self) -> dict[str, Component | Technology]:
if self._kind == "technologies":
return {
data.obj.name: data.obj
for data in self._project._technologies.values()
if data.parent_id == self._parent_id
}
return {
data.obj.name: data.obj
for data in self._project._components.values()
if data.parent_id == self._parent_id and data.external
}
def __getitem__(self, key: str) -> Component | Technology:
return self._items()[key]
def __iter__(self) -> Iterator[str]:
return iter(self._items())
def __len__(self) -> int:
return len(self._items())
[docs]
class Project:
"""Live view of a PDA project or library document."""
def __init__(self, document: Document) -> None:
self._doc: Document = document
self._technologies: dict[DocumentId, _TechnologyData] = {}
self._components: dict[DocumentId, _ComponentData] = {}
self._libraries: dict[DocumentId, Project] = {}
self._module_path: Path | None = None
self._refresh_state()
def _refresh_state(self) -> None:
self._data = self._doc.read()
self._doc_ref = self._doc.ref
_ProjectModel.model_validate(self._data)
self._data.setdefault("description", None)
self._data.setdefault("labels", [])
self._data.setdefault("config", {})
self._data.setdefault("applications", [])
self._data.setdefault("components", [])
self._data.setdefault("technologies", [])
self._data.setdefault("libraries", [])
self._data.setdefault("componentSchemas", {})
self._data.setdefault("modelSchemas", {})
module_name, _, _ = _python_module(self._data.get("pythonModule"))
self._module_name: str = module_name or _derive_module_name(
self._data["name"], self._doc.id
)
async def _reimport_cached_objects(
self, set_config: bool
) -> dict[DocumentId, Technology | Component]:
data = {}
top_content = []
for doc_ref in self._data["technologies"]:
obj_id = AutomergeRef.parse(doc_ref).document_id
technology_data = self._technologies[obj_id]
data[obj_id] = technology_data.doc.read()
top_content.append((obj_id, 11))
for component_ref in self._data["components"]:
obj_id = AutomergeRef.parse(component_ref["ref"]).document_id
component_data = self._components[obj_id]
data[obj_id] = component_data.native_data
top_content.append((obj_id, 26))
json_data = {
"type": 1,
"type_version": "0.0",
"config": self._data["config"],
"top_content": top_content,
"data": data,
"properties": None,
}
with _timing.time_block("cached_import.json_pack"):
byte_repr = json_dumps(json_data, ensure_ascii=False).encode("utf-8")
_timing.record("cached_import.bytes", bytes_count=len(byte_repr))
with _timing.time_block("cached_import.import"):
objects, _ = await asyncio.to_thread(
_import, byte_repr=byte_repr, set_config=set_config, use_json=True
)
result = {}
for (obj_id, _), obj in zip(top_content, objects, strict=True):
obj._pda_id = obj_id
result[obj_id] = obj
if isinstance(obj, Technology):
technology_data = self._technologies[obj_id]
self._technologies[obj_id] = _TechnologyData(
technology_data.doc, obj, technology_data.parent_id
)
else:
component_data = self._components[obj_id]
self._components[obj_id] = _ComponentData(
component_data.native_data,
component_data.ui_doc,
obj,
component_data.parent_id,
component_data.external,
)
return result
def __str__(self) -> str:
return f"Project {self._data['name']!r}"
def __repr__(self) -> str:
return f"Project(name={self._data['name']!r}, project_id={self._doc.id!r})"
[docs]
def load_latest(
self,
*,
module_path: str | Path | None = None,
set_config: bool = True,
reload_module: bool = True,
) -> Project:
"""Reload latest project data from PDA in place.
Args:
module_path: Root directory where modules are unpacked.
set_config: Whether to apply project config to PhotonForge.
reload_module: Whether to reload the project module.
Returns:
This project, reloaded in its latest, writable version."""
return _run(
self._load_latest(
module_path=module_path,
set_config=set_config,
reload_module=reload_module,
)
)
async def _load_latest(
self,
*,
module_path: str | Path | None = None,
set_config: bool = True,
reload_module: bool = True,
) -> Project:
"""Reload latest project data from PDA in place."""
loaded_modules = (
(
*tuple((lib._module_name, lib._module_path) for lib in self._libraries.values()),
(self._module_name, self._module_path),
)
if reload_module
else ()
)
self._doc = self._doc.latest()
self._refresh_state()
await self._load_data(set_config)
if reload_module:
if module_path is None:
module_path = self._module_path or _DEFAULT_MODULE_PATH
for loaded_module, loaded_path in loaded_modules:
_purge_module(loaded_module, loaded_path)
await asyncio.to_thread(self._load_module, module_path, True, False)
return self
@property
def id(self) -> DocumentId:
"""Project ID."""
return self._doc.id
@property
def name(self) -> str:
"""Project name."""
return self._data["name"]
@property
def description(self) -> str | None:
"""Project description."""
return self._data["description"]
@property
def labels(self) -> tuple[str]:
"""Project labels."""
return tuple(self._data["labels"])
@property
def is_read_only(self) -> bool:
"""Project read-only state."""
return self._doc.is_read_only
@property
def module_name(self) -> str:
"""Sanitized Python module name for this project."""
return self._module_name
@property
def module_path(self) -> Path:
"""Filesystem path where the project Python module is loaded."""
if self._module_path is None:
raise RuntimeError(
"Project module not loaded. Make sure the project was created using the functions "
"'create_project' or 'load_project'."
)
return self._module_path
def _find_target(
self, target: Component | Technology | str | None
) -> tuple[Component | Technology | Project, Document]:
self_id = self._doc.id
if target is None:
obj = self
doc = self._doc
elif isinstance(target, Technology):
data = self._technologies.get(target._pda_id)
if data is None:
for candidate in self._technologies.values():
if candidate.obj.name == target.name and candidate.parent_id == self_id:
obj = candidate.obj
doc = candidate.doc
break
else:
raise RuntimeError(f"{target.name!r} not found among project technologies.")
else:
if data.parent_id != self_id:
raise RuntimeError(
"Only technologies from this project can be targeted, not from libraries."
)
obj = data.obj
doc = data.doc
elif isinstance(target, Component):
data = self._components.get(target._pda_id)
if data is None:
for candidate in self._components.values():
if candidate.obj.name == target.name and candidate.parent_id == self_id:
obj = candidate.obj
doc = candidate.ui_doc
break
else:
raise RuntimeError(f"{target.name!r} not found among project components.")
else:
if data.parent_id != self_id:
raise RuntimeError(
"Only components from this project can be targeted, not from libraries."
)
obj = data.obj
doc = data.ui_doc
if isinstance(target, str):
for data in (*self._components.values(), *self._technologies.values()):
if data.obj.name == target and data.parent_id == self_id:
obj = data.obj
doc = data.doc if isinstance(obj, Technology) else data.ui_doc
break
else:
raise RuntimeError(
f"{target!r} not found among project components or technologies."
)
return obj, doc
[docs]
def get_info(self, target: Component | Technology | str | None = None) -> dict[str, object]:
"""Return metadata for the project or one contained object.
Args:
target: Optional object or name to query instead of the
project document.
Returns:
Metadata payload with document details, tags, and versions.
"""
return _run(self._get_info(target))
async def _get_info(
self, target: Component | Technology | str | None = None
) -> dict[str, object]:
target, doc = self._find_target(target)
info = await _get_document(doc.id)
if target is self:
info.update({"name": self.name, "labels": self._data["labels"]})
else:
info["name"] = target.name
if isinstance(target, Component):
info["external"] = self._components[doc.id].external
return await _add_version_and_tag_info(info, doc)
[docs]
def get_library_info(
self, name: str | None = None
) -> dict[str, object] | list[dict[str, object]]:
"""Return metadata for one library or all imported libraries.
Args:
name: Optional library name. If omitted, returns all libraries.
Returns:
Metadata payload for one library, or a list of payloads for all
imported libraries.
"""
return _run(self._get_library_info(name))
async def _get_library_info(
self, name: str | None = None
) -> dict[str, object] | list[dict[str, object]]:
if name is None:
return await _gather(*(library._get_info() for library in self._libraries.values()))
for library in self._libraries.values():
if library.name == name:
return await library._get_info()
raise RuntimeError(f"Library {name!r} not found in project.")
[docs]
def list_versions(self, target: Component | Technology | str | None = None) -> list[str]:
"""Return available versions for the project or one contained object.
Args:
target: Optional object or name to query instead of the
project document.
Returns:
List of version strings for the target.
"""
return _run(self._list_versions(target))
async def _list_versions(self, target: Component | Technology | str | None = None) -> list[str]:
_, doc = self._find_target(target)
return [version_info["version"] for version_info in await _list_versions(doc.id)]
async def _list_tags(self, target: Component | Technology | str | None = None) -> list[str]:
_, doc = self._find_target(target)
return [tag_info["tag"] for tag_info in await _list_tags(None, doc.id)]
[docs]
def set(
self,
*,
name: str | None = None,
description: str | None = None,
labels: Sequence[str] | None = None,
) -> None:
"""Update editable project fields in-place.
Args:
name: Optional replacement project name.
description: Optional replacement project description.
labels: Optional replacement label collection.
"""
_run(self._set(name=name, description=description, labels=labels))
async def _set(
self,
*,
name: str | None = None,
description: str | None = None,
labels: Sequence[str] | None = None,
) -> None:
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
changes = {}
if isinstance(name, str):
changes["name"] = name
if isinstance(description, str):
changes["description"] = description
if labels is not None and all(isinstance(lbl, str) for lbl in labels):
changes["labels"] = list(set(labels))
if len(changes) > 0:
old_module_name = self._module_name
stored_name, attachment_id, inline_archive = _python_module(
self._data.get("pythonModule")
)
if "name" in changes and stored_name is None:
if attachment_id is not None:
changes["pythonModule"] = [self._module_name, attachment_id]
elif inline_archive is not None:
module_id = await self._upload_module_archive(inline_archive)
changes["pythonModule"] = [self._module_name, module_id]
else:
new_module_name = _derive_module_name(changes["name"], self.id)
if new_module_name != old_module_name:
_check_module_name_available(new_module_name, self._module_path)
_check_module_name_available(old_module_name, self._module_path)
if any(
library.module_name == new_module_name
for library in self._libraries.values()
):
raise ValueError(
f"Module name {new_module_name!r} is already used by a library."
)
if (
self._module_path is not None
and (self._module_path / new_module_name).exists()
):
raise ValueError(
f"Module path {self._module_path / new_module_name} already exists."
)
self._doc.replace(changes)
self._refresh_state()
if (
"name" in changes
and stored_name is None
and attachment_id is None
and inline_archive is None
and self._module_name != old_module_name
):
self._move_module(old_module_name)
def _move_module(self, old_module_name: str) -> None:
_purge_module(old_module_name, self._module_path)
if self._module_path is None:
return
old_dir = self._module_path / old_module_name
if old_dir.is_dir():
try:
old_dir.rename(self._module_path / self._module_name)
except OSError as err:
warn(
f"Unable to move module directory {old_dir} to "
f"{self._module_path / self._module_name}: {err}. Reload the project to "
"recreate the module under the new name.",
RuntimeWarning,
2,
)
async def _load_data(self, set_config: bool) -> object:
"""Load the project from PDA."""
_timing.note("project_id", self._doc.id)
_timing.note("project_name", self.name)
_timing.note("technologies", len(self._data["technologies"]))
_timing.note("components", len(self._data["components"]))
_timing.note("libraries", len(self._data["libraries"]))
limit = asyncio.Semaphore(_PDA_BATCH_CONCURRENCY)
async def find_repo_document(ref: DocumentId | ReadOnlyURL) -> Document:
async with limit:
return await _find_repo_document(ref)
async def load_technology(
doc_ref: ReadOnlyURL,
) -> tuple[DocumentId, Document, dict[str, object]]:
doc_id = AutomergeRef.parse(doc_ref).document_id
doc = await find_repo_document(doc_ref)
with _timing.time_block("load_data.technology_read"):
obj_data = doc.read()
return doc_id, doc, obj_data
async def load_component(
flagged_component: Mapping[str, object],
) -> tuple[DocumentId, dict[str, object], Document, bool]:
ui_ref = flagged_component["ref"]
ui_id = AutomergeRef.parse(ui_ref).document_id
ui_doc = await find_repo_document(ui_ref)
with _timing.time_block("load_data.ui_read"):
ui_data = ui_doc.read()
return (
ui_id,
await _load_component_native_data(ui_data),
ui_doc,
flagged_component.get("external", False),
)
async def load_library(lib_ref: ReadOnlyURL) -> Project | None:
document = await find_repo_document(lib_ref)
try:
with _timing.time_block("load_data.library_project"):
return Project(document)
except ValidationError:
return None
self._technologies = {}
self._components = {}
parent_ids = [self._doc.id] + [
AutomergeRef.parse(lib_ref).document_id for lib_ref in self._data["libraries"]
]
with _timing.time_block("load_data.fetch"):
technology_rows, component_rows, library_rows, child_info_rows = await _gather(
_gather(*(load_technology(doc_ref) for doc_ref in self._data["technologies"])),
_limited_gather(
*(
load_component(flagged_component)
for flagged_component in self._data["components"]
)
),
_gather(*(load_library(lib_ref) for lib_ref in self._data["libraries"])),
_gather(
*(
_list_documents(
types=[_technology_type, _ui_component_type], parent_id=parent_id
)
for parent_id in parent_ids
),
*(
_list_documents(
types=[_technology_type, _ui_component_type],
parent_id=parent_id,
deleted=True,
)
for parent_id in parent_ids
),
),
)
parent_by_doc_id = {
info["documentId"]: info["parentId"]
for child_infos in child_info_rows
for info in child_infos
}
for doc_id, doc, _ in technology_rows:
parent_id = parent_by_doc_id.get(doc_id)
if parent_id is None:
info = await _get_document(doc_id)
parent_id = info["parentId"]
self._technologies[doc_id] = _TechnologyData(doc, None, parent_id)
for ui_id, obj_data, ui_doc, external in component_rows:
parent_id = parent_by_doc_id.get(ui_id)
if parent_id is None:
info = await _get_document(ui_id)
parent_id = info["parentId"]
self._components[ui_id] = _ComponentData(obj_data, ui_doc, None, parent_id, external)
await self._reimport_cached_objects(set_config)
self._libraries = {}
for lib in library_rows:
if lib is None:
continue
self._libraries[lib._doc.id] = lib
def _load_module(
self,
module_path: str | Path,
recursive: bool,
create_template: bool,
reserved_module_names: set[str] | None = None,
) -> None:
"""Materialize and load the project Python module on disk.
Args:
module_path: Root directory where modules are unpacked.
recursive: Whether dependency library modules are also loaded.
create_template: Whether to create a starter module when no
module archive exists.
"""
resolved_module_path = Path(module_path).resolve()
stored_name, attachment_id, inline_archive = _python_module(self._data.get("pythonModule"))
module_archive = None
if attachment_id is not None:
with _timing.time_block("load_module.download"):
module_archive = _run(_get_attachment_content(attachment_id))
elif inline_archive is not None:
module_archive = inline_archive
if module_archive is not None:
archive_name = _module_archive_name(module_archive)
if stored_name is None and archive_name is not None:
self._module_name = archive_name
elif archive_name is not None and archive_name != self._module_name:
raise RuntimeError(
f"Python module archive contains {archive_name!r}, "
f"expected {self._module_name!r}."
)
if reserved_module_names is not None and self._module_name in reserved_module_names:
raise RuntimeError(f"Module name {self._module_name!r} is already in use.")
_check_module_name_available(self._module_name, self._module_path, resolved_module_path)
self._module_path = resolved_module_path
if str(self._module_path) not in sys.path:
sys.path.insert(0, str(self._module_path))
if recursive:
module_names = {self._module_name}
for lib in self._libraries.values():
lib._load_module(self._module_path, False, False, module_names)
module_names.add(lib._module_name)
path = self._module_path / self._module_name
previous = None
if path.exists():
i = 0
while previous is None or previous.exists():
i += 1
previous = self._module_path / f"{self._module_name}.old{i}"
path.replace(previous)
path.mkdir(parents=True)
if module_archive is not None:
_extract_module_archive(module_archive, path, self._module_name)
elif create_template:
(path / "README.md").write_text(f"""# {self.name} module
Files in this directory make up the project/library module. Add custom parametric technologies and
components here to make them available when the project is loaded.
See existing files as examples.""")
(path / "__init__.py").write_text("from .component import example_component")
(path / "component.py").write_text(
'''import photonforge as pf
import photonforge.typing as pft
@pf.parametric_component(name_prefix="EXAMPLE")
def example_component(
*,
length: pft.PositiveDimension = 100,
) -> pf.Component:
"""Create an example parametric component.
Args:
length: Size of the square geometry.
Returns:
Component.
"""
c = pf.Component()
c.add(pf.Rectangle(size=(length, length)))
return c'''
)
# If current sources match previous, delete previous.
if previous is not None and previous.is_dir() and path.is_dir():
cmp = dircmp(path, previous, ignore=["__pycache__"])
if all(
len(getattr(cmp, attr)) == 0
for attr in ("left_only", "right_only", "common_funny", "diff_files", "funny_files")
):
for root, dirs, files in os.walk(previous, topdown=False):
root = Path(root)
for name in files:
(root / name).unlink()
for name in dirs:
(root / name).rmdir()
previous.rmdir()
[docs]
def save_module(self) -> None:
"""Pack the loaded module directory and store it in PDA.
The module includes all files in the directory
``project.module_path / project.module_name``.
"""
_run(self._save_module())
async def _save_module(self) -> None:
with _timing.operation("save_module", project_id=self.id, project_name=self.name):
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
if self._module_path is None:
raise RuntimeError(
"Project module not loaded. Make sure the project was created using the "
"functions 'create_project' or 'load_project'."
)
module_dir = self._module_path / self._module_name
if not module_dir.is_dir():
raise RuntimeError(f"Module directory {module_dir!s} does not exist.")
archive = io.BytesIO()
with (
_timing.time_block("save_module.pack_zip"),
zipfile.ZipFile(archive, mode="w", compression=zipfile.ZIP_DEFLATED) as zf,
):
for root, dirs, files in os.walk(module_dir, topdown=True):
root = Path(root)
dirs[:] = sorted(d for d in dirs if d != "__pycache__")
for filename in sorted(files):
if filename.endswith(".pyc"):
continue
file_path = root / filename
arcname = file_path.relative_to(self._module_path)
zf.write(file_path, arcname.as_posix())
module_archive = archive.getvalue()
_timing.record("save_module.bytes", bytes_count=len(module_archive))
with _timing.time_block("save_module.upload"):
attachment_id = await self._upload_module_archive(module_archive)
self._data["pythonModule"] = [self._module_name, attachment_id]
with _timing.time_block("save_module.change_project_doc"):
self._doc.replace({"pythonModule": self._data["pythonModule"]})
async def _upload_module_archive(self, module_archive: bytes) -> str:
record = await _upload_attachment(
module_archive,
_PYTHON_MODULE_FILENAME,
_PYTHON_MODULE_MIME_TYPE,
document_id=self.id,
)
if record.get("documentId") != self.id:
raise RuntimeError("Module attachment upload returned a mismatched documentId.")
return record["id"]
def _bind_module_objects(self, module: ModuleType, parent_id: DocumentId) -> None:
module.pda_components = _ProjectObjectMap(self, "components", parent_id)
module.pda_technologies = _ProjectObjectMap(self, "technologies", parent_id)
[docs]
def import_module(self, namespace: dict | None, *, reload: bool = True) -> dict[str, object]:
"""Import the project and library Python modules.
Args:
namespace: Optional mapping to populate with imported modules.
reload: Whether to hard-reload project/library modules from disk.
Returns:
Mapping from module name to imported module object.
Each imported module gets two read-only live mappings:
``pda_components`` and ``pda_technologies``.
Example:
Import parametric components from the project and its libraries
in the current namespace (equivalent to using
``import project_module`` for each dependency of the project):
>>> project.import_module(globals()) # doctest: +SKIP
"""
projects = (*self._libraries.values(), self)
for project in projects:
_check_module_name_available(project._module_name, project._module_path)
if reload:
importlib.invalidate_caches()
for lib in self._libraries.values():
_purge_module(lib._module_name, lib._module_path)
_purge_module(self._module_name, self._module_path)
modules = {}
for lib in self._libraries.values():
module = importlib.import_module(lib._module_name)
self._bind_module_objects(module, lib._doc.id)
modules[lib._module_name] = module
module = importlib.import_module(self._module_name)
self._bind_module_objects(module, self._doc.id)
modules[self._module_name] = module
if namespace is not None:
namespace.update(modules)
return modules
def _resolve_origin(self, origin: str | DocumentId) -> Iterable[DocumentId]:
if origin in ("self", self.name, self._module_name):
return [self._doc.id]
elif origin == "libraries":
return self._libraries.keys()
elif origin in self._libraries:
return [origin]
else:
for lib_id, lib in self._libraries.items():
if lib.name == origin or lib._module_name == origin:
return [lib_id]
raise RuntimeError(f"Origin {origin!r} does not map to any library in the project.")
def _filter_contents(
self,
values: Iterable[object],
name: str | None,
search: str | re.Pattern | None,
origin: str | DocumentId | None,
) -> (
dict[str, dict[str, Technology | Component]]
| dict[str, Technology | Component]
| Technology
| Component
):
"""Apply name/search/origin filters to component or technology sets.
Args:
values: Iterable of internal object metadata records.
name: Exact object name filter.
search: Regex filter for object names.
origin: Source scope selector.
Returns:
Filtered objects, grouped by origin when appropriate.
"""
data = {}
for item_data in values:
d = data.get(item_data.parent_id, {})
d[item_data.obj.name] = item_data.obj
data[item_data.parent_id] = d
single_origin = origin not in (None, "libraries")
if origin is not None:
data = {k: data[k] for k in self._resolve_origin(origin) if k in data}
if name is not None:
filtered = {}
for origin, origin_results in data.items():
if name in origin_results:
filtered[origin] = origin_results[name]
data = filtered
elif search is not None:
if not isinstance(search, re.Pattern):
search = re.compile(search)
filtered = {}
for origin, origin_results in data.items():
matches = {k: v for k, v in origin_results.items() if search.search(k) is not None}
if len(matches) > 0:
filtered[origin] = matches
data = filtered
if single_origin:
if len(data) == 1:
data = next(iter(data.values()))
else:
data = {
self._libraries[k].name if k in self._libraries else self.name: v
for k, v in data.items()
}
return data
[docs]
def technologies(
self,
*,
name: str | None = None,
search: str | re.Pattern | None = None,
origin: str | DocumentId | None = None,
) -> dict[str, dict[str, Technology]] | dict[str, Technology] | Technology:
"""Query technologies by exact name, regex, and origin scope.
Args:
name: Exact technology name filter.
search: Regex or pattern string filter for names.
origin: Source scope ("self", "libraries", or library name).
Returns:
Filtered technology objects grouped by origin when needed.
"""
return self._filter_contents(self._technologies.values(), name, search, origin)
[docs]
def components(
self,
*,
name: str | None = None,
search: str | re.Pattern | None = None,
origin: str | DocumentId | None = None,
external_only: bool = True,
) -> dict[str, dict[str, Component]] | dict[str, Component] | Component:
"""Query components by exact name, regex, and origin scope.
Args:
name: Exact component name filter.
search: Regex or pattern string filter for names.
origin: Source scope ("self", "libraries", or library name).
external_only: Whether listings include only components marked
external. Ignored when ``name`` or ``search`` is provided.
Returns:
Filtered component objects grouped by origin when needed.
"""
values = self._components.values()
if external_only and name is None and search is None:
values = (data for data in values if data.external)
return self._filter_contents(values, name, search, origin)
[docs]
def set_external(self, component: Component | str, external: bool = True) -> None:
"""Set/unset a component as external in the module mapping.
Args:
component: Component object or component name to set.
external: Set/unset flag.
"""
return _run(self._set_external(component, external))
async def _set_external(self, component: Component | str, external: bool = True):
"""Set/unset a component as external in the module mapping.
Args:
component: Component object or component name to set.
external: Set/unset flag.
"""
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
component, ui_doc = self._find_target(component)
if not isinstance(component, Component):
raise TypeError("Only components can be set/unset external.")
for index, flagged_component in enumerate(self._data["components"]):
if AutomergeRef.parse(flagged_component["ref"]).document_id == ui_doc.id:
value = flagged_component.get("external", False)
if external and not value:
self._components[ui_doc.id] = _ComponentData(
*self._components[ui_doc.id][:-1], True
)
flagged_component["external"] = True
with _timing.time_block("set_external.change_project_doc"):
with self._doc.change() as project_data:
project_data["components"][index]["external"] = True
elif not external and value:
self._components[ui_doc.id] = _ComponentData(
*self._components[ui_doc.id][:-1], False
)
del flagged_component["external"]
with _timing.time_block("set_external.change_project_doc"):
with self._doc.change() as project_data:
del project_data["components"][index]["external"]
break
def _reject_technology(self, obj: object, op: str) -> None:
if isinstance(obj, Technology):
raise RuntimeError(
f"Cannot {op} on a technology; technology attachments are not supported yet."
)
[docs]
def attach(
self,
path: str | Path,
*,
target: Component | str | None = None,
filename: str | None = None,
tags: Sequence[str] = (),
) -> dict[str, object]:
"""Upload a file and attach it to the project or one of its components.
The file is uploaded and bound to the target document on the server,
which grants read access to anyone with access to that document. The
binding is the association: nothing is stored in the document itself.
Args:
path: Path to the file to upload.
target: Component (object or name) to attach to; ``None`` (default)
attaches to the project itself. Technologies are not supported.
filename: Optional file name override (defaults to the path name).
tags: Optional tags stored on the attachment. Use a tag convention
(e.g. "design_manual", "drc_ruleset", "cross_section", "gds") to
categorize attachments and filter them with ``attachments(tags=)``.
Returns:
Attachment metadata for the uploaded file.
"""
return _run(self._attach(path, target=target, filename=filename, tags=tags))
async def _attach(
self,
path: str | Path,
*,
target: Component | str | None = None,
filename: str | None = None,
tags: Sequence[str] = (),
) -> dict[str, object]:
obj, doc = self._find_target(target)
self._reject_technology(obj, "attach")
path = Path(path)
with _timing.operation("attach", project_id=self.id, filename=filename or path.name):
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
name = filename or path.name
mime_type = mimetypes.guess_type(name)[0] or "application/octet-stream"
internal_keys = self._component_internal_attachment_keys()
if obj is self:
internal_keys = {(_PYTHON_MODULE_FILENAME, _PYTHON_MODULE_MIME_TYPE)}
if (name, mime_type) in internal_keys:
raise ValueError(
f"{name!r} collides with an internal attachment name; rename the file."
)
with _timing.time_block("attach.read_file"):
content = path.read_bytes()
_timing.record("attach.file_bytes", bytes_count=len(content))
# The document binding set at upload is the single source of truth for the
# attachment-document association; nothing is written to the CRDT document.
record = await _upload_attachment(
content, name, mime_type, list(tags) or None, document_id=doc.id
)
if record.get("documentId") != doc.id:
raise RuntimeError("Attachment upload returned a mismatched documentId.")
return record
[docs]
def attachments(
self,
*,
target: Component | str | None = None,
tags: Sequence[str] = (),
all: bool = False,
) -> list[dict[str, object]]:
"""List attachments bound to the project or one of its components.
Args:
target: Component (object or name) to list attachments for; ``None``
(default) lists the project's own attachments. Technologies are not
supported.
tags: Optional tag filter; only attachments carrying every given tag
are returned.
all: When ``True``, list attachments across the project and every
tracked component. Each returned record carries an extra
``owner`` field (``{"kind": "project"}`` or ``{"kind":
"component", "name": ...}``). Mutually exclusive with ``target``.
Returns:
Attachment metadata records (as returned by ``attach()``), each
with ``id``, ``filename``, ``mimeType``, ``tags``, and (when
``all=True``) an extra ``owner`` field.
"""
return _run(self._attachments(target=target, tags=tags, all=all))
@staticmethod
def _component_internal_attachment_keys() -> set[tuple[str, str]]:
return {
(_PF_NATIVE_FILENAME, _PF_NATIVE_MIME_TYPE),
(_PREVIEW_FILENAME, _PREVIEW_MIME_TYPE),
}
async def _document_snapshots(self, document: Document) -> list[dict[str, object]]:
snapshots = [document.read()]
latest = document.latest().read()
if latest != snapshots[0]:
snapshots.append(latest)
versions = await _list_versions(document.id)
tags = await _list_tags(None, document.id)
refs = {info["ref"] for info in (*versions, *tags) if isinstance(info.get("ref"), str)}
for ref in refs:
snapshots.append((await _find_repo_document(ref)).read())
return snapshots
async def _module_attachment_ids(self) -> set[str]:
attachment_ids = set()
for data in await self._document_snapshots(self._doc):
_, attachment_id, _ = _python_module(data.get("pythonModule"))
if attachment_id is not None:
attachment_ids.add(attachment_id)
return attachment_ids
async def _component_attachment_ids(self, document: Document) -> set[str]:
attachment_ids = set()
for data in await self._document_snapshots(document):
native_id = data.get("pfNative")
if isinstance(native_id, str):
attachment_ids.add(native_id)
preview = data.get("preview")
if isinstance(preview, str) and preview.startswith("att:"):
attachment_ids.add(preview.removeprefix("att:"))
return attachment_ids
async def _user_attachments(
self, document: Document, *, project: bool
) -> list[dict[str, object]]:
records = await _list_attachments(document_id=document.id)
if not project:
internal = self._component_internal_attachment_keys()
current = [
record
for record in records
if (record.get("filename"), record.get("mimeType")) not in internal
]
if len(current) == 0:
return current
attachment_ids = await self._component_attachment_ids(document)
return [record for record in current if record.get("id") not in attachment_ids]
current = [
record
for record in records
if (record.get("filename"), record.get("mimeType"))
!= (_PYTHON_MODULE_FILENAME, _PYTHON_MODULE_MIME_TYPE)
]
if not any(
record.get("mimeType") in (_PYTHON_MODULE_MIME_TYPE, _LEGACY_PYTHON_MODULE_MIME_TYPE)
for record in current
):
return current
attachment_ids = await self._module_attachment_ids()
return [record for record in current if record.get("id") not in attachment_ids]
async def _attachments(
self,
*,
target: Component | str | None = None,
tags: Sequence[str] = (),
all: bool = False,
) -> list[dict[str, object]]:
if all:
if target is not None:
raise ValueError("'all=True' cannot be combined with a 'target'.")
records = await self._all_attachments()
else:
obj, doc = self._find_target(target)
self._reject_technology(obj, "list attachments")
records = await self._user_attachments(doc, project=obj is self)
if tags:
wanted = set(tags)
records = [rec for rec in records if wanted <= set(rec.get("tags") or [])]
return records
async def _all_attachments(self) -> list[dict[str, object]]:
owners = [(self._doc, {"kind": "project"}, True)]
for component_data in self._components.values():
# self._components also holds components pulled in from imported libraries; those have
# a different parent_id. Only list this project's own components here.
if component_data.parent_id != self._doc.id:
continue
name = component_data.obj.name if component_data.obj is not None else None
owners.append((component_data.ui_doc, {"kind": "component", "name": name}, False))
record_lists = await _limited_gather(
*(self._user_attachments(doc, project=project) for doc, _, project in owners)
)
return [
{**record, "owner": owner}
for (_, owner, _), records in zip(owners, record_lists, strict=True)
for record in records
]
[docs]
def download_attachment(self, attachment: Mapping[str, object] | str) -> bytes:
"""Download the binary content of an attachment.
Args:
attachment: Raw attachment ID, attachment reference from
``attachments()``, or metadata returned by ``attach()``.
Returns:
Raw file bytes.
"""
return _run(self._download_attachment(attachment))
async def _download_attachment(self, attachment: Mapping[str, object] | str) -> bytes:
attachment_id = _attachment_id(attachment)
with _timing.operation("download_attachment", project_id=self.id):
return await _get_attachment_content(attachment_id)
[docs]
def detach(
self,
attachment: Mapping[str, object] | str,
*,
target: Component | str | None = None,
delete: bool = True,
) -> None:
"""Remove an attachment from the project or one of its components.
Args:
attachment: Raw attachment ID, attachment reference from
``attachments()``, or metadata returned by ``attach()``.
target: Component (object or name) the attachment is on; ``None``
(default) targets the project. Technologies are not supported.
delete: Soft-delete the attachment (default). When ``False`` the
attachment is only unlinked from the target and kept.
"""
return _run(self._detach(attachment, target=target, delete=delete))
async def _detach(
self,
attachment: Mapping[str, object] | str,
*,
target: Component | str | None = None,
delete: bool = True,
) -> None:
obj, doc = self._find_target(target)
self._reject_technology(obj, "detach")
with _timing.operation("detach", project_id=self.id, delete=delete):
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
attachment_id = _attachment_id(attachment)
# Guard against unlinking or soft-deleting a blob that is not a user attachment of
# this target (an ID copied from another document, or infrastructure data like the
# component's native blob).
current = await self._user_attachments(doc, project=obj is self)
if not any(record.get("id") == attachment_id for record in current):
raise ValueError(
f"Attachment {attachment_id!r} is not attached to the specified target."
)
if delete:
await _delete_attachment(attachment_id)
else:
await _update_attachment(attachment_id, document_id="")
[docs]
def add(
self,
new_obj: Component | Technology,
*,
tag: str | None = None,
bump_version: Literal["major", "minor", "patch"] | None = None,
update_existing_dependencies: bool = True,
update_config: bool = True,
project_tag: str | None = "{datetime:%Y%m%d-%H%M%S}-{name}-{version_and_tag}",
set_external: bool = True,
) -> dict[Literal["added", "updated"], list[Technology | Component]]:
"""Add an object and any missing dependencies to the project.
Args:
new_obj: Component or technology to add.
tag: Optional tag applied to changed objects.
bump_version: Optional semantic bump for changed objects.
update_existing_dependencies: Whether existing dependencies may be
updated.
update_config: Whether to update project config from current
settings.
project_tag: If either ``tag`` or ``bump_version`` is not ``None``,
tag to the project using this template.
set_external: Whether to mark this component to be external in the
module ``pda_components`` mapping. Dependencies are never marked.
Returns:
Mapping with ``added`` and ``updated`` object lists.
"""
return _run(
self._add_or_update(
True,
new_obj,
tag,
bump_version,
update_existing_dependencies,
update_config,
project_tag,
set_external,
False,
)
)
[docs]
def update(
self,
upd_obj: Component | Technology,
*,
tag: str | None = None,
bump_version: Literal["major", "minor", "patch"] | None = None,
update_existing_dependencies: bool = True,
update_config: bool = True,
project_tag: str | None = "{datetime:%Y%m%d-%H%M%S}-{name}-{version_and_tag}",
) -> dict[Literal["added", "updated"], list[Technology | Component]]:
"""Update an existing project object and optional dependencies.
Args:
upd_obj: Component or technology to update.
tag: Optional tag applied to changed objects.
bump_version: Optional semantic bump for changed objects.
update_existing_dependencies: Whether existing dependencies may
be updated.
update_config: Whether to update project config from current
setting.
project_tag: If either ``tag`` or ``bump_version`` is not ``None``,
tag to the project using this template.
Returns:
Mapping with ``added`` and ``updated`` object lists.
"""
return _run(
self._add_or_update(
False,
upd_obj,
tag,
bump_version,
update_existing_dependencies,
update_config,
project_tag,
False,
False,
)
)
async def _add_or_update(
self,
add: bool,
new_obj: Component | Technology,
tag: str | None,
bump_version: Literal["major", "minor", "patch"] | None,
update_existing_dependencies: bool,
update_config: bool,
project_tag: str | None,
set_external: bool,
keep_existing_ids: bool,
) -> dict[Literal["added", "updated"], list[Technology | Component]]:
with _timing.operation(
"add" if add else "update",
project_id=self.id,
object_name=getattr(new_obj, "name", None),
object_type=type(new_obj).__name__,
):
return await self._add_or_update_impl(
add,
new_obj,
tag,
bump_version,
update_existing_dependencies,
update_config,
project_tag,
set_external,
keep_existing_ids,
)
async def _add_or_update_impl(
self,
add: bool,
new_obj: Component | Technology,
tag: str | None,
bump_version: Literal["major", "minor", "patch"] | None,
update_existing_dependencies: bool,
update_config: bool,
project_tag: str | None,
set_external: bool,
keep_existing_ids: bool,
) -> dict[Literal["added", "updated"], list[Technology | Component]]:
"""Internal implementation for add/update synchronization.
Args:
add: ``True`` to add, ``False`` to update.
new_obj: Primary component or technology to synchronize.
tag: Optional tag applied to changed objects.
bump_version: Optional semantic bump for changed objects.
update_existing_dependencies: Whether existing dependencies may be
updated.
update_config: Whether to replace project config from export.
project_tag: If either ``tag`` or ``bump_version`` is not ``None``,
tag to the project using this template.
set_external: Whether to mark this component to be external in the
module ``pda_components`` mapping. Dependencies are never marked.
keep_existing_ids: Used by the backend only.
Returns:
Mapping with ``added`` and ``updated`` object lists.
"""
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
self_id = self._doc.id
tech_by_name = {
t.obj.name: t.obj for t in self._technologies.values() if t.parent_id == self_id
}
comp_by_name = {
c.obj.name: c.obj for c in self._components.values() if c.parent_id == self_id
}
current_tech = {k: v for k, v in self._technologies.items() if v.parent_id == self_id}
current_comp = {k: v for k, v in self._components.items() if v.parent_id == self_id}
if isinstance(new_obj, Technology):
if add:
if not keep_existing_ids:
new_obj._pda_id = ""
elif new_obj._pda_id not in current_tech:
if new_obj.name in tech_by_name and not keep_existing_ids:
new_obj._pda_id = tech_by_name[new_obj.name]._pda_id
else:
raise RuntimeError(f"{new_obj.name!r} not found in project. Use 'Project.add'.")
elif isinstance(new_obj, Component):
if add:
if not keep_existing_ids:
new_obj._pda_id = ""
elif new_obj._pda_id not in current_comp:
if new_obj.name in comp_by_name and not keep_existing_ids:
new_obj._pda_id = comp_by_name[new_obj.name]._pda_id
else:
raise RuntimeError(f"{new_obj.name!r} not found in project. Use 'Project.add'.")
else:
raise TypeError("Only Component and Technology can be added or updated.")
primary_update_id = new_obj._pda_id
future_technologies = {k: v.obj for k, v in current_tech.items()}
future_components = {k: v.obj for k, v in current_comp.items()}
module_names = {lib._module_name for lib in self._libraries.values()}
module_names.update([self._module_name, "photonforge"])
# Export to gather all required dependencies and create missing IDs
with _timing.time_block("add_update.export_initial"):
objects, data = await asyncio.to_thread(_export, new_obj, use_json=True)
data = json_loads(data)
sorted_initial_data = sort_exported_data(objects, data)
_timing.note("exported_objects", len(objects))
preserved_dependency_ids = set()
copied_dependency_ids = set()
component_dependency_ids = {}
for obj_id, _, obj in sorted_initial_data:
if not isinstance(obj, Component):
continue
dependencies = {
reference.component._pda_id
for reference in obj.references
if len(reference.component._pda_id) > 0
}
if len(obj.technology._pda_id) > 0:
dependencies.add(obj.technology._pda_id)
component_dependency_ids[obj_id] = dependencies
for obj_id, obj_data, obj in sorted_initial_data:
if (
obj.parametric_function is not None
and obj.parametric_function.partition(".")[0] not in module_names
):
warn(
f"The parametric function for {obj.name!r} ({obj.parametric_function}) does "
f"not come from a library within this project. Updating this component or "
f" technology will not be possible without the original source.",
RuntimeWarning,
2,
)
if isinstance(obj, Technology):
if obj is new_obj:
# obj will be added or updated
future_technologies[obj_id] = obj
continue
id_match = current_tech.get(obj._pda_id, None)
name_match = tech_by_name.get(obj.name, None)
library_match = self._technologies.get(obj._pda_id)
is_library_dependency = (
library_match is not None and library_match.parent_id != self_id
)
if id_match is not None:
if update_existing_dependencies:
# obj will be updated
future_technologies[obj_id] = obj
elif name_match is not None:
if is_library_dependency:
copied_dependency_ids.add(obj._pda_id)
obj._pda_id = name_match._pda_id
if update_existing_dependencies:
# obj will be updated
future_technologies[obj._pda_id] = obj
elif keep_existing_ids and is_library_dependency:
# We do not add library objects as project-owned when keep_existing_ids == True,
# because they are read-only
preserved_dependency_ids.add(obj._pda_id)
elif is_library_dependency and library_match.doc.read() == obj_data:
# Do not duplicate library objects into the project if they have not changed
preserved_dependency_ids.add(obj._pda_id)
else:
if is_library_dependency:
copied_dependency_ids.add(obj._pda_id)
# obj will be added
future_technologies[obj_id] = obj
elif isinstance(obj, Component):
if obj is new_obj:
# obj will be added
future_components[obj_id] = obj
continue
id_match = current_comp.get(obj._pda_id, None)
name_match = comp_by_name.get(obj.name, None)
library_match = self._components.get(obj._pda_id)
is_library_dependency = (
library_match is not None and library_match.parent_id != self_id
)
if id_match is not None:
if update_existing_dependencies:
# obj will be updated
future_components[obj_id] = obj
elif name_match is not None and not keep_existing_ids:
if is_library_dependency:
copied_dependency_ids.add(obj._pda_id)
obj._pda_id = name_match._pda_id
if update_existing_dependencies:
# obj will be updated
future_components[obj._pda_id] = obj
elif keep_existing_ids and is_library_dependency:
# We do not add library objects as project-owned when keep_existing_ids == True,
# because they are read-only
preserved_dependency_ids.add(obj._pda_id)
elif (
is_library_dependency
and not component_dependency_ids.get(obj_id, set()) & copied_dependency_ids
and library_match.native_data == obj_data
):
# Do not duplicate library objects into the project if they and their
# dependencies have not changed
preserved_dependency_ids.add(obj._pda_id)
else:
if is_library_dependency:
copied_dependency_ids.add(obj._pda_id)
# obj will be added
future_components[obj_id] = obj
else:
raise TypeError(f"Unsupported object in exported data: {obj!r}.")
_check_unique_names(future_technologies, "technologies")
_check_unique_names(future_components, "components")
create_limit = asyncio.Semaphore(_PDA_DOC_CREATE_CONCURRENCY)
async def create_object_docs(obj: Technology | Component):
async with create_limit:
return await create_object_docs_unlimited(obj)
async def create_object_docs_unlimited(obj: Technology | Component):
if obj._pda_id in preserved_dependency_ids:
return None
if isinstance(obj, Technology):
if obj._pda_id in current_tech:
return None
if keep_existing_ids and len(obj._pda_id) > 0:
doc_id = obj._pda_id
else:
doc_info = await _create_document(_technology_type, self_id)
doc_id = doc_info["documentId"]
obj._pda_id = doc_id
doc = await _find_repo_document(doc_id)
return "technology", doc_id, doc, obj
if obj._pda_id in current_comp:
return None
if keep_existing_ids and len(obj._pda_id) > 0:
ui_doc_id = obj._pda_id
else:
ui_doc_info = await _create_document(_ui_component_type, self_id)
ui_doc_id = ui_doc_info["documentId"]
obj._pda_id = ui_doc_id
ui_doc = await _find_repo_document(ui_doc_id)
return "component", ui_doc_id, ui_doc, obj
# Create new documents for all new objects
new_ids = set()
with _timing.time_block("add_update.create_object_docs"):
created_docs = await _limited_gather(
*(create_object_docs(obj) for obj in objects.values())
)
for created in created_docs:
if created is None:
continue
if created[0] == "technology":
_, doc_id, doc, obj = created
self._technologies[doc_id] = _TechnologyData(doc, obj, self_id)
new_ids.add(doc_id)
else:
_, ui_doc_id, ui_doc, obj = created
self._components[ui_doc_id] = _ComponentData({}, ui_doc, obj, self_id, False)
new_ids.add(ui_doc_id)
# Now re-export using existing doc ids for cross-referencing
with _timing.time_block("add_update.export_with_ids"):
objects, data = await asyncio.to_thread(_export, new_obj, use_json=True)
data = json_loads(data)
sorted_data = sort_exported_data(objects, data)
component_refs = {
AutomergeRef.parse_id(r): r
for r in (*self._data["technologies"], *(x["ref"] for x in self._data["components"]))
}
component_schemas_before = self._data["componentSchemas"].copy()
model_schemas_before = self._data["modelSchemas"].copy()
project_technology_appends: list[str] = []
project_technology_updates: list[tuple[int, str]] = []
project_component_appends: list[dict[str, object]] = []
project_component_ref_updates: list[tuple[int, str]] = []
project_config_update: dict[str, object] | None = None
external_set_in_append = False
# Add contents to new and updated docs
added = []
updated = []
updated_component_ids = set()
for obj_id, obj_data, obj in sorted_data:
if obj_id in preserved_dependency_ids:
continue
is_new = obj_id in new_ids
is_primary_update = obj_id == primary_update_id
if not is_new and not update_existing_dependencies and not is_primary_update:
continue
if isinstance(obj, Technology):
doc, _, parent_id = self._technologies[obj_id]
with _timing.time_block("add_update.read_existing_doc"):
is_update = not is_new and not doc.read() == obj_data
if is_new or is_update:
with _timing.time_block("add_update.change_doc"):
doc = doc.latest()
doc.change(obj_data)
self._technologies[obj_id] = _TechnologyData(doc.read_only(), obj, parent_id)
if is_new:
ref = doc.ref
self._data["technologies"].append(ref)
project_technology_appends.append(ref)
added.append(obj)
elif is_update:
ref = doc.ref
i = self._data["technologies"].index(component_refs[obj_id])
self._data["technologies"][i] = ref
project_technology_updates.append((i, ref))
updated.append(obj)
else:
native_data, ui_doc, _, parent_id, external = self._components[obj_id]
native_changed = is_new or native_data != obj_data
is_update = not is_new and (
native_changed
or any(
reference.component._pda_id in updated_component_ids
for reference in obj.references
)
)
if is_new or is_update:
previous_ui_ref = ui_doc.ref
ui_doc = ui_doc.latest()
existing_ui = None
try:
with _timing.time_block("add_update.read_existing_ui"):
existing_ui = UiComponent.model_validate(ui_doc.read())
except Exception as err:
if not is_new:
warn(
f"Unable to get existing UI data for {obj.name!r}: {err}",
RuntimeWarning,
3,
)
with _timing.time_block("component_ui.convert"):
component = await component_from_pf(
obj,
model_schemas=self._data["modelSchemas"],
component_schemas=self._data["componentSchemas"],
available_refs=component_refs,
project_id=self_id,
document_id=ui_doc.id,
existing_ui=existing_ui,
allow_schema_creation=True,
)
component.pfRef = obj.parametric_function
if native_changed or existing_ui is None or existing_ui.pfNative is None:
component.pfNative = await _upload_component_native_data(
ui_doc.id, obj_data
)
else:
component.pfNative = existing_ui.pfNative
with _timing.time_block("add_update.change_ui_doc"):
ui_doc.change(component.model_dump(mode="json"))
ref = ui_doc.ref
component_refs[obj_id] = ref
new_external = external or (is_new and set_external and obj is new_obj)
self._components[obj_id] = _ComponentData(
obj_data, ui_doc.read_only(), obj, parent_id, new_external
)
updated_component_ids.add(obj_id)
if is_new:
info = {"ref": ref}
if new_external:
info["external"] = True
external_set_in_append = True
self._data["components"].append(info)
project_component_appends.append(info.copy())
added.append(obj)
elif is_update:
for index, updated_component in enumerate(self._data["components"]):
if updated_component["ref"] == previous_ui_ref:
updated_component["ref"] = ref
project_component_ref_updates.append((index, ref))
break
updated.append(obj)
if update_config and self._data["config"] != data["config"]:
self._data["config"] = data["config"]
project_config_update = self._data["config"]
project_component_schema_updates = {
key: doc_id
for key, doc_id in self._data["componentSchemas"].items()
if component_schemas_before.get(key) != doc_id
}
project_model_schema_updates = {
key: doc_id
for key, doc_id in self._data["modelSchemas"].items()
if model_schemas_before.get(key) != doc_id
}
if (
project_technology_appends
or project_technology_updates
or project_component_appends
or project_component_ref_updates
or project_component_schema_updates
or project_model_schema_updates
or project_config_update is not None
):
with _timing.time_block("add_update.change_project_doc"):
with self._doc.change() as project_data:
technologies = project_data["technologies"]
for index, ref in project_technology_updates:
technologies[index] = ref
for ref in project_technology_appends:
technologies.append(ref)
components = project_data["components"]
for index, ref in project_component_ref_updates:
components[index]["ref"] = ref
for info in project_component_appends:
components.append(info)
component_schemas = project_data["componentSchemas"]
if component_schemas is None:
project_data["componentSchemas"] = {}
component_schemas = project_data["componentSchemas"]
for key, doc_id in project_component_schema_updates.items():
component_schemas[key] = _to_automerge_value(doc_id)
model_schemas = project_data["modelSchemas"]
if model_schemas is None:
project_data["modelSchemas"] = {}
model_schemas = project_data["modelSchemas"]
for key, doc_id in project_model_schema_updates.items():
model_schemas[key] = _to_automerge_value(doc_id)
if project_config_update is not None:
project_data["config"] = project_config_update
if set_external and isinstance(new_obj, Component) and not external_set_in_append:
await self._set_external(new_obj, True)
changed_docs = [
self._technologies[obj._pda_id].doc
if isinstance(obj, Technology)
else self._components[obj._pda_id].ui_doc
for obj in (*added, *updated)
]
version = ""
version_and_tag = None
if tag is None:
tag = ""
else:
with _timing.time_block("add_update.add_tags"):
await _limited_gather(*(_add_tag(doc.ref, tag) for doc in changed_docs))
version_and_tag = tag
if bump_version is not None:
with _timing.time_block("add_update.add_versions"):
await _limited_gather(
*(_add_version(doc.ref, None, bump_version) for doc in changed_docs)
)
with _timing.time_block("add_update.list_latest_versions"):
versions = await _list_latest_versions([new_obj._pda_id])
if len(versions) == 1:
version = versions[0]["version"]
version_and_tag = version if version_and_tag is None else f"{version}-{tag}"
if project_tag is not None and version_and_tag is not None and len(changed_docs) > 0:
project_tag = project_tag.format(
datetime=datetime.now().astimezone(),
name=new_obj.name,
version_and_tag=version_and_tag,
version=version,
tag=tag,
)
with _timing.time_block("add_update.add_project_tag"):
await _add_tag(self._doc.ref, project_tag)
if len(added) > 0 or len(updated) > 0:
cached_objects = await self._reimport_cached_objects(False)
added = [cached_objects[obj._pda_id] for obj in added]
updated = [cached_objects[obj._pda_id] for obj in updated]
return {"added": added, "updated": updated}
[docs]
def retire_unused(self) -> None:
"""Retire project-owned internal components unused by external ones."""
_run(self._retire_unused())
async def _retire_unused(self) -> None:
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
self_id = self._doc.id
internal_components = {
obj_id: data
for obj_id, data in self._components.items()
if data.parent_id == self_id and not data.external
}
if len(internal_components) == 0:
return
external_components = [
data.obj
for data in self._components.values()
if data.parent_id == self_id and data.external
]
if len(external_components) == 0:
raise RuntimeError(
"retire_unused() requires at least one external component to define reachability."
)
exported, _ = await asyncio.to_thread(_export, *external_components)
used_ids = {obj._pda_id for obj in exported.values()}
retired_ids = set(internal_components).difference(used_ids)
if len(retired_ids) > 0:
await self._retire([internal_components[obj_id].obj for obj_id in retired_ids])
[docs]
def retire(self, targets: Sequence[Component | Technology | str]) -> None:
"""Retire technologies/components by object instance or name.
Objects still referenced by remaining project content are kept and
reported through a runtime warning.
Args:
targets: Components/technologies (or names) to retire.
Returns:
``None``.
"""
_run(self._retire(targets))
async def _retire(self, targets: Sequence[Component | Technology | str]) -> None:
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
self_id = self._doc.id
tech_by_name = {
v.obj.name: (k, v) for k, v in self._technologies.items() if v.parent_id == self_id
}
comp_by_name = {
v.obj.name: (k, v) for k, v in self._components.items() if v.parent_id == self_id
}
retire_technologies = {}
retire_components = {}
for target in targets:
if isinstance(target, Technology):
data = self._technologies.get(target._pda_id)
obj_id = None
if data is None:
obj_id, data = tech_by_name.get(target.name, (None, None))
if data is None or data.parent_id != self_id or target != data.obj:
raise RuntimeError(
f"Only technologies from this project can be retired. {target.name!r} not "
"found (must match exactly)."
)
if obj_id is not None:
data.obj._pda_id = obj_id
retire_technologies[data.obj._pda_id] = data
elif isinstance(target, Component):
data = self._components.get(target._pda_id)
obj_id = None
if data is None:
obj_id, data = comp_by_name.get(target.name, (None, None))
if data is None or data.parent_id != self_id or target != data.obj:
raise RuntimeError(
f"Only components from this project can be retired. {target.name!r} not"
f"found (must match exactly)."
)
if obj_id is not None:
data.obj._pda_id = obj_id
retire_components[data.obj._pda_id] = data
elif isinstance(target, str):
obj_id, data = comp_by_name.get(target, (None, None))
if data is None:
obj_id, data = tech_by_name.get(target, (None, None))
if data is None:
raise RuntimeError(
f"{target!r} not found among project components or technologies."
)
else:
data.obj._pda_id = obj_id
retire_technologies[obj_id] = data
else:
data.obj._pda_id = obj_id
retire_components[obj_id] = data
else:
raise TypeError("Expected Component, Technology, or string name.")
retire_ids = set(retire_components) | set(retire_technologies)
keep_objects = [
data.obj
for obj_id, data in (*self._components.items(), *self._technologies.items())
if data.parent_id == self_id and obj_id not in retire_ids
]
exported, _ = await asyncio.to_thread(_export, *keep_objects)
exported_ids = {obj._pda_id for obj in exported.values() if len(obj._pda_id) > 0}
used_component_ids = set(retire_components).intersection(exported_ids)
used_technology_ids = set(retire_technologies).intersection(exported_ids)
if len(used_component_ids) > 0 or len(used_technology_ids) > 0:
used_names = sorted(
retire_components[obj_id].obj.name for obj_id in used_component_ids
) + sorted(retire_technologies[obj_id].obj.name for obj_id in used_technology_ids)
warn(
"Objects in use cannot be retired: "
+ ", ".join(repr(name) for name in used_names)
+ ".",
RuntimeWarning,
2,
)
retired_component_ids = set(retire_components).difference(used_component_ids)
retired_technology_ids = set(retire_technologies).difference(used_technology_ids)
if len(retired_component_ids) == 0 and len(retired_technology_ids) == 0:
return
with _timing.time_block("retire.change_project_doc"):
with self._doc.change() as project_data:
components = project_data["components"]
if components is not None:
for index in range(len(components) - 1, -1, -1):
ref = components[index].get("ref")
if AutomergeRef.parse(str(ref)).document_id in retired_component_ids:
del components[index]
technologies = project_data["technologies"]
if technologies is not None:
for index in range(len(technologies) - 1, -1, -1):
if (
AutomergeRef.parse(str(technologies[index])).document_id
in retired_technology_ids
):
del technologies[index]
self._data["components"] = [
x
for x in self._data["components"]
if AutomergeRef.parse(x["ref"]).document_id not in retired_component_ids
]
self._data["technologies"] = [
ref
for ref in self._data["technologies"]
if AutomergeRef.parse(ref).document_id not in retired_technology_ids
]
delete_tasks = []
for obj_id in retired_component_ids:
data = self._components.pop(obj_id)
delete_tasks.append(_delete_document(data.ui_doc.id))
for obj_id in retired_technology_ids:
data = self._technologies.pop(obj_id)
delete_tasks.append(_delete_document(data.doc.id))
await _limited_gather(*delete_tasks)
[docs]
def add_tag(
self, tag: str, *, target: Component | Technology | str | None = None
) -> Component | Technology | Project:
"""Attach a tag to the project or a specific contained object.
Args:
tag: Tag label to create.
target: Optional target object or name. Defaults to project.
Returns:
Target object that received the tag.
"""
obj, doc = self._find_target(target)
_run(_add_tag(doc.ref, tag))
return obj
[docs]
def remove_tag(
self, tag: str | Sequence[str], *, target: Component | Technology | str | None = None
) -> Component | Technology | Project:
"""Remove one or more tags from the selected target.
Args:
tag: Tag label or labels to remove.
target: Optional target object or name. Defaults to project.
Returns:
Target object from which tags were removed.
"""
return _run(self._remove_tag(tag, target=target))
async def _remove_tag(
self, tag: str | Sequence[str], *, target: Component | Technology | str | None = None
) -> Component | Technology | Project:
if isinstance(tag, str):
tag = [tag]
obj, doc = self._find_target(target)
tag_ids = [
tag_data["id"] for tag_data in await _list_tags(None, doc.id) if tag_data["tag"] in tag
]
await _gather(*(_delete_tag(tag_id) for tag_id in tag_ids))
return obj
[docs]
def add_version(
self,
version: str | None = None,
bump_version: Literal["major", "minor", "patch"] | None = None,
*,
target: Component | Technology | str | None = None,
) -> Component | Technology | Project:
"""Create a version for the selected target document reference.
Args:
version: Explicit semantic version string.
bump_version: Semantic bump strategy when ``version`` is
omitted.
target: Optional target object or name. Defaults to project.
Returns:
Target object that received the version.
"""
obj, doc = self._find_target(target)
_run(_add_version(doc.ref, version, bump_version))
return obj
[docs]
def list_permissions(self) -> list[dict[str, object]]:
"""List sharing permissions for the project document.
Args:
None.
Returns:
Permission metadata list.
"""
return _run(_list_permissions(self.id))
[docs]
def grant_permission(
self,
*,
visibility: Literal["private", "organization"],
grantee_id: str | None = None,
role: Literal["owner", "editor", "viewer"] = "viewer",
) -> dict[str, object]:
"""Grant access permissions for the project document.
Args:
visibility: Permission visibility scope.
grantee_id: User or organization ID who will receive the permission.
role: Access role to grant.
Returns:
Created permission payload.
Notes:
For organization visibility, ``grantee_id`` defaults to the current
organization when available.
Important:
Setting public visibility is not allowed, as it shares the contents
and all dependencies with **all** users in the platform. Please
contact support if you want to share a public library.
"""
if visibility == "public":
if grantee_id is not None:
raise ValueError("'grantee_id' must be None when visibility is 'public'.")
if role != "viewer":
raise ValueError("'role' must be 'viewer' when visibility is 'public'.")
elif visibility == "organization":
if not isinstance(grantee_id, str) or len(grantee_id.strip()) == 0:
_, grantee_id = user_info()
if not isinstance(grantee_id, str) or len(grantee_id.strip()) == 0:
raise RuntimeError(
"Current session has no organization. Cannot grant organization permission "
"without providing 'grantee_id'."
)
elif not isinstance(grantee_id, str) or len(grantee_id.strip()) == 0:
raise ValueError("'grantee_id' must be a non-empty string for private or organization.")
return _run(_grant_permission(self.id, visibility, grantee_id, role))
[docs]
def update_permission(
self, permission_id: str, *, role: Literal["editor", "viewer"]
) -> dict[str, object]:
"""Change the role associated with an existing permission.
Args:
permission_id: Permission identifier.
role: New role value.
Returns:
Updated permission payload.
"""
return _run(_update_permission(self.id, permission_id, role))
[docs]
def revoke_permission(self, permission_id: str) -> None:
"""Revoke one permission by its identifier.
Args:
permission_id: Permission identifier.
Returns:
``None``.
"""
_run(_revoke_permission(self.id, permission_id))
[docs]
def transfer_ownership(self, grantee_id: str) -> dict[str, object]:
"""Transfer project ownership to another user identifier.
Args:
grantee_id: Identifier of the new owner.
Returns:
Updated owner permission payload.
"""
if not isinstance(grantee_id, str) or len(grantee_id) == 0:
raise ValueError("'grantee_id' must be a non-empty string with the new owner's ID.")
return _run(_transfer_ownership(self.id, grantee_id))
[docs]
def check_access(self) -> tuple[bool, dict[str, object]]:
"""Check if the current user can access this project.
Args:
None.
Returns:
Tuple ``(has_access, payload)`` from the access endpoint.
"""
return _run(_check_access(self.id))
async def _load_obj_ref(
self, doc_id: DocumentId, doc_ref: ReadOnlyURL
) -> Component | Technology:
"""Load one object from a specific reference snapshot.
Args:
doc_id: Document ID of the selected object in this project.
doc_ref: Reference URL for the selected snapshot.
Returns:
Loaded component or technology object.
"""
data = {}
top_content = []
for obj_id, obj_data in self._technologies.items():
if obj_id == doc_id:
top_content.append((obj_id, 11))
doc = await _find_repo_document(doc_ref)
else:
doc = obj_data.doc
data[obj_id] = doc.read()
for obj_id, obj_data in self._components.items():
if obj_id == doc_id:
top_content.append((obj_id, 26))
ui_doc = await _find_repo_document(doc_ref)
data[obj_id] = await _load_component_native_data(ui_doc.read())
else:
data[obj_id] = obj_data.native_data
if len(top_content) == 0:
raise RuntimeError("Target object not found in project.")
json_data = {
"type": 1,
"type_version": "0.0",
"config": self._data["config"],
"top_content": top_content,
"data": data,
"properties": None,
}
byte_repr = json_dumps(json_data, ensure_ascii=False).encode("utf-8")
objects, _ = await asyncio.to_thread(
_import, byte_repr=byte_repr, set_config=False, use_json=True
)
return objects[0]
[docs]
def load(
self,
target: Component | Technology | str,
*,
tag: str | None = None,
version: str | None = None,
) -> Component | Technology:
"""Load a tagged/versioned snapshot of a project object.
Args:
target: Component/technology object or exact name.
tag: Optional tag filter.
version: Optional version filter.
Returns:
Object snapshot matching the search parameters.
"""
return _run(self._load(target, tag=tag, version=version))
async def _load(
self,
target: Component | Technology | str,
*,
tag: str | None = None,
version: str | None = None,
) -> Component | Technology:
if target is None:
raise ValueError(
"Invalid target. If you want to load a specific version of this project, use "
"'load_project' instead."
)
_, doc = self._find_target(target)
refs = set()
if tag is not None and version is not None:
tag_infos, version_infos = await _gather(
_list_tags(None, doc.id),
_list_versions(doc.id),
)
refs = {info["ref"] for info in tag_infos if info["tag"] == tag}.intersection(
info["ref"] for info in version_infos if info["version"] == version
)
elif tag is not None:
refs = {info["ref"] for info in await _list_tags(None, doc.id) if info["tag"] == tag}
elif version is not None:
refs = {
info["ref"] for info in await _list_versions(doc.id) if info["version"] == version
}
if len(refs) != 1:
raise RuntimeError(f"{len(refs)} references matching search parameters found.")
return await self._load_obj_ref(doc.id, next(iter(refs)))
[docs]
def add_library(
self,
name: str | None = None,
*,
version: str | None = None,
library_id: str | None = None,
) -> object:
"""Attach a versioned library and its dependencies to the project.
Args:
name: Library project name when selecting by name/version.
version: Library version when selecting by name/version.
library_id: Specific library version record ID.
Returns:
``None``.
"""
return _run(self._add_library(name, version=version, library_id=library_id))
async def _add_library(
self,
name: str | None = None,
*,
version: str | None = None,
library_id: str | None = None,
) -> object:
if self._doc.is_read_only:
raise RuntimeError(f"Read only project/library {self.name!r} cannot be modified.")
library, library_id = await _select_library(name, version, library_id, False)
if library._doc.id in self._libraries:
raise RuntimeError(
f"{library.name!r} already in project. Library updates will be implemented in the "
f"future."
)
module_names = {lib._module_name for lib in self._libraries.values()}
module_names.add(self._module_name)
all_libraries = []
for lib_ref in library._data["libraries"]:
document = await _find_repo_document(lib_ref)
try:
lib = Project(document)
except ValidationError:
continue
match = self._libraries.get(lib._doc.id)
if match is None:
all_libraries.append(lib)
elif lib._doc.ref != match._doc.ref:
raise RuntimeError(
f"{lib.name!r} cannot be added as a dependency because it is incompatible with "
f"{match.name!r} already in the project (different versions)."
)
# Make sure it loads last (depends on sub-libraries)
all_libraries.append(library)
for lib in all_libraries:
await lib._load_data(False)
await asyncio.to_thread(lib._load_module, self._module_path, False, False, module_names)
module_names.add(lib._module_name)
schema_updates: dict[str, dict[str, str]] = {"componentSchemas": {}, "modelSchemas": {}}
async def merge_schemas(field: str, schema_kind: str) -> None:
for lib in all_libraries:
for key, schema_doc_id in lib._data[field].items():
existing_doc_id = self._data[field].get(key)
if existing_doc_id is None:
self._data[field][key] = schema_doc_id
schema_updates[field][key] = schema_doc_id
continue
if existing_doc_id == schema_doc_id:
continue
existing_doc = await _find_repo_document(existing_doc_id)
schema_doc = await _find_repo_document(schema_doc_id)
if existing_doc is None or schema_doc is None:
raise RuntimeError(
f"Unable to resolve {schema_kind} schema {key!r} while adding "
f"library {lib.name!r}."
)
schema_selection = _select_schema(existing_doc.read(), schema_doc.read())
if schema_selection is None:
raise RuntimeError(
f"Library {lib.name!r} cannot be added because {schema_kind} schema "
f"{key!r} is incompatible with the version already in the project."
)
if schema_selection == 1:
self._data[field][key] = schema_doc_id
schema_updates[field][key] = schema_doc_id
await merge_schemas("componentSchemas", "component")
await merge_schemas("modelSchemas", "model")
library_appends = [lib._doc.ref for lib in all_libraries]
self._data["libraries"].extend(library_appends)
self._libraries.update({lib._doc.id: lib for lib in all_libraries})
component_appends: list[dict[str, object]] = []
existing = {x["ref"] for x in self._data["components"]}
for x in library._data["components"]:
if x["ref"] not in existing:
info = {"ref": x["ref"]}
if x.get("external", False):
info["external"] = True
self._data["components"].append(info)
component_appends.append(info)
self._components.update(
{k: v for k, v in library._components.items() if k not in self._components}
)
existing = set(self._data["technologies"])
technology_appends = [ref for ref in library._data["technologies"] if ref not in existing]
self._data["technologies"].extend(technology_appends)
self._technologies.update(
{k: v for k, v in library._technologies.items() if k not in self._technologies}
)
with _timing.time_block("add_library.change_project_doc"):
with self._doc.change() as project_data:
for field, updates in schema_updates.items():
schemas = project_data[field]
if schemas is None:
project_data[field] = {}
schemas = project_data[field]
for key, schema_doc_id in updates.items():
schemas[key] = _to_automerge_value(schema_doc_id)
libraries = project_data["libraries"]
if libraries is None:
project_data["libraries"] = []
libraries = project_data["libraries"]
for ref in library_appends:
libraries.append(_to_automerge_value(ref))
components = project_data["components"]
if components is None:
project_data["components"] = []
components = project_data["components"]
for info in component_appends:
components.append(_to_automerge_value(info))
technologies = project_data["technologies"]
if technologies is None:
project_data["technologies"] = []
technologies = project_data["technologies"]
for ref in technology_appends:
technologies.append(_to_automerge_value(ref))
def _resolved_create_permission(
visibility: Literal["private", "organization", "public"],
role: Literal["editor", "viewer"] | None,
) -> tuple[Literal["organization", "public"] | None, Literal["editor", "viewer"] | None]:
"""Normalize create-time permission arguments into grant payload values.
Args:
visibility: Requested visibility at project creation.
role: Optional role for non-private visibility.
Returns:
Tuple ``(visibility, role)`` for ``Project.grant_permission``.
"""
if visibility == "private":
if role is not None:
raise ValueError("'role' cannot be used with visibility='private'.")
return None, None
if visibility == "public":
if role is None:
role = "viewer"
if role != "viewer":
raise ValueError("'public' permissions only support role='viewer'.")
return visibility, role
if role is None:
role = "viewer"
return visibility, role
[docs]
def create_project(
name: str,
*,
description: str | None = None,
labels: Sequence[str] = (),
visibility: Literal["private", "organization"] = "private",
role: Literal["editor", "viewer"] | None = None,
module_path: str = _DEFAULT_MODULE_PATH,
create_template: bool = True,
) -> Project:
"""Create a new PDA project from current PhotonForge configuration.
Args:
name: Project name.
description: Optional project description.
labels: Project labels.
visibility: Initial sharing visibility.
role: Initial sharing role for non-private visibility.
module_path: Root path where modules are unpacked.
create_template: Whether to generate a starter module.
Returns:
Loaded project object.
The initial project includes exported config dependencies and an
optional template Python module directory.
Important:
Setting public visibility is not allowed, as it shares the contents
and all dependencies with **all** users in the platform. Please
contact support if you want to share a public library.
"""
return _run(
_create_project(
name,
description=description,
labels=labels,
visibility=visibility,
role=role,
module_path=module_path,
create_template=create_template,
)
)
def sort_exported_data(
objects: dict[str, Technology | Component], data: dict[str, object]
) -> list[tuple[str, dict[str, object], Technology | Component]]:
sorted_data = []
remaining = {}
for obj_id, obj_data in data["data"].items():
obj = objects[obj_id]
if isinstance(objects[obj_id], Technology) or len(obj.references) == 0:
sorted_data.append((obj_id, obj_data, obj))
else:
remaining[obj_id] = (obj_data, obj)
progress = len(sorted_data)
while len(remaining) > 0:
for obj_id, (obj_data, obj) in remaining.items():
if all(x.component._pda_id not in remaining for x in obj.references):
sorted_data.append((obj_id, obj_data, obj))
if len(sorted_data) == progress:
raise RuntimeError("Dependency cycle found in exported data.")
for obj_id, *_ in sorted_data[progress:]:
remaining.pop(obj_id)
progress = len(sorted_data)
return sorted_data
async def _create_project(
name: str,
*,
description: str | None = None,
labels: Sequence[str] = (),
visibility: Literal["private", "organization"] = "private",
role: Literal["editor", "viewer"] | None = None,
module_path: str = _DEFAULT_MODULE_PATH,
create_template: bool = True,
) -> Project:
with _timing.operation("create_project", name=name, visibility=visibility):
return await _create_project_impl(
name,
description=description,
labels=labels,
visibility=visibility,
role=role,
module_path=module_path,
create_template=create_template,
)
async def _create_project_impl(
name: str,
*,
description: str | None = None,
labels: Sequence[str] = (),
visibility: Literal["private", "organization"] = "private",
role: Literal["editor", "viewer"] | None = None,
module_path: str = _DEFAULT_MODULE_PATH,
create_template: bool = True,
) -> Project:
resolved_visibility, resolved_role = _resolved_create_permission(visibility, role)
grant_id = None
if resolved_visibility == "organization":
with _timing.time_block("create_project.user_info"):
_, grant_id = await _user_info()
if grant_id is None:
raise RuntimeError(
"Current session has no organization. Cannot grant organization permission."
)
# Create components and technologies used in config defaults.
# If they come from another project (already have document IDs), we overwrite
# them with new IDs because they can only belong to a single project.
with _timing.time_block("create_project.export_defaults"):
objects, _ = await asyncio.to_thread(_export)
if not all(isinstance(obj, (Component, Technology)) for obj in objects.values()):
raise RuntimeError("Unexpected type in config data.")
original_ids = [(obj, obj._pda_id) for obj in objects.values()]
project_id = None
try:
with _timing.time_block("create_project.create_root"):
project_info = await _create_document(_project_type, None)
project_id = project_info["documentId"]
_timing.note("project_id", project_id)
project_technologies = {}
project_components = {}
project_data = {
"name": name,
"description": description,
"labels": list(labels),
"applications": [],
"libraries": [],
"components": [],
"technologies": [],
"config": {},
"componentSchemas": {},
"modelSchemas": {},
}
# Create the documents ahead of time to define PDA IDs
with _timing.time_block("create_project.create_object_docs"):
for obj in objects.values():
if isinstance(obj, Technology):
doc_info = await _create_document(_technology_type, project_id)
doc_id = doc_info["documentId"]
doc = await _find_repo_document(doc_id)
obj._pda_id = doc_id
project_technologies[doc_id] = doc
else:
ui_doc_info = await _create_document(_ui_component_type, project_id)
ui_doc_id = ui_doc_info["documentId"]
ui_doc = await _find_repo_document(ui_doc_id)
obj._pda_id = ui_doc_id
project_components[ui_doc_id] = ui_doc
# Now re-export using doc ids for cross-referencing from _pda_id
with _timing.time_block("create_project.export_with_ids"):
objects, data = await asyncio.to_thread(_export, use_json=True)
data = json_loads(data)
sorted_data = sort_exported_data(objects, data)
component_refs = {}
technologies = []
with _timing.time_block("create_project.write_object_docs"):
for obj_id, obj_data, obj in sorted_data:
if isinstance(obj, Technology):
doc = project_technologies[obj_id]
doc.change(obj_data)
technologies.append(doc.ref)
project_technologies[obj_id] = _TechnologyData(doc.read_only(), obj, project_id)
else:
ui_doc = project_components[obj_id]
with _timing.time_block("component_ui.convert"):
component = await component_from_pf(
obj,
model_schemas=project_data["modelSchemas"],
component_schemas=project_data["componentSchemas"],
available_refs=component_refs,
project_id=project_id,
document_id=ui_doc.id,
allow_schema_creation=True,
)
component.pfRef = obj.parametric_function
component.pfNative = await _upload_component_native_data(ui_doc.id, obj_data)
ui_doc.change(component.model_dump(mode="json"))
component_refs[obj_id] = ui_doc.ref
project_components[obj_id] = _ComponentData(
obj_data, ui_doc.read_only(), obj, project_id, False
)
with _timing.time_block("create_project.find_root"):
document = await _find_repo_document(project_id)
project_data["components"] = [{"ref": ref} for ref in component_refs.values()]
project_data["technologies"] = technologies
project_data["config"] = data["config"]
with _timing.time_block("create_project.write_root"):
document.change(project_data)
project = Project(document)
project._technologies = project_technologies
project._components = project_components
if resolved_visibility is not None and resolved_role is not None:
with _timing.time_block("create_project.grant_permission"):
await _grant_permission(project.id, resolved_visibility, grant_id, resolved_role)
with _timing.time_block("create_project.load_module"):
await asyncio.to_thread(project._load_module, module_path, True, create_template)
return project
except BaseException as error:
for obj, original_id in original_ids:
obj._pda_id = original_id
if project_id is not None:
try:
with _timing.time_block("create_project.rollback"):
await asyncio.shield(_delete_document(project_id))
except Exception as rollback_error:
if isinstance(error, asyncio.CancelledError):
raise error from rollback_error
raise RuntimeError(
f"Project creation failed and project {project_id!r} could not be retired: "
f"{rollback_error}"
) from error
raise
[docs]
def list_projects(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
retired: bool = False,
) -> list[dict[str, object]]:
"""List projects matching metadata, sharing, and retirement filters.
Args:
name: Exact project name filter.
labels: Label filter; any overlap matches.
visibility: Visibility filter.
role: Permission role filter.
retired: Whether to search retired projects.
Returns:
Project metadata list.
"""
return _run(
_list_projects(
name,
labels=labels,
visibility=visibility,
role=role,
retired=retired,
)
)
async def _list_projects(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
retired: bool = False,
) -> list[dict[str, object]]:
with _timing.operation("list_projects", name=name, retired=retired):
return await _list_projects_impl(
name,
labels=labels,
visibility=visibility,
role=role,
retired=retired,
)
async def _list_projects_impl(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
retired: bool = False,
) -> list[dict[str, object]]:
result = []
labels = set(labels)
for project_data in await _list_documents(types=[_project_type], deleted=retired):
try:
document = await _find_repo_document(project_data["documentId"])
except RuntimeError as err:
warn(
f"Skipping unloadable project {project_data['documentId']}: {err}",
RuntimeWarning,
stacklevel=2,
)
continue
try:
project = Project(document)
except ValidationError:
continue
if (
(name is None or project._data.get("name") == name)
and (len(labels) == 0 or len(labels.intersection(project._data.get("labels", ()))) > 0)
and (role is None or project_data.get("role") == role)
and (visibility is None or project_data.get("visibility") == visibility)
):
project_data.update({"name": project.name, "labels": project.labels})
result.append(project_data)
return result
[docs]
def list_libraries(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
latest: bool = True,
retired: bool = False,
) -> list[dict[str, object]]:
"""List versioned project libraries with optional metadata filters.
Args:
name: Exact library name filter.
labels: Label filter; any overlap matches.
visibility: Visibility filter.
role: Permission role filter.
latests: If ``True``, only the latest version of each library is
returned.
retired: Whether to search retired libraries.
Returns:
Library metadata list.
"""
return _run(
_list_libraries(
name,
labels=labels,
visibility=visibility,
role=role,
latest=latest,
retired=retired,
)
)
async def _list_libraries(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
latest: bool = True,
retired: bool = False,
) -> list[dict[str, object]]:
with _timing.operation("list_libraries", name=name, latest=latest, retired=retired):
return await _list_libraries_impl(
name,
labels=labels,
visibility=visibility,
role=role,
latest=latest,
retired=retired,
)
async def _list_libraries_impl(
name: str | None = None,
*,
labels: Sequence[str] = (),
visibility: Literal["private", "organization", "public"] | None = None,
role: Literal["owner", "editor", "viewer"] | None = None,
latest: bool = True,
retired: bool = False,
) -> list[dict[str, object]]:
result = []
labels = set(labels)
for library_data in await _list_libs(latest=latest, deleted=retired):
try:
document = await _find_repo_document(library_data["documentId"])
except RuntimeError as err:
# Storage may contain a stub/empty snapshot for a half-created
# project (DB row exists, content was never written). Skip rather
# than aborting the whole listing so callers can still discover
# healthy libraries.
warn(
f"Skipping unloadable library {library_data['documentId']}: {err}",
RuntimeWarning,
stacklevel=2,
)
continue
try:
library = Project(document)
except ValidationError:
continue
if (
(name is None or library._data.get("name") == name)
and (len(labels) == 0 or len(labels.intersection(library._data.get("labels", ()))) > 0)
and (role is None or library_data.get("role") == role)
and (visibility is None or library_data.get("visibility") == visibility)
):
library_data.update({"name": library.name, "labels": library.labels})
result.append(library_data)
return result
async def _select_project(
name: str | None,
version: str | None,
tag: str | None,
project_id: DocumentId | ReadOnlyURL | None,
deleted: bool,
) -> Project:
"""Select one project by name/id and optional version/tag filters.
Args:
name: Project name used when ``project_id`` is not provided.
version: Optional version filter.
tag: Optional tag filter.
project_id: Project document ID or reference URL.
deleted: Whether to search retired projects.
Returns:
Matched project.
"""
if project_id is None:
if name is None:
raise RuntimeError("Either 'name' or 'project_id' must be defined.")
result = await _list_documents(types=[_project_type], deleted=deleted)
matches = []
for project_data in result:
try:
document = await _find_repo_document(project_data["documentId"])
except RuntimeError as err:
warn(
f"Skipping unloadable project {project_data['documentId']}: {err}",
RuntimeWarning,
stacklevel=2,
)
continue
try:
project = Project(document)
except ValidationError:
continue
if project._data.get("name") == name:
matches.append(project)
if len(matches) == 0:
raise RuntimeError(f"Project {name!r} not found.")
else:
if name is not None:
warn("Project name ignored. Using 'project_id' only.", RuntimeWarning, 3)
document = await _find_repo_document(project_id)
try:
matches = [Project(document)]
except ValidationError as err:
raise RuntimeError(f"Project {project_id!r} has invalid schema.") from err
refs = None
if version is not None:
refs = {
info["ref"]
for project in matches
for info in await _list_versions(project.id)
if info["version"] == version
}
if len(refs) == 0:
raise RuntimeError(f"Version {version} not found.")
if tag is not None:
tag_refs = {
info["ref"]
for project in matches
for info in await _list_tags(None, project.id)
if info["tag"] == tag
}
if len(tag_refs) == 0:
raise RuntimeError(f"Tag {tag!r} not found.")
refs = tag_refs if refs is None else refs.intersection(tag_refs)
if len(refs) == 0:
raise RuntimeError(
f"No project snapshot matching version {version} and tag {tag!r} found."
)
if refs is None:
if len(matches) != 1:
raise RuntimeError(f"{len(matches)} projects matching search parameters found.")
return matches[0]
if len(refs) != 1:
raise RuntimeError(f"{len(refs)} project snapshots matching search parameters found.")
return Project(await _find_repo_document(next(iter(refs))))
[docs]
def load_project(
name: str | None = None,
*,
project_id: DocumentId | ReadOnlyURL | None = None,
version: str | None = None,
tag: str | None = None,
module_path: str = _DEFAULT_MODULE_PATH,
set_config: bool = True,
create_template: bool = False,
) -> Project:
"""Load one project by name or document reference.
Args:
name: Project name to load.
project_id: Project document ID or reference URL.
version: Optional version filter.
tag: Optional tag filter.
module_path: Root path where modules are unpacked.
set_config: Whether to apply project config to PhotonForge.
create_template: Whether to create a starter module if absent.
Returns:
Loaded project object.
"""
return _run(
_load_project(
name,
project_id=project_id,
version=version,
tag=tag,
module_path=module_path,
set_config=set_config,
create_template=create_template,
)
)
async def _load_project(
name: str | None = None,
*,
project_id: DocumentId | ReadOnlyURL | None = None,
version: str | None = None,
tag: str | None = None,
module_path: str = _DEFAULT_MODULE_PATH,
set_config: bool = True,
create_template: bool = False,
) -> Project:
with _timing.operation(
"load_project", name=name, project_id=project_id, version=version, tag=tag
):
with _timing.time_block("load_project.select"):
project = await _select_project(name, version, tag, project_id, False)
_timing.note("project_id", project.id)
_timing.note("project_name", project.name)
with _timing.time_block("load_project.load_data"):
await project._load_data(set_config)
with _timing.time_block("load_project.load_module"):
await asyncio.to_thread(project._load_module, module_path, True, create_template)
return project
[docs]
def retire_project(
name: str | None = None, *, project_id: DocumentId | ReadOnlyURL | None = None
) -> object:
"""Retire a project document by name or identifier.
Args:
name: Project name.
project_id: Project document ID or reference URL.
Returns:
``None``.
Important:
Projects that are released (given a version) cannot be retired.
"""
return _run(_retire_project(name, project_id=project_id))
async def _retire_project(
name: str | None = None, *, project_id: DocumentId | ReadOnlyURL | None = None
) -> object:
project = await _select_project(name, None, None, project_id, False)
await _delete_document(project.id)
# def restore_project(
# name: str | None = None, *, project_id: DocumentId | ReadOnlyURL | None = None
# ) -> object:
# """Restore a previously retired project document.
#
# Args:
# name: Project name.
# project_id: Project document ID or reference URL.
#
# Returns:
# ``None``.
# """
# return _run(_restore_project(name, project_id=project_id))
#
#
# async def _restore_project(
# name: str | None = None, *, project_id: DocumentId | ReadOnlyURL | None = None
# ) -> object:
# project = await _select_project(name, None, None, project_id, True)
# await _restore_document(project.id)
async def _select_library(
name: str | None, version: str | None, library_id: str | None, deleted: bool
) -> (Project, str):
"""Resolve a specific library version and version record ID.
Args:
name: Library project name.
version: Library semantic version string.
library_id: Explicit library version record ID.
deleted: Whether to search retired projects.
Returns:
Tuple ``(library_project, library_version_id)``.
"""
if library_id is None:
if name is None or version is None:
raise RuntimeError("Either 'name' and 'version', or 'library_id' must be defined.")
document_ids = [
project_info["documentId"]
for project_info in await _list_projects(name=name, retired=deleted)
]
version_infos = await _list_versions(document_ids) if document_ids else []
matches = [
library_info for library_info in version_infos if library_info["version"] == version
]
if len(matches) == 0:
raise RuntimeError(f"Library {name!r} version {version} not found.")
if len(matches) > 1:
raise RuntimeError(
f"{len(matches)} libraries with name {name!r} and version {version} found."
)
library_id = matches[0]["id"]
document = await _find_repo_document(matches[0]["ref"])
library = Project(document)
else:
if name is not None or version is not None:
warn("Library name and version ignored. Using 'library_id' only.", RuntimeWarning, 3)
library_data = await _get_version(library_id)
document = await _find_repo_document(library_data["ref"])
library = Project(document)
return library, library_id