802 lines
42 KiB
Python
802 lines
42 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from enum import Enum
|
|
from itertools import pairwise
|
|
from math import hypot
|
|
|
|
from PySide6.QtCore import QEvent, QObject, QPointF, QRectF, QTimer, Qt, Signal
|
|
from PySide6.QtGui import QActionGroup, QBrush, QColor, QCursor, QDragEnterEvent, QDragMoveEvent, QDropEvent, QKeySequence, QMouseEvent, QPainter, QPainterPath, QPen, QShortcut, QWheelEvent
|
|
from PySide6.QtWidgets import QGraphicsEllipseItem, QGraphicsItem, QGraphicsPathItem, QGraphicsPixmapItem, QGraphicsScene, QGraphicsSceneMouseEvent, QGraphicsTextItem, QGraphicsView, QMenu, QWidget
|
|
|
|
from bedit_core.models import BondCausality, BondConnection, BondPort, Component, ComponentID, Connection, ConnectionID, GraphImplementation, Port, PortID, SignalConnection, SignalDirection, SignalPort
|
|
from bedit_gui.models import Graph, GraphComponentLabel, GraphConnection, Icon
|
|
from bedit_gui.services.component_clipboard import COMPONENTS_MIME
|
|
from bedit_gui.ui.generated.ui_graph_editor_widget import Ui_graphEditorWidget
|
|
from bedit_gui.utils.icon import get_natural_icon_size, get_pixmap_bounding_box, render_icon
|
|
|
|
GRID_SPACING = 64
|
|
SCENE_SIZE = 10000
|
|
MIN_ZOOM = 0.2
|
|
MAX_ZOOM = 4.0
|
|
ZOOM_STEP = 1.15
|
|
ZOOM_TO_FIT_PADDING = 32.0
|
|
COMPONENT_LABEL_FONT_SIZE = 24.0
|
|
FALLBACK_COMPONENT_SPACING = 128
|
|
CONNECTION_WIDTH = 4.0
|
|
BOND_CONNECTION_COLOR = "#000000"
|
|
SIGNAL_CONNECTION_COLOR = "#00007f"
|
|
BOND_CONNECTION_BOUNDING_BOX_SPACING = 16.0
|
|
SIGNAL_CONNECTION_BOUNDING_BOX_SPACING = 0.0
|
|
CONNECTION_STRAIGHTEN_TOLERANCE = 8.0
|
|
ARROW_LENGTH = 32.0
|
|
ARROW_HALF_WIDTH = 16.0
|
|
SIGNAL_ARROW_LENGTH = ARROW_LENGTH / 2
|
|
SIGNAL_ARROW_HALF_WIDTH = ARROW_HALF_WIDTH / 2
|
|
CAUSALITY_TICK_HALF_LENGTH = 16.0
|
|
|
|
|
|
class GraphEditorMode(Enum):
|
|
NORMAL = "normal"
|
|
CONNECTION = "connection"
|
|
|
|
|
|
class GraphGraphicsScene(QGraphicsScene):
|
|
"""Graph canvas with a lightweight dotted-line grid."""
|
|
|
|
def drawBackground(self, painter: QPainter, rect: QRectF) -> None:
|
|
painter.fillRect(rect, QColor("white"))
|
|
pen = QPen(QColor(205, 205, 205), 0, Qt.PenStyle.DotLine)
|
|
painter.setPen(pen)
|
|
|
|
scene_rect = self.sceneRect()
|
|
left = int(rect.left()) - int(rect.left()) % GRID_SPACING
|
|
top = int(rect.top()) - int(rect.top()) % GRID_SPACING
|
|
x = left
|
|
while x <= rect.right():
|
|
painter.drawLine(x, scene_rect.top(), x, scene_rect.bottom())
|
|
x += GRID_SPACING
|
|
y = top
|
|
while y <= rect.bottom():
|
|
painter.drawLine(scene_rect.left(), y, scene_rect.right(), y)
|
|
y += GRID_SPACING
|
|
|
|
|
|
class GraphConnectionItem(QGraphicsPathItem):
|
|
"""A routed connection with a full signal arrow or half bond arrow."""
|
|
|
|
def __init__(self, points: list[tuple[float, float]], *, half_arrow: bool, tick_at_source: bool | None = None, connection_id: ConnectionID | None = None, editor: GraphEditorWidget | None = None) -> None:
|
|
super().__init__()
|
|
self.connection_id = connection_id
|
|
self.editor = editor
|
|
self.setPath(self._connection_path(points, half_arrow, tick_at_source))
|
|
color = QColor(BOND_CONNECTION_COLOR if half_arrow else SIGNAL_CONNECTION_COLOR)
|
|
pen = QPen(color, CONNECTION_WIDTH)
|
|
self.setPen(pen)
|
|
self.setBrush(QBrush(color))
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsSelectable)
|
|
self.setZValue(-1)
|
|
|
|
def contextMenuEvent(self, event) -> None:
|
|
if self.editor is None or self.connection_id is None:
|
|
return
|
|
if not self.isSelected():
|
|
self.scene().clearSelection()
|
|
self.setSelected(True)
|
|
menu = QMenu(self.editor)
|
|
add_point = menu.addAction("Add Point")
|
|
delete_connection = menu.addAction("Delete Connection")
|
|
selected = menu.exec(event.screenPos())
|
|
if selected is add_point:
|
|
self.editor.add_connection_point(self.connection_id, event.scenePos())
|
|
elif selected is delete_connection:
|
|
self.editor.delete_selected_connections()
|
|
event.accept()
|
|
|
|
@staticmethod
|
|
def _connection_path(points: list[tuple[float, float]], half_arrow: bool, tick_at_source: bool | None = None) -> QPainterPath:
|
|
path = QPainterPath(QPointF(*points[0]))
|
|
for point in points[1:]:
|
|
path.lineTo(QPointF(*point))
|
|
|
|
target = QPointF(*points[-1])
|
|
previous = next((QPointF(*point) for point in reversed(points[:-1]) if QPointF(*point) != target), None)
|
|
if previous is None:
|
|
return path
|
|
dx = target.x() - previous.x()
|
|
dy = target.y() - previous.y()
|
|
length = hypot(dx, dy)
|
|
arrow_length = ARROW_LENGTH if half_arrow else SIGNAL_ARROW_LENGTH
|
|
arrow_half_width = ARROW_HALF_WIDTH if half_arrow else SIGNAL_ARROW_HALF_WIDTH
|
|
back_x = target.x() - arrow_length * dx / length
|
|
back_y = target.y() - arrow_length * dy / length
|
|
perpendicular_x = -arrow_half_width * dy / length
|
|
perpendicular_y = arrow_half_width * dx / length
|
|
path.moveTo(target)
|
|
path.lineTo(back_x + perpendicular_x, back_y + perpendicular_y)
|
|
if not half_arrow:
|
|
path.lineTo(back_x - perpendicular_x, back_y - perpendicular_y)
|
|
path.closeSubpath()
|
|
if tick_at_source is not None:
|
|
GraphConnectionItem._add_causality_tick(path, points, tick_at_source)
|
|
return path
|
|
|
|
@staticmethod
|
|
def _add_causality_tick(path: QPainterPath, points: list[tuple[float, float]], at_source: bool) -> None:
|
|
if at_source:
|
|
endpoint = QPointF(*points[0])
|
|
neighbor = next((QPointF(*point) for point in points[1:] if QPointF(*point) != endpoint), None)
|
|
else:
|
|
endpoint = QPointF(*points[-1])
|
|
neighbor = next((QPointF(*point) for point in reversed(points[:-1]) if QPointF(*point) != endpoint), None)
|
|
if neighbor is None:
|
|
return
|
|
dx = neighbor.x() - endpoint.x()
|
|
dy = neighbor.y() - endpoint.y()
|
|
length = hypot(dx, dy)
|
|
perpendicular_x = -CAUSALITY_TICK_HALF_LENGTH * dy / length
|
|
perpendicular_y = CAUSALITY_TICK_HALF_LENGTH * dx / length
|
|
path.moveTo(endpoint.x() - perpendicular_x, endpoint.y() - perpendicular_y)
|
|
path.lineTo(endpoint.x() + perpendicular_x, endpoint.y() + perpendicular_y)
|
|
|
|
|
|
class GraphComponentItem(QGraphicsPixmapItem):
|
|
def __init__(self, component_id: ComponentID, pixmap, editor: GraphEditorWidget) -> None:
|
|
super().__init__(pixmap)
|
|
self.component_id = component_id
|
|
self.editor = editor
|
|
self._drag_start = QPointF()
|
|
self._dragging = False
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsSelectable)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsMovable, editor.mode is GraphEditorMode.NORMAL)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemSendsGeometryChanges)
|
|
|
|
def itemChange(self, change: QGraphicsItem.GraphicsItemChange, value: object) -> object:
|
|
if change == QGraphicsItem.GraphicsItemChange.ItemPositionChange and self._dragging and isinstance(value, QPointF):
|
|
grid_size = self.editor.snap_to_grid_size
|
|
value = QPointF(round(value.x() / grid_size) * grid_size, round(value.y() / grid_size) * grid_size)
|
|
result = super().itemChange(change, value)
|
|
if change == QGraphicsItem.GraphicsItemChange.ItemPositionHasChanged:
|
|
self.editor.refresh_connections()
|
|
return result
|
|
|
|
def mousePressEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
if event.button() == Qt.MouseButton.LeftButton and self.editor.mode is GraphEditorMode.CONNECTION:
|
|
event.accept()
|
|
return
|
|
self._drag_start = QPointF(self.pos())
|
|
self._dragging = True
|
|
super().mousePressEvent(event)
|
|
|
|
def mouseReleaseEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
if self.editor.mode is GraphEditorMode.CONNECTION:
|
|
if event.button() == Qt.MouseButton.LeftButton:
|
|
self.editor.choose_connection_component(self.component_id, event.screenPos())
|
|
event.accept()
|
|
return
|
|
super().mouseReleaseEvent(event)
|
|
self._dragging = False
|
|
position = (round(self.pos().x()), round(self.pos().y()))
|
|
self.setPos(*position)
|
|
if self.pos() != self._drag_start:
|
|
self.editor.finish_component_move(self.component_id, position)
|
|
|
|
def mouseDoubleClickEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
if event.button() == Qt.MouseButton.LeftButton:
|
|
self.editor.open_component(self.component_id)
|
|
event.accept()
|
|
return
|
|
super().mouseDoubleClickEvent(event)
|
|
|
|
def contextMenuEvent(self, event) -> None:
|
|
if not self.isSelected():
|
|
self.scene().clearSelection()
|
|
self.setSelected(True)
|
|
self.editor.component_context_menu_requested.emit(self.component_id, event.screenPos())
|
|
event.accept()
|
|
|
|
|
|
class GraphComponentLabelItem(QGraphicsTextItem):
|
|
def __init__(self, component_id: ComponentID, text: str, label: GraphComponentLabel, component_item: GraphComponentItem, editor: GraphEditorWidget) -> None:
|
|
super().__init__(text, component_item)
|
|
self.component_id = component_id
|
|
self.editor = editor
|
|
self._dragging = False
|
|
self._drag_start = label.relative_position
|
|
font = self.font()
|
|
font.setItalic(True)
|
|
font.setPointSizeF(COMPONENT_LABEL_FONT_SIZE)
|
|
self.setFont(font)
|
|
self.setDefaultTextColor(QColor("#202020"))
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsSelectable)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsMovable, editor.mode is GraphEditorMode.NORMAL)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemSendsGeometryChanges)
|
|
self.set_relative_position(label.relative_position)
|
|
|
|
def relative_position(self) -> tuple[int, int]:
|
|
component_bounds = self.editor._component_bounds[self.component_id]
|
|
return round(self.pos().x() + self.boundingRect().width() / 2), round(self.pos().y() - component_bounds.bottom())
|
|
|
|
def set_relative_position(self, position: tuple[int, int]) -> None:
|
|
component_bounds = self.editor._component_bounds[self.component_id]
|
|
self.setPos(position[0] - self.boundingRect().width() / 2, component_bounds.bottom() + position[1])
|
|
|
|
def itemChange(self, change: QGraphicsItem.GraphicsItemChange, value: object) -> object:
|
|
if change == QGraphicsItem.GraphicsItemChange.ItemPositionChange and self._dragging and isinstance(value, QPointF):
|
|
component_bounds = self.editor._component_bounds[self.component_id]
|
|
size = self.editor.snap_to_grid_size
|
|
relative_x = value.x() + self.boundingRect().width() / 2
|
|
relative_y = value.y() - component_bounds.bottom()
|
|
value = QPointF(round(relative_x / size) * size - self.boundingRect().width() / 2, component_bounds.bottom() + round(relative_y / size) * size)
|
|
return super().itemChange(change, value)
|
|
|
|
def mousePressEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
self._drag_start = self.relative_position()
|
|
self._dragging = True
|
|
super().mousePressEvent(event)
|
|
|
|
def mouseReleaseEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
super().mouseReleaseEvent(event)
|
|
self._dragging = False
|
|
position = self.relative_position()
|
|
self.set_relative_position(position)
|
|
if position != self._drag_start:
|
|
self.editor.finish_component_label_move(self.component_id, position)
|
|
|
|
def contextMenuEvent(self, event) -> None:
|
|
self.editor.component_context_menu_requested.emit(self.component_id, event.screenPos())
|
|
event.accept()
|
|
|
|
|
|
class GraphConnectionPointItem(QGraphicsEllipseItem):
|
|
def __init__(self, connection_id: ConnectionID, index: int, position: tuple[int, int], editor: GraphEditorWidget) -> None:
|
|
radius = CONNECTION_WIDTH
|
|
super().__init__(-radius, -radius, radius * 2, radius * 2)
|
|
self.connection_id = connection_id
|
|
self.index = index
|
|
self.editor = editor
|
|
self._drag_start = QPointF()
|
|
self._dragging = False
|
|
self.setPos(*position)
|
|
self.setPen(QPen(Qt.PenStyle.NoPen))
|
|
self.setBrush(QBrush(QColor("#202020")))
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsMovable)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemIsSelectable)
|
|
self.setFlag(QGraphicsItem.GraphicsItemFlag.ItemSendsGeometryChanges)
|
|
|
|
def itemChange(self, change: QGraphicsItem.GraphicsItemChange, value: object) -> object:
|
|
if change == QGraphicsItem.GraphicsItemChange.ItemPositionChange and self._dragging and isinstance(value, QPointF):
|
|
size = self.editor.snap_to_grid_size
|
|
value = QPointF(round(value.x() / size) * size, round(value.y() / size) * size)
|
|
result = super().itemChange(change, value)
|
|
if change == QGraphicsItem.GraphicsItemChange.ItemPositionHasChanged:
|
|
self.editor.refresh_connections()
|
|
return result
|
|
|
|
def mousePressEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
self._drag_start = QPointF(self.pos())
|
|
self._dragging = True
|
|
super().mousePressEvent(event)
|
|
|
|
def mouseReleaseEvent(self, event: QGraphicsSceneMouseEvent) -> None:
|
|
super().mouseReleaseEvent(event)
|
|
self._dragging = False
|
|
position = (round(self.pos().x()), round(self.pos().y()))
|
|
self.setPos(*position)
|
|
if self.pos() != self._drag_start:
|
|
self.editor.finish_connection_point_move(self.connection_id)
|
|
|
|
def contextMenuEvent(self, event) -> None:
|
|
menu = QMenu(self.editor)
|
|
delete_point = menu.addAction("Delete Point")
|
|
if menu.exec(event.screenPos()) is delete_point:
|
|
self.editor.delete_connection_point(self.connection_id, self.index)
|
|
event.accept()
|
|
|
|
|
|
class GraphEditorWidget(QWidget):
|
|
component_move_requested = Signal(object, object, object)
|
|
component_context_menu_requested = Signal(object, object)
|
|
component_open_requested = Signal(object)
|
|
component_label_move_requested = Signal(object, object, object)
|
|
connection_points_change_requested = Signal(object, object, object, str)
|
|
connection_add_requested = Signal(object, object)
|
|
connections_delete_requested = Signal(object, object)
|
|
component_drop_requested = Signal(object, object, object)
|
|
|
|
def __init__(self, parent: QWidget | None = None, snap_to_grid_size: int = 4) -> None:
|
|
super().__init__(parent)
|
|
|
|
self.ui = Ui_graphEditorWidget()
|
|
self.ui.setupUi(self)
|
|
self._component: Component | None = None
|
|
self._graph = Graph()
|
|
self._icons: dict[ComponentID, Icon] = {}
|
|
self._mode = GraphEditorMode.NORMAL
|
|
self._connection_start: ComponentID | None = None
|
|
self._connection_preview: QGraphicsPathItem | None = None
|
|
self._component_items: dict[ComponentID, GraphComponentItem] = {}
|
|
self._component_bounds: dict[ComponentID, QRectF] = {}
|
|
self._component_label_items: dict[ComponentID, GraphComponentLabelItem] = {}
|
|
self._connection_items: dict[ConnectionID, GraphConnectionItem] = {}
|
|
self._connection_point_items: dict[ConnectionID, list[GraphConnectionPointItem]] = {}
|
|
self.set_snap_to_grid_size(snap_to_grid_size)
|
|
self.scene = GraphGraphicsScene(self)
|
|
self.scene.setSceneRect(-SCENE_SIZE / 2, -SCENE_SIZE / 2, SCENE_SIZE, SCENE_SIZE)
|
|
self.ui.graphicsView.setScene(self.scene)
|
|
self.ui.graphicsView.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
self.ui.graphicsView.setTransformationAnchor(QGraphicsView.ViewportAnchor.AnchorUnderMouse)
|
|
self.ui.graphicsView.viewport().installEventFilter(self)
|
|
self.ui.graphicsView.viewport().setAcceptDrops(True)
|
|
self._mode_actions = QActionGroup(self)
|
|
self._mode_actions.setExclusive(True)
|
|
self._mode_actions.addAction(self.ui.actionMouseMode)
|
|
self._mode_actions.addAction(self.ui.actionConnectionMode)
|
|
self.ui.actionMouseMode.triggered.connect(lambda: self.set_mode(GraphEditorMode.NORMAL))
|
|
self.ui.actionConnectionMode.triggered.connect(lambda: self.set_mode(GraphEditorMode.CONNECTION))
|
|
self.ui.actionMouseMode.setChecked(True)
|
|
self._mode_shortcut = QShortcut(QKeySequence(Qt.Key.Key_Space), self)
|
|
self._mode_shortcut.setContext(Qt.ShortcutContext.WidgetWithChildrenShortcut)
|
|
self._mode_shortcut.activated.connect(self.toggle_mode)
|
|
self.ui.actionZoomToFit.triggered.connect(self.zoom_to_fit)
|
|
self.ui.graphicsView.viewport().setMouseTracking(True)
|
|
self.ui.graphicsView.centerOn(0, 0)
|
|
|
|
@property
|
|
def snap_to_grid_size(self) -> int:
|
|
return self._snap_to_grid_size
|
|
|
|
def set_snap_to_grid_size(self, size: int) -> None:
|
|
if size < 1:
|
|
raise ValueError("snap-to-grid size must be positive")
|
|
self._snap_to_grid_size = size
|
|
|
|
@property
|
|
def mode(self) -> GraphEditorMode:
|
|
return self._mode
|
|
|
|
def set_mode(self, mode: GraphEditorMode) -> None:
|
|
if mode is self._mode:
|
|
return
|
|
self._mode = mode
|
|
self.ui.actionMouseMode.setChecked(mode is GraphEditorMode.NORMAL)
|
|
self.ui.actionConnectionMode.setChecked(mode is GraphEditorMode.CONNECTION)
|
|
self._clear_connection_start()
|
|
component = self._component
|
|
if component is not None:
|
|
self.set_component(component, self._graph, self._icons)
|
|
|
|
def toggle_mode(self) -> None:
|
|
self.set_mode(GraphEditorMode.CONNECTION if self._mode is GraphEditorMode.NORMAL else GraphEditorMode.NORMAL)
|
|
|
|
def set_component(self, component: Component | None, graph: Graph | None = None, icons: dict[ComponentID, Icon] | None = None) -> None:
|
|
if component is not None and not isinstance(component.implementation, GraphImplementation):
|
|
raise TypeError("GraphEditorWidget only supports components with a graph implementation")
|
|
component_changed = component is not self._component
|
|
self._clear_connection_start()
|
|
self._component = component
|
|
self._graph = graph or Graph()
|
|
self._icons = icons or {}
|
|
self._component_items = {}
|
|
self._component_bounds = {}
|
|
self._component_label_items = {}
|
|
self._connection_items = {}
|
|
self._connection_point_items = {}
|
|
self.scene.clear()
|
|
if component is None:
|
|
return
|
|
|
|
graph = self._graph
|
|
icons = self._icons
|
|
positions = {component_id: graph.component_positions.get(component_id, (index * FALLBACK_COMPONENT_SPACING, 0)) for index, component_id in enumerate(component.implementation.graph.components)}
|
|
for component_id, child in component.implementation.graph.components.items():
|
|
icon = icons.get(component_id, Icon())
|
|
icon_size = get_natural_icon_size(icon)
|
|
pixmap = render_icon(icon, child.interface.ports, icon_size, render_ports=self._mode is GraphEditorMode.CONNECTION).pixmap(icon_size)
|
|
item = GraphComponentItem(component_id, pixmap, self)
|
|
item.setOffset(-pixmap.width() / 2, -pixmap.height() / 2)
|
|
item.setPos(*positions[component_id])
|
|
item.setToolTip(child.name)
|
|
bounds = get_pixmap_bounding_box(pixmap)
|
|
self._component_items[component_id] = item
|
|
self._component_bounds[component_id] = bounds.translated(-pixmap.width() / 2, -pixmap.height() / 2)
|
|
self.scene.addItem(item)
|
|
|
|
for component_id, child in component.implementation.graph.components.items():
|
|
label = graph.component_labels.get(component_id, GraphComponentLabel())
|
|
if label.visible:
|
|
self._create_component_label_item(component_id, child.name, label)
|
|
|
|
for connection_id, connection in component.implementation.graph.connections.items():
|
|
if not isinstance(connection, (SignalConnection, BondConnection)):
|
|
continue
|
|
tick_at_source = None
|
|
if isinstance(connection, BondConnection):
|
|
if connection.causality is BondCausality.EFFORT_OUT:
|
|
tick_at_source = False
|
|
elif connection.causality is BondCausality.FLOW_OUT:
|
|
tick_at_source = True
|
|
connection_item = GraphConnectionItem([(0, 0), (1, 0)], half_arrow=isinstance(connection, BondConnection), tick_at_source=tick_at_source, connection_id=connection_id, editor=self)
|
|
connection_item.setData(0, str(connection_id))
|
|
self._connection_items[connection_id] = connection_item
|
|
self.scene.addItem(connection_item)
|
|
visual_connection = graph.connections.get(connection_id)
|
|
self._create_connection_point_items(connection_id, visual_connection.points[1:-1] if visual_connection is not None and len(visual_connection.points) >= 2 else [])
|
|
self.refresh_connections()
|
|
if component_changed:
|
|
QTimer.singleShot(0, self.ui.actionZoomToFit.trigger)
|
|
|
|
def refresh_connections(self) -> None:
|
|
component = self._component
|
|
if component is None or not self._connection_items:
|
|
return
|
|
port_owners = {port_id: component_id for component_id, child in component.implementation.graph.components.items() for port_id in child.interface.ports}
|
|
for connection_id, item in self._connection_items.items():
|
|
connection = component.implementation.graph.connections.get(connection_id)
|
|
if connection is None:
|
|
continue
|
|
source_component = port_owners.get(connection.source)
|
|
target_component = port_owners.get(connection.target)
|
|
if source_component is None or target_component is None:
|
|
continue
|
|
source_position = self._item_position(source_component)
|
|
target_position = self._item_position(target_component)
|
|
point_items = self._connection_point_items.get(connection_id, [])
|
|
points = [source_position, *((point.pos().x(), point.pos().y()) for point in point_items), target_position]
|
|
points = self._straighten_direct_connection(points)
|
|
source_bounds = self._component_bounds[source_component].translated(*source_position)
|
|
target_bounds = self._component_bounds[target_component].translated(*target_position)
|
|
spacing = BOND_CONNECTION_BOUNDING_BOX_SPACING if isinstance(connection, BondConnection) else SIGNAL_CONNECTION_BOUNDING_BOX_SPACING
|
|
points = self._clip_connection(points, source_bounds, target_bounds, spacing)
|
|
tick_at_source = None
|
|
if isinstance(connection, BondConnection):
|
|
tick_at_source = False if connection.causality is BondCausality.EFFORT_OUT else True if connection.causality is BondCausality.FLOW_OUT else None
|
|
item.setPath(item._connection_path(points, isinstance(connection, BondConnection), tick_at_source))
|
|
|
|
def add_connection_point(self, connection_id: ConnectionID, scene_position: QPointF) -> None:
|
|
points = self._connection_metadata_points(connection_id)
|
|
position = self._snap_position(scene_position)
|
|
index = self._nearest_segment_index(points, position) + 1
|
|
points.insert(index, position)
|
|
self._request_connection_points_change(connection_id, points, "Add connection point")
|
|
|
|
def delete_connection_point(self, connection_id: ConnectionID, index: int) -> None:
|
|
points = self._connection_metadata_points(connection_id)
|
|
if 0 < index < len(points) - 1:
|
|
points.pop(index)
|
|
self._request_connection_points_change(connection_id, points, "Delete connection point")
|
|
|
|
def finish_connection_point_move(self, connection_id: ConnectionID) -> None:
|
|
self._request_connection_points_change(connection_id, self._connection_metadata_points(connection_id), "Move connection point")
|
|
|
|
def set_connection_points(self, connection_id: ConnectionID, points: list[tuple[int, int]] | None) -> None:
|
|
if points is None:
|
|
self._graph.connections.pop(connection_id, None)
|
|
for item in self._connection_point_items.pop(connection_id, []):
|
|
self.scene.removeItem(item)
|
|
connection_item = self._connection_items.pop(connection_id, None)
|
|
if connection_item is not None:
|
|
self.scene.removeItem(connection_item)
|
|
return
|
|
self._graph.connections[connection_id] = GraphConnection(points=list(points))
|
|
interior_points = points[1:-1]
|
|
items = self._connection_point_items.get(connection_id, [])
|
|
if len(items) == len(interior_points):
|
|
for item, position in zip(items, interior_points):
|
|
item.setPos(*position)
|
|
else:
|
|
for item in items:
|
|
self.scene.removeItem(item)
|
|
self._create_connection_point_items(connection_id, interior_points)
|
|
self.refresh_connections()
|
|
|
|
def _create_connection_point_items(self, connection_id: ConnectionID, positions: list[tuple[int, int]]) -> None:
|
|
items = [GraphConnectionPointItem(connection_id, index, position, self) for index, position in enumerate(positions, 1)]
|
|
self._connection_point_items[connection_id] = items
|
|
for item in items:
|
|
self.scene.addItem(item)
|
|
|
|
def _connection_metadata_points(self, connection_id: ConnectionID) -> list[tuple[int, int]]:
|
|
component = self._component
|
|
if component is None:
|
|
return []
|
|
connection = component.implementation.graph.connections[connection_id]
|
|
port_owners = {port_id: component_id for component_id, child in component.implementation.graph.components.items() for port_id in child.interface.ports}
|
|
source = self._item_position(port_owners[connection.source])
|
|
target = self._item_position(port_owners[connection.target])
|
|
interior = [(round(item.pos().x()), round(item.pos().y())) for item in self._connection_point_items.get(connection_id, [])]
|
|
return [(round(source[0]), round(source[1])), *interior, (round(target[0]), round(target[1]))]
|
|
|
|
def _request_connection_points_change(self, connection_id: ConnectionID, points: list[tuple[int, int]], text: str) -> None:
|
|
if self._component is not None:
|
|
self.connection_points_change_requested.emit(self._component, connection_id, points, text)
|
|
|
|
def _snap_position(self, position: QPointF) -> tuple[int, int]:
|
|
size = self.snap_to_grid_size
|
|
return round(position.x() / size) * size, round(position.y() / size) * size
|
|
|
|
@staticmethod
|
|
def _nearest_segment_index(points: list[tuple[int, int]], position: tuple[int, int]) -> int:
|
|
best_index = 0
|
|
best_distance = float("inf")
|
|
for index, (start, end) in enumerate(pairwise(points)):
|
|
dx = end[0] - start[0]
|
|
dy = end[1] - start[1]
|
|
length_squared = dx * dx + dy * dy
|
|
ratio = 0.0 if not length_squared else max(0.0, min(1.0, ((position[0] - start[0]) * dx + (position[1] - start[1]) * dy) / length_squared))
|
|
closest_x = start[0] + ratio * dx
|
|
closest_y = start[1] + ratio * dy
|
|
distance = (position[0] - closest_x) ** 2 + (position[1] - closest_y) ** 2
|
|
if distance < best_distance:
|
|
best_index = index
|
|
best_distance = distance
|
|
return best_index
|
|
|
|
def finish_component_move(self, component_id: ComponentID, position: tuple[int, int]) -> None:
|
|
if self._component is not None:
|
|
self.component_move_requested.emit(self._component, component_id, position)
|
|
|
|
def finish_component_label_move(self, component_id: ComponentID, relative_position: tuple[int, int]) -> None:
|
|
if self._component is not None:
|
|
self.component_label_move_requested.emit(self._component, component_id, relative_position)
|
|
|
|
def component_label_visible(self, component_id: ComponentID) -> bool:
|
|
return self._graph.component_labels.get(component_id, GraphComponentLabel()).visible
|
|
|
|
def set_component_label(self, component_id: ComponentID, label: GraphComponentLabel | None) -> None:
|
|
if label is None:
|
|
self._graph.component_labels.pop(component_id, None)
|
|
label = GraphComponentLabel()
|
|
else:
|
|
self._graph.component_labels[component_id] = label
|
|
item = self._component_label_items.pop(component_id, None)
|
|
if item is not None:
|
|
item.setParentItem(None)
|
|
self.scene.removeItem(item)
|
|
component = self._component
|
|
if component is not None and label.visible and component_id in component.implementation.graph.components:
|
|
self._create_component_label_item(component_id, component.implementation.graph.components[component_id].name, label)
|
|
|
|
def _create_component_label_item(self, component_id: ComponentID, text: str, label: GraphComponentLabel) -> None:
|
|
item = GraphComponentLabelItem(component_id, text, label, self._component_items[component_id], self)
|
|
self._component_label_items[component_id] = item
|
|
|
|
def set_component_position(self, component_id: ComponentID, position: tuple[int, int] | None) -> None:
|
|
item = self._component_items.get(component_id)
|
|
if item is None:
|
|
return
|
|
if position is None:
|
|
self._graph.component_positions.pop(component_id, None)
|
|
index = list(self._component_items).index(component_id)
|
|
position = (index * FALLBACK_COMPONENT_SPACING, 0)
|
|
else:
|
|
self._graph.component_positions[component_id] = position
|
|
item.setPos(*position)
|
|
self.refresh_connections()
|
|
|
|
def _item_position(self, component_id: ComponentID) -> tuple[float, float]:
|
|
position = self._component_items[component_id].pos()
|
|
return position.x(), position.y()
|
|
|
|
@staticmethod
|
|
def _straighten_direct_connection(points: list[tuple[float, float]]) -> list[tuple[float, float]]:
|
|
if len(points) != 2:
|
|
return points
|
|
source, target = points
|
|
dx = target[0] - source[0]
|
|
dy = target[1] - source[1]
|
|
if abs(dx) <= CONNECTION_STRAIGHTEN_TOLERANCE and abs(dx) < abs(dy):
|
|
x = (source[0] + target[0]) / 2
|
|
return [(x, source[1]), (x, target[1])]
|
|
if abs(dy) <= CONNECTION_STRAIGHTEN_TOLERANCE and abs(dy) < abs(dx):
|
|
y = (source[1] + target[1]) / 2
|
|
return [(source[0], y), (target[0], y)]
|
|
return points
|
|
|
|
@staticmethod
|
|
def _clip_connection(points: list[tuple[float, float]], source_bounds: QRectF, target_bounds: QRectF, spacing: float) -> list[tuple[float, float]]:
|
|
source = points[0]
|
|
target = points[-1]
|
|
source_direction = next((point for point in points[1:] if point != source), None)
|
|
target_direction = next((point for point in reversed(points[:-1]) if point != target), None)
|
|
if source_direction is None or target_direction is None:
|
|
return points
|
|
clipped = list(points)
|
|
clipped[0] = GraphEditorWidget._bounding_box_edge(source_bounds, source, source_direction, spacing)
|
|
clipped[-1] = GraphEditorWidget._bounding_box_edge(target_bounds, target, target_direction, spacing)
|
|
return clipped
|
|
|
|
@staticmethod
|
|
def _bounding_box_edge(bounds: QRectF, origin: tuple[float, float], toward: tuple[float, float], spacing: float) -> tuple[float, float]:
|
|
bounds = bounds.adjusted(-spacing, -spacing, spacing, spacing)
|
|
dx = toward[0] - origin[0]
|
|
dy = toward[1] - origin[1]
|
|
horizontal_scale = (bounds.right() - origin[0]) / dx if dx > 0 else (bounds.left() - origin[0]) / dx if dx < 0 else float("inf")
|
|
vertical_scale = (bounds.bottom() - origin[1]) / dy if dy > 0 else (bounds.top() - origin[1]) / dy if dy < 0 else float("inf")
|
|
scale = min(horizontal_scale, vertical_scale)
|
|
return origin[0] + dx * scale, origin[1] + dy * scale
|
|
|
|
def component(self) -> Component | None:
|
|
return self._component
|
|
|
|
def selected_component_ids(self) -> list[ComponentID]:
|
|
return [item.component_id for item in self.scene.selectedItems() if isinstance(item, GraphComponentItem)]
|
|
|
|
def selected_connection_ids(self) -> list[ConnectionID]:
|
|
return [item.connection_id for item in self.scene.selectedItems() if isinstance(item, GraphConnectionItem) and item.connection_id is not None]
|
|
|
|
def delete_selected_connections(self) -> None:
|
|
if self._component is not None:
|
|
self.connections_delete_requested.emit(self._component, self.selected_connection_ids())
|
|
|
|
def open_component(self, component_id: ComponentID) -> None:
|
|
self._clear_connection_start()
|
|
self.component_open_requested.emit(component_id)
|
|
|
|
def choose_connection_component(self, component_id: ComponentID, screen_position) -> None:
|
|
if self._component is None or self._mode is not GraphEditorMode.CONNECTION:
|
|
return
|
|
if self._connection_start is None:
|
|
self.scene.clearSelection()
|
|
self._component_items[component_id].setSelected(True)
|
|
self._connection_start = component_id
|
|
self._connection_preview = QGraphicsPathItem()
|
|
self._connection_preview.setPen(QPen(QColor("#606060"), CONNECTION_WIDTH, Qt.PenStyle.DashLine))
|
|
self._connection_preview.setZValue(-0.5)
|
|
self.scene.addItem(self._connection_preview)
|
|
return
|
|
start = self._connection_start
|
|
self._clear_connection_start()
|
|
self.scene.clearSelection()
|
|
if start == component_id:
|
|
return
|
|
options = self._connection_options(start, component_id)
|
|
if len(options) == 1:
|
|
self.connection_add_requested.emit(self._component, options[0][1])
|
|
return
|
|
if not options:
|
|
return
|
|
menu = QMenu(self)
|
|
actions = []
|
|
for label, connection, _preferred in options:
|
|
action = menu.addAction(label)
|
|
actions.append((action, connection))
|
|
menu.setActiveAction(actions[0][0])
|
|
selected = menu.exec(screen_position)
|
|
for action, connection in actions:
|
|
if selected is action:
|
|
self.connection_add_requested.emit(self._component, connection)
|
|
break
|
|
|
|
def _update_connection_preview(self, mouse_position: QPointF) -> None:
|
|
if self._connection_start is None or self._connection_preview is None:
|
|
return
|
|
source = self._item_position(self._connection_start)
|
|
if mouse_position == QPointF(*source):
|
|
self._connection_preview.setPath(QPainterPath(mouse_position))
|
|
return
|
|
bounds = self._component_bounds[self._connection_start].translated(*source)
|
|
spacing = max(BOND_CONNECTION_BOUNDING_BOX_SPACING, SIGNAL_CONNECTION_BOUNDING_BOX_SPACING)
|
|
start = self._bounding_box_edge(bounds, source, (mouse_position.x(), mouse_position.y()), spacing)
|
|
path = QPainterPath(QPointF(*start))
|
|
path.lineTo(mouse_position)
|
|
self._connection_preview.setPath(path)
|
|
|
|
def _clear_connection_start(self) -> None:
|
|
self._connection_start = None
|
|
if self._connection_preview is not None and self._connection_preview.scene() is self.scene:
|
|
self.scene.removeItem(self._connection_preview)
|
|
self._connection_preview = None
|
|
|
|
def _connection_options(self, first_id: ComponentID, second_id: ComponentID) -> list[tuple[str, Connection, bool]]:
|
|
if self._component is None:
|
|
return []
|
|
components = self._component.implementation.graph.components
|
|
first = components[first_id]
|
|
second = components[second_id]
|
|
used_ports = {port_id for connection in self._component.implementation.graph.connections.values() for port_id in (connection.source, connection.target)}
|
|
options = []
|
|
for first_port_id, first_port in first.interface.ports.items():
|
|
for second_port_id, second_port in second.interface.ports.items():
|
|
if not self._compatible_ports(first_port, second_port) or not self._port_available(first_port_id, first_port, used_ports) or not self._port_available(second_port_id, second_port, used_ports):
|
|
continue
|
|
preferred = first_port.direction is SignalDirection.OUTPUT and second_port.direction is SignalDirection.INPUT
|
|
reverse_preferred = second_port.direction is SignalDirection.OUTPUT and first_port.direction is SignalDirection.INPUT
|
|
if reverse_preferred:
|
|
source_id, source_port, source_name = second_port_id, second_port, second.name
|
|
target_id, target_port, target_name = first_port_id, first_port, first.name
|
|
else:
|
|
source_id, source_port, source_name = first_port_id, first_port, first.name
|
|
target_id, target_port, target_name = second_port_id, second_port, second.name
|
|
connection_type = SignalConnection if isinstance(source_port, SignalPort) else BondConnection
|
|
label = f"{source_name}.{source_port.name} → {target_name}.{target_port.name}"
|
|
options.append((label, connection_type(source=source_id, target=target_id), preferred or reverse_preferred))
|
|
return sorted(options, key=lambda option: not option[2])
|
|
|
|
@staticmethod
|
|
def _compatible_ports(first: Port, second: Port) -> bool:
|
|
if isinstance(first, SignalPort) and isinstance(second, SignalPort):
|
|
return first.direction is not second.direction
|
|
if isinstance(first, BondPort) and isinstance(second, BondPort):
|
|
return not first.domain or not second.domain or first.domain == second.domain
|
|
return False
|
|
|
|
@staticmethod
|
|
def _port_available(port_id: PortID, port: Port, used_ports: set[PortID]) -> bool:
|
|
if isinstance(port, SignalPort):
|
|
return port.direction is SignalDirection.OUTPUT or port.multiplicity or port_id not in used_ports
|
|
if isinstance(port, BondPort):
|
|
return port.multiplicity or port_id not in used_ports
|
|
return False
|
|
|
|
def paste_position(self) -> tuple[int, int]:
|
|
viewport = self.ui.graphicsView.viewport()
|
|
viewport_position = viewport.mapFromGlobal(QCursor.pos())
|
|
if not viewport.rect().contains(viewport_position):
|
|
viewport_position = viewport.rect().center()
|
|
scene_position = self.ui.graphicsView.mapToScene(viewport_position)
|
|
size = self.snap_to_grid_size
|
|
return round(scene_position.x() / size) * size, round(scene_position.y() / size) * size
|
|
|
|
def zoom_to_fit(self) -> None:
|
|
bounds = self.scene.itemsBoundingRect()
|
|
if bounds.isEmpty():
|
|
self.ui.graphicsView.resetTransform()
|
|
self.ui.graphicsView.centerOn(0, 0)
|
|
return
|
|
bounds.adjust(-ZOOM_TO_FIT_PADDING, -ZOOM_TO_FIT_PADDING, ZOOM_TO_FIT_PADDING, ZOOM_TO_FIT_PADDING)
|
|
self.ui.graphicsView.fitInView(bounds, Qt.AspectRatioMode.KeepAspectRatio)
|
|
current_zoom = self.ui.graphicsView.transform().m11()
|
|
target_zoom = min(MAX_ZOOM, max(MIN_ZOOM, current_zoom))
|
|
if target_zoom != current_zoom:
|
|
self.ui.graphicsView.scale(target_zoom / current_zoom, target_zoom / current_zoom)
|
|
self.ui.graphicsView.centerOn(bounds.center())
|
|
|
|
def eventFilter(self, watched: QObject, event: QEvent) -> bool:
|
|
if watched is self.ui.graphicsView.viewport() and event.type() in (QEvent.Type.DragEnter, QEvent.Type.DragMove):
|
|
assert isinstance(event, (QDragEnterEvent, QDragMoveEvent))
|
|
if self._component is not None and event.mimeData().hasFormat(COMPONENTS_MIME):
|
|
event.setDropAction(Qt.DropAction.CopyAction)
|
|
event.accept()
|
|
return True
|
|
if watched is self.ui.graphicsView.viewport() and event.type() == QEvent.Type.Drop:
|
|
assert isinstance(event, QDropEvent)
|
|
if self._component is not None and event.mimeData().hasFormat(COMPONENTS_MIME):
|
|
try:
|
|
payload = json.loads(bytes(event.mimeData().data(COMPONENTS_MIME)).decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
return True
|
|
if isinstance(payload, dict):
|
|
position = self._snap_position(self.ui.graphicsView.mapToScene(event.position().toPoint()))
|
|
self.component_drop_requested.emit(self._component, payload, position)
|
|
event.setDropAction(Qt.DropAction.CopyAction)
|
|
event.accept()
|
|
return True
|
|
if watched is self.ui.graphicsView.viewport() and event.type() == QEvent.Type.MouseMove:
|
|
assert isinstance(event, QMouseEvent)
|
|
self._update_connection_preview(self.ui.graphicsView.mapToScene(event.position().toPoint()))
|
|
if watched is self.ui.graphicsView.viewport() and event.type() == QEvent.Type.Wheel:
|
|
assert isinstance(event, QWheelEvent)
|
|
if event.modifiers() & Qt.KeyboardModifier.ControlModifier:
|
|
self._zoom(event)
|
|
return True
|
|
if event.modifiers() & Qt.KeyboardModifier.ShiftModifier:
|
|
self._scroll_horizontally(event)
|
|
return True
|
|
return super().eventFilter(watched, event)
|
|
|
|
def _zoom(self, event: QWheelEvent) -> None:
|
|
delta = event.angleDelta().y() or event.pixelDelta().y()
|
|
if not delta:
|
|
return
|
|
current_zoom = self.ui.graphicsView.transform().m11()
|
|
requested_zoom = current_zoom * ZOOM_STEP ** (delta / 120)
|
|
target_zoom = min(MAX_ZOOM, max(MIN_ZOOM, requested_zoom))
|
|
self.ui.graphicsView.scale(target_zoom / current_zoom, target_zoom / current_zoom)
|
|
|
|
def _scroll_horizontally(self, event: QWheelEvent) -> None:
|
|
scrollbar = self.ui.graphicsView.horizontalScrollBar()
|
|
pixel_delta = event.pixelDelta().y()
|
|
distance = pixel_delta if pixel_delta else event.angleDelta().y() / 120 * scrollbar.singleStep() * 3
|
|
scrollbar.setValue(scrollbar.value() - round(distance))
|