Reorganized the application around a clear frontend/backend boundary.
src/bedit/
├── __main__.py
├── core/ # Pure Python, no PySide
│ ├── model.py
│ ├── port_types.py
│ ├── serializer.py
│ └── libraries.py
└── gui/ # All Qt-dependent code
├── app.py
├── main_window.py
├── preferences.py
├── controllers/
├── dialogs/
├── graphics/
├── models/
└── generated/ # Designer/resource output only
Notable improvements:
Domain models, serialization, port types, and library parsing are now Qt-free.
Qt signals and undo infrastructure are explicitly isolated under gui/controllers.
Library parsing is separated from the Qt repository and tree models.
All generated Python is contained in gui/generated.
Designer build tasks now write to the generated directory.
The application entry point and package metadata use the new paths.
README now documents the structure and dependency rules.
Removed the old mixed document, library, and workspace packages.
This commit is contained in:
5
BEdit/src/bedit/gui/controllers/__init__.py
Normal file
5
BEdit/src/bedit/gui/controllers/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Qt-aware application controllers and undo commands."""
|
||||
|
||||
from bedit.gui.controllers.document import DocumentController
|
||||
|
||||
__all__ = ["DocumentController"]
|
||||
262
BEdit/src/bedit/gui/controllers/commands.py
Normal file
262
BEdit/src/bedit/gui/controllers/commands.py
Normal file
@@ -0,0 +1,262 @@
|
||||
from PySide6.QtCore import QPointF
|
||||
from PySide6.QtGui import QUndoCommand
|
||||
|
||||
from bedit.core.model import Component, Connection, Port
|
||||
|
||||
|
||||
class AddComponentCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str | None, component: Component) -> None:
|
||||
super().__init__(f"Add {component.name}")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.component = component
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._insert_component(self.owner_id, self.component)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._remove_component(self.owner_id, self.component.id)
|
||||
|
||||
|
||||
class MoveComponentCommand(QUndoCommand):
|
||||
def __init__(
|
||||
self,
|
||||
controller,
|
||||
owner_id: str,
|
||||
component_id: str,
|
||||
old: QPointF,
|
||||
new: QPointF,
|
||||
) -> None:
|
||||
super().__init__("Move component")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.component_id = component_id
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._move_component(self.owner_id, self.component_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._move_component(self.owner_id, self.component_id, self.old)
|
||||
|
||||
|
||||
class RotateComponentsCommand(QUndoCommand):
|
||||
def __init__(
|
||||
self,
|
||||
controller,
|
||||
owner_id: str,
|
||||
rotations: dict[str, tuple[float, float]],
|
||||
) -> None:
|
||||
super().__init__("Rotate components")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.rotations = rotations
|
||||
|
||||
def redo(self) -> None:
|
||||
for component_id, (_old, new) in self.rotations.items():
|
||||
self.controller._rotate_component(self.owner_id, component_id, new)
|
||||
|
||||
def undo(self) -> None:
|
||||
for component_id, (old, _new) in self.rotations.items():
|
||||
self.controller._rotate_component(self.owner_id, component_id, old)
|
||||
|
||||
|
||||
class AddConnectionCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str, connection: Connection) -> None:
|
||||
super().__init__("Connect components")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.connection = connection
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._insert_connection(self.owner_id, self.connection)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._remove_connection(self.owner_id, self.connection.id)
|
||||
|
||||
|
||||
class ReplaceComponentCommand(QUndoCommand):
|
||||
def __init__(self, controller, old: Component, new: Component) -> None:
|
||||
super().__init__("Apply JSON changes")
|
||||
self.controller = controller
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._replace_component(self.old.id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._replace_component(self.new.id, self.old)
|
||||
|
||||
|
||||
class AddInterfacePortCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str, direction: str, port: Port) -> None:
|
||||
super().__init__(f"Add {direction}")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.direction = direction
|
||||
self.port = port
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._insert_interface_port(self.owner_id, self.direction, self.port)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._remove_interface_port(self.owner_id, self.direction, self.port.id)
|
||||
|
||||
|
||||
class ReplaceSourceCommand(QUndoCommand):
|
||||
def __init__(self, controller, component_id: str, old: dict, new: dict) -> None:
|
||||
super().__init__("Apply source JSON")
|
||||
self.controller = controller
|
||||
self.component_id = component_id
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._replace_source(self.component_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._replace_source(self.component_id, self.old)
|
||||
|
||||
|
||||
class MoveInterfacePortCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str, port_id: str, old: QPointF, new: QPointF) -> None:
|
||||
super().__init__("Move interface terminal")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.port_id = port_id
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._move_interface_port(self.owner_id, self.port_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._move_interface_port(self.owner_id, self.port_id, self.old)
|
||||
|
||||
|
||||
class EditComponentAppearanceCommand(QUndoCommand):
|
||||
def __init__(self, controller, component_id: str, old: dict, new: dict) -> None:
|
||||
super().__init__("Edit component appearance")
|
||||
self.controller = controller
|
||||
self.component_id = component_id
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._set_component_appearance(self.component_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._set_component_appearance(self.component_id, self.old)
|
||||
|
||||
|
||||
class RenameInterfacePortCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str, port_id: str, old: str, new: str) -> None:
|
||||
super().__init__("Rename interface port")
|
||||
self.controller, self.owner_id, self.port_id = controller, owner_id, port_id
|
||||
self.old, self.new = old, new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._rename_interface_port(self.owner_id, self.port_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._rename_interface_port(self.owner_id, self.port_id, self.old)
|
||||
|
||||
|
||||
class RenameConnectionCommand(QUndoCommand):
|
||||
def __init__(self, controller, owner_id: str, connection_id: str, old: str, new: str) -> None:
|
||||
super().__init__("Rename connection")
|
||||
self.controller, self.owner_id, self.connection_id = controller, owner_id, connection_id
|
||||
self.old, self.new = old, new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._rename_connection(self.owner_id, self.connection_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._rename_connection(self.owner_id, self.connection_id, self.old)
|
||||
|
||||
|
||||
class DeleteSelectionCommand(QUndoCommand):
|
||||
def __init__(
|
||||
self,
|
||||
controller,
|
||||
owner_id: str | None,
|
||||
blocks: dict[str, Component],
|
||||
connections: dict[str, Connection],
|
||||
inputs: list[Port],
|
||||
outputs: list[Port],
|
||||
) -> None:
|
||||
super().__init__("Delete selection")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.blocks = blocks
|
||||
self.connections = connections
|
||||
self.inputs = inputs
|
||||
self.outputs = outputs
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._delete_items(
|
||||
self.owner_id,
|
||||
set(self.blocks),
|
||||
set(self.connections),
|
||||
{port.id for port in self.inputs},
|
||||
{port.id for port in self.outputs},
|
||||
)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._restore_items(
|
||||
self.owner_id,
|
||||
self.blocks,
|
||||
self.connections,
|
||||
self.inputs,
|
||||
self.outputs,
|
||||
)
|
||||
|
||||
|
||||
class PasteSelectionCommand(QUndoCommand):
|
||||
def __init__(
|
||||
self,
|
||||
controller,
|
||||
owner_id: str,
|
||||
blocks: dict[str, Component],
|
||||
connections: dict[str, Connection],
|
||||
) -> None:
|
||||
super().__init__("Paste selection")
|
||||
self.controller = controller
|
||||
self.owner_id = owner_id
|
||||
self.blocks = blocks
|
||||
self.connections = connections
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._restore_items(
|
||||
self.owner_id,
|
||||
self.blocks,
|
||||
self.connections,
|
||||
[],
|
||||
[],
|
||||
)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._delete_items(
|
||||
self.owner_id,
|
||||
set(self.blocks),
|
||||
set(self.connections),
|
||||
set(),
|
||||
set(),
|
||||
)
|
||||
|
||||
|
||||
class EditTextDefinitionCommand(QUndoCommand):
|
||||
def __init__(self, controller, component_id: str, old: dict, new: dict) -> None:
|
||||
super().__init__("Edit text component")
|
||||
self.controller = controller
|
||||
self.component_id = component_id
|
||||
self.old = old
|
||||
self.new = new
|
||||
|
||||
def redo(self) -> None:
|
||||
self.controller._set_text_definition(self.component_id, self.new)
|
||||
|
||||
def undo(self) -> None:
|
||||
self.controller._set_text_definition(self.component_id, self.old)
|
||||
741
BEdit/src/bedit/gui/controllers/document.py
Normal file
741
BEdit/src/bedit/gui/controllers/document.py
Normal file
@@ -0,0 +1,741 @@
|
||||
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,
|
||||
AddInterfacePortCommand,
|
||||
DeleteSelectionCommand,
|
||||
EditTextDefinitionCommand,
|
||||
EditComponentAppearanceCommand,
|
||||
MoveComponentCommand,
|
||||
MoveInterfacePortCommand,
|
||||
PasteSelectionCommand,
|
||||
RenameConnectionCommand,
|
||||
RenameInterfacePortCommand,
|
||||
ReplaceSourceCommand,
|
||||
RotateComponentsCommand,
|
||||
)
|
||||
from bedit.core.model import (
|
||||
Component,
|
||||
Connection,
|
||||
Endpoint,
|
||||
GraphDocument,
|
||||
Icon,
|
||||
Port,
|
||||
clone_component,
|
||||
)
|
||||
from bedit.core.port_types import PortTypeRegistry
|
||||
from bedit.core.serializer import JsonDocumentSerializer
|
||||
|
||||
|
||||
class DocumentController(QObject):
|
||||
documentReset = Signal()
|
||||
documentOpenedChanged = Signal(bool)
|
||||
activeGraphChanged = Signal()
|
||||
componentAdded = Signal(str)
|
||||
componentRemoved = Signal(str)
|
||||
componentMoved = Signal(str, QPointF)
|
||||
componentRotated = Signal(str, float)
|
||||
connectionAdded = Signal(str)
|
||||
connectionRemoved = Signal(str)
|
||||
interfaceChanged = Signal()
|
||||
filePathChanged = Signal(object)
|
||||
modifiedChanged = Signal(bool)
|
||||
|
||||
def __init__(self, parent=None) -> None:
|
||||
super().__init__(parent)
|
||||
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 = JsonDocumentSerializer.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")
|
||||
JsonDocumentSerializer.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
|
||||
component = Component(
|
||||
id=str(uuid4()),
|
||||
name=f"New {'Graph' if kind == 'graph' else 'Text'} Block {number}",
|
||||
implementation_kind=kind,
|
||||
icon=Icon(text="Graph" if kind == "graph" else "Text"),
|
||||
source={"equations": [], "parameters": {}} 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
|
||||
component = Component(
|
||||
id=str(uuid4()),
|
||||
name=f"New {'Graph' if kind == 'graph' else 'Text'} Block {number}",
|
||||
implementation_kind=kind,
|
||||
icon=Icon(text="Graph" if kind == "graph" else "Text"),
|
||||
source={"equations": [], "parameters": {}} 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 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.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 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) -> 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):
|
||||
raise ValueError(
|
||||
f"Cannot connect {source_port.type!r} to {target_port.type!r}"
|
||||
)
|
||||
connection = Connection(str(uuid4()), source, target)
|
||||
self.undo_stack.push(
|
||||
AddConnectionCommand(self, self.active_component_id, connection)
|
||||
)
|
||||
return connection.id
|
||||
|
||||
def _port_for_endpoint(self, endpoint: Endpoint, role: str) -> Port | None:
|
||||
owner = self.active_component
|
||||
if owner is None:
|
||||
return None
|
||||
if endpoint.interface is not None:
|
||||
ports = owner.inputs if role == "source" else owner.outputs
|
||||
else:
|
||||
component = owner.graph.blocks.get(endpoint.block or "")
|
||||
if component is None:
|
||||
return None
|
||||
ports = component.outputs if role == "source" else component.inputs
|
||||
return next((port for port in ports if port.id == (endpoint.interface or endpoint.port)), None)
|
||||
|
||||
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 = component.inputs if direction == "input" else component.outputs
|
||||
port = Port(
|
||||
id=f"{direction}-{uuid4().hex[:8]}",
|
||||
name=f"{direction.title()} {len(ports) + 1}",
|
||||
x=position.x(),
|
||||
y=position.y(),
|
||||
)
|
||||
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((p for p in (*owner.inputs, *owner.outputs) if p.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,
|
||||
inputs: list[Port],
|
||||
outputs: list[Port],
|
||||
source: dict,
|
||||
) -> None:
|
||||
component = self.active_component
|
||||
if component is None or component.implementation_kind != "text":
|
||||
raise ValueError("Only text-defined components can be edited here")
|
||||
input_ids = [port.id for port in inputs]
|
||||
output_ids = [port.id for port in outputs]
|
||||
if len(set(input_ids)) != len(input_ids) or len(set(output_ids)) != len(output_ids):
|
||||
raise ValueError("Input and output IDs must be unique")
|
||||
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 output_ids:
|
||||
raise ValueError(
|
||||
f"Output {connection.source.port!r} is still connected in the containing graph"
|
||||
)
|
||||
old = {
|
||||
"inputs": [port.to_dict() for port in component.inputs],
|
||||
"outputs": [port.to_dict() for port in component.outputs],
|
||||
"source": deepcopy(component.source),
|
||||
}
|
||||
new = {
|
||||
"inputs": [port.to_dict() for port in inputs],
|
||||
"outputs": [port.to_dict() for port in outputs],
|
||||
"source": deepcopy(source),
|
||||
}
|
||||
if old != new:
|
||||
self.undo_stack.push(EditTextDefinitionCommand(self, component.id, old, new))
|
||||
|
||||
def edit_component_appearance(
|
||||
self,
|
||||
component_id: str,
|
||||
name: str,
|
||||
icon: Icon,
|
||||
inputs: list[Port],
|
||||
outputs: list[Port],
|
||||
show_subtree: bool,
|
||||
) -> None:
|
||||
if self.document is None:
|
||||
return
|
||||
component = self.document.find_component(component_id)
|
||||
if component is None:
|
||||
return
|
||||
old = {
|
||||
"name": component.name,
|
||||
"icon": component.icon.to_dict(),
|
||||
"inputs": [port.to_dict() for port in component.inputs],
|
||||
"outputs": [port.to_dict() for port in component.outputs],
|
||||
"show_subtree": component.show_subtree_in_library,
|
||||
}
|
||||
new = {
|
||||
"name": name,
|
||||
"icon": icon.to_dict(),
|
||||
"inputs": [port.to_dict() for port in inputs],
|
||||
"outputs": [port.to_dict() for port in outputs],
|
||||
"show_subtree": show_subtree,
|
||||
}
|
||||
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.inputs = deepcopy(inputs)
|
||||
candidate_component.outputs = deepcopy(outputs)
|
||||
candidate.validate()
|
||||
self.undo_stack.push(EditComponentAppearanceCommand(self, component_id, old, new))
|
||||
|
||||
def edit_component_ports(
|
||||
self, component_id: str, inputs: list[Port], outputs: 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 inputs}
|
||||
output_ids = {port.id for port in outputs}
|
||||
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 output_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.inputs = deepcopy(inputs)
|
||||
candidate_component.outputs = deepcopy(outputs)
|
||||
candidate.validate()
|
||||
old = {
|
||||
"name": component.name,
|
||||
"icon": component.icon.to_dict(),
|
||||
"inputs": [port.to_dict() for port in component.inputs],
|
||||
"outputs": [port.to_dict() for port in component.outputs],
|
||||
"show_subtree": component.show_subtree_in_library,
|
||||
}
|
||||
new = {
|
||||
**old,
|
||||
"inputs": [port.to_dict() for port in inputs],
|
||||
"outputs": [port.to_dict() for port in outputs],
|
||||
}
|
||||
if old != new:
|
||||
self.undo_stack.push(EditComponentAppearanceCommand(self, component_id, old, new))
|
||||
|
||||
def delete_selection(
|
||||
self,
|
||||
block_ids: set[str],
|
||||
connection_ids: set[str],
|
||||
input_ids: set[str],
|
||||
output_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 input_ids
|
||||
or connection.target.interface in output_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
|
||||
}
|
||||
inputs = [port for port in component.inputs if port.id in input_ids]
|
||||
outputs = [port for port in component.outputs if port.id in output_ids]
|
||||
if not (blocks or connections or inputs or outputs):
|
||||
return
|
||||
self.undo_stack.push(
|
||||
DeleteSelectionCommand(
|
||||
self,
|
||||
component.id,
|
||||
blocks,
|
||||
connections,
|
||||
inputs,
|
||||
outputs,
|
||||
)
|
||||
)
|
||||
|
||||
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 = {}
|
||||
for _source, clone in pairs:
|
||||
clone.x += offset.x()
|
||||
clone.y += offset.y()
|
||||
blocks[clone.id] = clone
|
||||
connections = {}
|
||||
for source in source_connections:
|
||||
if source.source.block not in id_map or source.target.block not in id_map:
|
||||
continue
|
||||
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),
|
||||
)
|
||||
connections[connection.id] = connection
|
||||
if blocks:
|
||||
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
|
||||
|
||||
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 owner_id == self.active_component_id:
|
||||
self.connectionAdded.emit(connection.id)
|
||||
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 owner_id == self.active_component_id:
|
||||
self.connectionRemoved.emit(connection_id)
|
||||
self.documentReset.emit()
|
||||
|
||||
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
|
||||
ports = owner.inputs if direction == "input" else owner.outputs
|
||||
if all(existing.id != port.id for existing in ports):
|
||||
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
|
||||
ports = owner.inputs if direction == "input" else owner.outputs
|
||||
ports[:] = [port for port in 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((p for p in (*owner.inputs, *owner.outputs) if p.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((p for p in (*owner.inputs, *owner.outputs) if p.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 _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.inputs = [Port.from_dict(port) for port in values["inputs"]]
|
||||
component.outputs = [Port.from_dict(port) for port in values["outputs"]]
|
||||
component.show_subtree_in_library = values["show_subtree"]
|
||||
self.documentReset.emit()
|
||||
if component_id == self.active_component_id:
|
||||
self.activeGraphChanged.emit()
|
||||
|
||||
def _delete_items(
|
||||
self,
|
||||
owner_id: str | None,
|
||||
block_ids: set[str],
|
||||
connection_ids: set[str],
|
||||
input_ids: set[str],
|
||||
output_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.inputs[:] = [port for port in owner.inputs if port.id not in input_ids]
|
||||
owner.outputs[:] = [port for port in owner.outputs if port.id not in output_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],
|
||||
inputs: list[Port],
|
||||
outputs: 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_inputs = {port.id for port in owner.inputs}
|
||||
existing_outputs = {port.id for port in owner.outputs}
|
||||
owner.inputs.extend(port for port in inputs if port.id not in existing_inputs)
|
||||
owner.outputs.extend(port for port in outputs if port.id not in existing_outputs)
|
||||
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.inputs = [Port.from_dict(item) for item in values["inputs"]]
|
||||
component.outputs = [Port.from_dict(item) for item in values["outputs"]]
|
||||
component.source = deepcopy(values["source"])
|
||||
self.interfaceChanged.emit()
|
||||
self.documentReset.emit()
|
||||
if component_id == self.active_component_id:
|
||||
self.activeGraphChanged.emit()
|
||||
Reference in New Issue
Block a user