ilya/diagrams-opus-qt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(view): graphics scene, node/port/edge items, undoable controller

Browse Source 
Add the QGraphicsView layer: NodeItem (SVG symbol + title + draggable,
snap-to-grid), PortItem (relation-colored anchors with compatibility
hints), EdgeItem (orthogonal routing, arc hops on crossings, line styles,
head/tail decorations, animated flow dashes), and DiagramScene binding the
document to items with edge-drawing interaction and a grid background.
Add EditorController with a QUndoStack (add/connect/delete/move/edit) and
scene interaction tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilya 2026-07-02 23:20:09 +02:00
parent 58f3c891ee
commit af66debcde
7 changed files with 1020 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

192
pipeline_editor/controller.py Normal file
View File

@ -0,0 +1,192 @@
"""EditorController: high-level, undoable editing operations.
All user edits go through here so they are uniformly undoable and reusable from
both the UI and tests. The controller owns a ``QUndoStack``; the scene reacts to
the resulting document signals.
"""
from __future__ import annotations
from typing import Optional
from PyQt6.QtGui import QUndoCommand, QUndoStack
from .model.document import DiagramDocument
from .model.edge import EdgeModel
from .model.node import NodeModel
from .model.node_library import NodeLibrary
# -- commands -------------------------------------------------------------
class AddNodeCommand(QUndoCommand):
def __init__(self, doc: DiagramDocument, node: NodeModel):
super().__init__(f"Add {node.title or node.template_key}")
self._doc = doc
self._node = node
def redo(self):
if self._doc.node(self._node.node_id) is None:
self._doc.add_node(self._node)
def undo(self):
self._doc.remove_node(self._node.node_id)
class AddEdgeCommand(QUndoCommand):
def __init__(self, doc: DiagramDocument, edge: EdgeModel):
super().__init__("Connect")
self._doc = doc
self._edge = edge
def redo(self):
if self._doc.edge(self._edge.edge_id) is None:
self._doc.add_edge(self._edge)
def undo(self):
self._doc.remove_edge(self._edge.edge_id)
class DeleteCommand(QUndoCommand):
"""Delete nodes and/or edges, restoring them (and cascaded edges) on undo."""
def __init__(self, doc: DiagramDocument, node_ids, edge_ids):
super().__init__("Delete")
self._doc = doc
node_ids = set(node_ids)
# any edge touching a deleted node is implicitly deleted too
edge_ids = set(edge_ids)
for e in doc.edges():
if e.source_node in node_ids or e.target_node in node_ids:
edge_ids.add(e.edge_id)
self._nodes = [doc.node(nid).to_dict() for nid in node_ids if doc.node(nid)]
self._edges = [doc.edge(eid).to_dict() for eid in edge_ids if doc.edge(eid)]
def redo(self):
for e in self._edges:
self._doc.remove_edge(e["id"])
for n in self._nodes:
self._doc.remove_node(n["id"])
def undo(self):
for n in self._nodes:
if self._doc.node(n["id"]) is None:
self._doc.add_node(NodeModel.from_dict(n))
for e in self._edges:
if self._doc.edge(e["id"]) is None:
self._doc.add_edge(EdgeModel.from_dict(e))
class MoveNodeCommand(QUndoCommand):
def __init__(self, doc: DiagramDocument, node_id: str, old, new):
super().__init__("Move")
self._doc = doc
self._id = node_id
self._old = old
self._new = new
def _apply(self, pos):
node = self._doc.node(self._id)
if node is not None:
node.x, node.y = pos
self._doc.notify_node_changed(self._id)
def redo(self):
self._apply(self._new)
def undo(self):
self._apply(self._old)
def id(self): # enable merging of consecutive moves of the same node
return 0xC0FFEE
def mergeWith(self, other: "QUndoCommand") -> bool:
if isinstance(other, MoveNodeCommand) and other._id == self._id:
self._new = other._new
return True
return False
class SetPropertyCommand(QUndoCommand):
def __init__(self, doc: DiagramDocument, kind: str, owner_id: str,
key: str, old, new):
super().__init__(f"Edit {key}")
self._doc = doc
self._kind = kind
self._owner_id = owner_id
self._key = key
self._old = old
self._new = new
def _owner(self):
return (self._doc.node(self._owner_id) if self._kind == "node"
else self._doc.edge(self._owner_id))
def _apply(self, value):
owner = self._owner()
if owner is None:
return
owner.properties.set_value(self._key, value)
if self._kind == "node":
self._doc.notify_node_changed(self._owner_id)
else:
self._doc.notify_edge_changed(self._owner_id)
def redo(self):
self._apply(self._new)
def undo(self):
self._apply(self._old)
# -- controller -----------------------------------------------------------
class EditorController:
def __init__(self, document: DiagramDocument, library: NodeLibrary):
self.document = document
self.library = library
self.undo_stack = QUndoStack()
def add_node(self, template_key: str, x: float, y: float) -> Optional[NodeModel]:
tmpl = self.library.get(template_key)
if tmpl is None:
return None
node = tmpl.instantiate(self.document.next_id("n"), x, y)
self.undo_stack.push(AddNodeCommand(self.document, node))
return node
def connect(self, s_node, s_port, t_node, t_port) -> Optional[EdgeModel]:
if not self.document.can_connect(s_node, s_port, t_node, t_port):
return None
edge = EdgeModel(self.document.next_id("e"), s_node, s_port, t_node, t_port)
self.undo_stack.push(AddEdgeCommand(self.document, edge))
return edge
def delete(self, node_ids=(), edge_ids=()) -> None:
node_ids = list(node_ids)
edge_ids = list(edge_ids)
if not node_ids and not edge_ids:
return
self.undo_stack.push(DeleteCommand(self.document, node_ids, edge_ids))
def record_move(self, node_id: str, old, new) -> None:
if old == new:
return
self.undo_stack.push(MoveNodeCommand(self.document, node_id, old, new))
def set_property(self, kind: str, owner_id: str, key: str, value) -> None:
owner = (self.document.node(owner_id) if kind == "node"
else self.document.edge(owner_id))
if owner is None:
return
prop = owner.properties.find(key)
if prop is None:
return
old = prop.value
if old == value:
return
self.undo_stack.push(SetPropertyCommand(self.document, kind, owner_id, key, old, value))
# convenience
def undo(self):
self.undo_stack.undo()
def redo(self):
self.undo_stack.redo()

