diff --git a/pipeline_editor/controller.py b/pipeline_editor/controller.py new file mode 100644 index 0000000..ced793b --- /dev/null +++ b/pipeline_editor/controller.py @@ -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() diff --git a/pipeline_editor/view/edge_item.py b/pipeline_editor/view/edge_item.py new file mode 100644 index 0000000..e07df9e --- /dev/null +++ b/pipeline_editor/view/edge_item.py @@ -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), + ) diff --git a/pipeline_editor/view/node_item.py b/pipeline_editor/view/node_item.py new file mode 100644 index 0000000..5a9a2b8 --- /dev/null +++ b/pipeline_editor/view/node_item.py @@ -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) diff --git a/pipeline_editor/view/port_item.py b/pipeline_editor/view/port_item.py new file mode 100644 index 0000000..7ec21b5 --- /dev/null +++ b/pipeline_editor/view/port_item.py @@ -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) diff --git a/pipeline_editor/view/scene.py b/pipeline_editor/view/scene.py new file mode 100644 index 0000000..cfe01e9 --- /dev/null +++ b/pipeline_editor/view/scene.py @@ -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 diff --git a/pipeline_editor/view/svg_cache.py b/pipeline_editor/view/svg_cache.py new file mode 100644 index 0000000..470617f --- /dev/null +++ b/pipeline_editor/view/svg_cache.py @@ -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 diff --git a/tests/test_scene.py b/tests/test_scene.py new file mode 100644 index 0000000..f8a2646 --- /dev/null +++ b/tests/test_scene.py @@ -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