620 lines
23 KiB
Python
620 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from bedit.core.port_types import PortTypeRegistry
|
|
|
|
|
|
@dataclass
|
|
class Port:
|
|
id: str
|
|
name: str
|
|
x: float = 0.0
|
|
y: float = 0.0
|
|
properties: dict[str, Any] = field(default_factory=dict)
|
|
type: str = "signal"
|
|
allows_multiple_connections: bool = False
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"position": {"x": self.x, "y": self.y},
|
|
"properties": self.properties,
|
|
"type": self.type,
|
|
"multipleConnections": self.allows_multiple_connections,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Port":
|
|
position = data.get("position", {})
|
|
return cls(
|
|
id=str(data["id"]),
|
|
name=str(data.get("name", data["id"])),
|
|
x=float(position.get("x", 0.0)),
|
|
y=float(position.get("y", 0.0)),
|
|
properties=dict(data.get("properties", {})),
|
|
type=str(data.get("type", "signal")),
|
|
allows_multiple_connections=bool(data.get("multipleConnections", False)),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Icon:
|
|
shape: str = "rectangle"
|
|
fill: str = "#f4f4f4"
|
|
border: str = "#303030"
|
|
text: str = ""
|
|
width: float = 128.0
|
|
height: float = 128.0
|
|
elements: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.elements:
|
|
self.elements = [
|
|
{
|
|
"type": "ellipse" if self.shape == "ellipse" else "rectangle",
|
|
"x": 32.0,
|
|
"y": 32.0,
|
|
"width": 64.0,
|
|
"height": 64.0,
|
|
"fill": self.fill,
|
|
"stroke": self.border,
|
|
"lineWidth": 1.5,
|
|
"lineStyle": "solid",
|
|
"cornerRadius": 5.0,
|
|
}
|
|
]
|
|
if self.text:
|
|
self.elements.append(
|
|
{
|
|
"type": "text",
|
|
"x": 40.0,
|
|
"y": 40.0,
|
|
"width": 48.0,
|
|
"height": 48.0,
|
|
"text": self.text,
|
|
"color": "#202020",
|
|
"fontSize": 12.0,
|
|
}
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"shape": self.shape,
|
|
"fill": self.fill,
|
|
"border": self.border,
|
|
"text": self.text,
|
|
"size": {"width": self.width, "height": self.height},
|
|
"elements": deepcopy(self.elements),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | None) -> "Icon":
|
|
data = data or {}
|
|
icon = cls(
|
|
shape=str(data.get("shape", "rectangle")),
|
|
fill=str(data.get("fill", "#f4f4f4")),
|
|
border=str(data.get("border", "#303030")),
|
|
text=str(data.get("text", "")),
|
|
width=128.0,
|
|
height=128.0,
|
|
elements=deepcopy(data.get("elements", [])),
|
|
)
|
|
return icon
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Endpoint:
|
|
block: str | None = None
|
|
port: str | None = None
|
|
interface: str | None = None
|
|
junction: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
if self.interface is not None:
|
|
return {"interface": self.interface}
|
|
if self.junction is not None:
|
|
return {"junction": self.junction}
|
|
if self.block is None or self.port is None:
|
|
raise ValueError("A block endpoint requires both block and port")
|
|
return {"block": self.block, "port": self.port}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Endpoint":
|
|
if "interface" in data:
|
|
return cls(interface=str(data["interface"]))
|
|
if "junction" in data:
|
|
return cls(junction=str(data["junction"]))
|
|
return cls(block=str(data["block"]), port=str(data["port"]))
|
|
|
|
|
|
@dataclass
|
|
class Connection:
|
|
id: str
|
|
source: Endpoint
|
|
target: Endpoint
|
|
name: str = ""
|
|
properties: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"source": self.source.to_dict(),
|
|
"target": self.target.to_dict(),
|
|
"name": self.name,
|
|
"properties": self.properties,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Connection":
|
|
return cls(
|
|
id=str(data["id"]),
|
|
source=Endpoint.from_dict(data["source"]),
|
|
target=Endpoint.from_dict(data["target"]),
|
|
name=str(data.get("name", "")),
|
|
properties=dict(data.get("properties", {})),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Parameter:
|
|
id: str
|
|
name: str
|
|
type: str = "real"
|
|
value: str = "0"
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
return {"id": self.id, "name": self.name, "type": self.type, "value": self.value}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Parameter":
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Each text component parameter must be an object")
|
|
if "id" not in data:
|
|
raise ValueError("Each text component parameter must have an ID")
|
|
return cls(
|
|
id=str(data["id"]),
|
|
name=str(data.get("name", "")),
|
|
type=str(data.get("type", "real")),
|
|
value=str(data.get("value", "0")),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Junction:
|
|
id: str
|
|
x: float
|
|
y: float
|
|
type: str = "signal"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"position": {"x": self.x, "y": self.y},
|
|
"type": self.type,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Junction":
|
|
position = data.get("position", {})
|
|
return cls(
|
|
id=str(data["id"]),
|
|
x=float(position.get("x", 0)),
|
|
y=float(position.get("y", 0)),
|
|
type=str(data.get("type", "signal")),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Annotation:
|
|
id: str
|
|
kind: str
|
|
x: float = 0.0
|
|
y: float = 0.0
|
|
width: float = 0.0
|
|
height: float = 0.0
|
|
text: str = ""
|
|
layer: int = -1
|
|
properties: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"kind": self.kind,
|
|
"position": {"x": self.x, "y": self.y},
|
|
"size": {"width": self.width, "height": self.height},
|
|
"text": self.text,
|
|
"layer": self.layer,
|
|
"properties": deepcopy(self.properties),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Annotation":
|
|
position = data.get("position", {})
|
|
size = data.get("size", {})
|
|
return cls(
|
|
id=str(data["id"]),
|
|
kind=str(data["kind"]),
|
|
x=float(position.get("x", 0)),
|
|
y=float(position.get("y", 0)),
|
|
width=float(size.get("width", 0)),
|
|
height=float(size.get("height", 0)),
|
|
text=str(data.get("text", "")),
|
|
layer=int(data.get("layer", -1)),
|
|
properties=deepcopy(data.get("properties", {})),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Graph:
|
|
blocks: dict[str, Component] = field(default_factory=dict)
|
|
connections: dict[str, Connection] = field(default_factory=dict)
|
|
annotations: dict[str, Annotation] = field(default_factory=dict)
|
|
junctions: dict[str, Junction] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"blocks": [block.to_dict() for block in self.blocks.values()],
|
|
"connections": [connection.to_dict() for connection in self.connections.values()],
|
|
"annotations": [item.to_dict() for item in self.annotations.values()],
|
|
"junctions": [junction.to_dict() for junction in self.junctions.values()],
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | None) -> "Graph":
|
|
data = data or {}
|
|
blocks = [Component.from_dict(item) for item in data.get("blocks", [])]
|
|
connections = [Connection.from_dict(item) for item in data.get("connections", [])]
|
|
annotations = [Annotation.from_dict(item) for item in data.get("annotations", [])]
|
|
junctions = [Junction.from_dict(item) for item in data.get("junctions", [])]
|
|
if len({block.id for block in blocks}) != len(blocks):
|
|
raise ValueError("A graph contains duplicate component IDs")
|
|
if len({connection.id for connection in connections}) != len(connections):
|
|
raise ValueError("A graph contains duplicate connection IDs")
|
|
if len({item.id for item in annotations}) != len(annotations):
|
|
raise ValueError("A graph contains duplicate annotation IDs")
|
|
if len({item.id for item in junctions}) != len(junctions):
|
|
raise ValueError("A graph contains duplicate junction IDs")
|
|
if any(item.kind not in {"box", "line", "text"} for item in annotations):
|
|
raise ValueError("A graph contains an unknown annotation kind")
|
|
return cls(
|
|
blocks={block.id: block for block in blocks},
|
|
connections={connection.id: connection for connection in connections},
|
|
annotations={item.id: item for item in annotations},
|
|
junctions={junction.id: junction for junction in junctions},
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Component:
|
|
id: str
|
|
name: str
|
|
x: float = 0.0
|
|
y: float = 0.0
|
|
rotation: float = 0.0
|
|
inputs: list[Port] = field(default_factory=list)
|
|
outputs: list[Port] = field(default_factory=list)
|
|
icon: Icon = field(default_factory=Icon)
|
|
properties: dict[str, Any] = field(default_factory=dict)
|
|
implementation_kind: str = "graph"
|
|
graph: Graph = field(default_factory=Graph)
|
|
source: dict[str, Any] = field(default_factory=dict)
|
|
show_subtree_in_library: bool = True
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
implementation = {"kind": self.implementation_kind}
|
|
if self.implementation_kind == "text":
|
|
implementation["source"] = self.source
|
|
else:
|
|
implementation["graph"] = self.graph.to_dict()
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"position": {"x": self.x, "y": self.y},
|
|
"rotation": self.rotation,
|
|
"interface": {
|
|
"inputs": [port.to_dict() for port in self.inputs],
|
|
"outputs": [port.to_dict() for port in self.outputs],
|
|
},
|
|
"icon": self.icon.to_dict(),
|
|
"properties": self.properties,
|
|
"library": {"showSubtree": self.show_subtree_in_library},
|
|
"implementation": implementation,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Component":
|
|
interface = data.get("interface", {})
|
|
position = data.get("position", {})
|
|
implementation = data["implementation"]
|
|
kind = str(implementation.get("kind", "graph"))
|
|
if kind not in {"graph", "text"}:
|
|
raise ValueError(f"Unknown component implementation kind: {kind}")
|
|
source: dict[str, Any] = {}
|
|
if kind == "text":
|
|
raw_source = implementation.get("source", {})
|
|
equations = raw_source.get("equations", "")
|
|
parameters = raw_source.get("parameters", [])
|
|
if not isinstance(equations, str):
|
|
raise ValueError("Text component equations must be a string")
|
|
if not isinstance(parameters, list):
|
|
raise ValueError("Text component parameters must be a list")
|
|
source = {
|
|
"equations": equations,
|
|
"parameters": [Parameter.from_dict(item).to_dict() for item in parameters],
|
|
}
|
|
return cls(
|
|
id=str(data["id"]),
|
|
name=str(data.get("name", "Unnamed")),
|
|
x=float(position.get("x", 0.0)),
|
|
y=float(position.get("y", 0.0)),
|
|
rotation=float(data.get("rotation", 0.0)),
|
|
inputs=[Port.from_dict(item) for item in interface.get("inputs", [])],
|
|
outputs=[Port.from_dict(item) for item in interface.get("outputs", [])],
|
|
icon=Icon.from_dict(data.get("icon")),
|
|
properties=dict(data.get("properties", {})),
|
|
show_subtree_in_library=bool(data.get("library", {}).get("showSubtree", True)),
|
|
implementation_kind=kind,
|
|
graph=Graph.from_dict(implementation.get("graph")) if kind == "graph" else Graph(),
|
|
source=source,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GraphDocument:
|
|
roots: dict[str, Component] = field(default_factory=dict)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def empty(cls) -> "GraphDocument":
|
|
return cls(metadata={"name": "Current Document"})
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"format": "bedit-document",
|
|
"version": 1,
|
|
"metadata": self.metadata,
|
|
"roots": [root.to_dict() for root in self.roots.values()],
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "GraphDocument":
|
|
if data.get("format") != "bedit-document":
|
|
raise ValueError("This is not a BEdit document")
|
|
if data.get("version") != 1:
|
|
raise ValueError(f"Unsupported BEdit document version: {data.get('version')}")
|
|
roots = [Component.from_dict(item) for item in data["roots"]]
|
|
if len({root.id for root in roots}) != len(roots):
|
|
raise ValueError("The document contains duplicate root IDs")
|
|
document = cls(
|
|
roots={root.id: root for root in roots},
|
|
metadata=dict(data.get("metadata", {})),
|
|
)
|
|
document.validate()
|
|
return document
|
|
|
|
def all_components(self):
|
|
def walk(component: Component):
|
|
yield component
|
|
if component.implementation_kind == "graph":
|
|
for child in component.graph.blocks.values():
|
|
yield from walk(child)
|
|
|
|
def all_roots():
|
|
for root in self.roots.values():
|
|
yield from walk(root)
|
|
|
|
return all_roots()
|
|
|
|
def find_component(self, component_id: str) -> Component | None:
|
|
return next(
|
|
(component for component in self.all_components() if component.id == component_id),
|
|
None,
|
|
)
|
|
|
|
def find_parent(self, component_id: str) -> Component | None:
|
|
for component in self.all_components():
|
|
if component_id in component.graph.blocks:
|
|
return component
|
|
return None
|
|
|
|
def validate(self) -> None:
|
|
seen: set[str] = set()
|
|
root_names = [component.name for component in self.roots.values()]
|
|
if len(set(root_names)) != len(root_names):
|
|
raise ValueError("Root component names must be unique")
|
|
for component in self.all_components():
|
|
if component.id in seen:
|
|
raise ValueError(f"Duplicate component ID: {component.id}")
|
|
seen.add(component.id)
|
|
if component.implementation_kind == "text" and component.graph.blocks:
|
|
raise ValueError(f"Text component {component.name} cannot contain a graph")
|
|
self._validate_graph(component)
|
|
|
|
@staticmethod
|
|
def _validate_graph(owner: Component) -> None:
|
|
child_names = [component.name for component in owner.graph.blocks.values()]
|
|
if len(set(child_names)) != len(child_names):
|
|
raise ValueError(
|
|
f"Component names inside {owner.name!r} must be unique"
|
|
)
|
|
input_ids = {port.id for port in owner.inputs}
|
|
output_ids = {port.id for port in owner.outputs}
|
|
if len(input_ids) != len(owner.inputs) or len(output_ids) != len(owner.outputs):
|
|
raise ValueError(f"Component {owner.name} contains duplicate port IDs")
|
|
for port in (*owner.inputs, *owner.outputs):
|
|
PortTypeRegistry.get(port.type)
|
|
for junction in owner.graph.junctions.values():
|
|
PortTypeRegistry.get(junction.type)
|
|
endpoint_counts: dict[tuple[str, str, str], int] = {}
|
|
for connection in owner.graph.connections.values():
|
|
if connection.source.junction is not None:
|
|
junction = owner.graph.junctions.get(connection.source.junction)
|
|
if junction is None:
|
|
raise ValueError(
|
|
f"Connection {connection.id} uses an unknown source junction"
|
|
)
|
|
source_port = Port(junction.id, "Junction", type=junction.type)
|
|
elif connection.source.interface is not None:
|
|
if connection.source.interface not in input_ids:
|
|
raise ValueError(f"Connection {connection.id} uses an unknown interface input")
|
|
source_port = next(p for p in owner.inputs if p.id == connection.source.interface)
|
|
else:
|
|
source = owner.graph.blocks.get(connection.source.block or "")
|
|
if source is None or connection.source.port not in {p.id for p in source.outputs}:
|
|
raise ValueError(f"Connection {connection.id} uses an unknown block output")
|
|
source_port = next(p for p in source.outputs if p.id == connection.source.port)
|
|
if connection.target.junction is not None:
|
|
junction = owner.graph.junctions.get(connection.target.junction)
|
|
if junction is None:
|
|
raise ValueError(
|
|
f"Connection {connection.id} uses an unknown target junction"
|
|
)
|
|
target_port = Port(
|
|
junction.id,
|
|
"Junction",
|
|
type=junction.type,
|
|
allows_multiple_connections=False,
|
|
)
|
|
elif connection.target.interface is not None:
|
|
if connection.target.interface not in output_ids:
|
|
raise ValueError(f"Connection {connection.id} uses an unknown interface output")
|
|
target_port = next(p for p in owner.outputs if p.id == connection.target.interface)
|
|
else:
|
|
target = owner.graph.blocks.get(connection.target.block or "")
|
|
if target is None or connection.target.port not in {p.id for p in target.inputs}:
|
|
raise ValueError(f"Connection {connection.id} uses an unknown block input")
|
|
target_port = next(p for p in target.inputs if p.id == connection.target.port)
|
|
if not PortTypeRegistry.compatible(source_port.type, target_port.type):
|
|
raise ValueError(f"Connection {connection.id} joins incompatible port types")
|
|
source_key = (
|
|
"source-junction"
|
|
if connection.source.junction is not None
|
|
else "source-interface"
|
|
if connection.source.interface is not None
|
|
else "source-block",
|
|
connection.source.block or "",
|
|
connection.source.junction
|
|
or connection.source.interface
|
|
or connection.source.port
|
|
or "",
|
|
)
|
|
target_key = (
|
|
"target-junction"
|
|
if connection.target.junction is not None
|
|
else "target-interface"
|
|
if connection.target.interface is not None
|
|
else "target-block",
|
|
connection.target.block or "",
|
|
connection.target.junction
|
|
or connection.target.interface
|
|
or connection.target.port
|
|
or "",
|
|
)
|
|
endpoint_counts[source_key] = endpoint_counts.get(source_key, 0) + 1
|
|
endpoint_counts[target_key] = endpoint_counts.get(target_key, 0) + 1
|
|
if (
|
|
endpoint_counts[target_key] > 1
|
|
and not target_port.allows_multiple_connections
|
|
):
|
|
raise ValueError(
|
|
f"Input {target_port.name!r} has multiple incoming connections "
|
|
"but does not allow them"
|
|
)
|
|
|
|
|
|
def clone_component(source: Component) -> Component:
|
|
"""Deep-copy a component tree and remap every owned object ID."""
|
|
|
|
def clone(current: Component) -> Component:
|
|
child_pairs = [(child, clone(child)) for child in current.graph.blocks.values()]
|
|
child_ids = {old.id: new.id for old, new in child_pairs}
|
|
junction_ids = {
|
|
junction.id: str(uuid4()) for junction in current.graph.junctions.values()
|
|
}
|
|
|
|
def remap(endpoint: Endpoint) -> Endpoint:
|
|
if endpoint.interface is not None:
|
|
return endpoint
|
|
if endpoint.junction is not None:
|
|
return Endpoint(junction=junction_ids[endpoint.junction])
|
|
return Endpoint(block=child_ids[endpoint.block or ""], port=endpoint.port)
|
|
|
|
graph = Graph(
|
|
blocks={new.id: new for _old, new in child_pairs},
|
|
connections={
|
|
new_id: Connection(
|
|
new_id,
|
|
remap(connection.source),
|
|
remap(connection.target),
|
|
connection.name,
|
|
deepcopy(connection.properties),
|
|
)
|
|
for connection in current.graph.connections.values()
|
|
for new_id in [str(uuid4())]
|
|
},
|
|
annotations={
|
|
new_id: Annotation(
|
|
new_id,
|
|
annotation.kind,
|
|
annotation.x,
|
|
annotation.y,
|
|
annotation.width,
|
|
annotation.height,
|
|
annotation.text,
|
|
annotation.layer,
|
|
deepcopy(annotation.properties),
|
|
)
|
|
for annotation in current.graph.annotations.values()
|
|
for new_id in [str(uuid4())]
|
|
},
|
|
junctions={
|
|
junction_ids[junction.id]: Junction(
|
|
junction_ids[junction.id], junction.x, junction.y, junction.type
|
|
)
|
|
for junction in current.graph.junctions.values()
|
|
},
|
|
)
|
|
return Component(
|
|
id=str(uuid4()),
|
|
name=current.name,
|
|
x=current.x,
|
|
y=current.y,
|
|
inputs=[
|
|
Port(
|
|
port.id,
|
|
port.name,
|
|
port.x,
|
|
port.y,
|
|
deepcopy(port.properties),
|
|
port.type,
|
|
port.allows_multiple_connections,
|
|
)
|
|
for port in current.inputs
|
|
],
|
|
outputs=[
|
|
Port(
|
|
port.id,
|
|
port.name,
|
|
port.x,
|
|
port.y,
|
|
deepcopy(port.properties),
|
|
port.type,
|
|
port.allows_multiple_connections,
|
|
)
|
|
for port in current.outputs
|
|
],
|
|
icon=Icon.from_dict(current.icon.to_dict()),
|
|
properties=deepcopy(current.properties),
|
|
implementation_kind=current.implementation_kind,
|
|
graph=graph if current.implementation_kind == "graph" else Graph(),
|
|
source=deepcopy(current.source),
|
|
show_subtree_in_library=current.show_subtree_in_library,
|
|
)
|
|
|
|
return clone(source)
|