223
pipeline_editor/view/edge_item.py Normal file
View File

@ -0,0 +1,223 @@
"""EdgeItem: orthogonal connector with styles, arc hops and flow animation."""
from __future__ import annotations
import math
from PyQt6.QtCore import QLineF, QPointF, QRectF, Qt
from PyQt6.QtGui import QBrush, QColor, QPainterPath, QPen, QPolygonF
from PyQt6.QtWidgets import QGraphicsObject, QGraphicsItem, QStyle
from ..model.edge import EdgeModel, EndpointDecoration, LineStyle
from . import routing
_PEN_STYLES = {
LineStyle.SOLID: Qt.PenStyle.SolidLine,
LineStyle.DASHED: Qt.PenStyle.DashLine,
LineStyle.DOTTED: Qt.PenStyle.DotLine,
LineStyle.DASH_DOT: Qt.PenStyle.DashDotLine,
}
class EdgeItem(QGraphicsObject):
"""Graphics item bound to an :class:`EdgeModel`."""
def __init__(self, scene_ref, edge: EdgeModel) -> None:
super().__init__()
self._scene = scene_ref
self.edge = edge
self.setFlags(QGraphicsItem.GraphicsItemFlag.ItemIsSelectable)
self.setAcceptHoverEvents(True)
self.setZValue(0)
self._points: list[routing.Point] = []
self._path = QPainterPath()
self._anim_phase = 0.0
self.reroute()
# -- data -------------------------------------------------------------
def relation_color(self) -> QColor:
if self.edge.style.color:
return QColor(self.edge.style.color)
doc = getattr(self._scene, "document", None)
if doc is not None and self.edge.relation:
return QColor(doc.relations.color(self.edge.relation))
return QColor("#607d8b")
def endpoints_scene(self):
"""Return ((sx,sy),(snx,sny)),((ex,ey),(enx,eny)) in scene coords."""
return self._scene.edge_endpoints(self.edge)
# -- routing ----------------------------------------------------------
def reroute(self) -> None:
data = self.endpoints_scene()
if data is None:
return
(start, snorm), (end, enorm) = data
cfg = getattr(self._scene, "config", None)
grid = cfg.grid_size if cfg else 20
stub = cfg.route_stub if cfg else 20
self.prepareGeometryChange()
self._points = routing.orthogonal_route(
start, snorm, end, enorm, grid=grid, stub=stub)
self._rebuild_path()
self.update()
def _crossings(self) -> list[routing.Point]:
cfg = getattr(self._scene, "config", None)
if not (cfg and cfg.arc_on_crossing):
return []
my_segs = routing.polyline_segments(self._points)
others = self._scene.other_edge_segments(self.edge.edge_id)
return routing.segment_crossings(my_segs, others)
def _rebuild_path(self) -> None:
pts = [QPointF(*p) for p in self._points]
path = QPainterPath()
if len(pts) < 2:
self._path = path
return
cfg = getattr(self._scene, "config", None)
radius = cfg.arc_radius if cfg else 5.0
crossings = self._crossings()
path.moveTo(pts[0])
for i in range(len(pts) - 1):
a, b = pts[i], pts[i + 1]
hops = self._hops_on_segment(a, b, crossings, radius)
for cx, cy in hops:
self._draw_hop(path, a, b, QPointF(cx, cy), radius)
path.lineTo(b)
self._path = path
@staticmethod
def _hops_on_segment(a: QPointF, b: QPointF, crossings, radius):
"""Crossings lying on horizontal segment a->b, ordered along it."""
eps = 1e-6
if abs(a.y() - b.y()) > eps: # only horizontal segments hop
return []
y = a.y()
lo, hi = sorted((a.x(), b.x()))
on = [(cx, cy) for (cx, cy) in crossings
if abs(cy - y) < 1e-3 and lo + radius < cx < hi - radius]
on.sort(key=lambda c: c[0] if b.x() >= a.x() else -c[0])
return on
@staticmethod
def _draw_hop(path: QPainterPath, a: QPointF, b: QPointF, c: QPointF, radius: float):
sign = 1.0 if b.x() >= a.x() else -1.0
path.lineTo(c.x() - sign * radius, c.y())
rect = QRectF(c.x() - radius, c.y() - radius, 2 * radius, 2 * radius)
# semicircular hop above the line
start_angle = 180 if sign > 0 else 0
sweep = -180 if sign > 0 else 180
path.arcTo(rect, start_angle, sweep)
# -- geometry for the scene ------------------------------------------
def path_points(self) -> list[routing.Point]:
return list(self._points)
def boundingRect(self) -> QRectF:
if self._path.isEmpty():
return QRectF()
extra = self.edge.style.width + 12
return self._path.boundingRect().adjusted(-extra, -extra, extra, extra)
def shape(self) -> QPainterPath:
stroker = QPen()
from PyQt6.QtGui import QPainterPathStroker
s = QPainterPathStroker()
s.setWidth(max(10.0, self.edge.style.width + 8))
return s.createStroke(self._path)
# -- animation --------------------------------------------------------
def set_phase(self, phase: float) -> None:
self._anim_phase = phase
if self.edge.flow_direction != 0:
self.update()
# -- painting ---------------------------------------------------------
def paint(self, painter, option, widget=None):
if self._path.isEmpty():
return
painter.setRenderHint(painter.RenderHint.Antialiasing, True)
color = self.relation_color()
style = self.edge.style
selected = bool(option.state & QStyle.StateFlag.State_Selected)
# base line
pen = QPen(color, style.width, _PEN_STYLES.get(style.line_style, Qt.PenStyle.SolidLine))
pen.setCapStyle(Qt.PenCapStyle.RoundCap)
pen.setJoinStyle(Qt.PenJoinStyle.RoundJoin)
if selected:
halo = QPen(QColor(25, 118, 210, 120), style.width + 6)
halo.setCapStyle(Qt.PenCapStyle.RoundCap)
painter.setPen(halo)
painter.setBrush(Qt.BrushStyle.NoBrush)
painter.drawPath(self._path)
painter.setPen(pen)
painter.setBrush(Qt.BrushStyle.NoBrush)
painter.drawPath(self._path)
# flow animation: marching dashes overlaid, moving with flow direction
if self.edge.flow_direction != 0:
self._paint_flow(painter, color)
# endpoint decorations
if len(self._points) >= 2:
self._paint_decoration(painter, color, style.tail,
self._points[1], self._points[0])
self._paint_decoration(painter, color, style.head,
self._points[-2], self._points[-1])
def _paint_flow(self, painter, color: QColor):
width = max(2.0, min(8.0, 1.5 + self.edge.flow / 25.0))
dash = QPen(color.lighter(150), width, Qt.PenStyle.CustomDashLine)
dash.setCapStyle(Qt.PenCapStyle.RoundCap)
dash.setDashPattern([2.0, 4.0])
offset = self._anim_phase * (2.0 + self.edge.flow / 20.0)
# dashes travel source->target for +1, reverse for -1
dash.setDashOffset(-offset if self.edge.flow_direction > 0 else offset)
painter.setPen(dash)
painter.setBrush(Qt.BrushStyle.NoBrush)
painter.drawPath(self._path)
def _paint_decoration(self, painter, color: QColor,
deco: EndpointDecoration, from_pt, to_pt):
if deco == EndpointDecoration.NONE:
return
p_from = QPointF(*from_pt)
p_to = QPointF(*to_pt)
line = QLineF(p_from, p_to)
angle = math.radians(line.angle()) # QLineF angle is CCW from +x
size = 9.0
painter.setPen(QPen(color, 1.5))
painter.setBrush(QBrush(color))
if deco == EndpointDecoration.ARROW:
a1 = angle + math.radians(150)
a2 = angle - math.radians(150)
p1 = QPointF(p_to.x() + size * math.cos(a1), p_to.y() - size * math.sin(a1))
p2 = QPointF(p_to.x() + size * math.cos(a2), p_to.y() - size * math.sin(a2))
painter.drawPolygon(QPolygonF([p_to, p1, p2]))
elif deco == EndpointDecoration.CIRCLE:
painter.drawEllipse(p_to, size * 0.55, size * 0.55)
elif deco == EndpointDecoration.DIAMOND:
d = size * 0.7
perp = angle + math.radians(90)
fx, fy = math.cos(angle), -math.sin(angle)
px, py = math.cos(perp), -math.sin(perp)
c = QPointF(p_to.x() - d * fx, p_to.y() - d * fy)
poly = QPolygonF([
QPointF(c.x() + d * fx, c.y() + d * fy),
QPointF(c.x() + d * px, c.y() + d * py),
QPointF(c.x() - d * fx, c.y() - d * fy),
QPointF(c.x() - d * px, c.y() - d * py),
])
painter.drawPolygon(poly)
elif deco == EndpointDecoration.BAR:
perp = angle + math.radians(90)
px, py = math.cos(perp), -math.sin(perp)
painter.setPen(QPen(color, 2.5))
painter.drawLine(
QPointF(p_to.x() + size * 0.7 * px, p_to.y() + size * 0.7 * py),
QPointF(p_to.x() - size * 0.7 * px, p_to.y() - size * 0.7 * py),
)

