joppe/BondGraph
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new connection editing

Browse Source
This commit is contained in:
Joppe Blondel
2026-07-20 13:50:33 +02:00
parent 6a3e9f01d0
commit 8f87a3b477
21 changed files with 1571 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

383
BEdit/src/bedit/gui/graphics/workspace.py
View File

@@ -1,10 +1,12 @@
import json
from dataclasses import dataclass
from PySide6.QtCore import QMimeData, QPointF, QRectF, Qt, Signal
from PySide6.QtGui import (
QColor,
QDragEnterEvent,
QDropEvent,
QKeyEvent,
QMouseEvent,
QWheelEvent,
QPainter,
@@ -32,17 +34,26 @@ from PySide6.QtWidgets import (
from bedit.core.model import Annotation, Component, Connection, Endpoint, Port
from bedit.gui.controllers.document import DocumentController
from bedit.gui.dialogs.connection_chooser import ConnectionChooserDialog
from bedit.gui.models.library_tree import COMPONENT_MIME_TYPE
from bedit.gui.graphics.icon_renderer import icon_bounds, paint_icon
from bedit.gui.graphics.icon_editor import ShapeOptionsDialog
from bedit.gui.graphics.icon_renderer import shape_pen
from bedit.gui.graphics.connection_styles import ConnectionStyle, connection_style
from bedit.gui.preferences import application_settings
from bedit.core.port_types import PortTypeRegistry
SELECTION_MIME_TYPE = "application/x-bedit-selection"
@dataclass(frozen=True)
class ConnectionChoice:
source: Endpoint
target: Endpoint
label: str
def _graph_snap_size() -> int:
return application_settings().value("grid/graphSnapSize", 8, type=int)
@@ -690,11 +701,10 @@ class GraphScene(QGraphicsScene):
self.output_items: dict[str, InterfaceTerminalItem] = {}
self.connection_items: dict[str, ConnectionGraphicsItem] = {}
self.annotation_items: dict[str, QGraphicsItem] = {}
self.pending_source: ConnectionPortItem | None = None
self.pending_connection_item: ComponentGraphicsItem | ConnectionPortItem | None = None
self.pending_waypoints: list[QPointF] = []
self.pending_preview: QGraphicsPathItem | None = None
self.interaction_mode = "pointer"
self.connection_routing = "angled"
self.drawing_start: QPointF | None = None
self.drawing_waypoints: list[QPointF] = []
self.setSceneRect(-2000, -2000, 4000, 4000)
@@ -713,7 +723,7 @@ class GraphScene(QGraphicsScene):
self.output_items.clear()
self.connection_items.clear()
self.annotation_items.clear()
self.pending_source = None
self.pending_connection_item = None
self.pending_waypoints.clear()
self.pending_preview = None
owner = self.controller.active_component
@@ -753,6 +763,7 @@ class GraphScene(QGraphicsScene):
item = AnnotationGraphicsItem(annotation, self.controller)
self.addItem(item)
self.annotation_items[annotation.id] = item
self._set_port_hints_visible(self.interaction_mode == "connect")
def set_component_position(self, component_id: str, position: QPointF) -> None:
item = self.component_items.get(component_id)
@@ -837,23 +848,65 @@ class GraphScene(QGraphicsScene):
ports = component.output_ports if role == "source" else component.input_ports
return ports.get(endpoint.port or "")
@staticmethod
def _hitbox_intersection(component: ComponentGraphicsItem, reference: QPointF) -> QPointF:
rect = component.hitbox.normalized()
local = component.mapFromScene(reference)
center = rect.center()
delta = local - center
if delta.isNull():
return component.mapToScene(QPointF(rect.right(), center.y()))
scales: list[float] = []
if delta.x() != 0:
scales.append((rect.width() / 2) / abs(delta.x()))
if delta.y() != 0:
scales.append((rect.height() / 2) / abs(delta.y()))
return component.mapToScene(center + delta * min(scales))
def _endpoint_center(self, endpoint: Endpoint, role: str) -> QPointF | None:
if endpoint.block is not None:
component = self.component_items.get(endpoint.block)
return component.mapToScene(component.hitbox.center()) if component else None
port = self._endpoint_item(endpoint, role)
return port.scenePos() if port else None
def _visible_endpoint(
self, endpoint: Endpoint, role: str, reference: QPointF
) -> QPointF | None:
if endpoint.block is not None:
component = self.component_items.get(endpoint.block)
return self._hitbox_intersection(component, reference) if component else None
port = self._endpoint_item(endpoint, role)
return port.scenePos() if port else None
def _connection_geometry(
self, connection: Connection, waypoints: list[QPointF]
) -> tuple[QPointF, QPointF] | None:
source_center = self._endpoint_center(connection.source, "source")
target_center = self._endpoint_center(connection.target, "target")
if source_center is None or target_center is None:
return None
start_reference = waypoints[0] if waypoints else target_center
end_reference = waypoints[-1] if waypoints else source_center
start = self._visible_endpoint(connection.source, "source", start_reference)
end = self._visible_endpoint(connection.target, "target", end_reference)
return (start, end) if start is not None and end is not None else None
def update_connection(self, connection_id: str) -> None:
connection = self.controller.active_graph.connections.get(connection_id)
graphics = self.connection_items.get(connection_id)
if connection is None or graphics is None:
return
source = self._endpoint_item(connection.source, "source")
target = self._endpoint_item(connection.target, "target")
if source is None or target is None:
return
start, end = source.scenePos(), target.scenePos()
routing = str(connection.properties.get("routing", "angled"))
waypoints = [
QPointF(float(point["x"]), float(point["y"]))
for point in connection.properties.get("waypoints", [])
if isinstance(point, dict) and "x" in point and "y" in point
]
path, direction_points = self._route_path(start, end, routing, waypoints)
geometry = self._connection_geometry(connection, waypoints)
if geometry is None:
return
start, end = geometry
path, direction_points = self._route_path(start, end, waypoints)
graphics.set_connection_path(path, direction_points)
graphics.set_waypoints(waypoints)
@@ -868,112 +921,193 @@ class GraphScene(QGraphicsScene):
QPointF(float(p["x"]), float(p["y"]))
for p in annotation.properties.get("waypoints", [])
]
routing = str(annotation.properties.get("routing", "angled"))
path, points = self._route_path(start, end, routing, waypoints)
path, points = self._route_path(start, end, waypoints)
graphics.setPath(path)
graphics.set_waypoints(waypoints, start, end)
@staticmethod
def _orthogonal_points(start: QPointF, end: QPointF, waypoints: list[QPointF]) -> list[QPointF]:
points = [start]
for waypoint in [*waypoints, end]:
previous = points[-1]
if previous.x() != waypoint.x() and previous.y() != waypoint.y():
elbow = QPointF(waypoint.x(), previous.y())
points.append(elbow)
if waypoint != points[-1]:
points.append(waypoint)
return points
@staticmethod
def _route_path(
start: QPointF,
end: QPointF,
routing: str,
waypoints: list[QPointF] | None = None,
) -> tuple[QPainterPath, list[QPointF]]:
waypoints = waypoints or []
points = [start, *(waypoints or []), end]
path = QPainterPath(start)
if routing == "angled":
points = GraphScene._orthogonal_points(start, end, waypoints)
for point in points[1:]:
path.lineTo(point)
return path, points
if routing == "direct":
path.lineTo(end)
return path, [start, end]
if waypoints:
points = [start, *waypoints, end]
for index, point in enumerate(points[1:-1], start=1):
following = points[index + 1]
midpoint = QPointF((point.x() + following.x()) / 2, (point.y() + following.y()) / 2)
path.quadTo(point, midpoint)
path.quadTo(points[-2], end)
return path, points
distance = max(50.0, abs(end.x() - start.x()) * 0.5)
first_control, second_control = start + QPointF(distance, 0), end - QPointF(distance, 0)
path.cubicTo(first_control, second_control, end)
return path, [start, first_control, second_control, end]
for point in points[1:]:
path.lineTo(point)
return path, points
def set_interaction_mode(self, mode: str) -> None:
self.interaction_mode = mode
if mode != "connect":
self._clear_pending_source()
self._clear_pending_connection()
self._clear_drawing()
self._set_port_hints_visible(mode == "connect")
def set_connection_routing(self, routing: str) -> None:
self.connection_routing = routing
self._clear_pending_source()
def _set_port_hints_visible(self, visible: bool) -> None:
for component in self.component_items.values():
for port in (*component.input_ports.values(), *component.output_ports.values()):
port.setVisible(visible)
for terminal in (*self.input_items.values(), *self.output_items.values()):
terminal.connection_port.setVisible(visible)
@staticmethod
def _clicked_component(
item: QGraphicsItem | None,
) -> ComponentGraphicsItem | None:
if isinstance(item, ComponentGraphicsItem):
return item
if isinstance(item, ConnectionPortItem) and isinstance(
item.parentItem(), ComponentGraphicsItem
):
return item.parentItem()
return None
@staticmethod
def _click_endpoint(item: QGraphicsItem | None) -> Endpoint | None:
return item.endpoint if isinstance(item, ConnectionPortItem) else None
@staticmethod
def _connection_anchor(
item: ComponentGraphicsItem | ConnectionPortItem,
) -> QPointF:
return (
item.scenePos()
if isinstance(item, ConnectionPortItem)
else item.sceneBoundingRect().center()
)
def _connection_choices(
self,
first: ComponentGraphicsItem,
second: ComponentGraphicsItem,
) -> list[ConnectionChoice]:
choices: list[ConnectionChoice] = []
def add_pairs(
source_item: ComponentGraphicsItem, target_item: ComponentGraphicsItem
) -> None:
for output in source_item.component.outputs:
for input_port in target_item.component.inputs:
if not PortTypeRegistry.compatible(output.type, input_port.type):
continue
choices.append(
ConnectionChoice(
Endpoint(block=source_item.component_id, port=output.id),
Endpoint(block=target_item.component_id, port=input_port.id),
f"{source_item.component.name}.{output.name} → "
f"{target_item.component.name}.{input_port.name}",
)
)
add_pairs(first, second)
add_pairs(second, first)
return choices
@staticmethod
def _default_choice_index(
choices: list[ConnectionChoice],
first_component: ComponentGraphicsItem,
second_component: ComponentGraphicsItem,
first_endpoint: Endpoint | None,
second_endpoint: Endpoint | None,
) -> int:
def score(choice: ConnectionChoice) -> int:
value = 0
if first_endpoint is not None and first_endpoint in (choice.source, choice.target):
value += 8
if second_endpoint is not None and second_endpoint in (choice.source, choice.target):
value += 8
if choice.source.block == first_component.component_id:
value += 2
if choice.target.block == second_component.component_id:
value += 1
return value
return max(range(len(choices)), key=lambda index: score(choices[index]))
def _finish_block_connection(
self,
second_item: ComponentGraphicsItem | ConnectionPortItem,
) -> None:
first_item = self.pending_connection_item
if first_item is None:
return
first_component = self._clicked_component(first_item)
second_component = self._clicked_component(second_item)
if (
first_component is None
or second_component is None
or first_component is second_component
):
self._clear_pending_connection()
return
choices = self._connection_choices(first_component, second_component)
if not choices:
QToolTip.showText(
self.views()[0].mapToGlobal(self.views()[0].viewport().rect().center())
if self.views()
else QPointF().toPoint(),
"These blocks have no compatible input/output pairs",
)
self._clear_pending_connection()
return
default = self._default_choice_index(
choices,
first_component,
second_component,
self._click_endpoint(first_item),
self._click_endpoint(second_item),
)
dialog = ConnectionChooserDialog(
[choice.label for choice in choices],
default,
self.views()[0] if self.views() else None,
)
if dialog.exec() == dialog.DialogCode.Accepted:
choice = choices[dialog.selected_index]
try:
self.controller.connect(
choice.source, choice.target, waypoints=self.pending_waypoints
)
except ValueError as error:
QToolTip.showText(dialog.mapToGlobal(dialog.rect().center()), str(error))
self._clear_pending_connection()
def mousePressEvent(self, event: QGraphicsSceneMouseEvent) -> None: # noqa: N802
if self.interaction_mode in {"box", "text", "line"}:
self._drawing_press(event)
return
if self.interaction_mode != "connect":
self._clear_pending_source()
self._clear_pending_connection()
super().mousePressEvent(event)
return
if event.button() == Qt.MouseButton.RightButton:
self._clear_pending_source()
self._clear_pending_connection()
event.accept()
return
item = self.itemAt(event.scenePos(), QTransform())
if isinstance(item, ConnectionPortItem):
if item.role == "source":
self._clear_pending_source()
self.pending_source = item
component = self._clicked_component(item)
if component is not None:
clicked = item if isinstance(item, ConnectionPortItem) else component
if self.pending_connection_item is None:
self.pending_connection_item = clicked
self.pending_waypoints = []
item.setBrush(QColor("#f5b642"))
if isinstance(clicked, ConnectionPortItem):
clicked.setBrush(QColor("#f5b642"))
self._create_preview()
elif self.pending_source is not None:
if self.pending_source.endpoint != item.endpoint:
try:
self.controller.connect(
self.pending_source.endpoint,
item.endpoint,
routing=self.connection_routing,
waypoints=(
self._completed_angled_waypoints(
self.pending_source.scenePos(),
item.scenePos(),
self.pending_waypoints,
)
if self.connection_routing == "angled"
else self.pending_waypoints
),
)
except ValueError as error:
QToolTip.showText(event.screenPos(), str(error))
self._clear_pending_source()
self._update_preview(event.scenePos())
else:
self._finish_block_connection(clicked)
event.accept()
return
if self.pending_source is not None and self.connection_routing == "angled":
if self.pending_connection_item is not None:
self.pending_waypoints.append(_snapped(event.scenePos()))
self._update_preview(event.scenePos())
event.accept()
def mouseMoveEvent(self, event: QGraphicsSceneMouseEvent) -> None: # noqa: N802
if self.interaction_mode == "connect" and self.pending_source is not None:
if self.interaction_mode == "connect" and self.pending_connection_item is not None:
self._update_preview(event.scenePos())
event.accept()
return
@@ -1011,14 +1145,9 @@ class GraphScene(QGraphicsScene):
self.drawing_start = point
self.drawing_waypoints = []
self._create_preview()
elif self.interaction_mode == "line" and self.connection_routing == "angled":
elif self.interaction_mode == "line":
self.drawing_waypoints.append(point)
self._update_drawing_preview(point)
elif self.interaction_mode == "line":
self.controller.add_annotation(
"line", self.drawing_start, point, routing=self.connection_routing
)
self._clear_drawing()
event.accept()
def mouseDoubleClickEvent(self, event: QGraphicsSceneMouseEvent) -> None: # noqa: N802
@@ -1029,21 +1158,12 @@ class GraphScene(QGraphicsScene):
if self.drawing_waypoints and self.drawing_waypoints[-1] == end
else self.drawing_waypoints
)
completed = self._completed_angled_waypoints(self.drawing_start, end, waypoints)
self.controller.add_annotation(
"line", self.drawing_start, end, routing="angled", waypoints=completed
)
self.controller.add_annotation("line", self.drawing_start, end, waypoints=waypoints)
self._clear_drawing()
event.accept()
return
super().mouseDoubleClickEvent(event)
@staticmethod
def _completed_angled_waypoints(
start: QPointF, end: QPointF, waypoints: list[QPointF]
) -> list[QPointF]:
return GraphScene._orthogonal_points(start, end, waypoints)[1:-1]
def _update_drawing_preview(self, cursor: QPointF) -> None:
if self.drawing_start is None or self.pending_preview is None:
return
@@ -1052,13 +1172,11 @@ class GraphScene(QGraphicsScene):
path = QPainterPath()
path.addRect(QRectF(self.drawing_start, end).normalized())
else:
path, _ = self._route_path(
self.drawing_start, end, self.connection_routing, self.drawing_waypoints
)
path, _ = self._route_path(self.drawing_start, end, self.drawing_waypoints)
self.pending_preview.setPath(path)
def _clear_drawing(self) -> None:
if self.pending_preview is not None and self.pending_source is None:
if self.pending_preview is not None and self.pending_connection_item is None:
self.removeItem(self.pending_preview)
self.pending_preview = None
self.drawing_start = None
@@ -1118,10 +1236,7 @@ class GraphScene(QGraphicsScene):
key=lambda index: segment_distance(snapped, anchors[index], anchors[index + 1]),
)
points.insert(insertion, snapped)
routing = "angled" if item.properties.get("routing") == "direct" else None
if routing == "angled" or item.properties.get("routing") == "angled":
points = self._orthogonal_points(start, end, points)[1:-1]
self.controller.set_route_waypoints(item_kind, item_id, points, routing)
self.controller.set_route_waypoints(item_kind, item_id, points)
def move_route_node(self, item_kind: str, item_id: str, index: int, position: QPointF) -> None:
item, points = self._route_values(item_kind, item_id)
@@ -1147,14 +1262,6 @@ class GraphScene(QGraphicsScene):
return
if index < len(points):
points[index] = position
if item.properties.get("routing") == "angled":
if item_kind == "annotation":
start = QPointF(item.x, item.y)
end = QPointF(item.x + item.width, item.y + item.height)
else:
graphics = self.connection_items[item_id]
start, end = graphics.start, graphics.end
points = self._orthogonal_points(start, end, points)[1:-1]
self.controller.set_route_waypoints(item_kind, item_id, points)
def preview_route_node(
@@ -1169,23 +1276,20 @@ class GraphScene(QGraphicsScene):
return
start = position if index == -1 else QPointF(item.x, item.y)
end = position if index == -2 else QPointF(item.x + item.width, item.y + item.height)
path, _direction_points = self._route_path(
start,
end,
str(item.properties.get("routing", "angled")),
points,
)
path, _direction_points = self._route_path(start, end, points)
graphics.setPath(path)
return
if index >= len(points):
return
points[index] = position
routing = str(item.properties.get("routing", "angled"))
if item_kind == "connection":
graphics = self.connection_items.get(item_id)
if graphics is None:
return
path, direction_points = self._route_path(graphics.start, graphics.end, routing, points)
geometry = self._connection_geometry(item, points)
if geometry is None:
return
path, direction_points = self._route_path(*geometry, points)
graphics.set_connection_path(path, direction_points)
return
graphics = self.annotation_items.get(item_id)
@@ -1193,7 +1297,7 @@ class GraphScene(QGraphicsScene):
return
start = QPointF(item.x, item.y)
end = QPointF(item.x + item.width, item.y + item.height)
path, _direction_points = self._route_path(start, end, routing, points)
path, _direction_points = self._route_path(start, end, points)
graphics.setPath(path)
def delete_route_node(self, item_kind: str, item_id: str, index: int) -> None:
@@ -1217,29 +1321,29 @@ class GraphScene(QGraphicsScene):
self.addItem(self.pending_preview)
def _update_preview(self, cursor: QPointF) -> None:
if self.pending_source is None or self.pending_preview is None:
if self.pending_connection_item is None or self.pending_preview is None:
return
preview_cursor = _snapped(cursor) if self.connection_routing == "angled" else cursor
preview_cursor = _snapped(cursor)
path, _points = self._route_path(
self.pending_source.scenePos(),
self._connection_anchor(self.pending_connection_item),
preview_cursor,
self.connection_routing,
self.pending_waypoints,
)
self.pending_preview.setPath(path)
def _clear_pending_source(self) -> None:
if self.pending_source is not None:
self.pending_source.setBrush(QColor("#ffffff"))
def _clear_pending_connection(self) -> None:
if isinstance(self.pending_connection_item, ConnectionPortItem):
self.pending_connection_item.setBrush(QColor("#ffffff"))
if self.pending_preview is not None:
self.removeItem(self.pending_preview)
self.pending_source = None
self.pending_connection_item = None
self.pending_waypoints.clear()
self.pending_preview = None
class GraphWorkspaceView(QGraphicsView):
toolUsed = Signal()
toolModeShortcutRequested = Signal(str)
componentOptionsRequested = Signal(str)
componentPortOptionsRequested = Signal(str)
portOptionsRequested = Signal(str, str)
@@ -1252,6 +1356,7 @@ class GraphWorkspaceView(QGraphicsView):
self.tool_mode = "pointer"
self.paste_count = 0
self.setAcceptDrops(True)
self.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
self.setMouseTracking(True)
self.setRenderHint(QPainter.RenderHint.Antialiasing)
self.setDragMode(QGraphicsView.DragMode.RubberBandDrag)
@@ -1304,6 +1409,23 @@ class GraphWorkspaceView(QGraphicsView):
self._zoom(1.2 if event.angleDelta().y() > 0 else 1 / 1.2)
event.accept()
def keyPressEvent(self, event: QKeyEvent) -> None: # noqa: N802
if not event.isAutoRepeat() and event.key() == Qt.Key.Key_Space:
self.toolModeShortcutRequested.emit(
"connect" if self.tool_mode == "pointer" else "pointer"
)
event.accept()
return
if (
not event.isAutoRepeat()
and event.key() == Qt.Key.Key_Escape
and self.tool_mode != "pointer"
):
self.toolModeShortcutRequested.emit("pointer")
event.accept()
return
super().keyPressEvent(event)
def set_model(self, controller: DocumentController) -> None:
self.controller = controller
scene = GraphScene(controller, self)
@@ -1445,11 +1567,6 @@ class GraphWorkspaceView(QGraphicsView):
if isinstance(scene, GraphScene):
scene.set_interaction_mode(mode)
def set_connection_routing(self, routing: str) -> None:
scene = self.scene()
if isinstance(scene, GraphScene):
scene.set_connection_routing(routing)
def mousePressEvent(self, event: QMouseEvent) -> None: # noqa: N802
super().mousePressEvent(event)