forked from 1b-agent/opencadstudio-upstream
General improvements on API
This commit is contained in:
Sebastian committed
1 parent
56e09cd3f5
commit
55f61a4327
30 files changed
+1820
-2565
No files matched your search
@@ -1,166 +1,198 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate a Python facade over a caller-supplied document transport.
|
||||
"""Reference binding generator for ocs_doc_api (plan §10.2).
|
||||
|
||||
Usage: python generate.py ../../src/gen/binding_schema.json > ocs_doc.py
|
||||
The transport converts these Python DTOs to/from the Rust wire representation.
|
||||
Reads `gen/binding_schema.json` and emits a Python module `ocs_doc.py` with the
|
||||
object model (DocApi -> Document -> collections -> handles) whose methods build
|
||||
`DocApiEnvelope` ops/queries and ship them through a caller-supplied transport.
|
||||
|
||||
This is the *contract* in executable form (bindings/README.md): one versioned
|
||||
schema, no Rust FFI, thin serializer. Lua/JS generators follow the same shape.
|
||||
|
||||
Usage:
|
||||
python bindings/python/generate.py gen/binding_schema.json > ocs_doc.py
|
||||
|
||||
The emitted module expects a `transport` object with:
|
||||
def apply(self, envelope: DocApiEnvelope) -> Receipt
|
||||
which the bridge author implements (the session glue over the plugin channel).
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any
|
||||
|
||||
HEADER = '''"""Generated document facade. See bindings/README.md for the transport contract."""
|
||||
HEADER = '''"""Auto-generated ocs_doc_api Python bindings. Do NOT edit by hand.
|
||||
Generated by bindings/python/generate.py from binding_schema.json (envelope_version {ver}).
|
||||
Check envelope_version before use; see bindings/README.md."""
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from typing import Any, List, Optional
|
||||
|
||||
@dataclass
|
||||
class DocApiEnvelope:
|
||||
version: int
|
||||
body: Any
|
||||
# --- wire DTOs (thin, schema-driven) -----------------------------------------
|
||||
ObjectId = int # u64
|
||||
|
||||
@dataclass
|
||||
class Receipt:
|
||||
outcome: Any
|
||||
query_results: list
|
||||
query_results: List[Any]
|
||||
new_revision: int
|
||||
|
||||
@dataclass
|
||||
class Entity:
|
||||
id: int
|
||||
_doc: "Document"
|
||||
class DocApiEnvelope:
|
||||
version: int
|
||||
body: Any # ("Op", operation_dict) | ("Queries", [query_dict])
|
||||
|
||||
def bounds(self):
|
||||
return self._doc._apply_query({"GetBounds": {"id": self.id}})
|
||||
ENVELOPE_VERSION = {ver}
|
||||
|
||||
def transform(self, placement):
|
||||
self._doc._apply_op({"Transform": {"id": self.id, "placement": placement}})
|
||||
|
||||
def delete(self):
|
||||
self._doc._apply_op({"Delete": {"id": self.id}})
|
||||
def _check_version(schema_version: int) -> None:
|
||||
if schema_version != ENVELOPE_VERSION:
|
||||
raise RuntimeError(
|
||||
f"binding_schema envelope_version {{schema_version}} != {{ENVELOPE_VERSION}}"
|
||||
)
|
||||
'''
|
||||
|
||||
DOCUMENT = '''
|
||||
HANDLE_TEMPLATE = '''
|
||||
@dataclass
|
||||
class {handle}:
|
||||
id: ObjectId
|
||||
_doc: "Document"
|
||||
|
||||
{methods}
|
||||
'''
|
||||
|
||||
# Rust type -> Python annotation (schema types are Rust-flavoured).
|
||||
PY_TYPES = {
|
||||
"Solid": "'Solid'", "Line": "'Line'", "Circle": "'Circle'",
|
||||
"Polyline": "'Polyline'", "Point": "'Point'", "Entity": "'Entity'",
|
||||
"ArcCurve": "'ArcCurve'", "Ellipse": "'Ellipse'", "Spline": "'Spline'",
|
||||
"Ray": "'Ray'", "XLine": "'XLine'", "Text": "'Text'", "MText": "'MText'",
|
||||
"Dimension": "'Dimension'",
|
||||
"f64": "float", "bool": "bool", "usize": "int", "i32": "int",
|
||||
"[f64; 3]": "tuple", "[f64; 2]": "tuple", "ObjectId": "ObjectId",
|
||||
"String": "str", "&str": "str", "()": "None", "None": "None",
|
||||
"Vec<(String, String)>": "list", "Vec<Vec<[f64; 2]>>": "list",
|
||||
"([f64; 3], f64)": "tuple", "Vec<Point>": "list", "Vec<ObjectId>": "list",
|
||||
}
|
||||
|
||||
def py_type(rust_ty: str) -> str:
|
||||
return PY_TYPES.get(rust_ty, "Any")
|
||||
|
||||
|
||||
def py_lit(v: Any) -> str:
|
||||
"""A JSON value as a Python literal (handles booleans/None)."""
|
||||
if v is True:
|
||||
return "True"
|
||||
if v is False:
|
||||
return "False"
|
||||
if v is None:
|
||||
return "None"
|
||||
return json.dumps(v)
|
||||
|
||||
|
||||
METHOD_OP = ''' def {name}(self{sig}) -> {ret}:
|
||||
op = {{"{variant}": {body}}}
|
||||
return self._doc._apply_op(op)
|
||||
'''
|
||||
|
||||
METHOD_QUERY = ''' def {name}(self{sig}) -> {ret}:
|
||||
q = {{"{variant}": {body}}}
|
||||
return self._doc._apply_query(q)
|
||||
'''
|
||||
|
||||
|
||||
def render_method(m: Dict[str, Any]) -> str:
|
||||
"""Emit one facade method. `self.id` is the receiver's ObjectId (`a`/`id`);
|
||||
each handle-typed arg contributes its `.id`. Fixed fields are injected from
|
||||
the schema node."""
|
||||
name = m["name"]
|
||||
args = m.get("args", [])
|
||||
sig = "".join(f", {a['name']}" for a in args)
|
||||
ret = py_type(m.get("returns") or "None")
|
||||
|
||||
# Receiver is `a` for SolidBoolean, `id` otherwise. Handle args map to .id.
|
||||
field_for_arg = {}
|
||||
for a in args:
|
||||
# a handle arg becomes `<name>_id` in the payload when the variant needs ids
|
||||
field_for_arg[a["name"]] = a["name"]
|
||||
|
||||
body_items: List[str] = []
|
||||
variant = m["op"] if m["kind"] == "op" else m["query"]
|
||||
if variant == "SolidBoolean":
|
||||
body_items.append("'op': '%s'" % (m.get("fixed", {}).get("op", "")))
|
||||
body_items.append("'a': self.id")
|
||||
if args:
|
||||
body_items.append(f"'b': {args[0]['name']}.id")
|
||||
body_items.append(f"'erase_sources': {py_lit((m.get('fixed') or {}).get('erase_sources', True))}")
|
||||
elif variant in ("GetVolume", "GetCentroid", "GetBounds", "GetEntity", "GetTextContent", "GetHatchBoundary", "GetDimensionMeasurement", "GetAttributes", "GetViewportView"):
|
||||
body_items.append("'id': self.id")
|
||||
elif variant in ("SetTextContent", "SetAttribute", "SetViewportView", "Transform", "Delete", "AddVertex"):
|
||||
# Op variants that carry the receiver ObjectId: inject it first.
|
||||
body_items.append("'id': self.id")
|
||||
for a in args:
|
||||
body_items.append(f"'{a['name']}': {a['name']}")
|
||||
elif variant == "GetIntersects":
|
||||
body_items.append("'a': self.id")
|
||||
if args:
|
||||
body_items.append(f"'b': {args[0]['name']}.id")
|
||||
else:
|
||||
for a in args:
|
||||
body_items.append(f"'{a['name']}': {a['name']}")
|
||||
body = "{" + ", ".join(body_items) + "}"
|
||||
|
||||
if m["kind"] == "op":
|
||||
return METHOD_OP.format(name=name, sig=sig, ret=ret, variant=variant, body=body)
|
||||
return METHOD_QUERY.format(name=name, sig=sig, ret=ret, variant=variant, body=body)
|
||||
|
||||
|
||||
def render(schema: Dict[str, Any]) -> str:
|
||||
out = [HEADER.format(ver=schema["envelope_version"])]
|
||||
handles = set(schema["object_model"]["handles"])
|
||||
for fam in schema["families"]:
|
||||
handle = fam.get("handle") or fam["name"]
|
||||
methods = "\n".join(render_method(m) for m in fam.get("methods", []))
|
||||
if not methods.strip():
|
||||
methods = " pass\n"
|
||||
out.append(HANDLE_TEMPLATE.format(handle=handle, methods=methods))
|
||||
out.append(
|
||||
'''
|
||||
@dataclass
|
||||
class Document:
|
||||
_transport: Any
|
||||
_tab: int
|
||||
|
||||
def _id(self, value):
|
||||
if isinstance(value, Entity):
|
||||
if value._doc._transport is not self._transport or value._doc._tab != self._tab:
|
||||
raise ValueError("handle belongs to another document")
|
||||
return value.id
|
||||
return int(value)
|
||||
|
||||
def _apply_op(self, op, want=None):
|
||||
receipt = self._transport.apply(DocApiEnvelope(ENVELOPE_VERSION, ("Op", op)))
|
||||
if want is None:
|
||||
return None
|
||||
if want.startswith("Vec<"):
|
||||
handle = globals()[want[4:-1]]
|
||||
return [handle(i, self) for i in receipt.outcome["NewIds"]]
|
||||
return globals()[want](receipt.outcome["NewId"], self)
|
||||
env = DocApiEnvelope(ENVELOPE_VERSION, ("Op", op))
|
||||
return self._transport.apply(env)
|
||||
|
||||
def _apply_query(self, query):
|
||||
receipt = self._transport.apply(DocApiEnvelope(ENVELOPE_VERSION, ("Queries", [query])))
|
||||
if len(receipt.query_results) != 1:
|
||||
raise ValueError("expected one query result")
|
||||
result = receipt.query_results[0]
|
||||
if not isinstance(result, dict) or len(result) != 1:
|
||||
raise ValueError("invalid query result")
|
||||
return next(iter(result.values()))
|
||||
|
||||
def revision(self):
|
||||
return self._apply_query("GetGeometryRevision")
|
||||
|
||||
def block_entities(self, block_name):
|
||||
return self._apply_query({"GetBlockEntities": {"block_name": block_name}})
|
||||
def _apply_query(self, q, want=None):
|
||||
env = DocApiEnvelope(ENVELOPE_VERSION, ("Queries", [q]))
|
||||
return self._transport.apply(env)
|
||||
|
||||
def solids(self): return Solids(self)
|
||||
def curves(self): return Curves(self)
|
||||
def entities(self): return Entities(self)
|
||||
|
||||
class Solids:
|
||||
def __init__(self, doc): self._doc = doc
|
||||
class Curves:
|
||||
def __init__(self, doc): self._doc = doc
|
||||
class Entities:
|
||||
def __init__(self, doc): self._doc = doc
|
||||
|
||||
@dataclass
|
||||
class DocApi:
|
||||
_transport: Any
|
||||
active_tab: int = 0
|
||||
|
||||
def document(self, tab):
|
||||
if tab != self.active_tab:
|
||||
raise ValueError("transport is bound to another tab")
|
||||
def document(self, tab: int) -> Document:
|
||||
return Document(self._transport, tab)
|
||||
'''
|
||||
|
||||
|
||||
def render_method(method, constructor=False):
|
||||
args = method.get("args", [])
|
||||
signature = "".join(", " + a["name"] for a in args)
|
||||
variant = method.get("op") or method["query"]
|
||||
fields = {}
|
||||
for arg in args:
|
||||
name, ty = arg["name"], arg["ty"]
|
||||
if ty in ("EntityRef", "Solid", "ObjectId"):
|
||||
fields[name] = f"self._doc._id({name})"
|
||||
elif ty == "&[EntityRef]":
|
||||
fields[name] = f"[self._doc._id(p) for p in {name}]"
|
||||
else:
|
||||
fields[name] = name
|
||||
fixed = dict(method.get("fixed") or {})
|
||||
if variant in ("SolidBoolean", "GetIntersects"):
|
||||
fields = {"a": "self.id", "b": fields.pop("other"), **fields}
|
||||
fields.pop("other", None)
|
||||
elif not constructor:
|
||||
fields = {"id": "self.id", **fields}
|
||||
if variant == "Revolve":
|
||||
fields["axis"] = "(pivot, axis)"
|
||||
fields.pop("pivot")
|
||||
nested = fixed.pop("primitive", fixed.pop("curve", None))
|
||||
fields.update({key: repr(value) for key, value in fixed.items()})
|
||||
payload = "{" + ", ".join(repr(k) + ": " + v for k, v in fields.items()) + "}"
|
||||
if variant == "CreateMany":
|
||||
payload = "[{'Curve': {'Point': {'position': p}}} for p in positions]"
|
||||
elif nested:
|
||||
payload = "{" + repr(nested) + ": " + payload + "}"
|
||||
message = "{" + repr(variant) + ": " + payload + "}"
|
||||
if method.get("kind") == "query":
|
||||
call = f"self._doc._apply_query({message})"
|
||||
else:
|
||||
want = method.get("returns")
|
||||
if want == "()":
|
||||
want = None
|
||||
call = f"self._doc._apply_op({message}, {want!r})"
|
||||
return f" def {method['name']}(self{signature}):\n return {call}\n"
|
||||
|
||||
|
||||
def render(schema):
|
||||
out = [HEADER, f"ENVELOPE_VERSION = {schema['envelope_version']}\n"]
|
||||
handles = {name: {} for name in schema["object_model"]["handles"]}
|
||||
collections = {"solids": {}, "curves": {}, "entities": {}}
|
||||
for family in schema["families"]:
|
||||
handle = family.get("handle") or family["name"]
|
||||
for method in family.get("methods", []):
|
||||
existing = handles[handle].get(method["name"])
|
||||
if existing is not None and existing != method:
|
||||
raise ValueError(f"conflicting method: {handle}.{method['name']}")
|
||||
handles[handle][method["name"]] = method
|
||||
for constructor in family.get("constructors", []):
|
||||
collections[family["collection"]][constructor["name"]] = constructor
|
||||
# Families sharing Entity contribute to one class, preserving all methods.
|
||||
entity_methods = "\n".join(render_method(m) for m in handles.pop("Entity").values())
|
||||
out[0] += entity_methods
|
||||
for handle, methods in handles.items():
|
||||
out.append(f"\nclass {handle}(Entity):\n" + (
|
||||
"\n".join(render_method(m) for m in methods.values()) or " pass\n"))
|
||||
for collection, constructors in collections.items():
|
||||
out.append(f"\nclass {collection.title()}:\n def __init__(self, doc):\n self._doc = doc\n")
|
||||
out.extend(render_method(c, constructor=True) for c in constructors.values())
|
||||
out.append(DOCUMENT)
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def main():
|
||||
default = Path(__file__).resolve().parents[2] / "src/gen/binding_schema.json"
|
||||
path = Path(sys.argv[1]) if len(sys.argv) > 1 else default
|
||||
sys.stdout.write(render(json.loads(path.read_text(encoding="utf-8"))))
|
||||
def main() -> None:
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "gen/binding_schema.json"
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
schema = json.load(f)
|
||||
sys.stdout.write(render(schema))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in new issue
Block a user