123
pipeline_editor/view/node_item.py Normal file
View File

@ -0,0 +1,123 @@
"""NodeItem: renders a node's SVG symbol, title and ports; draggable/selectable."""
from __future__ import annotations
from PyQt6.QtCore import QRectF, Qt, QPointF
from PyQt6.QtGui import QColor, QFont, QPen
from PyQt6.QtWidgets import QGraphicsObject, QGraphicsItem, QStyle
from ..model.node import NodeModel
from . import svg_cache
from .port_item import PortItem
LABEL_HEIGHT = 18.0
class NodeItem(QGraphicsObject):
"""Graphics item bound to a :class:`NodeModel`."""
def __init__(self, scene_ref, node: NodeModel) -> None:
super().__init__()
self._scene = scene_ref
self.node = node
self.setFlags(
QGraphicsItem.GraphicsItemFlag.ItemIsMovable
| QGraphicsItem.GraphicsItemFlag.ItemIsSelectable
| QGraphicsItem.GraphicsItemFlag.ItemSendsGeometryChanges
)
self.setAcceptHoverEvents(True)
self.setZValue(1)
self.setPos(node.x, node.y)
self._renderer = svg_cache.renderer(node.svg_name)
self._port_items: dict[str, PortItem] = {}
self._press_pos: QPointF | None = None
self._build_ports()
# -- helpers used by children ----------------------------------------
def scene_relations(self):
return getattr(self._scene, "document", None) and self._scene.document.relations
def port_item(self, port_id: str) -> PortItem | None:
return self._port_items.get(port_id)
def port_items(self):
return self._port_items.values()
def _build_ports(self) -> None:
for p in self.node.ports:
self._port_items[p.port_id] = PortItem(self, p)
def refresh_ports(self) -> None:
for pi in self._port_items.values():
pi.update_position()
# -- geometry ---------------------------------------------------------
def body_rect(self) -> QRectF:
return QRectF(0, 0, self.node.width, self.node.height)
def boundingRect(self) -> QRectF:
m = 6
return QRectF(-m, -m, self.node.width + 2 * m,
self.node.height + LABEL_HEIGHT + 2 * m)
def set_model_size(self, w: float, h: float) -> None:
self.prepareGeometryChange()
self.node.width = w
self.node.height = h
self.refresh_ports()
self.update()
# -- interaction ------------------------------------------------------
def mousePressEvent(self, event):
self._press_pos = self.pos()
super().mousePressEvent(event)
def mouseReleaseEvent(self, event):
super().mouseReleaseEvent(event)
if self._press_pos is not None and self.pos() != self._press_pos:
old = (self._press_pos.x(), self._press_pos.y())
new = (self.pos().x(), self.pos().y())
self._scene.on_node_move_finished(self.node.node_id, old, new)
self._press_pos = None
def itemChange(self, change, value):
if change == QGraphicsItem.GraphicsItemChange.ItemPositionChange and self.scene():
cfg = getattr(self._scene, "config", None)
if cfg is not None and cfg.snap_to_grid:
value = QPointF(cfg.snap(value.x()), cfg.snap(value.y()))
return value
if change == QGraphicsItem.GraphicsItemChange.ItemPositionHasChanged:
self.node.x = self.pos().x()
self.node.y = self.pos().y()
self._scene.reroute_edges_for_node(self.node.node_id)
if change == QGraphicsItem.GraphicsItemChange.ItemSelectedHasChanged:
self.update()
return super().itemChange(change, value)
# -- painting ---------------------------------------------------------
def paint(self, painter, option, widget=None):
painter.setRenderHint(painter.RenderHint.Antialiasing, True)
body = self.body_rect()
selected = bool(option.state & QStyle.StateFlag.State_Selected)
if selected:
pen = QPen(QColor("#1976d2"), 1.5, Qt.PenStyle.DashLine)
painter.setPen(pen)
painter.setBrush(QColor(25, 118, 210, 30))
painter.drawRoundedRect(body.adjusted(-3, -3, 3, 3), 4, 4)
if self._renderer is not None:
self._renderer.render(painter, body)
else:
painter.setPen(QPen(QColor("#555"), 2))
painter.setBrush(QColor("#eceff1"))
painter.drawRoundedRect(body, 6, 6)
# title label
title = self.node.title
if title:
painter.setPen(QColor("#263238"))
f = QFont()
f.setPointSizeF(8.5)
painter.setFont(f)
rect = QRectF(-10, self.node.height + 2, self.node.width + 20, LABEL_HEIGHT)
painter.drawText(rect, Qt.AlignmentFlag.AlignHCenter | Qt.AlignmentFlag.AlignTop, title)

