1514 lines
60 KiB
Python
1514 lines
60 KiB
Python
from copy import deepcopy
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from PySide6.QtCore import QObject, QPointF, Signal
|
|
from PySide6.QtGui import QUndoStack
|
|
|
|
from bedit.gui.controllers.commands import (
|
|
AddComponentCommand,
|
|
AddConnectionCommand,
|
|
AddAnnotationCommand,
|
|
AddInterfacePortCommand,
|
|
DeleteSelectionCommand,
|
|
DeleteAnnotationsCommand,
|
|
EditGraphItemCommand,
|
|
EditGraphParametersCommand,
|
|
EditSimulationSettingsCommand,
|
|
EditTextDefinitionCommand,
|
|
EditComponentAppearanceCommand,
|
|
EditComponentPropertiesCommand,
|
|
EditComponentParametersCommand,
|
|
MoveComponentCommand,
|
|
MoveInterfacePortCommand,
|
|
PasteSelectionCommand,
|
|
RenameConnectionCommand,
|
|
RenameDocumentCommand,
|
|
RenameInterfacePortCommand,
|
|
ReplaceSourceCommand,
|
|
RotateComponentsCommand,
|
|
SplitConnectionCommand,
|
|
)
|
|
from bedit.core.model import (
|
|
Annotation,
|
|
Component,
|
|
Connection,
|
|
Endpoint,
|
|
GraphDocument,
|
|
Icon,
|
|
Junction,
|
|
Parameter,
|
|
Port,
|
|
clone_component,
|
|
)
|
|
from bedit.core.bond_graph import infer_causality
|
|
from bedit.core.simulation import Simulation
|
|
from bedit.core.port_types import PortTypeRegistry
|
|
from bedit.core.serializer import DocumentSerializer
|
|
from bedit.gui.preferences import application_settings
|
|
|
|
|
|
class DocumentController(QObject):
|
|
documentReset = Signal()
|
|
documentOpenedChanged = Signal(bool)
|
|
documentNameChanged = Signal(str)
|
|
activeGraphChanged = Signal()
|
|
componentAdded = Signal(str)
|
|
componentRemoved = Signal(str)
|
|
componentMoved = Signal(str, QPointF)
|
|
componentRotated = Signal(str, float)
|
|
componentPropertiesChanged = Signal(str)
|
|
textDefinitionChanged = Signal(str)
|
|
connectionAdded = Signal(str)
|
|
connectionRemoved = Signal(str)
|
|
graphItemChanged = Signal(str, str)
|
|
annotationAdded = Signal(str)
|
|
annotationRemoved = Signal(str)
|
|
interfaceChanged = Signal()
|
|
filePathChanged = Signal(object)
|
|
modifiedChanged = Signal(bool)
|
|
|
|
def __init__(self, parent=None, *, simulation: Simulation | None = None) -> None:
|
|
super().__init__(parent)
|
|
self.simulation = simulation or Simulation()
|
|
self.document: GraphDocument | None = None
|
|
self.active_component_id: str | None = None
|
|
self.file_path: Path | None = None
|
|
self.undo_stack = QUndoStack(self)
|
|
self.undo_stack.cleanChanged.connect(self._clean_changed)
|
|
|
|
def _clean_changed(self, clean: bool) -> None:
|
|
self.modifiedChanged.emit(not clean)
|
|
|
|
@property
|
|
def active_component(self) -> Component | None:
|
|
if self.document is None or self.active_component_id is None:
|
|
return None
|
|
return self.document.find_component(self.active_component_id)
|
|
|
|
@property
|
|
def active_graph(self):
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("There is no active graph")
|
|
return component.graph
|
|
|
|
def new_document(self) -> None:
|
|
self.document = GraphDocument.empty()
|
|
self.active_component_id = None
|
|
self.file_path = None
|
|
self.undo_stack.clear()
|
|
self.documentOpenedChanged.emit(True)
|
|
self.documentReset.emit()
|
|
self.activeGraphChanged.emit()
|
|
self.filePathChanged.emit(None)
|
|
|
|
def close_document(self) -> None:
|
|
self.document = None
|
|
self.active_component_id = None
|
|
self.file_path = None
|
|
self.undo_stack.clear()
|
|
self.documentOpenedChanged.emit(False)
|
|
self.documentReset.emit()
|
|
self.activeGraphChanged.emit()
|
|
self.filePathChanged.emit(None)
|
|
|
|
def load(self, path: Path) -> None:
|
|
self.document = DocumentSerializer.load(path)
|
|
self.active_component_id = next(iter(self.document.roots), None)
|
|
self.file_path = path
|
|
self.undo_stack.clear()
|
|
self.undo_stack.setClean()
|
|
self.documentOpenedChanged.emit(True)
|
|
self.documentReset.emit()
|
|
self.activeGraphChanged.emit()
|
|
self.filePathChanged.emit(path)
|
|
|
|
def save(self, path: Path | None = None) -> Path:
|
|
if self.document is None:
|
|
raise ValueError("There is no open document")
|
|
target = path or self.file_path
|
|
if target is None:
|
|
raise ValueError("No file path has been selected")
|
|
DocumentSerializer.save(self.document, target)
|
|
self.file_path = target
|
|
self.undo_stack.setClean()
|
|
self.filePathChanged.emit(target)
|
|
return target
|
|
|
|
def activate_component(self, component_id: str) -> None:
|
|
if self.document is None or self.document.find_component(component_id) is None:
|
|
return
|
|
self.active_component_id = component_id
|
|
self.activeGraphChanged.emit()
|
|
|
|
def navigate_up(self) -> None:
|
|
if self.document is None or self.active_component_id is None:
|
|
return
|
|
parent = self.document.find_parent(self.active_component_id)
|
|
if parent is not None:
|
|
self.activate_component(parent.id)
|
|
|
|
def breadcrumb(self) -> list[str]:
|
|
if self.document is None or self.active_component is None:
|
|
return []
|
|
names = [self.active_component.name]
|
|
current = self.active_component
|
|
while True:
|
|
parent = self.document.find_parent(current.id)
|
|
if parent is None:
|
|
break
|
|
names.append(parent.name)
|
|
current = parent
|
|
return list(reversed(names))
|
|
|
|
def add_root(self, kind: str) -> str:
|
|
if self.document is None:
|
|
raise ValueError("Open or create a document first")
|
|
number = len(self.document.roots) + 1
|
|
base_name = f"New {'Graph' if kind == 'graph' else 'Text'} Block "
|
|
component = Component(
|
|
id=str(uuid4()),
|
|
name=self._available_component_name(base_name, self.document.roots.values(), number),
|
|
implementation_kind=kind,
|
|
icon=Icon(text="Graph" if kind == "graph" else "Text"),
|
|
source={
|
|
"declarations": "",
|
|
"initialEquations": "",
|
|
"equations": "",
|
|
}
|
|
if kind == "text"
|
|
else {},
|
|
)
|
|
self.undo_stack.push(AddComponentCommand(self, None, component))
|
|
self.activate_component(component.id)
|
|
return component.id
|
|
|
|
def add_child(self, owner_id: str, kind: str) -> str:
|
|
if self.document is None:
|
|
raise ValueError("Open or create a document first")
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None or owner.implementation_kind != "graph":
|
|
raise ValueError("Children can only be added to graph components")
|
|
number = len(owner.graph.blocks) + 1
|
|
base_name = f"New {'Graph' if kind == 'graph' else 'Text'} Block "
|
|
component = Component(
|
|
id=str(uuid4()),
|
|
name=self._available_component_name(base_name, owner.graph.blocks.values(), number),
|
|
implementation_kind=kind,
|
|
icon=Icon(text="Graph" if kind == "graph" else "Text"),
|
|
source={
|
|
"declarations": "",
|
|
"initialEquations": "",
|
|
"equations": "",
|
|
}
|
|
if kind == "text"
|
|
else {},
|
|
)
|
|
self.undo_stack.push(AddComponentCommand(self, owner_id, component))
|
|
return component.id
|
|
|
|
def delete_component(self, component_id: str) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
parent = self.document.find_parent(component_id)
|
|
owner_id = parent.id if parent else None
|
|
connections = {}
|
|
if parent is not None:
|
|
connections = {
|
|
connection.id: connection
|
|
for connection in parent.graph.connections.values()
|
|
if component_id in (connection.source.block, connection.target.block)
|
|
}
|
|
self.undo_stack.push(
|
|
DeleteSelectionCommand(
|
|
self,
|
|
owner_id,
|
|
{component_id: component},
|
|
connections,
|
|
[],
|
|
)
|
|
)
|
|
|
|
def rename_document(self, name: str) -> None:
|
|
if self.document is None:
|
|
return
|
|
name = name.strip()
|
|
if not name:
|
|
raise ValueError("The document name cannot be empty")
|
|
old = str(self.document.metadata.get("name") or "Current Document")
|
|
if name != old:
|
|
self.undo_stack.push(RenameDocumentCommand(self, old, name))
|
|
|
|
def add_component_copy(self, source: Component, position: QPointF) -> str:
|
|
if self.active_component is None or self.active_component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before placing components")
|
|
component = clone_component(source)
|
|
component.name = self._available_component_name(
|
|
source.name, self.active_component.graph.blocks.values(), 0
|
|
)
|
|
component.x, component.y = position.x(), position.y()
|
|
self.undo_stack.push(AddComponentCommand(self, self.active_component_id, component))
|
|
return component.id
|
|
|
|
def move_component(self, component_id: str, old: QPointF, new: QPointF) -> None:
|
|
if old != new and self.active_component_id is not None:
|
|
self.undo_stack.push(
|
|
MoveComponentCommand(self, self.active_component_id, component_id, old, new)
|
|
)
|
|
|
|
def move_junction(self, junction_id: str, old: QPointF, new: QPointF) -> None:
|
|
if old != new and self.active_component_id is not None:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"junction_geometry",
|
|
junction_id,
|
|
{"x": old.x(), "y": old.y()},
|
|
{"x": new.x(), "y": new.y()},
|
|
"Move connection junction",
|
|
)
|
|
)
|
|
|
|
def rotate_components(self, component_ids: set[str]) -> None:
|
|
if self.active_component is None or self.active_component_id is None:
|
|
return
|
|
rotations = {
|
|
component_id: (component.rotation, (component.rotation + 90.0) % 360.0)
|
|
for component_id in component_ids
|
|
if (component := self.active_component.graph.blocks.get(component_id)) is not None
|
|
}
|
|
if rotations:
|
|
self.undo_stack.push(RotateComponentsCommand(self, self.active_component_id, rotations))
|
|
|
|
def connect(
|
|
self,
|
|
source: Endpoint,
|
|
target: Endpoint,
|
|
*,
|
|
waypoints: list[QPointF] | None = None,
|
|
) -> str:
|
|
if self.active_component_id is None:
|
|
raise ValueError("There is no active graph")
|
|
source_port = self._port_for_endpoint(source, "source")
|
|
target_port = self._port_for_endpoint(target, "target")
|
|
if source_port is None or target_port is None:
|
|
raise ValueError("A connection endpoint no longer exists")
|
|
if not PortTypeRegistry.compatible(
|
|
source_port.type,
|
|
target_port.type,
|
|
source_port.domain,
|
|
target_port.domain,
|
|
):
|
|
raise ValueError(f"Cannot connect {source_port.type!r} to {target_port.type!r}")
|
|
if not self.endpoint_accepts_connection(source, "source"):
|
|
raise ValueError(
|
|
f"Port {source_port.name!r} already has a connection; enable multiple connections first"
|
|
)
|
|
if not self.endpoint_accepts_connection(target, "target"):
|
|
raise ValueError(
|
|
f"Port {target_port.name!r} already has a connection; enable multiple connections first"
|
|
)
|
|
connection = Connection(
|
|
str(uuid4()),
|
|
source,
|
|
target,
|
|
properties={
|
|
"waypoints": [{"x": point.x(), "y": point.y()} for point in (waypoints or [])],
|
|
},
|
|
type=source_port.type,
|
|
)
|
|
self.undo_stack.push(AddConnectionCommand(self, self.active_component_id, connection))
|
|
return connection.id
|
|
|
|
def split_connection(
|
|
self,
|
|
connection_id: str,
|
|
position: QPointF,
|
|
first_waypoints: list[QPointF],
|
|
second_waypoints: list[QPointF],
|
|
) -> str:
|
|
if self.active_component_id is None:
|
|
raise ValueError("There is no active graph")
|
|
original = self.active_graph.connections.get(connection_id)
|
|
if original is None:
|
|
raise ValueError("The connection no longer exists")
|
|
port_type = self.connection_port_type(original)
|
|
if port_type == "power":
|
|
raise ValueError("Power bond connections cannot contain junctions")
|
|
junction = Junction(str(uuid4()), position.x(), position.y(), port_type)
|
|
first_properties = deepcopy(original.properties)
|
|
first_properties["waypoints"] = [
|
|
{"x": point.x(), "y": point.y()} for point in first_waypoints
|
|
]
|
|
second_properties = {
|
|
"waypoints": [{"x": point.x(), "y": point.y()} for point in second_waypoints]
|
|
}
|
|
first = Connection(
|
|
str(uuid4()),
|
|
original.source,
|
|
Endpoint(junction=junction.id),
|
|
original.name,
|
|
first_properties,
|
|
original.type,
|
|
original.causality,
|
|
)
|
|
second = Connection(
|
|
str(uuid4()),
|
|
Endpoint(junction=junction.id),
|
|
original.target,
|
|
"",
|
|
second_properties,
|
|
original.type,
|
|
original.causality,
|
|
)
|
|
self.undo_stack.push(
|
|
SplitConnectionCommand(
|
|
self,
|
|
self.active_component_id,
|
|
original,
|
|
junction,
|
|
first,
|
|
second,
|
|
)
|
|
)
|
|
return junction.id
|
|
|
|
def add_annotation(
|
|
self,
|
|
kind: str,
|
|
start: QPointF,
|
|
end: QPointF,
|
|
*,
|
|
text: str = "",
|
|
waypoints: list[QPointF] | None = None,
|
|
) -> str:
|
|
if self.active_component_id is None:
|
|
raise ValueError("There is no active graph")
|
|
if kind != "line":
|
|
left, right = sorted((start.x(), end.x()))
|
|
top, bottom = sorted((start.y(), end.y()))
|
|
start, end = QPointF(left, top), QPointF(right, bottom)
|
|
style = {
|
|
"stroke": "#303030",
|
|
"lineWidth": 1.5,
|
|
"lineStyle": "solid",
|
|
"fill": "none" if kind in {"line", "text"} else "#dbeafe",
|
|
}
|
|
if kind == "box":
|
|
style["cornerRadius"] = 0.0
|
|
if kind == "text":
|
|
style.update({"fontSize": 12.0, "color": "#202020"})
|
|
annotation = Annotation(
|
|
id=str(uuid4()),
|
|
kind=kind,
|
|
x=start.x(),
|
|
y=start.y(),
|
|
width=end.x() - start.x(),
|
|
height=end.y() - start.y(),
|
|
text=text,
|
|
layer=-1,
|
|
properties={
|
|
**style,
|
|
"waypoints": [{"x": point.x(), "y": point.y()} for point in (waypoints or [])],
|
|
},
|
|
)
|
|
self.undo_stack.push(AddAnnotationCommand(self, self.active_component_id, annotation))
|
|
return annotation.id
|
|
|
|
def edit_simulation_settings(self, settings: dict) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before editing simulation settings")
|
|
old = deepcopy(component.graph.simulation_settings)
|
|
new = deepcopy(settings)
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditSimulationSettingsCommand(self, component.id, old, new)
|
|
)
|
|
|
|
def edit_graph_parameter_values(
|
|
self, root_id: str, values: dict[str, dict[str, str]]
|
|
) -> None:
|
|
root = self.document.find_component(root_id) if self.document else None
|
|
if root is None:
|
|
return
|
|
subtree_ids = {component.id for component in self._component_subtree(root)}
|
|
if not set(values) <= subtree_ids:
|
|
raise ValueError("Parameter changes contain a component outside the active graph")
|
|
|
|
old: dict[str, list[dict]] = {}
|
|
new: dict[str, list[dict]] = {}
|
|
for component_id, parameter_values in values.items():
|
|
component = self.document.find_component(component_id)
|
|
known_ids = {parameter.id for parameter in component.parameters}
|
|
if not set(parameter_values) <= known_ids:
|
|
raise ValueError(f"Component {component.name!r} contains an unknown parameter")
|
|
updated = deepcopy(component.parameters)
|
|
for parameter in updated:
|
|
if parameter.id in parameter_values:
|
|
parameter.value = parameter_values[parameter.id]
|
|
old[component_id] = [parameter.to_dict() for parameter in component.parameters]
|
|
new[component_id] = [parameter.to_dict() for parameter in updated]
|
|
|
|
if old != new:
|
|
self.undo_stack.push(EditGraphParametersCommand(self, old, new))
|
|
|
|
def compose_active_graph(self) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before composing")
|
|
self._infer_active_graph_causality(component)
|
|
self.simulation.compose(component.to_dict())
|
|
|
|
def compose_active_graph_source(self) -> tuple[str, str]:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before exporting a model")
|
|
self._infer_active_graph_causality(component)
|
|
return self.simulation.compose_source(component.to_dict())
|
|
|
|
def run_simulation(
|
|
self,
|
|
progress_callback=None,
|
|
message_callback=None,
|
|
callback=None,
|
|
error_callback=None,
|
|
) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before running a simulation")
|
|
self._infer_active_graph_causality(component)
|
|
self.simulation.run_simulation(
|
|
component.to_dict(),
|
|
progress_callback,
|
|
message_callback,
|
|
callback,
|
|
error_callback,
|
|
)
|
|
|
|
def _infer_active_graph_causality(
|
|
self, component: Component, *, emit_reset: bool = True
|
|
) -> None:
|
|
"""Infer causality and copy the derived values into the live model."""
|
|
|
|
inferred = infer_causality(component.to_dict())
|
|
causalities: dict[str, str] = {}
|
|
|
|
def collect(serialized_component: dict) -> None:
|
|
graph = serialized_component.get("implementation", {}).get("graph", {})
|
|
for connection in graph.get("connections", []):
|
|
causalities[str(connection["id"])] = str(
|
|
connection.get("causality", "none")
|
|
)
|
|
for block in graph.get("blocks", []):
|
|
collect(block)
|
|
|
|
collect(inferred)
|
|
changed = False
|
|
for nested_component in self._component_subtree(component):
|
|
for connection in nested_component.graph.connections.values():
|
|
causality = causalities.get(connection.id, "none")
|
|
if connection.causality != causality:
|
|
connection.causality = causality
|
|
changed = True
|
|
if changed:
|
|
if self.document is not None:
|
|
self.document.validate()
|
|
if emit_reset:
|
|
self.documentReset.emit()
|
|
|
|
def set_route_waypoints(self, item_kind: str, item_id: str, waypoints: list[QPointF]) -> None:
|
|
item = (
|
|
self.active_graph.connections
|
|
if item_kind == "connection"
|
|
else self.active_graph.annotations
|
|
).get(item_id)
|
|
if item is None or self.active_component_id is None:
|
|
return
|
|
old = deepcopy(item.properties)
|
|
new = deepcopy(old)
|
|
new["waypoints"] = [{"x": point.x(), "y": point.y()} for point in waypoints]
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self, self.active_component_id, item_kind, item_id, old, new, "Edit line nodes"
|
|
)
|
|
)
|
|
|
|
def set_annotation_geometry(self, annotation_id: str, old: dict, new: dict) -> None:
|
|
if old != new and self.active_component_id is not None:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"annotation_geometry",
|
|
annotation_id,
|
|
old,
|
|
new,
|
|
"Move annotation",
|
|
)
|
|
)
|
|
|
|
def edit_annotation(self, annotation_id: str, values: dict) -> None:
|
|
annotation = self.active_graph.annotations.get(annotation_id)
|
|
if annotation is None or self.active_component_id is None:
|
|
return
|
|
old = annotation.to_dict()
|
|
if old != values:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"annotation_data",
|
|
annotation_id,
|
|
old,
|
|
deepcopy(values),
|
|
"Edit shape",
|
|
)
|
|
)
|
|
|
|
def reorder_annotations(self, annotation_ids: set[str], operation: str) -> None:
|
|
if not annotation_ids or self.active_component_id is None:
|
|
return
|
|
graph = self.active_graph
|
|
layers = [item.layer for item in graph.annotations.values()]
|
|
minimum, maximum = min(layers, default=-1), max(layers, default=1)
|
|
for annotation_id in annotation_ids:
|
|
item = graph.annotations.get(annotation_id)
|
|
if item is None:
|
|
continue
|
|
old = {"layer": item.layer}
|
|
forward = item.layer + 1
|
|
backward = item.layer - 1
|
|
if forward == 0:
|
|
forward = 1
|
|
if backward == 0:
|
|
backward = -1
|
|
layer = {
|
|
"forward": forward,
|
|
"backward": backward,
|
|
"front": max(1, maximum + 1),
|
|
"back": min(-1, minimum - 1),
|
|
}[operation]
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"annotation_layer",
|
|
annotation_id,
|
|
old,
|
|
{"layer": layer},
|
|
"Reorder annotation",
|
|
)
|
|
)
|
|
|
|
def delete_annotations(self, annotation_ids: set[str]) -> None:
|
|
items = {
|
|
key: self.active_graph.annotations[key]
|
|
for key in annotation_ids
|
|
if key in self.active_graph.annotations
|
|
}
|
|
if items and self.active_component_id is not None:
|
|
self.undo_stack.push(DeleteAnnotationsCommand(self, self.active_component_id, items))
|
|
|
|
def _port_for_endpoint(self, endpoint: Endpoint, role: str) -> Port | None:
|
|
owner = self.active_component
|
|
if owner is None:
|
|
return None
|
|
if endpoint.junction is not None:
|
|
junction = owner.graph.junctions.get(endpoint.junction)
|
|
if junction is None:
|
|
return None
|
|
return Port(
|
|
junction.id,
|
|
"Junction",
|
|
type=junction.type,
|
|
allows_multiple_connections=role == "source",
|
|
)
|
|
if endpoint.interface is not None:
|
|
orientations = (
|
|
{"input", "indifferent"}
|
|
if role == "source"
|
|
else {"output", "indifferent"}
|
|
)
|
|
ports = [
|
|
port for port in owner.ports if port.orientation in orientations
|
|
]
|
|
else:
|
|
component = owner.graph.blocks.get(endpoint.block or "")
|
|
if component is None:
|
|
return None
|
|
orientations = (
|
|
{"output", "indifferent"}
|
|
if role == "source"
|
|
else {"input", "indifferent"}
|
|
)
|
|
ports = [
|
|
port for port in component.ports if port.orientation in orientations
|
|
]
|
|
return next(
|
|
(port for port in ports if port.id == (endpoint.interface or endpoint.port)), None
|
|
)
|
|
|
|
def connection_port_type(self, connection: Connection) -> str:
|
|
port = self._port_for_endpoint(connection.source, "source")
|
|
return port.type if port is not None else "signal"
|
|
|
|
def endpoint_accepts_connection(self, endpoint: Endpoint, role: str) -> bool:
|
|
port = self._port_for_endpoint(endpoint, role)
|
|
if port is None:
|
|
return False
|
|
if role == "source" or port.allows_multiple_connections:
|
|
return True
|
|
return not any(
|
|
endpoint == (connection.source if role == "source" else connection.target)
|
|
for connection in self.active_graph.connections.values()
|
|
)
|
|
|
|
def add_interface_port(self, direction: str, position: QPointF) -> str:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
raise ValueError("Open a graph component before adding an interface")
|
|
ports = [port for port in component.ports if port.orientation == direction]
|
|
port = Port(
|
|
id=f"{direction}-{uuid4().hex[:8]}",
|
|
name=f"{direction.title()} {len(ports) + 1}",
|
|
x=position.x(),
|
|
y=position.y(),
|
|
orientation=direction,
|
|
)
|
|
self.undo_stack.push(AddInterfacePortCommand(self, component.id, direction, port))
|
|
return port.id
|
|
|
|
def move_interface_port(self, port_id: str, old: QPointF, new: QPointF) -> None:
|
|
if old != new and self.active_component_id is not None:
|
|
self.undo_stack.push(
|
|
MoveInterfacePortCommand(self, self.active_component_id, port_id, old, new)
|
|
)
|
|
|
|
def rename_interface_port(self, port_id: str, name: str) -> None:
|
|
owner = self.active_component
|
|
if owner is None:
|
|
return
|
|
port = next((port for port in owner.ports if port.id == port_id), None)
|
|
if port is not None and port.name != name:
|
|
self.undo_stack.push(
|
|
RenameInterfacePortCommand(self, owner.id, port_id, port.name, name)
|
|
)
|
|
|
|
def rename_connection(self, connection_id: str, name: str) -> None:
|
|
owner = self.active_component
|
|
if owner is None or owner.implementation_kind != "graph":
|
|
return
|
|
connection = owner.graph.connections.get(connection_id)
|
|
if connection is not None and connection.name != name:
|
|
self.undo_stack.push(
|
|
RenameConnectionCommand(self, owner.id, connection_id, connection.name, name)
|
|
)
|
|
|
|
def replace_active_source(self, source: dict) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "text":
|
|
raise ValueError("Only text-defined components have source JSON")
|
|
self.undo_stack.push(
|
|
ReplaceSourceCommand(
|
|
self,
|
|
component.id,
|
|
deepcopy(component.source),
|
|
deepcopy(source),
|
|
)
|
|
)
|
|
|
|
def replace_active_text_definition(
|
|
self,
|
|
ports: list[Port],
|
|
declarations: str,
|
|
initial_equations: str,
|
|
equations: str,
|
|
parameters: list[Parameter],
|
|
) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "text":
|
|
raise ValueError("Only text-defined components can be edited here")
|
|
port_ids = [port.id for port in ports]
|
|
input_ids = {
|
|
port.id for port in ports if port.orientation in {"input", "indifferent"}
|
|
}
|
|
output_ids = {
|
|
port.id for port in ports if port.orientation in {"output", "indifferent"}
|
|
}
|
|
source_ids = output_ids
|
|
if len(set(port_ids)) != len(port_ids):
|
|
raise ValueError("Port IDs must be unique")
|
|
if any(not port.name.strip() for port in ports):
|
|
raise ValueError("Every port must have a name")
|
|
parameter_ids = [parameter.id for parameter in parameters]
|
|
if len(set(parameter_ids)) != len(parameter_ids):
|
|
raise ValueError("Parameter IDs must be unique")
|
|
if any(not parameter.name.strip() for parameter in parameters):
|
|
raise ValueError("Every parameter must have a name")
|
|
if self.document is not None:
|
|
parent = self.document.find_parent(component.id)
|
|
if parent is not None:
|
|
for connection in parent.graph.connections.values():
|
|
if (
|
|
connection.target.block == component.id
|
|
and connection.target.port not in input_ids
|
|
):
|
|
raise ValueError(
|
|
f"Input {connection.target.port!r} is still connected in the containing graph"
|
|
)
|
|
if (
|
|
connection.source.block == component.id
|
|
and connection.source.port not in source_ids
|
|
):
|
|
raise ValueError(
|
|
f"Output {connection.source.port!r} is still connected in the containing graph"
|
|
)
|
|
old = {
|
|
"ports": [port.to_dict() for port in component.ports],
|
|
"source": deepcopy(component.source),
|
|
"parameters": [parameter.to_dict() for parameter in component.parameters],
|
|
}
|
|
new = {
|
|
"ports": [port.to_dict() for port in ports],
|
|
"source": {
|
|
"equations": equations,
|
|
"declarations": declarations,
|
|
"initialEquations": initial_equations,
|
|
},
|
|
"parameters": [parameter.to_dict() for parameter in parameters],
|
|
}
|
|
if old != new:
|
|
candidate = deepcopy(self.document)
|
|
candidate_component = candidate.find_component(component.id)
|
|
candidate_component.ports = deepcopy(ports)
|
|
candidate_component.source = deepcopy(new["source"])
|
|
candidate_component.parameters = deepcopy(parameters)
|
|
candidate.validate()
|
|
self.undo_stack.push(EditTextDefinitionCommand(self, component.id, old, new))
|
|
|
|
def edit_component_appearance(
|
|
self,
|
|
component_id: str,
|
|
name: str,
|
|
icon: Icon,
|
|
ports: list[Port],
|
|
show_subtree: bool,
|
|
show_name: bool,
|
|
) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
siblings = self._component_siblings(component_id)
|
|
if any(item.id != component_id and item.name == name for item in siblings):
|
|
raise ValueError(f"A component named {name!r} already exists at this level")
|
|
old = {
|
|
"name": component.name,
|
|
"icon": component.icon.to_dict(),
|
|
"ports": [port.to_dict() for port in component.ports],
|
|
"show_subtree": component.show_subtree_in_library,
|
|
"properties": deepcopy(component.properties),
|
|
}
|
|
properties = deepcopy(component.properties)
|
|
was_visible = bool(properties.get("showName", False))
|
|
properties["showName"] = show_name
|
|
if show_name and not was_visible:
|
|
properties.pop("nameLabelPosition", None)
|
|
new = {
|
|
"name": name,
|
|
"icon": icon.to_dict(),
|
|
"ports": [port.to_dict() for port in ports],
|
|
"show_subtree": show_subtree,
|
|
"properties": properties,
|
|
}
|
|
if old != new:
|
|
candidate = deepcopy(self.document)
|
|
candidate_component = candidate.find_component(component_id)
|
|
candidate_component.name = name
|
|
candidate_component.icon = Icon.from_dict(icon.to_dict())
|
|
candidate_component.ports = deepcopy(ports)
|
|
candidate_component.properties = deepcopy(properties)
|
|
candidate.validate()
|
|
self.undo_stack.push(EditComponentAppearanceCommand(self, component_id, old, new))
|
|
|
|
def move_component_name_label(self, component_id: str, position: QPointF) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
old = deepcopy(component.properties)
|
|
new = deepcopy(old)
|
|
new["nameLabelPosition"] = {"x": position.x(), "y": position.y()}
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditComponentPropertiesCommand(
|
|
self, component_id, old, new, "Move component name"
|
|
)
|
|
)
|
|
|
|
def edit_connection_options(
|
|
self, connection_id: str, name: str, show_name: bool
|
|
) -> None:
|
|
owner = self.active_component
|
|
if owner is None or self.active_component_id is None:
|
|
return
|
|
connection = owner.graph.connections.get(connection_id)
|
|
if connection is None:
|
|
return
|
|
old = {"name": connection.name, "properties": deepcopy(connection.properties)}
|
|
properties = deepcopy(connection.properties)
|
|
was_visible = bool(properties.get("showName", False))
|
|
properties["showName"] = show_name
|
|
if show_name and not was_visible:
|
|
properties.pop("nameLabelPosition", None)
|
|
new = {"name": name, "properties": properties}
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"connection_data",
|
|
connection_id,
|
|
old,
|
|
new,
|
|
"Edit connection options",
|
|
)
|
|
)
|
|
|
|
def move_connection_name_label(self, connection_id: str, position: QPointF) -> None:
|
|
connection = self.active_graph.connections.get(connection_id)
|
|
if connection is None or self.active_component_id is None:
|
|
return
|
|
old = deepcopy(connection.properties)
|
|
new = deepcopy(old)
|
|
new["nameLabelPosition"] = {"x": position.x(), "y": position.y()}
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditGraphItemCommand(
|
|
self,
|
|
self.active_component_id,
|
|
"connection",
|
|
connection_id,
|
|
old,
|
|
new,
|
|
"Move connection name",
|
|
)
|
|
)
|
|
|
|
def edit_component_ports(self, component_id: str, ports: list[Port]) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
input_ids = {
|
|
port.id for port in ports if port.orientation in {"input", "indifferent"}
|
|
}
|
|
output_ids = {
|
|
port.id for port in ports if port.orientation in {"output", "indifferent"}
|
|
}
|
|
source_ids = output_ids
|
|
parent = self.document.find_parent(component_id)
|
|
if parent is not None:
|
|
for connection in parent.graph.connections.values():
|
|
if (
|
|
connection.target.block == component_id
|
|
and connection.target.port not in input_ids
|
|
):
|
|
raise ValueError("An input cannot be removed or reoriented while connected")
|
|
if (
|
|
connection.source.block == component_id
|
|
and connection.source.port not in source_ids
|
|
):
|
|
raise ValueError("An output cannot be removed or reoriented while connected")
|
|
for connection in component.graph.connections.values():
|
|
if connection.source.interface and connection.source.interface not in input_ids:
|
|
raise ValueError("An interface input cannot be removed while connected")
|
|
if connection.target.interface and connection.target.interface not in output_ids:
|
|
raise ValueError("An interface output cannot be removed while connected")
|
|
candidate = deepcopy(self.document)
|
|
candidate_component = candidate.find_component(component_id)
|
|
candidate_component.ports = deepcopy(ports)
|
|
candidate.validate()
|
|
old = {
|
|
"name": component.name,
|
|
"icon": component.icon.to_dict(),
|
|
"ports": [port.to_dict() for port in component.ports],
|
|
"show_subtree": component.show_subtree_in_library,
|
|
"properties": deepcopy(component.properties),
|
|
}
|
|
new = {
|
|
**old,
|
|
"ports": [port.to_dict() for port in ports],
|
|
}
|
|
if old != new:
|
|
self.undo_stack.push(EditComponentAppearanceCommand(self, component_id, old, new))
|
|
|
|
def edit_component_parameters(
|
|
self, component_id: str, parameters: list[Parameter]
|
|
) -> None:
|
|
component = self.document.find_component(component_id) if self.document else None
|
|
if component is None:
|
|
return
|
|
ids = [parameter.id for parameter in parameters]
|
|
names = [parameter.name for parameter in parameters]
|
|
if len(set(ids)) != len(ids):
|
|
raise ValueError("Parameter IDs must be unique")
|
|
if len(set(names)) != len(names):
|
|
raise ValueError("Parameter names must be unique")
|
|
if any(not name.strip() for name in names):
|
|
raise ValueError("Every parameter must have a name")
|
|
old = [parameter.to_dict() for parameter in component.parameters]
|
|
new = [parameter.to_dict() for parameter in parameters]
|
|
if old != new:
|
|
self.undo_stack.push(
|
|
EditComponentParametersCommand(self, component_id, old, new)
|
|
)
|
|
|
|
def delete_selection(
|
|
self,
|
|
block_ids: set[str],
|
|
connection_ids: set[str],
|
|
port_ids: set[str],
|
|
) -> None:
|
|
component = self.active_component
|
|
if component is None or component.implementation_kind != "graph":
|
|
return
|
|
graph = component.graph
|
|
all_connection_ids = set(connection_ids)
|
|
for connection in graph.connections.values():
|
|
if (
|
|
connection.source.block in block_ids
|
|
or connection.target.block in block_ids
|
|
or connection.source.interface in port_ids
|
|
or connection.target.interface in port_ids
|
|
):
|
|
all_connection_ids.add(connection.id)
|
|
blocks = {
|
|
block_id: graph.blocks[block_id] for block_id in block_ids if block_id in graph.blocks
|
|
}
|
|
connections = {
|
|
connection_id: graph.connections[connection_id]
|
|
for connection_id in all_connection_ids
|
|
if connection_id in graph.connections
|
|
}
|
|
ports = [port for port in component.ports if port.id in port_ids]
|
|
if not (blocks or connections or ports):
|
|
return
|
|
self.undo_stack.push(
|
|
DeleteSelectionCommand(
|
|
self,
|
|
component.id,
|
|
blocks,
|
|
connections,
|
|
ports,
|
|
)
|
|
)
|
|
|
|
def paste_selection(
|
|
self,
|
|
source_components: list[Component],
|
|
source_connections: list[Connection],
|
|
offset: QPointF,
|
|
) -> list[str]:
|
|
owner = self.active_component
|
|
if owner is None or owner.implementation_kind != "graph":
|
|
return []
|
|
pairs = [(source, clone_component(source)) for source in source_components]
|
|
id_map = {source.id: clone.id for source, clone in pairs}
|
|
blocks = {}
|
|
used = list(owner.graph.blocks.values())
|
|
for source, clone in pairs:
|
|
clone.name = self._available_component_name(source.name, used, 0)
|
|
clone.x += offset.x()
|
|
clone.y += offset.y()
|
|
blocks[clone.id] = clone
|
|
used.append(clone)
|
|
connections = {}
|
|
for source in source_connections:
|
|
if source.source.block not in id_map or source.target.block not in id_map:
|
|
continue
|
|
properties = deepcopy(source.properties)
|
|
for point in properties.get("waypoints", []):
|
|
if isinstance(point, dict):
|
|
point["x"] = float(point.get("x", 0)) + offset.x()
|
|
point["y"] = float(point.get("y", 0)) + offset.y()
|
|
connection = Connection(
|
|
id=str(uuid4()),
|
|
source=Endpoint(block=id_map[source.source.block], port=source.source.port),
|
|
target=Endpoint(block=id_map[source.target.block], port=source.target.port),
|
|
name=source.name,
|
|
properties=properties,
|
|
type=source.type,
|
|
causality=source.causality,
|
|
)
|
|
connections[connection.id] = connection
|
|
if blocks:
|
|
self.undo_stack.push(PasteSelectionCommand(self, owner.id, blocks, connections))
|
|
return list(blocks)
|
|
|
|
def paste_components_to(
|
|
self,
|
|
owner_id: str | None,
|
|
source_components: list[Component],
|
|
source_connections: list[Connection] | None = None,
|
|
) -> list[str]:
|
|
"""Clone components into the document root or a graph at origin."""
|
|
|
|
if self.document is None:
|
|
raise ValueError("Open or create a document before pasting components")
|
|
if owner_id is None:
|
|
siblings = self.document.roots.values()
|
|
else:
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None or owner.implementation_kind != "graph":
|
|
raise ValueError("Components can only be pasted into a graph")
|
|
siblings = owner.graph.blocks.values()
|
|
|
|
pairs = [(source, clone_component(source)) for source in source_components]
|
|
if not pairs:
|
|
return []
|
|
id_map = {source.id: clone.id for source, clone in pairs}
|
|
used = list(siblings)
|
|
minimum_x = min(source.x for source, _clone in pairs)
|
|
minimum_y = min(source.y for source, _clone in pairs)
|
|
blocks: dict[str, Component] = {}
|
|
for source, clone in pairs:
|
|
clone.name = self._available_component_name(source.name, used, 0)
|
|
if owner_id is not None:
|
|
clone.x = source.x - minimum_x
|
|
clone.y = source.y - minimum_y
|
|
blocks[clone.id] = clone
|
|
used.append(clone)
|
|
|
|
connections: dict[str, Connection] = {}
|
|
if owner_id is not None:
|
|
for source in source_connections or []:
|
|
if source.source.block not in id_map or source.target.block not in id_map:
|
|
continue
|
|
properties = deepcopy(source.properties)
|
|
for point in properties.get("waypoints", []):
|
|
if isinstance(point, dict):
|
|
point["x"] = float(point.get("x", 0)) - minimum_x
|
|
point["y"] = float(point.get("y", 0)) - minimum_y
|
|
connection = Connection(
|
|
id=str(uuid4()),
|
|
source=Endpoint(
|
|
block=id_map[source.source.block], port=source.source.port
|
|
),
|
|
target=Endpoint(
|
|
block=id_map[source.target.block], port=source.target.port
|
|
),
|
|
name=source.name,
|
|
properties=properties,
|
|
type=source.type,
|
|
causality=source.causality,
|
|
)
|
|
connections[connection.id] = connection
|
|
|
|
self.undo_stack.push(
|
|
PasteSelectionCommand(self, owner_id, blocks, connections)
|
|
)
|
|
return list(blocks)
|
|
|
|
def _graph_for(self, owner_id: str):
|
|
if self.document is None:
|
|
raise ValueError("There is no open document")
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
raise ValueError("The containing component is no longer in the document")
|
|
return owner.graph
|
|
|
|
@staticmethod
|
|
def _available_component_name(
|
|
base: str, components, start: int = 0
|
|
) -> str:
|
|
used = {component.name for component in components}
|
|
number = start
|
|
while f"{base}{number}" in used:
|
|
number += 1
|
|
return f"{base}{number}"
|
|
|
|
def _component_siblings(self, component_id: str):
|
|
if self.document is None:
|
|
return ()
|
|
parent = self.document.find_parent(component_id)
|
|
return (
|
|
parent.graph.blocks.values()
|
|
if parent is not None
|
|
else self.document.roots.values()
|
|
)
|
|
|
|
def _insert_component(self, owner_id: str | None, component: Component) -> None:
|
|
if self.document is None:
|
|
raise ValueError("There is no open document")
|
|
if owner_id is None:
|
|
self.document.roots[component.id] = component
|
|
else:
|
|
self._graph_for(owner_id).blocks[component.id] = component
|
|
if owner_id == self.active_component_id:
|
|
self.componentAdded.emit(component.id)
|
|
self.documentReset.emit()
|
|
|
|
def _remove_component(self, owner_id: str | None, component_id: str) -> None:
|
|
if self.document is None:
|
|
return
|
|
if owner_id is None:
|
|
self.document.roots.pop(component_id, None)
|
|
if self.active_component_id == component_id:
|
|
self.active_component_id = next(iter(self.document.roots), None)
|
|
self.activeGraphChanged.emit()
|
|
else:
|
|
self._graph_for(owner_id).blocks.pop(component_id, None)
|
|
if owner_id == self.active_component_id:
|
|
self.componentRemoved.emit(component_id)
|
|
self.documentReset.emit()
|
|
|
|
def _move_component(self, owner_id: str, component_id: str, position: QPointF) -> None:
|
|
component = self._graph_for(owner_id).blocks[component_id]
|
|
component.x, component.y = position.x(), position.y()
|
|
if owner_id == self.active_component_id:
|
|
self.componentMoved.emit(component_id, position)
|
|
self.documentReset.emit()
|
|
|
|
def _rotate_component(self, owner_id: str, component_id: str, rotation: float) -> None:
|
|
component = self._graph_for(owner_id).blocks.get(component_id)
|
|
if component is None:
|
|
return
|
|
component.rotation = rotation
|
|
if owner_id == self.active_component_id:
|
|
self.componentRotated.emit(component_id, rotation)
|
|
|
|
def _insert_connection(self, owner_id: str, connection: Connection) -> None:
|
|
self._graph_for(owner_id).connections[connection.id] = connection
|
|
if self.document is not None and self._infer_causality_on_connection_change():
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is not None:
|
|
self._infer_active_graph_causality(owner, emit_reset=False)
|
|
if owner_id == self.active_component_id:
|
|
self.connectionAdded.emit(connection.id)
|
|
self.documentReset.emit()
|
|
|
|
def _split_connection(
|
|
self,
|
|
owner_id: str,
|
|
original_id: str,
|
|
junction: Junction,
|
|
first: Connection,
|
|
second: Connection,
|
|
) -> None:
|
|
graph = self._graph_for(owner_id)
|
|
graph.connections.pop(original_id, None)
|
|
graph.junctions[junction.id] = junction
|
|
graph.connections[first.id] = first
|
|
graph.connections[second.id] = second
|
|
self.documentReset.emit()
|
|
|
|
def _restore_split_connection(
|
|
self,
|
|
owner_id: str,
|
|
original: Connection,
|
|
junction_id: str,
|
|
first_id: str,
|
|
second_id: str,
|
|
) -> None:
|
|
graph = self._graph_for(owner_id)
|
|
graph.connections.pop(first_id, None)
|
|
graph.connections.pop(second_id, None)
|
|
graph.junctions.pop(junction_id, None)
|
|
graph.connections[original.id] = original
|
|
self.documentReset.emit()
|
|
|
|
def _remove_connection(self, owner_id: str, connection_id: str) -> None:
|
|
self._graph_for(owner_id).connections.pop(connection_id, None)
|
|
if self.document is not None and self._infer_causality_on_connection_change():
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is not None:
|
|
self._infer_active_graph_causality(owner, emit_reset=False)
|
|
if owner_id == self.active_component_id:
|
|
self.connectionRemoved.emit(connection_id)
|
|
self.documentReset.emit()
|
|
|
|
@staticmethod
|
|
def _infer_causality_on_connection_change() -> bool:
|
|
return application_settings().value(
|
|
"bondGraph/inferCausalityOnConnectionChange", True, type=bool
|
|
)
|
|
|
|
def _insert_annotation(self, owner_id: str, annotation: Annotation) -> None:
|
|
self._graph_for(owner_id).annotations[annotation.id] = annotation
|
|
if owner_id == self.active_component_id:
|
|
self.annotationAdded.emit(annotation.id)
|
|
self.documentReset.emit()
|
|
|
|
def _remove_annotation(self, owner_id: str, annotation_id: str) -> None:
|
|
self._graph_for(owner_id).annotations.pop(annotation_id, None)
|
|
if owner_id == self.active_component_id:
|
|
self.annotationRemoved.emit(annotation_id)
|
|
self.documentReset.emit()
|
|
|
|
def _set_graph_item_data(
|
|
self, owner_id: str, item_kind: str, item_id: str, values: dict
|
|
) -> None:
|
|
graph = self._graph_for(owner_id)
|
|
if item_kind == "junction_geometry":
|
|
item = graph.junctions.get(item_id)
|
|
if item is not None:
|
|
item.x, item.y = float(values["x"]), float(values["y"])
|
|
elif item_kind == "connection_data":
|
|
item = graph.connections.get(item_id)
|
|
if item is not None:
|
|
item.name = values["name"]
|
|
item.properties = deepcopy(values["properties"])
|
|
elif item_kind == "connection":
|
|
item = graph.connections.get(item_id)
|
|
if item is not None:
|
|
item.properties = deepcopy(values)
|
|
else:
|
|
item = graph.annotations.get(item_id)
|
|
if item is not None:
|
|
if item_kind == "annotation_geometry":
|
|
item.x, item.y = float(values["x"]), float(values["y"])
|
|
item.width, item.height = float(values["width"]), float(values["height"])
|
|
elif item_kind == "annotation_layer":
|
|
item.layer = int(values["layer"])
|
|
elif item_kind == "annotation_data":
|
|
replacement = Annotation.from_dict(values)
|
|
item.kind = replacement.kind
|
|
item.x, item.y = replacement.x, replacement.y
|
|
item.width, item.height = replacement.width, replacement.height
|
|
item.text, item.layer = replacement.text, replacement.layer
|
|
item.properties = replacement.properties
|
|
else:
|
|
item.properties = deepcopy(values)
|
|
if owner_id == self.active_component_id:
|
|
self.graphItemChanged.emit(item_kind, item_id)
|
|
|
|
def _set_simulation_settings(self, owner_id: str, settings: dict) -> None:
|
|
owner = self.document.find_component(owner_id) if self.document else None
|
|
if owner is not None and owner.implementation_kind == "graph":
|
|
owner.graph.simulation_settings = deepcopy(settings)
|
|
self.documentReset.emit()
|
|
|
|
def _set_graph_parameters(self, values: dict[str, list[dict]]) -> None:
|
|
if self.document is None:
|
|
return
|
|
changed_text_components: list[str] = []
|
|
for component_id, parameters in values.items():
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
continue
|
|
component.parameters = [Parameter.from_dict(item) for item in parameters]
|
|
if component.implementation_kind == "text":
|
|
changed_text_components.append(component_id)
|
|
self.document.validate()
|
|
self.documentReset.emit()
|
|
for component_id in changed_text_components:
|
|
self.textDefinitionChanged.emit(component_id)
|
|
|
|
def _replace_component(self, old_id: str, replacement: Component) -> None:
|
|
if self.document is None:
|
|
return
|
|
was_active = self.active_component_id == old_id
|
|
if old_id in self.document.roots:
|
|
self.document.roots.pop(old_id)
|
|
self.document.roots[replacement.id] = replacement
|
|
else:
|
|
parent = self.document.find_parent(old_id)
|
|
if parent is None:
|
|
raise ValueError("The component is no longer in this document")
|
|
parent.graph.blocks.pop(old_id)
|
|
parent.graph.blocks[replacement.id] = replacement
|
|
if was_active:
|
|
self.active_component_id = replacement.id
|
|
self.document.validate()
|
|
self.documentReset.emit()
|
|
if was_active:
|
|
self.activeGraphChanged.emit()
|
|
|
|
def _insert_interface_port(self, owner_id: str, direction: str, port: Port) -> None:
|
|
if self.document is None:
|
|
return
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
return
|
|
port.orientation = direction
|
|
if all(existing.id != port.id for existing in owner.ports):
|
|
owner.ports.append(port)
|
|
self.interfaceChanged.emit()
|
|
self.documentReset.emit()
|
|
|
|
def _remove_interface_port(self, owner_id: str, direction: str, port_id: str) -> None:
|
|
if self.document is None:
|
|
return
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
return
|
|
owner.ports[:] = [port for port in owner.ports if port.id != port_id]
|
|
self.interfaceChanged.emit()
|
|
self.documentReset.emit()
|
|
|
|
def _move_interface_port(self, owner_id: str, port_id: str, position: QPointF) -> None:
|
|
if self.document is None:
|
|
return
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
return
|
|
port = next((port for port in owner.ports if port.id == port_id), None)
|
|
if port is not None:
|
|
port.x, port.y = position.x(), position.y()
|
|
self.interfaceChanged.emit()
|
|
self.documentReset.emit()
|
|
|
|
def _rename_interface_port(self, owner_id: str, port_id: str, name: str) -> None:
|
|
owner = self.document.find_component(owner_id) if self.document else None
|
|
if owner is None:
|
|
return
|
|
port = next((port for port in owner.ports if port.id == port_id), None)
|
|
if port is not None:
|
|
port.name = name
|
|
self.interfaceChanged.emit()
|
|
self.documentReset.emit()
|
|
|
|
def _rename_connection(self, owner_id: str, connection_id: str, name: str) -> None:
|
|
connection = self._graph_for(owner_id).connections.get(connection_id)
|
|
if connection is not None:
|
|
connection.name = name
|
|
self.documentReset.emit()
|
|
|
|
def _rename_document(self, name: str) -> None:
|
|
if self.document is None:
|
|
return
|
|
self.document.metadata["name"] = name
|
|
self.documentNameChanged.emit(name)
|
|
|
|
def _replace_source(self, component_id: str, source: dict) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
component.source = deepcopy(source)
|
|
self.documentReset.emit()
|
|
if component_id == self.active_component_id:
|
|
self.activeGraphChanged.emit()
|
|
|
|
def _set_component_appearance(self, component_id: str, values: dict) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
component.name = values["name"]
|
|
component.icon = Icon.from_dict(values["icon"])
|
|
component.ports = [Port.from_dict(port) for port in values["ports"]]
|
|
component.show_subtree_in_library = values["show_subtree"]
|
|
component.properties = deepcopy(values["properties"])
|
|
self.documentReset.emit()
|
|
if component_id == self.active_component_id:
|
|
self.activeGraphChanged.emit()
|
|
|
|
def _set_component_properties(self, component_id: str, properties: dict) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is not None:
|
|
component.properties = deepcopy(properties)
|
|
self.componentPropertiesChanged.emit(component_id)
|
|
|
|
def _set_component_parameters(self, component_id: str, values: list[dict]) -> None:
|
|
component = self.document.find_component(component_id) if self.document else None
|
|
if component is not None:
|
|
component.parameters = [Parameter.from_dict(item) for item in values]
|
|
self.documentReset.emit()
|
|
if (
|
|
component_id == self.active_component_id
|
|
and component.implementation_kind == "text"
|
|
):
|
|
self.textDefinitionChanged.emit(component_id)
|
|
|
|
def _delete_items(
|
|
self,
|
|
owner_id: str | None,
|
|
block_ids: set[str],
|
|
connection_ids: set[str],
|
|
port_ids: set[str],
|
|
) -> None:
|
|
if self.document is None:
|
|
return
|
|
deleted_component_ids: set[str] = set()
|
|
for block_id in block_ids:
|
|
component = self.document.find_component(block_id)
|
|
if component is not None:
|
|
deleted_component_ids.update(
|
|
child.id for child in self._component_subtree(component)
|
|
)
|
|
active_was_deleted = self.active_component_id in deleted_component_ids
|
|
if owner_id is None:
|
|
for block_id in block_ids:
|
|
self.document.roots.pop(block_id, None)
|
|
else:
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
return
|
|
for block_id in block_ids:
|
|
owner.graph.blocks.pop(block_id, None)
|
|
for connection_id in connection_ids:
|
|
owner.graph.connections.pop(connection_id, None)
|
|
owner.ports[:] = [port for port in owner.ports if port.id not in port_ids]
|
|
if active_was_deleted:
|
|
self.active_component_id = owner_id or next(iter(self.document.roots), None)
|
|
self.activeGraphChanged.emit()
|
|
self.documentReset.emit()
|
|
|
|
def _restore_items(
|
|
self,
|
|
owner_id: str | None,
|
|
blocks: dict[str, Component],
|
|
connections: dict[str, Connection],
|
|
ports: list[Port],
|
|
) -> None:
|
|
if self.document is None:
|
|
return
|
|
if owner_id is None:
|
|
self.document.roots.update(blocks)
|
|
else:
|
|
owner = self.document.find_component(owner_id)
|
|
if owner is None:
|
|
return
|
|
owner.graph.blocks.update(blocks)
|
|
owner.graph.connections.update(connections)
|
|
existing_ports = {port.id for port in owner.ports}
|
|
owner.ports.extend(port for port in ports if port.id not in existing_ports)
|
|
self.documentReset.emit()
|
|
|
|
@staticmethod
|
|
def _component_subtree(component: Component):
|
|
yield component
|
|
for child in component.graph.blocks.values():
|
|
yield from DocumentController._component_subtree(child)
|
|
|
|
def _set_text_definition(self, component_id: str, values: dict) -> None:
|
|
if self.document is None:
|
|
return
|
|
component = self.document.find_component(component_id)
|
|
if component is None:
|
|
return
|
|
component.ports = [Port.from_dict(item) for item in values["ports"]]
|
|
component.source = deepcopy(values["source"])
|
|
component.parameters = [
|
|
Parameter.from_dict(item) for item in values.get("parameters", [])
|
|
]
|
|
self.interfaceChanged.emit()
|
|
self.documentReset.emit()
|
|
self.textDefinitionChanged.emit(component_id)
|