74
pipeline_editor/view/port_item.py Normal file
View File

@ -0,0 +1,74 @@
"""PortItem: a clickable, relation-colored connection anchor on a node."""
from __future__ import annotations
from PyQt6.QtCore import QPointF, QRectF, Qt
from PyQt6.QtGui import QBrush, QColor, QPen
from PyQt6.QtWidgets import QGraphicsItem
from ..model.port import Port
PORT_RADIUS = 6.0
class PortItem(QGraphicsItem):
"""Rendered anchor for a single :class:`Port`, child of its NodeItem."""
def __init__(self, node_item, port: Port) -> None:
super().__init__(node_item)
self.node_item = node_item
self.port = port
self.setAcceptHoverEvents(True)
self.setZValue(2)
self._hover = False
# Visual affordance flags set by the scene during an edge drag.
self.compatible_hint = False
self.setToolTip(f"{port.name} ({port.relation})")
self.update_position()
# -- geometry ---------------------------------------------------------
def update_position(self) -> None:
n = self.node_item.node
lx, ly = self.port.local_point(n.width, n.height)
self.setPos(lx, ly)
def boundingRect(self) -> QRectF:
r = PORT_RADIUS + 3
return QRectF(-r, -r, 2 * r, 2 * r)
def scene_center(self) -> QPointF:
return self.mapToScene(QPointF(0, 0))
def relation_color(self) -> QColor:
reg = self.node_item.scene_relations()
return QColor(reg.color(self.port.relation)) if reg else QColor("#888")
# -- interaction ------------------------------------------------------
def hoverEnterEvent(self, event):
self._hover = True
self.update()
super().hoverEnterEvent(event)
def hoverLeaveEvent(self, event):
self._hover = False
self.update()
super().hoverLeaveEvent(event)
def set_compatible_hint(self, value: bool) -> None:
if self.compatible_hint != value:
self.compatible_hint = value
self.update()
# -- painting ---------------------------------------------------------
def paint(self, painter, option, widget=None):
color = self.relation_color()
r = PORT_RADIUS
if self._hover or self.compatible_hint:
r = PORT_RADIUS + 2
painter.setRenderHint(painter.RenderHint.Antialiasing, True)
pen = QPen(color.darker(140), 1.5)
if self.compatible_hint:
pen = QPen(QColor("#000000"), 2)
painter.setPen(pen)
fill = color.lighter(120) if self._hover else color
painter.setBrush(QBrush(fill))
painter.drawEllipse(QPointF(0, 0), r, r)

280
pipeline_editor/view/scene.py Normal file
View File

@ -0,0 +1,280 @@
"""DiagramScene: binds the document to graphics items and handles interaction."""
from __future__ import annotations
from typing import Optional
from PyQt6.QtCore import QPointF, Qt, pyqtSignal
from PyQt6.QtGui import QColor, QPainter, QPen
from PyQt6.QtWidgets import QGraphicsPathItem, QGraphicsScene
from PyQt6.QtGui import QPainterPath
from ..config import AppConfig
from ..model.document import DiagramDocument
from ..model.edge import EdgeModel
from .edge_item import EdgeItem
from .node_item import NodeItem
from .port_item import PortItem
from . import routing
class DiagramScene(QGraphicsScene):
"""Keeps :class:`NodeItem`/:class:`EdgeItem` in sync with a document."""
# Emitted with the selected NodeModel / EdgeModel, or None.
selection_changed = pyqtSignal(object)
def __init__(self, document: DiagramDocument, config: AppConfig,
controller=None, parent=None) -> None:
super().__init__(parent)
self.document = document
self.config = config
self.controller = controller
self.node_items: dict[str, NodeItem] = {}
self.edge_items: dict[str, EdgeItem] = {}
self.setSceneRect(-2000, -2000, 4000, 4000)
self._drag_port: Optional[PortItem] = None
self._temp_edge: Optional[QGraphicsPathItem] = None
self._connect_document()
self._connect_config()
self.selectionChanged.connect(self._on_selection_changed)
self.rebuild()
# -- document wiring --------------------------------------------------
def _connect_document(self) -> None:
d = self.document
d.node_added.connect(self._on_node_added)
d.node_removed.connect(self._on_node_removed)
d.node_changed.connect(self._on_node_changed)
d.edge_added.connect(self._on_edge_added)
d.edge_removed.connect(self._on_edge_removed)
d.edge_changed.connect(self._on_edge_changed)
d.cleared.connect(self.rebuild)
def _connect_config(self) -> None:
self.config.changed.connect(self._on_config_changed)
def rebuild(self) -> None:
for it in list(self.node_items.values()) + list(self.edge_items.values()):
self.removeItem(it)
self.node_items.clear()
self.edge_items.clear()
for node in self.document.nodes():
self._on_node_added(node.node_id)
for edge in self.document.edges():
self._on_edge_added(edge.edge_id)
self.update()
def _on_node_added(self, node_id: str) -> None:
node = self.document.node(node_id)
if node is None or node_id in self.node_items:
return
item = NodeItem(self, node)
self.node_items[node_id] = item
self.addItem(item)
def _on_node_removed(self, node_id: str) -> None:
item = self.node_items.pop(node_id, None)
if item is not None:
self.removeItem(item)
def _on_node_changed(self, node_id: str) -> None:
item = self.node_items.get(node_id)
if item is None:
return
if item.pos() != QPointF(item.node.x, item.node.y):
item.setPos(item.node.x, item.node.y)
item.refresh_ports()
item.update()
self.reroute_edges_for_node(node_id)
def _on_edge_added(self, edge_id: str) -> None:
edge = self.document.edge(edge_id)
if edge is None or edge_id in self.edge_items:
return
item = EdgeItem(self, edge)
self.edge_items[edge_id] = item
self.addItem(item)
def _on_edge_removed(self, edge_id: str) -> None:
item = self.edge_items.pop(edge_id, None)
if item is not None:
self.removeItem(item)
def _on_edge_changed(self, edge_id: str) -> None:
item = self.edge_items.get(edge_id)
if item is not None:
item.reroute()
item.update()
def _on_config_changed(self) -> None:
for item in self.edge_items.values():
item.reroute()
self.update()
# -- geometry services for items -------------------------------------
def port_scene_point(self, node_id: str, port_id: str) -> Optional[QPointF]:
item = self.node_items.get(node_id)
if item is not None:
pi = item.port_item(port_id)
if pi is not None:
return pi.scene_center()
node = self.document.node(node_id)
if node is not None:
pt = node.port_scene_point(port_id)
if pt is not None:
return QPointF(*pt)
return None
def edge_endpoints(self, edge: EdgeModel):
s = self.port_scene_point(edge.source_node, edge.source_port)
t = self.port_scene_point(edge.target_node, edge.target_port)
if s is None or t is None:
return None
sn = self.document.node(edge.source_node)
tn = self.document.node(edge.target_node)
snorm = sn.port(edge.source_port).normal() if sn else (1, 0)
tnorm = tn.port(edge.target_port).normal() if tn else (-1, 0)
return ((s.x(), s.y()), snorm), ((t.x(), t.y()), tnorm)
def other_edge_segments(self, edge_id: str):
segs = []
for eid, item in self.edge_items.items():
if eid == edge_id:
continue
segs.extend(routing.polyline_segments(item.path_points()))
return segs
def reroute_edges_for_node(self, node_id: str) -> None:
for edge in self.document.edges_for_node(node_id):
item = self.edge_items.get(edge.edge_id)
if item is not None:
item.reroute()
def on_node_move_finished(self, node_id: str, old, new) -> None:
if self.controller is not None:
self.controller.record_move(node_id, old, new)
# -- selection --------------------------------------------------------
def _on_selection_changed(self) -> None:
model = None
items = self.selectedItems()
if len(items) == 1:
it = items[0]
if isinstance(it, NodeItem):
model = it.node
elif isinstance(it, EdgeItem):
model = it.edge
self.selection_changed.emit(model)
# -- node creation from a template (drop / programmatic) --------------
def create_node(self, template_key: str, scene_x: float, scene_y: float):
x, y = self.config.snap_point(scene_x, scene_y)
if self.controller is not None:
return self.controller.add_node(template_key, x, y)
return None
# -- edge drawing interaction ----------------------------------------
def _port_at(self, scene_pos: QPointF) -> Optional[PortItem]:
for it in self.items(scene_pos):
if isinstance(it, PortItem):
return it
return None
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
port = self._port_at(event.scenePos())
if port is not None:
self._begin_edge(port)
event.accept()
return
super().mousePressEvent(event)
def mouseMoveEvent(self, event):
if self._drag_port is not None:
self._update_temp_edge(event.scenePos())
event.accept()
return
super().mouseMoveEvent(event)
def mouseReleaseEvent(self, event):
if self._drag_port is not None:
target = self._port_at(event.scenePos())
self._finish_edge(target)
event.accept()
return
super().mouseReleaseEvent(event)
def _begin_edge(self, port: PortItem) -> None:
self._drag_port = port
self._temp_edge = QGraphicsPathItem()
pen = QPen(QColor("#1976d2"), 2, Qt.PenStyle.DashLine)
self._temp_edge.setPen(pen)
self._temp_edge.setZValue(10)
self.addItem(self._temp_edge)
self._highlight_compatible(port, True)
def _update_temp_edge(self, scene_pos: QPointF) -> None:
if self._temp_edge is None or self._drag_port is None:
return
start = self._drag_port.scene_center()
path = QPainterPath(start)
# simple elbow preview
mid = QPointF(scene_pos.x(), start.y())
path.lineTo(mid)
path.lineTo(scene_pos)
self._temp_edge.setPath(path)
def _finish_edge(self, target: Optional[PortItem]) -> None:
source = self._drag_port
self._highlight_compatible(source, False)
if self._temp_edge is not None:
self.removeItem(self._temp_edge)
self._temp_edge = None
self._drag_port = None
if source is None or target is None or target is source:
return
s_node = source.node_item.node.node_id
s_port = source.port.port_id
t_node = target.node_item.node.node_id
t_port = target.port.port_id
if not self.document.can_connect(s_node, s_port, t_node, t_port):
return
if self.controller is not None:
self.controller.connect(s_node, s_port, t_node, t_port)
def _highlight_compatible(self, source: PortItem, on: bool) -> None:
if not on:
for ni in self.node_items.values():
for pi in ni.port_items():
pi.set_compatible_hint(False)
return
s_node = source.node_item.node.node_id
s_port = source.port.port_id
for ni in self.node_items.values():
for pi in ni.port_items():
if pi is source:
continue
ok = self.document.can_connect(
s_node, s_port, ni.node.node_id, pi.port.port_id)
pi.set_compatible_hint(ok)
# -- grid background --------------------------------------------------
def drawBackground(self, painter: QPainter, rect) -> None:
painter.fillRect(rect, QColor(self.config.background))
if not self.config.show_grid or self.config.grid_size <= 0:
return
g = self.config.grid_size
left = int(rect.left()) - (int(rect.left()) % g)
top = int(rect.top()) - (int(rect.top()) % g)
pen = QPen(QColor(self.config.grid_color), 0)
painter.setPen(pen)
x = left
while x < rect.right():
painter.drawLine(int(x), int(rect.top()), int(x), int(rect.bottom()))
x += g
y = top
while y < rect.bottom():
painter.drawLine(int(rect.left()), int(y), int(rect.right()), int(y))
y += g

29
pipeline_editor/view/svg_cache.py Normal file
View File

@ -0,0 +1,29 @@
"""Cached access to bundled SVG node symbols."""
from __future__ import annotations
import os
from PyQt6.QtSvg import QSvgRenderer
_NODES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "resources", "nodes")
_cache: dict[str, QSvgRenderer] = {}
def nodes_dir() -> str:
return _NODES_DIR
def renderer(svg_name: str) -> QSvgRenderer | None:
"""Return a cached, valid QSvgRenderer for ``svg_name`` or None."""
if not svg_name:
return None
r = _cache.get(svg_name)
if r is None:
path = os.path.join(_NODES_DIR, svg_name)
if not os.path.exists(path):
return None
r = QSvgRenderer(path)
if not r.isValid():
return None
_cache[svg_name] = r
return r

99
tests/test_scene.py Normal file
View File

@ -0,0 +1,99 @@
"""Scene/interaction tests: node items, port connect, move reroute, delete, undo."""
import pytest
from pipeline_editor.config import AppConfig
from pipeline_editor.controller import EditorController
from pipeline_editor.model.document import DiagramDocument
from pipeline_editor.model.node_library import default_library
from pipeline_editor.view.node_item import NodeItem
from pipeline_editor.view.edge_item import EdgeItem
from pipeline_editor.view.scene import DiagramScene
@pytest.fixture
def env(qapp):
doc = DiagramDocument()
lib = default_library()
ctrl = EditorController(doc, lib)
cfg = AppConfig()
scene = DiagramScene(doc, cfg, controller=ctrl)
return doc, lib, ctrl, cfg, scene
def test_add_node_creates_item(env):
doc, lib, ctrl, cfg, scene = env
node = scene.create_node("pump", 100, 100)
assert node is not None
assert node.node_id in scene.node_items
assert isinstance(scene.node_items[node.node_id], NodeItem)
def test_connect_ports_creates_edge_item(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("source", 0, 100)
b = ctrl.add_node("pump", 200, 100)
src_port = scene.node_items[a.node_id].port_item("out")
dst_port = scene.node_items[b.node_id].port_item("in")
scene._begin_edge(src_port)
scene._finish_edge(dst_port)
assert len(scene.edge_items) == 1
edge = next(doc.edges())
assert edge.source_node == a.node_id and edge.target_node == b.node_id
def test_incompatible_ports_do_not_connect(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("source", 0, 100) # water
t = ctrl.add_node("transformer", 200, 100) # power
scene._begin_edge(scene.node_items[a.node_id].port_item("out"))
scene._finish_edge(scene.node_items[t.node_id].port_item("hv"))
assert len(scene.edge_items) == 0
def test_move_node_reroutes_edge(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("source", 0, 100)
b = ctrl.add_node("pump", 200, 100)
ctrl.connect(a.node_id, "out", b.node_id, "in")
edge_item = next(iter(scene.edge_items.values()))
before = edge_item.path_points()[-1]
# move target node down; edge end should follow
scene.node_items[b.node_id].setPos(200, 300)
after = edge_item.path_points()[-1]
assert after != before
assert after[1] > before[1]
def test_selection_emits_model(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("pump", 0, 0)
received = []
scene.selection_changed.connect(lambda m: received.append(m))
scene.node_items[a.node_id].setSelected(True)
assert received and received[-1] is a
def test_delete_and_undo_roundtrip(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("source", 0, 100)
b = ctrl.add_node("pump", 200, 100)
ctrl.connect(a.node_id, "out", b.node_id, "in")
assert len(scene.edge_items) == 1
ctrl.delete(node_ids=[a.node_id])
assert a.node_id not in scene.node_items
assert len(scene.edge_items) == 0 # cascaded
ctrl.undo()
assert a.node_id in scene.node_items
assert len(scene.edge_items) == 1 # edge restored
def test_move_undo(env):
doc, lib, ctrl, cfg, scene = env
a = ctrl.add_node("pump", 40, 40)
ctrl.record_move(a.node_id, (40, 40), (120, 200))
assert (a.x, a.y) == (120, 200)
ctrl.undo()
assert (a.x, a.y) == (40, 40)
assert scene.node_items[a.node_id].pos().x() == 40